diff --git a/backend/src/main/java/com/opensource/docgrid/domain/document/repository/DocumentChunkRepository.java b/backend/src/main/java/com/opensource/docgrid/domain/document/repository/DocumentChunkRepository.java index 12c32ab..b66b4a2 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/document/repository/DocumentChunkRepository.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/document/repository/DocumentChunkRepository.java @@ -27,7 +27,7 @@ public interface DocumentChunkRepository extends JpaRepository현재 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만 유효한 시각으로 종결할 수 있습니다."); + } + } +} diff --git a/backend/src/main/java/com/opensource/docgrid/domain/sync/enums/SyncEventDeliveryAttemptStatus.java b/backend/src/main/java/com/opensource/docgrid/domain/sync/enums/SyncEventDeliveryAttemptStatus.java new file mode 100644 index 0000000..64f76e0 --- /dev/null +++ b/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 +} diff --git a/backend/src/main/java/com/opensource/docgrid/domain/sync/repository/SyncEventDeliveryAttemptRepository.java b/backend/src/main/java/com/opensource/docgrid/domain/sync/repository/SyncEventDeliveryAttemptRepository.java new file mode 100644 index 0000000..0820ea8 --- /dev/null +++ b/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 { + + @Lock(LockModeType.PESSIMISTIC_WRITE) + @Query(""" + SELECT attempt + FROM SyncEventDeliveryAttempt attempt + WHERE attempt.eventId = :eventId + AND attempt.claimToken = :claimToken + """) + Optional findByEventIdAndClaimTokenForUpdate( + @Param("eventId") UUID eventId, + @Param("claimToken") UUID claimToken + ); +} diff --git a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventClaimService.java b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventClaimService.java index bca9bb0..165579a 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventClaimService.java +++ b/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 claim() { @@ -37,13 +38,17 @@ public Optional 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); } } diff --git a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDeliveryAttemptService.java b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDeliveryAttemptService.java new file mode 100644 index 0000000..b3903c0 --- /dev/null +++ b/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과 함께 기록한다. + * + *

이 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)); + } +} diff --git a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchService.java b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchService.java index 1d0d521..d3b9ff2 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchService.java +++ b/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( diff --git a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureService.java b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureService.java index 3d71478..b8e42a4 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureService.java +++ b/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()) { diff --git a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryService.java b/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryService.java index 5b43841..bc41f2b 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryService.java +++ b/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, diff --git a/backend/src/main/resources/db/migration/V39__create_sync_event_delivery_attempts.sql b/backend/src/main/resources/db/migration/V39__create_sync_event_delivery_attempts.sql new file mode 100644 index 0000000..d9a14a9 --- /dev/null +++ b/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); diff --git a/backend/src/test/java/com/opensource/docgrid/domain/sync/entity/SyncEventDeliveryAttemptTest.java b/backend/src/test/java/com/opensource/docgrid/domain/sync/entity/SyncEventDeliveryAttemptTest.java new file mode 100644 index 0000000..d4ddaae --- /dev/null +++ b/backend/src/test/java/com/opensource/docgrid/domain/sync/entity/SyncEventDeliveryAttemptTest.java @@ -0,0 +1,59 @@ +package com.opensource.docgrid.domain.sync.entity; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.time.LocalDateTime; +import java.util.UUID; + +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; + +import com.opensource.docgrid.domain.sync.enums.SyncEventDeliveryAttemptStatus; + +/** + * Sync Event 전달 Attempt가 성공·실패 원인을 한 번만 종결하는지 검증한다. + */ +@DisplayName("SyncEventDeliveryAttempt 단위 테스트") +class SyncEventDeliveryAttemptTest { + + @Test + @DisplayName("실패 종결은 원인과 완료 시각을 보존하고 재종결을 거부한다") + void fail_preservesCauseAndRejectsSecondCompletion() { + LocalDateTime startedAt = LocalDateTime.of(2026, 8, 13, 19, 0); + SyncEventDeliveryAttempt attempt = attempt(startedAt); + + attempt.fail("SYNC_LEASE_EXPIRED", "Lease 만료", startedAt.plusSeconds(30)); + + assertThat(attempt.getStatus()).isEqualTo(SyncEventDeliveryAttemptStatus.FAILED); + assertThat(attempt.getErrorCode()).isEqualTo("SYNC_LEASE_EXPIRED"); + assertThat(attempt.getErrorMessage()).isEqualTo("Lease 만료"); + assertThat(attempt.getCompletedAt()).isEqualTo(startedAt.plusSeconds(30)); + assertThatThrownBy(() -> attempt.succeed(startedAt.plusSeconds(31))) + .isInstanceOf(IllegalStateException.class); + } + + @Test + @DisplayName("성공 종결은 오류 없이 완료 시각을 기록한다") + void succeed_recordsCompletionWithoutError() { + LocalDateTime startedAt = LocalDateTime.of(2026, 8, 13, 19, 0); + SyncEventDeliveryAttempt attempt = attempt(startedAt); + + attempt.succeed(startedAt.plusSeconds(2)); + + assertThat(attempt.getStatus()).isEqualTo(SyncEventDeliveryAttemptStatus.SUCCEEDED); + assertThat(attempt.getCompletedAt()).isEqualTo(startedAt.plusSeconds(2)); + assertThat(attempt.getErrorCode()).isNull(); + assertThat(attempt.getErrorMessage()).isNull(); + } + + private SyncEventDeliveryAttempt attempt(LocalDateTime startedAt) { + return SyncEventDeliveryAttempt.builder() + .eventId(UUID.randomUUID()) + .claimToken(UUID.randomUUID()) + .attemptNo(1) + .dispatcherName("sync-dispatcher-test") + .startedAt(startedAt) + .build(); + } +} diff --git a/backend/src/test/java/com/opensource/docgrid/domain/sync/integration/EmbeddingFailureRecoveryIntegrationTest.java b/backend/src/test/java/com/opensource/docgrid/domain/sync/integration/EmbeddingFailureRecoveryIntegrationTest.java new file mode 100644 index 0000000..6a2c192 --- /dev/null +++ b/backend/src/test/java/com/opensource/docgrid/domain/sync/integration/EmbeddingFailureRecoveryIntegrationTest.java @@ -0,0 +1,404 @@ +package com.opensource.docgrid.domain.sync.integration; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.UUID; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.dao.DataAccessException; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; + +import com.opensource.docgrid.domain.embedding.dto.request.CompleteDocumentIndexingRequest; +import com.opensource.docgrid.domain.embedding.dto.request.StartEmbeddingJobAttemptRequest; +import com.opensource.docgrid.domain.embedding.dto.response.ClaimedEmbeddingJobResponse; +import com.opensource.docgrid.domain.embedding.dto.response.StartedEmbeddingJobAttemptResponse; +import com.opensource.docgrid.domain.embedding.enums.EmbeddingJobStatus; +import com.opensource.docgrid.domain.embedding.service.DocumentEmbeddingDraft; +import com.opensource.docgrid.domain.embedding.service.EmbeddingVectorSupport; +import com.opensource.docgrid.domain.embedding.service.command.DocumentEmbeddingTransactionService; +import com.opensource.docgrid.domain.embedding.service.command.DocumentEmbeddingTransactionService.EmbeddingWork; +import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobAttemptService; +import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobClaimService; +import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobLeaseRecoveryService; +import com.opensource.docgrid.domain.embedding.service.command.DocumentIndexingCompletionService; +import com.opensource.docgrid.domain.sync.dto.SyncReconciliationBatchResult; +import com.opensource.docgrid.domain.sync.enums.SyncReconciliationMode; +import com.opensource.docgrid.domain.sync.service.SyncReconciliationOrchestrator; + +/** + * Embedding Set 저장 중 프로세스 중단을 주입하고 Worker Lease Recovery부터 최종 정합성까지 검증한다. + * + *

두 번째 Vector INSERT에서 PostgreSQL 예외를 발생시켜 부분 저장을 차단하고, 만료 Claim 회수, + * 새 Attempt 재실행, 인덱싱 완료와 Reconciler 무결성 확인을 하나의 장애 시나리오로 연결한다. + */ +@Tag("integration") +@ActiveProfiles("test") +@SpringBootTest +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +@DisplayName("Embedding 저장 장애·Lease·Reconciler 복구 통합 테스트") +class EmbeddingFailureRecoveryIntegrationTest { + + private static final String TEST_SCHEMA = "docgrid_embedding_failure_recovery_test"; + private static final String EMBEDDING_TRIGGER = "docgrid_test_fail_second_embedding"; + private static final String EMBEDDING_FUNCTION = "docgrid_test_raise_second_embedding_failure"; + private static final int VECTOR_DIMENSION = 1024; + private static final String FIRST_CLAIM_TOKEN = "34c19d16-6ae1-4f6a-a35d-0123456789ab"; + private static final String CONTENT_HASH = + "26e4a23eec4241e034f1b4631f0222f1895847637c35e77687d5945f75edb42c"; + + @Autowired private DocumentEmbeddingTransactionService documentEmbeddingTransactionService; + @Autowired private EmbeddingJobLeaseRecoveryService embeddingJobLeaseRecoveryService; + @Autowired private EmbeddingJobClaimService embeddingJobClaimService; + @Autowired private EmbeddingJobAttemptService embeddingJobAttemptService; + @Autowired private DocumentIndexingCompletionService documentIndexingCompletionService; + @Autowired private SyncReconciliationOrchestrator syncReconciliationOrchestrator; + @Autowired private JdbcTemplate jdbcTemplate; + + @DynamicPropertySource + static void configureSchema(DynamicPropertyRegistry registry) { + registry.add("TEST_DB_SCHEMA", () -> TEST_SCHEMA); + registry.add("jwt.secret", () -> "docgrid-embedding-failure-recovery-secret-key-2026"); + registry.add("indexing.worker.retry-initial-delay", () -> "1ms"); + registry.add("indexing.worker.retry-max-delay", () -> "1ms"); + } + + @BeforeEach + void resetState() { + dropEmbeddingFailureTrigger(); + jdbcTemplate.execute(""" + TRUNCATE TABLE + sync_admin_actions, + sync_consistency_issues, + sync_reconciliation_runs, + sync_event_delivery_attempts, + embeddings, + indexing_events, + document_chunks, + embedding_job_attempts, + embedding_jobs, + sync_outbox_events, + document_versions, + documents, + worker_nodes + RESTART IDENTITY CASCADE + """); + } + + @AfterEach + void dropTriggerAfterTest() { + dropEmbeddingFailureTrigger(); + } + + @AfterAll + void dropSchema() { + jdbcTemplate.execute("DROP SCHEMA IF EXISTS " + TEST_SCHEMA + " CASCADE"); + } + + @Test + @DisplayName("Embedding 저장 도중 장애는 부분 Vector 없이 Lease 회수·재실행·Reconciliation으로 정상화된다") + void failureDuringEmbeddingSave_recoversToConsistentIndexedState() { + ExecutionContext context = insertExecution(); + EmbeddingWork firstWork = documentEmbeddingTransactionService.prepare( + context.jobId(), + context.attemptId(), + context.workerId(), + FIRST_CLAIM_TOKEN + ).work(); + List firstDrafts = drafts(firstWork); + installSecondEmbeddingFailure(); + + // 1. 두 번째 Vector INSERT를 실패시켜 saveAllAndFlush Transaction 전체를 Rollback한다. + assertThatThrownBy(() -> documentEmbeddingTransactionService.complete( + context.jobId(), + context.attemptId(), + context.workerId(), + FIRST_CLAIM_TOKEN, + firstWork, + firstDrafts + )).isInstanceOf(DataAccessException.class); + assertThat(embeddingCount(context.versionId())).isZero(); + assertThat(queryStatus("document_versions", context.versionId())).isEqualTo("EMBEDDING"); + assertThat(queryStatus("embedding_jobs", context.jobId())).isEqualTo("PROCESSING"); + + // 2. 프로세스 중단으로 실패 응답도 유실된 상태에서 Worker Lease가 Attempt와 Job을 회수한다. + dropEmbeddingFailureTrigger(); + LocalDateTime recoveredAt = LocalDateTime.now(); + jdbcTemplate.update(""" + UPDATE embedding_jobs + SET locked_at = ?, lock_expires_at = ? + WHERE id = ? + """, recoveredAt.minusMinutes(2), recoveredAt.minusMinutes(1), context.jobId()); + EmbeddingJobLeaseRecoveryService.RecoveryResult recovery = + embeddingJobLeaseRecoveryService.recover(context.jobId(), recoveredAt); + assertThat(recovery.recovered()).isTrue(); + assertThat(recovery.status()).isEqualTo(EmbeddingJobStatus.PENDING); + assertThat(queryStatus("embedding_job_attempts", context.attemptId())).isEqualTo("FAILED"); + assertThat(queryString( + "SELECT error_code FROM embedding_job_attempts WHERE id = ?", + context.attemptId() + )).isEqualTo("WORKER_LEASE_EXPIRED"); + assertThat(eventCount(context.jobId(), "LEASE_EXPIRED")).isOne(); + + // 3. 새 Claim·Attempt가 기존 EMBEDDING Version을 재개해 전체 Vector Set을 한 번만 저장한다. + jdbcTemplate.update( + "UPDATE embedding_jobs SET next_retry_at = CURRENT_TIMESTAMP - INTERVAL '1 second' WHERE id = ?", + context.jobId() + ); + ClaimedEmbeddingJobResponse secondClaim = embeddingJobClaimService + .claim(context.workerId()) + .orElseThrow(); + StartedEmbeddingJobAttemptResponse secondAttempt = embeddingJobAttemptService.start( + context.jobId(), + new StartEmbeddingJobAttemptRequest(context.workerId(), secondClaim.claimToken()) + ).response(); + EmbeddingWork secondWork = documentEmbeddingTransactionService.prepare( + context.jobId(), + secondAttempt.attemptId(), + context.workerId(), + secondClaim.claimToken() + ).work(); + documentEmbeddingTransactionService.complete( + context.jobId(), + secondAttempt.attemptId(), + context.workerId(), + secondClaim.claimToken(), + secondWork, + drafts(secondWork) + ); + documentIndexingCompletionService.complete( + context.jobId(), + secondAttempt.attemptId(), + new CompleteDocumentIndexingRequest(context.workerId(), secondClaim.claimToken()) + ); + + // 4. 최종 원장·파생 상태와 Reconciler 결과가 모두 정상이며 중복 Vector가 없음을 확인한다. + assertThat(queryStatus("sync_outbox_events", context.eventDatabaseId())).isEqualTo("PROCESSED"); + assertThat(queryStatus("embedding_jobs", context.jobId())).isEqualTo("INDEXED"); + assertThat(queryStatus("embedding_job_attempts", secondAttempt.attemptId())).isEqualTo("SUCCESS"); + assertThat(queryStatus("document_versions", context.versionId())).isEqualTo("INDEXED"); + assertThat(queryStatus("documents", context.documentId())).isEqualTo("INDEXED"); + assertThat(jdbcTemplate.queryForObject( + "SELECT current_version_id FROM documents WHERE id = ?", + Long.class, + context.documentId() + )).isEqualTo(context.versionId()); + assertThat(embeddingCount(context.versionId())).isEqualTo(3); + assertThat(duplicateVectorGroups(context.versionId(), context.embeddingModelId())).isZero(); + + SyncReconciliationBatchResult reconciliation = syncReconciliationOrchestrator.reconcileBatch( + 0L, + SyncReconciliationMode.DRY_RUN + ); + assertThat(reconciliation.detectedCount()).isZero(); + assertThat(reconciliation.repairRequestedCount()).isZero(); + assertThat(count("SELECT COUNT(*) FROM sync_consistency_issues")).isZero(); + } + + private ExecutionContext insertExecution() { + String suffix = UUID.randomUUID().toString(); + Long userId = jdbcTemplate.queryForObject(""" + SELECT id FROM users WHERE email = 'kcw130502@gmail.com' + """, Long.class); + Long workerId = jdbcTemplate.queryForObject(""" + INSERT INTO worker_nodes ( + worker_name, instance_id, status, last_heartbeat_at, started_at, created_at, updated_at + ) VALUES ('embedding-recovery-worker', ?, 'ACTIVE', CURRENT_TIMESTAMP, + CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, suffix); + Long documentId = jdbcTemplate.queryForObject(""" + INSERT INTO documents ( + owner_user_id, title, document_type, source_type, status, visibility, created_at, updated_at + ) VALUES (?, 'Embedding Recovery Document', 'TXT', 'UPLOAD', 'UPLOADED', 'PRIVATE', + CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, userId); + Long versionId = jdbcTemplate.queryForObject(""" + INSERT INTO document_versions ( + document_id, version_no, title_snapshot, content_type, status, + created_by, created_at, updated_at + ) VALUES (?, 1, 'Embedding Recovery Version', 'text/plain', 'CHUNKED', ?, + CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, documentId, userId); + jdbcTemplate.update("UPDATE documents SET current_version_id = ? WHERE id = ?", versionId, documentId); + insertChunks(versionId); + Long embeddingModelId = jdbcTemplate.queryForObject(""" + SELECT id FROM embedding_models WHERE is_active = TRUE AND is_searchable = TRUE + """, Long.class); + UUID eventId = UUID.randomUUID(); + Long eventDatabaseId = jdbcTemplate.queryForObject(""" + INSERT INTO sync_outbox_events ( + event_id, idempotency_key, aggregate_type, aggregate_id, aggregate_version, + event_type, payload_json, status, available_at, occurred_at, processed_at, + retry_count, max_retry_count, created_at, updated_at + ) VALUES (?, ?, 'DOCUMENT_VERSION', ?, 1, 'DOCUMENT_VERSION_CREATED', ?, + 'PROCESSED', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, + 0, 5, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, + Long.class, + eventId, + "embedding-failure:" + suffix, + versionId, + "{\"embeddingModelId\":" + embeddingModelId + "}" + ); + Long jobId = jdbcTemplate.queryForObject(""" + INSERT INTO embedding_jobs ( + document_version_id, embedding_model_id, status, priority, retry_count, max_retry_count, + locked_by_worker_id, locked_at, lock_expires_at, claim_token, started_at, + source_event_id, created_at, updated_at + ) VALUES (?, ?, 'PROCESSING', 0, 0, 3, ?, CURRENT_TIMESTAMP, + TIMESTAMP '2099-01-01 00:00:00', ?, CURRENT_TIMESTAMP, ?, + CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, versionId, embeddingModelId, workerId, FIRST_CLAIM_TOKEN, eventId); + Long attemptId = jdbcTemplate.queryForObject(""" + INSERT INTO embedding_job_attempts ( + embedding_job_id, worker_node_id, attempt_no, claim_token, status, + started_at, created_at, updated_at + ) VALUES (?, ?, 1, ?, 'STARTED', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, jobId, workerId, FIRST_CLAIM_TOKEN); + return new ExecutionContext( + workerId, + jobId, + attemptId, + documentId, + versionId, + embeddingModelId, + eventDatabaseId + ); + } + + private void insertChunks(Long versionId) { + jdbcTemplate.update(""" + INSERT INTO document_chunks ( + document_version_id, chunk_index, chunk_text, token_count, char_start, char_end, + content_hash, created_at, updated_at + ) VALUES + (?, 0, '첫 번째', 1, 0, 4, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), + (?, 1, '두 번째', 1, 4, 8, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP), + (?, 2, '세 번째', 1, 8, 12, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + """, + versionId, CONTENT_HASH, + versionId, CONTENT_HASH, + versionId, CONTENT_HASH + ); + } + + private List drafts(EmbeddingWork work) { + return work.chunks().stream() + .map(chunk -> { + float[] vector = new float[VECTOR_DIMENSION]; + vector[0] = chunk.chunkIndex() + 1; + vector[1] = chunk.chunkText().length(); + return new DocumentEmbeddingDraft( + chunk.chunkId(), + chunk.chunkIndex(), + chunk.contentHash(), + vector, + EmbeddingVectorSupport.calculateHash(vector) + ); + }) + .toList(); + } + + private void installSecondEmbeddingFailure() { + jdbcTemplate.execute(""" + CREATE OR REPLACE FUNCTION docgrid_test_raise_second_embedding_failure() + RETURNS trigger AS $$ + BEGIN + IF EXISTS ( + SELECT 1 FROM document_chunks chunk + WHERE chunk.id = NEW.chunk_id AND chunk.chunk_index = 1 + ) THEN + RAISE EXCEPTION 'forced failure during second embedding insert'; + END IF; + RETURN NEW; + END; + $$ LANGUAGE plpgsql + """); + jdbcTemplate.execute(""" + CREATE TRIGGER docgrid_test_fail_second_embedding + BEFORE INSERT ON embeddings + FOR EACH ROW EXECUTE FUNCTION docgrid_test_raise_second_embedding_failure() + """); + } + + private void dropEmbeddingFailureTrigger() { + jdbcTemplate.execute("DROP TRIGGER IF EXISTS " + EMBEDDING_TRIGGER + " ON embeddings"); + jdbcTemplate.execute("DROP FUNCTION IF EXISTS " + EMBEDDING_FUNCTION + "()"); + } + + private int embeddingCount(Long versionId) { + return count("SELECT COUNT(*) FROM embeddings WHERE document_version_id = ?", versionId); + } + + private int duplicateVectorGroups(Long versionId, Long modelId) { + return count(""" + SELECT COUNT(*) + FROM ( + SELECT chunk_id + FROM embeddings + WHERE document_version_id = ? AND embedding_model_id = ? + GROUP BY chunk_id + HAVING COUNT(*) > 1 + ) duplicate + """, versionId, modelId); + } + + private int eventCount(Long jobId, String eventType) { + return count( + "SELECT COUNT(*) FROM indexing_events WHERE embedding_job_id = ? AND event_type = ?", + jobId, + eventType + ); + } + + private String queryStatus(String table, Long id) { + return jdbcTemplate.queryForObject( + "SELECT status FROM " + table + " WHERE id = ?", + String.class, + id + ); + } + + private String queryString(String sql, Long id) { + return jdbcTemplate.queryForObject(sql, String.class, id); + } + + private int count(String sql, Object... arguments) { + return jdbcTemplate.queryForObject(sql, Integer.class, arguments); + } + + /** + * 장애·복구 실행의 Worker, Job, Attempt, 원장 대상과 Event DB 식별자를 묶는다. + */ + private record ExecutionContext( + Long workerId, + Long jobId, + Long attemptId, + Long documentId, + Long versionId, + Long embeddingModelId, + Long eventDatabaseId + ) { + } +} diff --git a/backend/src/test/java/com/opensource/docgrid/domain/sync/integration/SyncDispatchFailureRecoveryIntegrationTest.java b/backend/src/test/java/com/opensource/docgrid/domain/sync/integration/SyncDispatchFailureRecoveryIntegrationTest.java new file mode 100644 index 0000000..a6d26bf --- /dev/null +++ b/backend/src/test/java/com/opensource/docgrid/domain/sync/integration/SyncDispatchFailureRecoveryIntegrationTest.java @@ -0,0 +1,422 @@ +package com.opensource.docgrid.domain.sync.integration; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.BDDMockito.given; +import static org.mockito.Mockito.reset; + +import java.io.InputStream; +import java.time.LocalDateTime; +import java.util.List; +import java.util.Map; +import java.util.UUID; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.mock.web.MockMultipartFile; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; +import org.springframework.test.context.bean.override.mockito.MockitoBean; + +import com.opensource.docgrid.domain.document.dto.request.DocumentUploadRequest; +import com.opensource.docgrid.domain.document.dto.response.DocumentUploadResponse; +import com.opensource.docgrid.domain.document.enums.VisibilityType; +import com.opensource.docgrid.domain.document.service.DocumentUploadFacade; +import com.opensource.docgrid.domain.document.storage.FileStorageService; +import com.opensource.docgrid.domain.document.storage.StoredFile; +import com.opensource.docgrid.domain.sync.dto.ClaimedSyncEvent; +import com.opensource.docgrid.domain.sync.dto.SyncReconciliationBatchResult; +import com.opensource.docgrid.domain.sync.entity.SyncOutboxEvent; +import com.opensource.docgrid.domain.sync.enums.SyncEventStatus; +import com.opensource.docgrid.domain.sync.enums.SyncReconciliationMode; +import com.opensource.docgrid.domain.sync.repository.SyncOutboxEventRepository; +import com.opensource.docgrid.domain.sync.service.SyncReconciliationOrchestrator; +import com.opensource.docgrid.domain.sync.service.command.SyncEventClaimService; +import com.opensource.docgrid.domain.sync.service.command.SyncEventDispatchService; +import com.opensource.docgrid.domain.sync.service.command.SyncEventFailureService; +import com.opensource.docgrid.domain.sync.service.command.SyncEventLeaseRecoveryService; +import com.opensource.docgrid.domain.user.repository.UserRepository; + +/** + * Sync Event Claim부터 완료까지 주요 중단 지점에서 Retry와 Lease Recovery 수렴을 검증한다. + * + *

실제 PostgreSQL Trigger로 Job INSERT와 Event 완료 UPDATE를 실패시키며, 모든 복구 뒤 단일 Job, + * 정리된 소유권, 성공·실패 전달 Attempt 이력과 장애 원인을 직접 확인한다. + */ +@Tag("integration") +@ActiveProfiles("test") +@SpringBootTest +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +@DisplayName("Sync Dispatch 장애 주입·복구 통합 테스트") +class SyncDispatchFailureRecoveryIntegrationTest { + + private static final String TEST_SCHEMA = "docgrid_sync_dispatch_failure_test"; + private static final String JOB_TRIGGER = "docgrid_test_fail_sync_job_insert"; + private static final String JOB_FUNCTION = "docgrid_test_raise_sync_job_insert_failure"; + private static final String COMPLETE_TRIGGER = "docgrid_test_fail_sync_event_complete"; + private static final String COMPLETE_FUNCTION = "docgrid_test_raise_sync_event_complete_failure"; + + @Autowired private DocumentUploadFacade documentUploadFacade; + @Autowired private UserRepository userRepository; + @Autowired private SyncOutboxEventRepository syncOutboxEventRepository; + @Autowired private SyncEventClaimService syncEventClaimService; + @Autowired private SyncEventDispatchService syncEventDispatchService; + @Autowired private SyncEventFailureService syncEventFailureService; + @Autowired private SyncEventLeaseRecoveryService syncEventLeaseRecoveryService; + @Autowired private SyncReconciliationOrchestrator syncReconciliationOrchestrator; + @Autowired private JdbcTemplate jdbcTemplate; + + @MockitoBean + private FileStorageService fileStorageService; + + private Long userId; + + @DynamicPropertySource + static void configureSchema(DynamicPropertyRegistry registry) { + registry.add("TEST_DB_SCHEMA", () -> TEST_SCHEMA); + registry.add("jwt.secret", () -> "docgrid-sync-dispatch-failure-secret-key-2026"); + registry.add("sync.dispatcher.retry-initial-delay", () -> "1ms"); + registry.add("sync.dispatcher.retry-max-delay", () -> "1ms"); + } + + @BeforeEach + void resetState() { + reset(fileStorageService); + dropFailureTriggers(); + jdbcTemplate.execute(""" + TRUNCATE TABLE + sync_admin_actions, + sync_consistency_issues, + sync_reconciliation_runs, + sync_event_delivery_attempts, + embeddings, + indexing_events, + document_chunks, + embedding_job_attempts, + embedding_jobs, + sync_outbox_events, + document_versions, + documents, + file_objects + RESTART IDENTITY CASCADE + """); + userId = userRepository.findByEmail("kcw130502@gmail.com").orElseThrow().getId(); + given(fileStorageService.store(any(InputStream.class), anyLong(), anyString(), anyString())) + .willAnswer(invocation -> new StoredFile( + "test-bucket", + "documents/sync-failure/" + UUID.randomUUID() + )); + } + + @AfterEach + void dropTriggersAfterTest() { + dropFailureTriggers(); + } + + @AfterAll + void dropSchema() { + jdbcTemplate.execute("DROP SCHEMA IF EXISTS " + TEST_SCHEMA + " CASCADE"); + } + + @Test + @DisplayName("Event Claim 직후 프로세스 중단은 Lease 만료 이력 후 새 Claim으로 완료된다") + void crashAfterClaim_isRecoveredByLeaseAndRedispatch() { + EventContext context = eventWithoutJob("crash-after-claim"); + + // 1. Claim Transaction만 Commit된 직후 프로세스가 중단된 상황을 강제로 만든다. + assertThatThrownBy(() -> claimAndCrash()) + .isInstanceOf(ForcedFailure.class) + .hasMessage("forced crash after event claim"); + SyncOutboxEvent claimed = event(context.eventId()); + assertThat(claimed.getStatus()).isEqualTo(SyncEventStatus.PROCESSING); + + // 2. Lease 만료 시각 뒤 Recovery가 소유권을 회수하고 실패 Attempt 원인을 보존한다. + LocalDateTime recoveredAt = claimed.getLockExpiresAt().plusSeconds(1); + jdbcTemplate.update( + "UPDATE sync_outbox_events SET lock_expires_at = ? WHERE event_id = ?", + recoveredAt.minusSeconds(1), + context.eventId() + ); + assertThat(syncEventLeaseRecoveryService.recover(context.eventId(), recoveredAt).recovered()).isTrue(); + assertThat(event(context.eventId()).getStatus()).isEqualTo(SyncEventStatus.PENDING); + assertAttempt(1, "FAILED", "SYNC_LEASE_EXPIRED", context.eventId()); + + // 3. 새 Claim 세대가 Handler를 다시 실행해 단일 Job과 성공 Attempt로 수렴한다. + dispatchNext(); + assertRecovered(context, 1); + } + + @Test + @DisplayName("Handler 시작 직후 실패는 원인을 기록하고 수정된 Event 재시도로 완료된다") + void failureAtHandlerStart_isRetriedWithoutDuplicateJob() { + EventContext context = eventWithoutJob("fail-handler-start"); + Long modelId = activeModelId(); + jdbcTemplate.update( + "UPDATE sync_outbox_events SET payload_json = '{}' WHERE event_id = ?", + context.eventId() + ); + ClaimedSyncEvent firstClaim = syncEventClaimService.claim().orElseThrow(); + + assertThatThrownBy(() -> syncEventDispatchService.dispatch(firstClaim)) + .isInstanceOf(RuntimeException.class); + syncEventFailureService.recordFailure( + firstClaim.eventId(), + firstClaim.claimToken(), + "FORCED_HANDLER_START", + "Handler 시작 직후 강제 실패" + ); + assertAttempt(1, "FAILED", "FORCED_HANDLER_START", context.eventId()); + + jdbcTemplate.update( + "UPDATE sync_outbox_events SET payload_json = ? WHERE event_id = ?", + "{\"embeddingModelId\":" + modelId + "}", + context.eventId() + ); + dispatchNext(); + assertRecovered(context, 1); + } + + @Test + @DisplayName("Job 생성 직후 실패는 Job을 롤백하고 Retry에서 한 건만 다시 만든다") + void failureAfterJobCreation_rollsBackAndRetries() { + EventContext context = eventWithoutJob("fail-after-job"); + installJobInsertFailure(context.eventId()); + ClaimedSyncEvent firstClaim = syncEventClaimService.claim().orElseThrow(); + + assertThatThrownBy(() -> syncEventDispatchService.dispatch(firstClaim)) + .isInstanceOf(RuntimeException.class); + assertThat(jobCount(context.eventId())).isZero(); + syncEventFailureService.recordFailure( + firstClaim.eventId(), + firstClaim.claimToken(), + "FORCED_AFTER_JOB_CREATED", + "Job 생성 직후 강제 실패" + ); + assertAttempt(1, "FAILED", "FORCED_AFTER_JOB_CREATED", context.eventId()); + + dropFailureTriggers(); + dispatchNext(); + assertRecovered(context, 1); + } + + @Test + @DisplayName("Event 완료 직전 실패는 Handler Job과 성공 전이를 롤백하고 Retry에서 완료된다") + void failureBeforeEventCompletion_rollsBackSideEffectAndRetries() { + EventContext context = eventWithoutJob("fail-before-complete"); + installEventCompletionFailure(context.eventId()); + ClaimedSyncEvent firstClaim = syncEventClaimService.claim().orElseThrow(); + + assertThatThrownBy(() -> syncEventDispatchService.dispatch(firstClaim)) + .isInstanceOf(RuntimeException.class); + assertThat(jobCount(context.eventId())).isZero(); + assertThat(event(context.eventId()).getStatus()).isEqualTo(SyncEventStatus.PROCESSING); + syncEventFailureService.recordFailure( + firstClaim.eventId(), + firstClaim.claimToken(), + "FORCED_BEFORE_EVENT_COMPLETE", + "Event 완료 직전 강제 실패" + ); + assertAttempt(1, "FAILED", "FORCED_BEFORE_EVENT_COMPLETE", context.eventId()); + + dropFailureTriggers(); + dispatchNext(); + assertRecovered(context, 1); + } + + @Test + @DisplayName("과거 부분 Commit으로 Job이 유실돼도 Reconciler가 Repair Event로 복구하고 Issue를 해결한다") + void missingJob_isRecoveredAndResolvedByReconciler() { + EventContext context = eventWithoutJob("reconcile-missing-job"); + jdbcTemplate.update(""" + UPDATE sync_outbox_events + SET status = 'PROCESSED', processed_at = CURRENT_TIMESTAMP + WHERE event_id = ? + """, context.eventId()); + + // 1. 처리 중 Version인데 원인 Event만 완료되고 Job이 없는 과거 손상 상태를 검사한다. + SyncReconciliationBatchResult repair = syncReconciliationOrchestrator.reconcileBatch( + 0L, + SyncReconciliationMode.REPAIR + ); + assertThat(repair.detectedCount()).isOne(); + assertThat(repair.repairRequestedCount()).isOne(); + assertThat(count(""" + SELECT COUNT(*) FROM sync_consistency_issues + WHERE issue_type = 'MISSING_JOB' AND status = 'REPAIRING' + """)).isOne(); + + // 2. Dispatcher가 Reconciler의 Repair Event를 처리해 단일 Job을 복원한다. + dispatchNext(); + assertThat(count( + "SELECT COUNT(*) FROM embedding_jobs WHERE document_version_id = ?", + context.versionId() + )).isOne(); + + // 3. 재검사는 불일치가 사라졌음을 확인하고 기존 Issue를 RESOLVED로 종결한다. + SyncReconciliationBatchResult verify = syncReconciliationOrchestrator.reconcileBatch( + 0L, + SyncReconciliationMode.DRY_RUN + ); + assertThat(verify.detectedCount()).isZero(); + assertThat(count(""" + SELECT COUNT(*) FROM sync_consistency_issues + WHERE issue_type = 'MISSING_JOB' AND status = 'RESOLVED' + """)).isOne(); + } + + private void claimAndCrash() { + syncEventClaimService.claim().orElseThrow(); + throw new ForcedFailure("forced crash after event claim"); + } + + private void dispatchNext() { + jdbcTemplate.update( + "UPDATE sync_outbox_events SET available_at = CURRENT_TIMESTAMP - INTERVAL '1 second' " + + "WHERE status = 'PENDING'" + ); + ClaimedSyncEvent claim = syncEventClaimService.claim().orElseThrow(); + syncEventDispatchService.dispatch(claim); + } + + private EventContext eventWithoutJob(String label) { + String marker = label + "-" + UUID.randomUUID(); + DocumentUploadResponse response = documentUploadFacade.upload( + userId, + new DocumentUploadRequest( + new MockMultipartFile("file", marker + ".txt", "text/plain", marker.getBytes()), + marker, + "Sync 장애 복구 통합 테스트", + VisibilityType.PRIVATE + ) + ); + UUID eventId = jdbcTemplate.queryForObject( + "SELECT source_event_id FROM embedding_jobs WHERE id = ?", + UUID.class, + response.embeddingJobId() + ); + jdbcTemplate.update("DELETE FROM embedding_jobs WHERE id = ?", response.embeddingJobId()); + return new EventContext(eventId, response.documentVersionId()); + } + + private void assertRecovered(EventContext context, int expectedRetryCount) { + SyncOutboxEvent recovered = event(context.eventId()); + assertThat(recovered.getStatus()).isEqualTo(SyncEventStatus.PROCESSED); + assertThat(recovered.getProcessedAt()).isNotNull(); + assertThat(recovered.getRetryCount()).isEqualTo(expectedRetryCount); + assertThat(recovered.getClaimToken()).isNull(); + assertThat(recovered.getLockedBy()).isNull(); + assertThat(recovered.getLockExpiresAt()).isNull(); + assertThat(jobCount(context.eventId())).isOne(); + assertThat(count( + "SELECT COUNT(*) FROM embedding_jobs WHERE document_version_id = ?", + context.versionId() + )).isOne(); + assertAttempt(2, "SUCCEEDED", null, context.eventId()); + } + + private void assertAttempt(int attemptNo, String status, String errorCode, UUID eventId) { + List> attempts = jdbcTemplate.queryForList(""" + SELECT status, error_code, completed_at + FROM sync_event_delivery_attempts + WHERE event_id = ? AND attempt_no = ? + """, eventId, attemptNo); + assertThat(attempts).hasSize(1); + assertThat(attempts.get(0).get("status")).isEqualTo(status); + assertThat(attempts.get(0).get("error_code")).isEqualTo(errorCode); + assertThat(attempts.get(0).get("completed_at")).isNotNull(); + } + + private SyncOutboxEvent event(UUID eventId) { + return syncOutboxEventRepository.findByEventId(eventId).orElseThrow(); + } + + private int jobCount(UUID eventId) { + return count("SELECT COUNT(*) FROM embedding_jobs WHERE source_event_id = ?", eventId); + } + + private int count(String sql, Object... arguments) { + return jdbcTemplate.queryForObject(sql, Integer.class, arguments); + } + + private Long activeModelId() { + return jdbcTemplate.queryForObject( + "SELECT id FROM embedding_models WHERE is_active = TRUE AND is_searchable = TRUE", + Long.class + ); + } + + private void installJobInsertFailure(UUID eventId) { + jdbcTemplate.execute(""" + CREATE OR REPLACE FUNCTION docgrid_test_raise_sync_job_insert_failure() + RETURNS trigger AS $$ + BEGIN + IF NEW.source_event_id = '%s'::uuid THEN + RAISE EXCEPTION 'forced failure after sync job insert'; + END IF; + RETURN NEW; + END; + $$ LANGUAGE plpgsql + """.formatted(eventId)); + jdbcTemplate.execute(""" + CREATE TRIGGER docgrid_test_fail_sync_job_insert + AFTER INSERT ON embedding_jobs + FOR EACH ROW EXECUTE FUNCTION docgrid_test_raise_sync_job_insert_failure() + """); + } + + private void installEventCompletionFailure(UUID eventId) { + jdbcTemplate.execute(""" + CREATE OR REPLACE FUNCTION docgrid_test_raise_sync_event_complete_failure() + RETURNS trigger AS $$ + BEGIN + IF NEW.event_id = '%s'::uuid AND NEW.status = 'PROCESSED' THEN + RAISE EXCEPTION 'forced failure before sync event completion'; + END IF; + RETURN NEW; + END; + $$ LANGUAGE plpgsql + """.formatted(eventId)); + jdbcTemplate.execute(""" + CREATE TRIGGER docgrid_test_fail_sync_event_complete + BEFORE UPDATE ON sync_outbox_events + FOR EACH ROW EXECUTE FUNCTION docgrid_test_raise_sync_event_complete_failure() + """); + } + + private void dropFailureTriggers() { + jdbcTemplate.execute("DROP TRIGGER IF EXISTS " + JOB_TRIGGER + " ON embedding_jobs"); + jdbcTemplate.execute("DROP FUNCTION IF EXISTS " + JOB_FUNCTION + "()"); + jdbcTemplate.execute("DROP TRIGGER IF EXISTS " + COMPLETE_TRIGGER + " ON sync_outbox_events"); + jdbcTemplate.execute("DROP FUNCTION IF EXISTS " + COMPLETE_FUNCTION + "()"); + } + + /** + * 복구 대상 Sync Event와 파생 Job이 가리켜야 할 Version 식별자다. + */ + private record EventContext(UUID eventId, Long versionId) { + } + + /** + * 프로세스가 Claim 직후 종료된 경계를 테스트 흐름에서 명시하는 강제 예외다. + */ + private static final class ForcedFailure extends RuntimeException { + + private ForcedFailure(String message) { + super(message); + } + } +} diff --git a/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventClaimServiceTest.java b/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventClaimServiceTest.java index cd68d4e..5280b0f 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventClaimServiceTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventClaimServiceTest.java @@ -2,6 +2,7 @@ import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.then; import java.time.Clock; import java.time.Duration; @@ -37,6 +38,7 @@ class SyncEventClaimServiceTest { private static final ZoneId ZONE_ID = ZoneId.of("Asia/Seoul"); @Mock private SyncOutboxEventRepository syncOutboxEventRepository; + @Mock private SyncEventDeliveryAttemptService syncEventDeliveryAttemptService; private SyncEventClaimService service; private SyncDispatcherProperties properties; @@ -49,6 +51,7 @@ void setUp() { service = new SyncEventClaimService( syncOutboxEventRepository, properties, + syncEventDeliveryAttemptService, Clock.fixed(NOW, ZONE_ID) ); } @@ -68,6 +71,11 @@ void claim_assignsOwnershipAndLease() { assertThat(event.getStatus()).isEqualTo(SyncEventStatus.PROCESSING); assertThat(event.getLockedBy()).isEqualTo("dispatcher-test"); assertThat(event.getLockExpiresAt()).isEqualTo(claimedAt.plusSeconds(30)); + then(syncEventDeliveryAttemptService).should().start( + event, + result.orElseThrow().claimToken(), + claimedAt + ); } private SyncOutboxEvent event(LocalDateTime availableAt) { diff --git a/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchServiceTest.java b/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchServiceTest.java index b8839f7..246ed8e 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchServiceTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchServiceTest.java @@ -5,6 +5,7 @@ import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.then; import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.never; import java.time.Clock; import java.time.Instant; @@ -40,6 +41,7 @@ class SyncEventDispatchServiceTest { @Mock private SyncOutboxEventRepository syncOutboxEventRepository; @Mock private SyncEventHandlerRegistry syncEventHandlerRegistry; + @Mock private SyncEventDeliveryAttemptService syncEventDeliveryAttemptService; private SyncEventDispatchService service; private SyncOutboxEvent event; @@ -48,7 +50,12 @@ class SyncEventDispatchServiceTest { @BeforeEach void setUp() { Clock clock = Clock.fixed(NOW, ZONE_ID); - service = new SyncEventDispatchService(syncOutboxEventRepository, syncEventHandlerRegistry, clock); + service = new SyncEventDispatchService( + syncOutboxEventRepository, + syncEventHandlerRegistry, + syncEventDeliveryAttemptService, + clock + ); LocalDateTime now = LocalDateTime.ofInstant(NOW, ZONE_ID); event = event(now); UUID claimToken = UUID.randomUUID(); @@ -64,6 +71,11 @@ void dispatch_completesEvent_afterHandlerSucceeds() { service.dispatch(claim); then(syncEventHandlerRegistry).should().handle(event); + then(syncEventDeliveryAttemptService).should().succeed( + event.getEventId(), + claim.claimToken(), + LocalDateTime.ofInstant(NOW, ZONE_ID) + ); assertThat(event.getStatus()).isEqualTo(SyncEventStatus.PROCESSED); assertThat(event.getProcessedAt()).isNotNull(); } @@ -76,6 +88,11 @@ void dispatch_doesNotComplete_whenHandlerFails() { assertThatThrownBy(() -> service.dispatch(claim)) .isInstanceOf(IllegalStateException.class); + then(syncEventDeliveryAttemptService).should(never()).succeed( + event.getEventId(), + claim.claimToken(), + LocalDateTime.ofInstant(NOW, ZONE_ID) + ); assertThat(event.getStatus()).isEqualTo(SyncEventStatus.PROCESSING); assertThat(event.getProcessedAt()).isNull(); } diff --git a/docs/test-results/gimin-#168-sync-failure-recovery.md b/docs/test-results/gimin-#168-sync-failure-recovery.md new file mode 100644 index 0000000..48a8f8d --- /dev/null +++ b/docs/test-results/gimin-#168-sync-failure-recovery.md @@ -0,0 +1,71 @@ +# Sync 장애 주입·복구 검증 결과 + +- 관련 이슈: #168 +- 실행 일시: 2026-08-13 (Asia/Seoul) +- 환경: Java 17, Spring Boot test profile, PostgreSQL 17.8 + pgvector +- 격리 스키마: + - `docgrid_sync_dispatch_failure_test` + - `docgrid_embedding_failure_recovery_test` + +## 성공 기준 + +- 장애 뒤 Event, Job, Attempt, Version, Document가 유효한 상태로 수렴한다. +- Job, Chunk, Vector 중복이 없다. +- Handler 부작용과 Event 완료 사이의 부분 Commit이 없다. +- 실패 원인과 복구 실행 이력이 남는다. +- Reconciler 재검사 결과 활성 불일치가 없다. + +## 장애 지점별 결과 + +| 장애 지점 | 주입 방법 | 장애 직후 상태 | 복구 경로 | 최종 결과 | +|---|---|---|---|---| +| Event Claim 직후 | Claim Commit 뒤 테스트 프로세스 예외 | Event `PROCESSING`, Job 0건 | Sync Lease Recovery → 재Claim | Event `PROCESSED`, Job 1건 | +| Handler 시작 직후 | 필수 payload 제거 | Event 완료 안 됨, Job 0건 | 실패 기록 → payload 복원 → Retry | Event `PROCESSED`, Job 1건 | +| Job 생성 직후 | `embedding_jobs` AFTER INSERT Trigger 예외 | Job INSERT와 Dispatch Transaction Rollback | 실패 기록 → Retry | Job 1건, 중복 0건 | +| Embedding 저장 도중 | 두 번째 `embeddings` INSERT Trigger 예외 | Vector 0건, Version `EMBEDDING` | Worker Lease Recovery → 새 Attempt | Vector 3건, 중복 0건 | +| Event 완료 직전 | `PROCESSED` UPDATE Trigger 예외 | Handler Job과 Attempt 성공 전이 Rollback | 실패 기록 → Retry | Event `PROCESSED`, Job 1건 | + +## 복구 이력 검증 + +- Sync Event Claim마다 `sync_event_delivery_attempts`에 별도 실행 이력을 남겼다. +- Lease 만료 Attempt는 `FAILED / SYNC_LEASE_EXPIRED`로 보존됐다. +- Handler·Job·완료 장애는 각각 주입 지점별 오류 코드와 완료 시각을 보존했다. +- 복구 Claim은 별도 Attempt로 생성되어 `SUCCEEDED`로 종결됐다. +- Embedding 첫 Attempt는 `FAILED / WORKER_LEASE_EXPIRED`, 두 번째 Attempt는 `SUCCESS`로 종결됐다. +- `indexing_events`에 `LEASE_EXPIRED`, 단계 실패, `RETRY`, 최종 `INDEXED` 흐름이 남았다. + +## Reconciliation 검증 + +1. 완료 Event만 있고 Job이 없는 손상 상태를 만들었다. +2. `REPAIR` 검사에서 `MISSING_JOB` 한 건을 탐지하고 Repair Event 한 건을 생성했다. +3. Dispatcher가 Repair Event를 처리해 Job 한 건을 복원했다. +4. `DRY_RUN` 재검사에서 탐지 0건, 복구 요청 0건을 확인했다. +5. 기존 `MISSING_JOB` Issue는 `RESOLVED`로 종결됐다. +6. Embedding 장애 복구 완료 후에도 탐지 0건, `current_version_id` 일치, Vector 중복 0건을 확인했다. + +이 과정에서 실제 Reconciler 실행이 `document_chunks`에 존재하지 않는 `document_id`를 직접 참조하는 +고아 검사 오류를 발견했다. Document 원장은 `document_versions.document_id`를 통해 조인하도록 수정했고, +장애 복구 후 실제 Reconciliation 통합 테스트로 재검증했다. + +## 실행 명령과 결과 + +```bash +DB_PORT=5432 \ +JWT_SECRET=test-only-secret-key-with-at-least-32-characters \ +./gradlew test \ + --tests '*SyncDispatchFailureRecoveryIntegrationTest' \ + --tests '*EmbeddingFailureRecoveryIntegrationTest' \ + --tests '*SyncEventClaimServiceTest' \ + --tests '*SyncEventDispatchServiceTest' +``` + +- 결과: `BUILD SUCCESSFUL` + +```bash +DB_PORT=5432 \ +JWT_SECRET=test-only-secret-key-with-at-least-32-characters \ +./gradlew test +``` + +- 결과: `BUILD SUCCESSFUL in 34s` +- PostgreSQL 장애 Trigger는 테스트별 격리 스키마에만 만들고 각 테스트와 클래스 종료 시 제거했다.