Dispatch·Embedding 장애 지점 복구 검증

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

@@ -27,7 +27,7 @@ public interface DocumentChunkRepository extends JpaRepository<DocumentChunk, Lo
         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
+          LEFT JOIN documents document ON document.id = version.document_id
          WHERE version.id IS NULL OR document.id IS NULL
         """, nativeQuery = true)
     long countOrphanedRows();

backend/src/main/java/com/opensource/docgrid/domain/sync/entity/SyncEventDeliveryAttempt.java

@@ -0,0 +1,144 @@
+package com.opensource.docgrid.domain.sync.entity;
+
+import java.time.LocalDateTime;
+import java.util.UUID;
+
+import com.opensource.docgrid.domain.sync.enums.SyncEventDeliveryAttemptStatus;
+import com.opensource.docgrid.global.common.entity.BaseEntity;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.EnumType;
+import jakarta.persistence.Enumerated;
+import jakarta.persistence.GeneratedValue;
+import jakarta.persistence.GenerationType;
+import jakarta.persistence.Id;
+import jakarta.persistence.Index;
+import jakarta.persistence.Table;
+import jakarta.persistence.UniqueConstraint;
+import lombok.AccessLevel;
+import lombok.Builder;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+
+/**
+ * Dispatcher Claim Token별 처리 결과와 장애 원인을 보존하는 append-only 실행 이력이다.
+ *
+ * <p>현재 Queue 상태는 {@link SyncOutboxEvent}가 담당하고, 이 Entity는 성공 후 Event의 오류 Snapshot이
+ * 정리돼도 과거 Handler 실패와 Lease 만료 원인을 운영자가 추적할 수 있게 한다.
+ */
+@Getter
+@Entity
+@NoArgsConstructor(access = AccessLevel.PROTECTED)
+@Table(
+    name = "sync_event_delivery_attempts",
+    uniqueConstraints = {
+        @UniqueConstraint(
+            name = "uk_sync_event_delivery_attempts_claim_token",
+            columnNames = "claim_token"
+        ),
+        @UniqueConstraint(
+            name = "uk_sync_event_delivery_attempts_event_attempt",
+            columnNames = {"event_id", "attempt_no"}
+        )
+    },
+    indexes = {
+        @Index(name = "idx_sync_event_delivery_attempts_event_started", columnList = "event_id, started_at"),
+        @Index(name = "idx_sync_event_delivery_attempts_status_started", columnList = "status, started_at")
+    }
+)
+public class SyncEventDeliveryAttempt extends BaseEntity {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    @Column(name = "event_id", nullable = false, updatable = false)
+    private UUID eventId;
+
+    @Column(name = "claim_token", nullable = false, updatable = false)
+    private UUID claimToken;
+
+    @Column(name = "attempt_no", nullable = false, updatable = false)
+    private int attemptNo;
+
+    @Column(name = "dispatcher_name", nullable = false, updatable = false, length = 200)
+    private String dispatcherName;
+
+    @Enumerated(EnumType.STRING)
+    @Column(nullable = false, length = 20)
+    private SyncEventDeliveryAttemptStatus status;
+
+    @Column(name = "error_code", length = 100)
+    private String errorCode;
+
+    @Column(name = "error_message", columnDefinition = "TEXT")
+    private String errorMessage;
+
+    @Column(name = "started_at", nullable = false, updatable = false)
+    private LocalDateTime startedAt;
+
+    @Column(name = "completed_at")
+    private LocalDateTime completedAt;
+
+    @Builder
+    public SyncEventDeliveryAttempt(
+        UUID eventId,
+        UUID claimToken,
+        int attemptNo,
+        String dispatcherName,
+        LocalDateTime startedAt
+    ) {
+        if (eventId == null
+            || claimToken == null
+            || attemptNo <= 0
+            || dispatcherName == null
+            || dispatcherName.isBlank()
+            || startedAt == null) {
+            throw new IllegalArgumentException("유효한 Event Claim 실행 정보가 필요합니다.");
+        }
+        this.eventId = eventId;
+        this.claimToken = claimToken;
+        this.attemptNo = attemptNo;
+        this.dispatcherName = dispatcherName;
+        this.status = SyncEventDeliveryAttemptStatus.STARTED;
+        this.startedAt = startedAt;
+    }
+
+    /**
+     * Handler 부작용과 Event 완료가 Commit될 때 현재 실행을 성공으로 종결한다.
+     */
+    public void succeed(LocalDateTime succeededAt) {
+        validateStarted(succeededAt);
+        status = SyncEventDeliveryAttemptStatus.SUCCEEDED;
+        completedAt = succeededAt;
+        errorCode = null;
+        errorMessage = null;
+    }
+
+    /**
+     * Handler 실패나 Lease 만료 원인을 보존하고 현재 실행을 실패로 종결한다.
+     */
+    public void fail(String failureCode, String failureMessage, LocalDateTime failedAt) {
+        validateStarted(failedAt);
+        if (failureCode == null
+            || failureCode.isBlank()
+            || failureMessage == null
+            || failureMessage.isBlank()) {
+            throw new IllegalArgumentException("실패 Attempt에는 오류 코드와 안전한 진단 문구가 필요합니다.");
+        }
+        status = SyncEventDeliveryAttemptStatus.FAILED;
+        completedAt = failedAt;
+        errorCode = failureCode;
+        errorMessage = failureMessage;
+    }
+
+    private void validateStarted(LocalDateTime completedAt) {
+        if (status != SyncEventDeliveryAttemptStatus.STARTED
+            || completedAt == null
+            || startedAt == null
+            || completedAt.isBefore(startedAt)) {
+            throw new IllegalStateException("STARTED Sync Event Attempt만 유효한 시각으로 종결할 수 있습니다.");
+        }
+    }
+}

backend/src/main/java/com/opensource/docgrid/domain/sync/enums/SyncEventDeliveryAttemptStatus.java

@@ -0,0 +1,10 @@
+package com.opensource.docgrid.domain.sync.enums;
+
+/**
+ * 한 Sync Event Claim 세대의 실행·성공·실패 결과를 나타낸다.
+ */
+public enum SyncEventDeliveryAttemptStatus {
+    STARTED,
+    SUCCEEDED,
+    FAILED
+}

backend/src/main/java/com/opensource/docgrid/domain/sync/repository/SyncEventDeliveryAttemptRepository.java

@@ -0,0 +1,31 @@
+package com.opensource.docgrid.domain.sync.repository;
+
+import java.util.Optional;
+import java.util.UUID;
+
+import jakarta.persistence.LockModeType;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Lock;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+
+import com.opensource.docgrid.domain.sync.entity.SyncEventDeliveryAttempt;
+
+/**
+ * Sync Event Claim 세대별 append-only 실행 이력의 저장과 현재 Attempt 잠금 조회를 담당한다.
+ */
+public interface SyncEventDeliveryAttemptRepository extends JpaRepository<SyncEventDeliveryAttempt, Long> {
+
+    @Lock(LockModeType.PESSIMISTIC_WRITE)
+    @Query("""
+        SELECT attempt
+          FROM SyncEventDeliveryAttempt attempt
+         WHERE attempt.eventId = :eventId
+           AND attempt.claimToken = :claimToken
+        """)
+    Optional<SyncEventDeliveryAttempt> findByEventIdAndClaimTokenForUpdate(
+        @Param("eventId") UUID eventId,
+        @Param("claimToken") UUID claimToken
+    );
+}

backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventClaimService.java

@@ -28,6 +28,7 @@ public class SyncEventClaimService {
 
     private final SyncOutboxEventRepository syncOutboxEventRepository;
     private final SyncDispatcherProperties syncDispatcherProperties;
+    private final SyncEventDeliveryAttemptService syncEventDeliveryAttemptService;
     private final Clock clock;
 
     public Optional<ClaimedSyncEvent> claim() {
@@ -37,13 +38,17 @@ public Optional<ClaimedSyncEvent> claim() {
     }
 
     private ClaimedSyncEvent claim(SyncOutboxEvent event, LocalDateTime claimedAt) {
+        // 1. 이전 실행과 구분되는 Claim 세대 Token을 발급한다.
         UUID claimToken = UUID.randomUUID();
+        // 2. Queue 상태와 현재 Dispatcher Lease 소유권을 함께 설정한다.
         event.claim(
             syncDispatcherProperties.getName(),
             claimToken,
             claimedAt,
             claimedAt.plus(syncDispatcherProperties.getLeaseDuration())
         );
+        // 3. Queue 소유권과 같은 Transaction에 Claim 세대 실행 이력을 시작한다.
+        syncEventDeliveryAttemptService.start(event, claimToken, claimedAt);
         return new ClaimedSyncEvent(event.getEventId(), claimToken);
     }
 }

backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDeliveryAttemptService.java

@@ -0,0 +1,64 @@
+package com.opensource.docgrid.domain.sync.service.command;
+
+import java.time.LocalDateTime;
+import java.util.Objects;
+import java.util.UUID;
+
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import com.opensource.docgrid.domain.sync.entity.SyncEventDeliveryAttempt;
+import com.opensource.docgrid.domain.sync.entity.SyncOutboxEvent;
+import com.opensource.docgrid.domain.sync.enums.SyncEventStatus;
+import com.opensource.docgrid.domain.sync.repository.SyncEventDeliveryAttemptRepository;
+
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Sync Event Claim·성공·실패 이력을 호출한 Queue 상태 Transaction과 함께 기록한다.
+ *
+ * <p>이 Service는 별도 Transaction을 열지 않으며 Claim, Dispatch, Failure, Recovery Service의 기존
+ * 경계에 참여해 Queue 상태와 이력이 서로 다른 결과로 Commit되지 않도록 한다.
+ */
+@Service
+@RequiredArgsConstructor
+@Transactional
+public class SyncEventDeliveryAttemptService {
+
+    private final SyncEventDeliveryAttemptRepository repository;
+
+    public void start(SyncOutboxEvent event, UUID claimToken, LocalDateTime startedAt) {
+        if (event == null
+            || event.getStatus() != SyncEventStatus.PROCESSING
+            || event.getEventId() == null
+            || !Objects.equals(event.getClaimToken(), claimToken)
+            || event.getLockedBy() == null
+            || event.getLockedBy().isBlank()
+            || startedAt == null) {
+            throw new IllegalArgumentException("현재 Event Claim과 일치하는 실행 정보가 필요합니다.");
+        }
+        repository.save(SyncEventDeliveryAttempt.builder()
+            .eventId(event.getEventId())
+            .claimToken(claimToken)
+            .attemptNo(event.getRetryCount() + 1)
+            .dispatcherName(event.getLockedBy())
+            .startedAt(startedAt)
+            .build());
+    }
+
+    public void succeed(UUID eventId, UUID claimToken, LocalDateTime succeededAt) {
+        repository.findByEventIdAndClaimTokenForUpdate(eventId, claimToken)
+            .ifPresent(attempt -> attempt.succeed(succeededAt));
+    }
+
+    public void fail(
+        UUID eventId,
+        UUID claimToken,
+        String errorCode,
+        String errorMessage,
+        LocalDateTime failedAt
+    ) {
+        repository.findByEventIdAndClaimTokenForUpdate(eventId, claimToken)
+            .ifPresent(attempt -> attempt.fail(errorCode, errorMessage, failedAt));
+    }
+}

backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchService.java

@@ -30,6 +30,7 @@ public class SyncEventDispatchService {
 
     private final SyncOutboxEventRepository syncOutboxEventRepository;
     private final SyncEventHandlerRegistry syncEventHandlerRegistry;
+    private final SyncEventDeliveryAttemptService syncEventDeliveryAttemptService;
     private final Clock clock;
 
     @Transactional(propagation = Propagation.REQUIRES_NEW)
@@ -41,7 +42,9 @@ public void dispatch(ClaimedSyncEvent claimedEvent) {
 
         // Handler 부작용과 완료 상태가 같은 Commit 경계를 공유해야 부분 완료가 남지 않는다.
         syncEventHandlerRegistry.handle(event);
-        event.complete(claimedEvent.claimToken(), LocalDateTime.now(clock));
+        LocalDateTime completedAt = LocalDateTime.now(clock);
+        syncEventDeliveryAttemptService.succeed(event.getEventId(), claimedEvent.claimToken(), completedAt);
+        event.complete(claimedEvent.claimToken(), completedAt);
     }
 
     private void validateOwnership(

backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureService.java

@@ -29,6 +29,7 @@ public class SyncEventFailureService {
 
     private final SyncOutboxEventRepository syncOutboxEventRepository;
     private final SyncEventRetrySchedule syncEventRetrySchedule;
+    private final SyncEventDeliveryAttemptService syncEventDeliveryAttemptService;
     private final Clock clock;
 
     @Transactional(propagation = Propagation.REQUIRES_NEW)
@@ -37,6 +38,7 @@ public void recordFailure(UUID eventId, UUID claimToken, String errorCode, Strin
         SyncOutboxEvent event = syncOutboxEventRepository.findByEventIdForUpdate(eventId)
             .orElseThrow(() -> new DocGridException(ErrorCode.SYNC_EVENT_NOT_FOUND));
         validateOwnership(event, claimToken, failedAt);
+        syncEventDeliveryAttemptService.fail(eventId, claimToken, errorCode, errorMessage, failedAt);
 
         // 이번 실패가 허용 횟수를 채우면 다시 Claim되지 않는 최종 상태로 종결한다.
         if (event.getRetryCount() + 1 >= event.getMaxRetryCount()) {

backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryService.java

@@ -27,6 +27,7 @@ public class SyncEventLeaseRecoveryService {
 
     private final SyncOutboxEventRepository syncOutboxEventRepository;
     private final SyncEventRetrySchedule syncEventRetrySchedule;
+    private final SyncEventDeliveryAttemptService syncEventDeliveryAttemptService;
 
     @Transactional(propagation = Propagation.REQUIRES_NEW)
     public RecoveryResult recover(UUID eventId, LocalDateTime recoveredAt) {
@@ -37,6 +38,13 @@ public RecoveryResult recover(UUID eventId, LocalDateTime recoveredAt) {
         }
 
         SyncOutboxEvent event = candidate.get();
+        syncEventDeliveryAttemptService.fail(
+            event.getEventId(),
+            event.getClaimToken(),
+            LEASE_EXPIRED_CODE,
+            LEASE_EXPIRED_MESSAGE,
+            recoveredAt
+        );
         event.recoverExpiredLease(
             LEASE_EXPIRED_CODE,
             LEASE_EXPIRED_MESSAGE,

backend/src/main/resources/db/migration/V39__create_sync_event_delivery_attempts.sql

@@ -0,0 +1,23 @@
+-- sync_event_delivery_attempts: 완료 후에도 보존되는 Dispatcher Claim 세대별 처리·복구 이력
+CREATE TABLE sync_event_delivery_attempts (
+    id BIGSERIAL PRIMARY KEY,
+    event_id UUID NOT NULL REFERENCES sync_outbox_events (event_id) ON DELETE CASCADE,
+    claim_token UUID NOT NULL,
+    attempt_no INT NOT NULL,
+    dispatcher_name VARCHAR(200) NOT NULL,
+    status VARCHAR(20) NOT NULL,
+    error_code VARCHAR(100),
+    error_message TEXT,
+    started_at TIMESTAMP NOT NULL,
+    completed_at TIMESTAMP,
+    created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+
+    CONSTRAINT uk_sync_event_delivery_attempts_claim_token UNIQUE (claim_token),
+    CONSTRAINT uk_sync_event_delivery_attempts_event_attempt UNIQUE (event_id, attempt_no)
+);
+
+CREATE INDEX idx_sync_event_delivery_attempts_event_started
+    ON sync_event_delivery_attempts (event_id, started_at DESC);
+CREATE INDEX idx_sync_event_delivery_attempts_status_started
+    ON sync_event_delivery_attempts (status, started_at);