Index Consistency Reconciler와 안전한 복구 구현

backend/src/main/java/com/opensource/docgrid/domain/document/entity/DocumentVersion.java

@@ -155,6 +155,19 @@ public void markIndexed(LocalDateTime indexedAt) {
         this.indexedAt = indexedAt;
     }
 
+    /**
+     * 현재 검색 Version의 Vector 손상이 확인됐을 때 기존 Chunk Set부터 다시 임베딩하도록 되돌린다.
+     *
+     * <p>호출 Service가 현재 Version·Chunk 존재·Job 부재를 잠금 상태에서 검증해야 한다.
+     */
+    public void reopenIndexedForVectorRepair() {
+        if (status != DocumentVersionStatus.INDEXED) {
+            throw new IllegalStateException("INDEXED 상태의 문서 버전만 Vector 복구를 시작할 수 있습니다.");
+        }
+        status = DocumentVersionStatus.CHUNKED;
+        indexedAt = null;
+    }
+
     /**
      * 최종 실패한 Version을 수동 재처리가 다시 진행할 수 있는 재개 지점으로 되돌린다.
      *

backend/src/main/java/com/opensource/docgrid/domain/document/repository/DocumentChunkRepository.java

@@ -3,6 +3,7 @@
 import java.util.List;
 
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Query;
 
 import com.opensource.docgrid.domain.document.entity.DocumentChunk;
 
@@ -19,5 +20,17 @@ public interface DocumentChunkRepository extends JpaRepository<DocumentChunk, Lo
 
     long countByDocumentVersionId(Long documentVersionId);
 
+    /**
+     * FK가 비정상적으로 우회된 경우 Version 또는 Document 원장이 없는 Chunk를 탐지한다.
+     */
+    @Query(value = """
+        SELECT COUNT(*)
+          FROM document_chunks chunk
+          LEFT JOIN document_versions version ON version.id = chunk.document_version_id
+          LEFT JOIN documents document ON document.id = chunk.document_id
+         WHERE version.id IS NULL OR document.id IS NULL
+        """, nativeQuery = true)
+    long countOrphanedRows();
+
     List<DocumentChunk> findAllByDocumentVersionIdOrderByChunkIndexAsc(Long documentVersionId);
 }

backend/src/main/java/com/opensource/docgrid/domain/document/repository/DocumentVersionRepository.java

@@ -1,6 +1,7 @@
 package com.opensource.docgrid.domain.document.repository;
 
 import java.util.Collection;
+import java.util.List;
 import java.util.Optional;
 
 import jakarta.persistence.LockModeType;
@@ -9,6 +10,7 @@
 import org.springframework.data.jpa.repository.Lock;
 import org.springframework.data.jpa.repository.Query;
 import org.springframework.data.repository.query.Param;
+import org.springframework.data.domain.Pageable;
 
 import com.opensource.docgrid.domain.document.entity.DocumentVersion;
 import com.opensource.docgrid.domain.document.enums.DocumentVersionStatus;
@@ -21,6 +23,22 @@
  */
 public interface DocumentVersionRepository extends JpaRepository<DocumentVersion, Long> {
 
+    /**
+     * Reconciler가 전체 Version을 Offset 없이 작은 ID Cursor Batch로 순회한다.
+     */
+    @Query("""
+        SELECT version
+        FROM DocumentVersion version
+        JOIN FETCH version.document document
+        LEFT JOIN FETCH document.currentVersion
+        WHERE version.id > :cursor
+        ORDER BY version.id ASC
+        """)
+    List<DocumentVersion> findReconciliationBatchAfterId(
+        @Param("cursor") Long cursor,
+        Pageable pageable
+    );
+
     boolean existsByDocumentIdAndStatusIn(Long documentId, Collection<DocumentVersionStatus> statuses);
 
     Optional<DocumentVersion> findTopByDocumentIdOrderByVersionNoDesc(Long documentId);

backend/src/main/java/com/opensource/docgrid/domain/embedding/repository/EmbeddingJobRepository.java

@@ -33,6 +33,11 @@ Optional<EmbeddingJob> findTopByDocumentVersionIdAndEmbeddingModelIdOrderByIdDes
         Long embeddingModelId
     );
 
+    boolean existsByDocumentVersionIdAndStatusIn(
+        Long documentVersionId,
+        Collection<EmbeddingJobStatus> statuses
+    );
+
     /**
      * 관리자 목록 화면에 필요한 연관관계를 함께 조회하면서 선택 필터와 Pagination을 적용한다.
      *

backend/src/main/java/com/opensource/docgrid/domain/embedding/repository/EmbeddingRepository.java

@@ -1,5 +1,7 @@
 package com.opensource.docgrid.domain.embedding.repository;
 
+import java.util.List;
+
 import org.springframework.data.jpa.repository.JpaRepository;
 import org.springframework.data.jpa.repository.Modifying;
 import org.springframework.data.jpa.repository.Query;
@@ -31,6 +33,36 @@ long countByDocumentVersionIdAndEmbeddingModelIdAndStatus(
         EmbeddingStatus status
     );
 
+    @Query("""
+        SELECT embedding.chunk.id
+          FROM Embedding embedding
+         WHERE embedding.documentVersion.id = :documentVersionId
+           AND embedding.embeddingModel.id = :embeddingModelId
+        """)
+    List<Long> findChunkIdsByDocumentVersionIdAndEmbeddingModelId(
+        @Param("documentVersionId") Long documentVersionId,
+        @Param("embeddingModelId") Long embeddingModelId
+    );
+
+    long countByDocumentIdAndStatus(Long documentId, EmbeddingStatus status);
+
+    /**
+     * Chunk·Document·Version·Model 원장 중 하나라도 사라진 Embedding 행을 탐지한다.
+     */
+    @Query(value = """
+        SELECT COUNT(*)
+          FROM embeddings embedding
+          LEFT JOIN document_chunks chunk ON chunk.id = embedding.chunk_id
+          LEFT JOIN documents document ON document.id = embedding.document_id
+          LEFT JOIN document_versions version ON version.id = embedding.document_version_id
+          LEFT JOIN embedding_models model ON model.id = embedding.embedding_model_id
+         WHERE chunk.id IS NULL
+            OR document.id IS NULL
+            OR version.id IS NULL
+            OR model.id IS NULL
+        """, nativeQuery = true)
+    long countOrphanedRows();
+
     /**
      * 인덱싱 완료 시 이전 현재 Version 또는 최종 실패 대상의 검색 가능한 Embedding을 한 SQL로 비활성화한다.
      *

backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentEmbeddingTransactionService.java

@@ -5,6 +5,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
+import java.util.Set;
 
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -39,6 +40,7 @@
  *
  * <p>두 단계는 Job을 먼저, Version을 다음 순서로 잠가 Claim 교체와 같은 Version의 동시 실행을
  * 직렬화한다. 준비 단계는 Job에 고정된 Model과 Chunk 불변 Snapshot만 외부 호출 구간에 전달한다.
+ * Reconciler 복구 Job은 기존 Vector를 보존하고 같은 Version·Model에서 실제 누락된 Chunk만 채운다.
  */
 @Service
 @RequiredArgsConstructor
@@ -150,7 +152,7 @@ public CompletionResult complete(
         validateChunks(documentVersion, chunks);
         validatePreparedChunks(preparedWork, chunks);
 
-        // 4. 동시 요청이 먼저 전체 저장했으면 기존 결과를 재생하고 부분 저장은 내부 모순으로 거부한다.
+        // 4. 동시 요청이 먼저 전체 저장했으면 재생하고 Reconciler 복구의 부분 Set은 누락분만 채운다.
         EmbeddingState state = resolveState(documentVersion, embeddingModel, chunks.size());
         if (state == EmbeddingState.REPLAY) {
             return result(jobId, attemptId, documentVersion, embeddingModel, chunks.size(), false);
@@ -159,16 +161,18 @@ public CompletionResult complete(
             throw new DocGridException(ErrorCode.DOCUMENT_VERSION_EMBEDDING_NOT_ALLOWED);
         }
 
-        // 5. 모든 Draft를 다시 검증하고 같은 Version·Model의 ACTIVE Embedding Set으로 원자 저장한다.
+        // 5. 모든 Draft를 다시 검증하고 아직 없는 Chunk의 ACTIVE Embedding만 원자 저장한다.
         List<Embedding> embeddings = toEntities(
             documentVersion,
             embeddingModel,
             chunks,
             drafts
         );
-        embeddingRepository.saveAllAndFlush(embeddings);
+        if (!embeddings.isEmpty()) {
+            embeddingRepository.saveAllAndFlush(embeddings);
+        }
 
-        return result(jobId, attemptId, documentVersion, embeddingModel, embeddings.size(), true);
+        return result(jobId, attemptId, documentVersion, embeddingModel, chunks.size(), true);
     }
 
     private EmbeddingJob findLockedJob(Long jobId) {
@@ -282,14 +286,17 @@ private EmbeddingState resolveState(
             embeddingModel.getId()
         );
 
+        if (embeddingCount > chunkCount) {
+            throw new DocGridException(ErrorCode.DOCUMENT_EMBEDDINGS_INCONSISTENT);
+        }
         if (documentVersion.getStatus() == DocumentVersionStatus.CHUNKED) {
-            if (embeddingCount != 0) {
-                throw new DocGridException(ErrorCode.DOCUMENT_EMBEDDINGS_INCONSISTENT);
+            if (embeddingCount == chunkCount) {
+                return EmbeddingState.REPLAY;
             }
             return EmbeddingState.WORK;
         }
         if (documentVersion.getStatus() == DocumentVersionStatus.EMBEDDING) {
-            if (embeddingCount == 0) {
+            if (embeddingCount < chunkCount) {
                 return EmbeddingState.WORK;
             }
             if (embeddingCount == chunkCount) {
@@ -317,10 +324,20 @@ private List<Embedding> toEntities(
         }
 
         List<Embedding> embeddings = new ArrayList<>(drafts.size());
+        Set<Long> existingChunkIds = Set.copyOf(
+            embeddingRepository.findChunkIdsByDocumentVersionIdAndEmbeddingModelId(
+                documentVersion.getId(),
+                embeddingModel.getId()
+            )
+        );
         for (int index = 0; index < drafts.size(); index++) {
             DocumentChunk chunk = chunks.get(index);
             DocumentEmbeddingDraft draft = drafts.get(index);
             float[] vector = validateDraft(draft, chunk, embeddingModel.getDimension());
+            if (existingChunkIds.contains(chunk.getId())) {
+                // 기존 Vector는 보존하고 누락 Chunk만 채워 자동복구가 물리 삭제를 요구하지 않게 한다.
+                continue;
+            }
             embeddings.add(Embedding.builder()
                 .chunk(chunk)
                 .document(documentVersion.getDocument())

backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingCompletionService.java

@@ -410,7 +410,6 @@ private CompletionCounts validateEmbeddingSet(
 
         if (activeJobCount != 1
             || chunkCount <= 0
-            || allEmbeddingCount != chunkCount
             || modelEmbeddingCount != chunkCount
             || activeEmbeddingCount != chunkCount
             || invalidEmbeddingCount != 0) {

backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/IndexedVersionVectorRepairService.java

@@ -0,0 +1,114 @@
+package com.opensource.docgrid.domain.embedding.service.command;
+
+import java.util.EnumSet;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
+
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import com.opensource.docgrid.domain.document.entity.Document;
+import com.opensource.docgrid.domain.document.entity.DocumentVersion;
+import com.opensource.docgrid.domain.document.enums.DocumentStatus;
+import com.opensource.docgrid.domain.document.enums.DocumentVersionStatus;
+import com.opensource.docgrid.domain.document.repository.DocumentChunkRepository;
+import com.opensource.docgrid.domain.document.repository.DocumentRepository;
+import com.opensource.docgrid.domain.document.repository.DocumentVersionRepository;
+import com.opensource.docgrid.domain.embedding.entity.EmbeddingJob;
+import com.opensource.docgrid.domain.embedding.entity.EmbeddingModel;
+import com.opensource.docgrid.domain.embedding.enums.EmbeddingJobStatus;
+import com.opensource.docgrid.domain.embedding.enums.EmbeddingStatus;
+import com.opensource.docgrid.domain.embedding.repository.EmbeddingJobRepository;
+import com.opensource.docgrid.domain.embedding.repository.EmbeddingRepository;
+import com.opensource.docgrid.global.exception.DocGridException;
+import com.opensource.docgrid.global.exception.ErrorCode;
+
+import lombok.RequiredArgsConstructor;
+
+/**
+ * 현재 INDEXED Version을 기존 Vector 보존 상태로 Chunk 기반 재임베딩 Queue에 되돌린다.
+ *
+ * <p>Version·Document 잠금 안에서 현재 검색 대상과 부분 Set을 검증하고 새 Job만 만든다. Worker는 기존
+ * Vector를 덮거나 지우지 않고 누락 Chunk만 채우며, 다른 모델의 Vector도 모델 전환 이력으로 보존한다.
+ */
+@Service
+@RequiredArgsConstructor
+@Transactional
+public class IndexedVersionVectorRepairService {
+
+    private static final int DEFAULT_JOB_PRIORITY = 0;
+    private static final int MAX_RETRY_COUNT = 3;
+    private static final Set<EmbeddingJobStatus> LIVE_JOB_STATUSES = EnumSet.of(
+        EmbeddingJobStatus.PENDING,
+        EmbeddingJobStatus.PROCESSING
+    );
+
+    private final DocumentVersionRepository documentVersionRepository;
+    private final DocumentRepository documentRepository;
+    private final DocumentChunkRepository documentChunkRepository;
+    private final EmbeddingJobRepository embeddingJobRepository;
+    private final EmbeddingRepository embeddingRepository;
+
+    public EmbeddingJob repair(
+        Long documentVersionId,
+        EmbeddingModel embeddingModel,
+        UUID sourceEventId
+    ) {
+        // 1. Worker 완료 경로와 같은 Version → Document 순서로 잠그고 현재 검색 대상인지 검증한다.
+        DocumentVersion version = documentVersionRepository.findByIdForUpdate(documentVersionId)
+            .orElseThrow(() -> new DocGridException(ErrorCode.SYNC_EVENT_INCONSISTENT));
+        Document document = documentRepository.findByIdForUpdate(version.getDocument().getId())
+            .orElseThrow(() -> new DocGridException(ErrorCode.SYNC_EVENT_INCONSISTENT));
+        validateTarget(document, version, embeddingModel);
+
+        // 2. 검색 노출을 중단하고 기존 Chunk Set을 재사용하는 상태로 원자 전환한다.
+        version.reopenIndexedForVectorRepair();
+        document.markIndexing();
+
+        // 3. Repair Event를 원인으로 가진 Job을 만들어 Event 재전달에서도 한 건만 유지한다.
+        return embeddingJobRepository.save(
+            EmbeddingJob.builder()
+                .documentVersion(version)
+                .embeddingModel(embeddingModel)
+                .sourceEventId(sourceEventId)
+                .status(EmbeddingJobStatus.PENDING)
+                .priority(DEFAULT_JOB_PRIORITY)
+                .maxRetryCount(MAX_RETRY_COUNT)
+                .build()
+        );
+    }
+
+    private void validateTarget(
+        Document document,
+        DocumentVersion version,
+        EmbeddingModel embeddingModel
+    ) {
+        if (version.getStatus() != DocumentVersionStatus.INDEXED
+            || document.getStatus() != DocumentStatus.INDEXED
+            || document.getCurrentVersion() == null
+            || !Objects.equals(document.getCurrentVersion().getId(), version.getId())
+            || !documentChunkRepository.existsByDocumentVersionId(version.getId())
+            || embeddingJobRepository.existsByDocumentVersionIdAndStatusIn(
+                version.getId(), LIVE_JOB_STATUSES
+            )) {
+            throw new DocGridException(ErrorCode.SYNC_EVENT_INCONSISTENT);
+        }
+        long chunkCount = documentChunkRepository.countByDocumentVersionId(version.getId());
+        long allModelEmbeddingCount = embeddingRepository.countByDocumentVersionIdAndEmbeddingModelId(
+            version.getId(),
+            embeddingModel.getId()
+        );
+        long activeEmbeddingCount = embeddingRepository
+            .countByDocumentVersionIdAndEmbeddingModelIdAndStatus(
+                version.getId(),
+                embeddingModel.getId(),
+                EmbeddingStatus.ACTIVE
+            );
+        if (allModelEmbeddingCount >= chunkCount
+            || activeEmbeddingCount >= chunkCount
+            || allModelEmbeddingCount != activeEmbeddingCount) {
+            throw new DocGridException(ErrorCode.SYNC_EVENT_INCONSISTENT);
+        }
+    }
+}