diff --git a/docs/design/Gimini-3-#108-manual-retry-failed-indexing-job.md b/docs/design/Gimini-3-#108-manual-retry-failed-indexing-job.md new file mode 100644 index 0000000..ddd5f7e --- /dev/null +++ b/docs/design/Gimini-3-#108-manual-retry-failed-indexing-job.md @@ -0,0 +1,249 @@ +# Issue #108 최종 실패 인덱싱 Job 수동 재처리 상세 설계 + +closes #108 + +## 1. 문서 목적 + +이 문서는 이슈 [#108](https://github.com/DocGrid/backend/issues/108)의 구현 기준을 정의한다. + +인덱싱 Job은 실패 유형이 재시도 가능하고 남은 횟수가 있을 때만 `PENDING` Queue로 재예약된다. 재시도 +가능 횟수를 모두 소진했거나 재시도 불가 유형으로 종료된 Job은 `FAILED`로 종결되고, 대상 Document +Version은 `FAILED`, 그 Version의 Embedding Set은 `STALE`이 되어 검색에서 제외된다. + +`FAILED` Job을 다시 처리할 경로는 현재 존재하지 않는다. Claim은 `PENDING`만, Lease 만료 복구는 +`PROCESSING`만 후보로 삼기 때문에 어떤 자동 경로도 `FAILED` Job을 되살리지 않는다. 외부 Embedding +서버 장애나 일시적인 Storage 장애처럼 원인이 이미 해소된 뒤에도 같은 문서를 다시 인덱싱하려면 새 +Version을 업로드하는 방법밖에 없다. + +이 작업은 최종 실패로 종결된 Job만 관리자가 명시적으로 Queue에 되돌릴 수 있는 수동 재처리 경로를 +추가한다. + +### 1.1 성공 기준 + +- 최종 `FAILED` Job만 수동 재처리 대상이 된다. +- 처리 중이거나 자동 재시도가 예정된 Job은 명시적으로 거부한다. +- 최신 처리 대상 Version이 아니면 거부한다. +- 현재 검색 가능한 이전 Version과 `current_version` 포인터를 보존한다. +- 이전 Worker, Claim Token, Lease 등 소유권 정보를 초기화한다. +- Retry Count와 기존 Attempt 이력을 삭제하지 않는다. +- 수동 재처리를 새 상태 전이와 감사 Event로 남긴다. +- 기존 Chunk를 무조건 삭제하지 않는다. +- 중복 요청과 동시 요청이 하나의 상태 전이로 수렴한다. +- Lease 만료 복구 및 자동 재시도와 경합하지 않는다. +- 관리자 권한을 요구하고 Claim Token 등 민감 정보를 응답에 노출하지 않는다. + +## 2. 범위 + +### 2.1 포함 + +- 최종 실패 Job 한 건을 다시 Queue에 넣는 관리자 API +- Job, Document Version, Document의 재처리 상태 전이 +- 재개 지점 결정과 대상 Version Embedding 정리 +- 수동 재처리 감사 Event 기록 + +### 2.2 제외 + +- 재처리 대상 Job 목록·상세 조회 API (관리자 인덱싱 조회 작업에서 진행) +- 자동 재시도 정책과 Backoff 계산 변경 +- 여러 Job을 한 번에 재처리하는 Batch API +- 재처리 예약, 스케줄링, 자동 트리거 +- OCR 등 실패 원인 자체를 해결하는 파싱 기능 + +## 3. 현재 구조 분석 + +### 3.1 상태 모델 + +`EmbeddingJobStatus`는 `PENDING`, `PROCESSING`, `INDEXED`, `FAILED`, `CANCELED`로 구성된다. 별도의 +재시도 예약 상태는 없고, 자동 재시도가 예정된 Job은 `PENDING` + 미래의 `next_retry_at`으로 표현된다. +Claim Query는 `next_retry_at IS NULL OR next_retry_at <= :claimedAt` 조건을 사용하므로 예약 시각 전에는 +후보가 되지 않는다. + +### 3.2 최종 실패 시점의 데이터 상태 + +`IndexingFailureTransitionService`의 최종 실패 경로는 하나의 Transaction에서 다음을 수행한다. + +1. 대상 Version의 `ACTIVE` Embedding을 모두 `STALE`로 전환 +2. `document_versions.status`를 `FAILED`로 전환 +3. 이전 `INDEXED` Version이 현재 검색 대상이면 Document를 그대로 두고, 아니면 `FAILED`로 전환 +4. `embedding_jobs.status`를 `FAILED`로 전환하고 `failed_at`, 오류 Snapshot 기록 +5. 단계 실패 Event와 `FAILED` Event를 같은 시각으로 append + +`markFailed`는 `locked_by_worker_id`, `claim_token`, `locked_at`, `lock_expires_at`을 감사 목적으로 +남긴다. `document_chunks`는 삭제하지 않는다. + +### 3.3 재개 지점 계약 + +파이프라인 각 단계는 Version 상태로 재개 지점을 판단한다. + +| Version 상태 | 동작 | +|---|---| +| `UPLOADED`, `PARSING` | 원본을 다시 읽어 파싱하고 Chunk Set 저장 | +| `CHUNKED` | 파싱을 생략하고 Embedding 생성 | +| `EMBEDDING` | 저장된 Embedding 수에 따라 재생 또는 재작업 | + +Chunk Set 저장과 `CHUNKED` 전이는 같은 Transaction에서 일어나므로 Chunk가 존재하면 항상 완전한 +Set이다. 반면 `CHUNKED` 상태에서 대상 Version·Model의 Embedding 행이 0이 아니면 +`DOCUMENT_EMBEDDINGS_INCONSISTENT`로 차단된다. 따라서 최종 실패가 남긴 `STALE` Embedding을 정리하지 +않으면 재처리 자체가 불가능하다. + +### 3.4 검색 보호 장치 + +Vector 검색 Query는 `e.status = 'ACTIVE' AND d.status = 'INDEXED' AND d.current_version_id = +e.document_version_id` 조건을 사용한다. 실패한 Version의 Embedding은 `STALE`이므로 원래 검색에 노출될 +수 없고, 이전 `INDEXED` Version은 `current_version_id`가 유지되는 한 계속 검색된다. + +## 4. 설계 + +### 4.1 상태 전이 계약 + +```text +사전조건: embedding_jobs.status = FAILED + document_versions = 해당 문서의 최신 Version, status = FAILED + documents.deleted_at IS NULL + documents.status ∈ {UPLOADED, INDEXING, INDEXED, FAILED} + 같은 Version에 PENDING/PROCESSING Job 없음 + +전이: Job: FAILED -> PENDING, next_retry_at = NULL + locked_by_worker_id, claim_token, locked_at, lock_expires_at, failed_at = NULL + retry_count, max_retry_count, error_code, error_message 보존 + Version: FAILED -> CHUNKED (Chunk가 이미 있는 경우) + -> UPLOADED (Chunk가 없는 경우) + Document: 이전 INDEXED Version이 현재 검색 대상이면 변경 없음 + 그 외에는 INDEXING + Event: MANUAL_RETRY (FAILED -> PENDING) 1건 append + Attempt: 변경 없음 +``` + +`retry_count`를 유지하므로 수동 재처리는 추가 실행 1회만 부여한다. 이번 실행이 다시 실패하면 +`hasRemainingRetries()`가 거짓이 되어 자동 재시도 없이 즉시 최종 실패로 종결되고, 필요하면 관리자가 +다시 수동 재처리를 요청한다. 이 선택은 재시도 이력을 지우지 않으면서 무한 자동 재시도를 만들지 +않기 위한 것이다. + +### 4.2 Chunk와 Embedding 처리 정책 + +- Chunk는 삭제하지 않는다. 존재하면 완전한 Set이므로 파싱을 생략하고 재사용한다. +- 대상 Version의 Embedding 행만 삭제한다. 최종 실패 시점에 이미 `STALE`이라 검색에 노출되지 않으며, + 남겨두면 Embedding 개수 불변식 검증에서 재처리가 차단된다. +- 다른 Version의 Chunk와 Embedding은 조회하지도 변경하지도 않는다. + +실패한 실행이 남긴 Vector를 다시 `ACTIVE`로 되살리는 방식은 채택하지 않았다. 완료 검증 단계에서 +실패한 경우 그 Vector Set이 실제로 불완전할 수 있고, 이를 판별하려면 완료 Transaction과 같은 수준의 +검증을 재처리 경로에 중복 구현해야 하기 때문이다. + +### 4.3 Transaction 경계와 잠금 순서 + +`EmbeddingJobManualRetryService`는 단일 `@Transactional` 경계에서 외부 I/O 없이 동작한다. + +1. `findByIdForUpdate`로 Job 행을 잠근다. +2. Job 상태가 `FAILED`인지 확인한다. +3. Version, Document를 기존 경로와 같은 순서로 잠근다. +4. 재처리 대상 조건을 모두 검증한다. +5. 재개 지점을 정하고 대상 Version Embedding을 삭제한다. +6. Job, Version, Document 상태를 바꾸고 `MANUAL_RETRY` Event를 append한다. + +Job 행 잠금이 Claim, 완료, 협력적 실패, Lease 복구와의 단일 직렬화 지점이다. Lease 복구는 +`PROCESSING` + 만료 행만, Claim은 `PENDING` 행만 후보로 삼으므로 커밋 전에는 이 Transaction과 경합하지 +않고, 커밋 후에는 정상 Claim 경로로 흡수된다. + +### 4.4 API 계약 + +```text +POST /admin/indexing-jobs/{jobId}/retry +Request Body 없음, ADMIN 권한 필요 + +200 OK +{ + "success": true, + "data": { + "jobId": 10, + "status": "PENDING", + "documentId": 3, + "documentVersionId": 5, + "documentVersionStatus": "CHUNKED", + "retryCount": 3, + "maxRetryCount": 3, + "requeuedAt": "2026-08-06T15:00:00" + } +} +``` + +`/admin/**`은 `SecurityConfig`에서 이미 `hasRole("ADMIN")`으로 보호되므로 Security 설정은 변경하지 +않는다. 응답에는 Claim Token, 실패 원인 상세, 내부 예외 정보를 포함하지 않는다. + +## 5. 오류 케이스 + +| 상황 | HTTP | 코드 | +|---|---|---| +| Job 없음 | 404 | `EMBEDDING-JOB-001` | +| Job이 `PENDING`·`PROCESSING`·`INDEXED`·`CANCELED` (중복 요청 포함) | 409 | `EMBEDDING-JOB-008` | +| 최신 Version이 아님 | 409 | `EMBEDDING-JOB-009` | +| 삭제된 문서이거나 재처리 불가 문서 상태 | 409 | `EMBEDDING-JOB-009` | +| 같은 Version에 살아 있는 Job 존재 | 409 | `EMBEDDING-JOB-009` | +| Version이 `FAILED`가 아니거나 현재 Version 포인터 불일치 | 500 | `DOCUMENT-INDEXING-004` | +| Job ID가 양수가 아님 | 400 | `COMMON-002` | + +중복 요청은 멱등 재생 대신 명시적 충돌로 처리한다. 현재 Schema에는 `PENDING` Job이 자동 재시도 +예약인지 수동 재처리 결과인지 구분하는 식별자가 없어, 멱등 재생을 지원하려면 추가 Column이나 Event +조회가 필요하기 때문이다. + +## 6. 테스트 설계 + +### 6.1 단위 테스트 + +`EmbeddingJobManualRetryServiceTest` (Mockito) + +- Chunk 존재 시 `CHUNKED` 재개, 미존재 시 `UPLOADED` 재개 +- 소유권 필드와 종료 시각 초기화, `retry_count` 보존 +- Claim Token 없는 `MANUAL_RETRY` Event 기록 +- 이전 `INDEXED` Version이 있을 때 문서 상태·포인터 보존 +- Job 없음, `PENDING`·`PROCESSING`·`INDEXED` 거부 +- 최신 Version 아님, 삭제된 문서, 살아 있는 Job 존재 거부 +- Version이 `FAILED`가 아닐 때 불변식 오류 + +### 6.2 Controller 테스트 + +`IndexingJobAdminControllerTest` (`@WebMvcTest`) + +- 정상 응답 필드와 민감 정보 미노출 +- Job ID Validation +- 정의된 오류 코드와 HTTP 상태 매핑 +- ADMIN 외 사용자와 미인증 요청 차단 + +### 6.3 통합 테스트 + +`EmbeddingJobManualRetryIntegrationTest` (`@Tag("integration")`, 실제 PostgreSQL) + +- 소유권 초기화 후 즉시 Claim 후보가 되는지 확인 +- Chunk 유지와 대상 Version Embedding 삭제 +- Chunk 없는 Job의 `UPLOADED` 재개 +- Attempt 이력·재시도 횟수 보존과 `MANUAL_RETRY` Event 1건 +- 이전 `INDEXED` Version의 검색 결과와 현재 포인터 보존 +- 동시 요청 2건이 전이 1회 + 충돌 1회로 수렴 +- 자동 재시도 예정 Job 거부 시 예약 유지 +- 최신 Version이 아닐 때 거부하고 기존 데이터 유지 + +## 7. 커밋 분할 + +1. `feat: #108 최종 실패 Job 수동 재처리 도메인 규칙 추가` +2. `feat: #108 수동 재처리 대상 Embedding 삭제 쿼리 추가` +3. `feat: #108 최종 실패 Job 수동 재처리 Command Service 구현` +4. `feat: #108 관리자 수동 재처리 API 추가` +5. `test: #108 수동 재처리 단위·Controller 테스트 추가` +6. `test: #108 수동 재처리 PostgreSQL 통합 테스트와 검증 결과 추가` +7. `docs: #108 최종 실패 Job 수동 재처리 설계 문서 추가` + +## 8. 완료 조건 + +- 최종 `FAILED` Job만 수동 재처리 가능 +- `PENDING`·`PROCESSING`·`INDEXED`·`CANCELED` 거부 +- 현재 검색 가능한 Version 유지 +- Attempt와 Retry 감사 이력 유지 +- 소유권 정보 초기화 +- 동시 재처리 요청이 하나의 상태 전이로 수렴 +- 전체 회귀 테스트 통과 + +## 9. 참고 + +- Flyway 마이그레이션 없음. `indexing_events.event_type`은 CHECK 제약이 없는 `VARCHAR(30)`이라 + `MANUAL_RETRY` 값을 그대로 저장할 수 있다. +- `SecurityConfig` 변경 없음. diff --git a/docs/test-results/Gimini-3-#108-manual-retry-failed-indexing-job.md b/docs/test-results/Gimini-3-#108-manual-retry-failed-indexing-job.md new file mode 100644 index 0000000..b7b7d83 --- /dev/null +++ b/docs/test-results/Gimini-3-#108-manual-retry-failed-indexing-job.md @@ -0,0 +1,148 @@ +# #108 최종 실패 인덱싱 Job 수동 재처리 검증 결과 + +## 1. 검증 정보 + +- 실행일: 2026-08-06 (Asia/Seoul) +- 대상 브랜치: `feature/108` +- 애플리케이션: Spring Boot 3.5.16, Java 17 +- 데이터베이스: 로컬 PostgreSQL 17.8 + pgvector 0.8.1 컨테이너 +- 검증 범위: 전체 회귀 Test와 수동 재처리 전용 단위·Controller·PostgreSQL 통합 Test +- 최종 결과: 653개 통과, 실패·오류·Skip 0개 + +Test Class별 격리 Schema와 Flyway Migration을 사용했고, 기존 개발 데이터는 변경하지 않았다. DB 접속 +정보와 인증 값은 실행 Process 환경변수로만 주입했으며 실제 값은 기록하지 않는다. + +## 2. 전체 회귀 검증 + +실행 명령의 환경 값은 Placeholder로 대체한다. + +```bash +DB_PORT='' \ +DB_SSLMODE=disable \ +JWT_SECRET='' \ +MINIO_ENDPOINT='' \ +MINIO_ACCESS_KEY='' \ +MINIO_SECRET_KEY='' \ +MINIO_BUCKET='' \ +./gradlew test +``` + +결과: + +```text +BUILD SUCCESSFUL +tests=653 failures=0 errors=0 skipped=0 +``` + +같은 명령을 `develop`(`e1bd2d2`)에서 실행한 기준 Test 수는 625개이며, 이번 작업으로 단위 13개, 통합 +8개, Controller 7개가 추가되어 653개가 됐다. + +```text +develop : tests=625 failures=0 errors=0 +feature : tests=653 failures=0 errors=0 (625 + 28) +``` + +## 3. 단위 검증 + +`EmbeddingJobManualRetryServiceTest` 13개 통과. + +| 시나리오 | 기대 | 결과 | +|---|---|---| +| Chunk가 있는 최종 실패 Job | `CHUNKED` 재개, 문서 `INDEXING` | 통과 | +| Chunk가 없는 최종 실패 Job | `UPLOADED` 재개 | 통과 | +| 재처리 후 소유권과 재시도 이력 | Worker·Token·Lease·`failed_at` 모두 `null`, `retry_count = 3` 유지, 잔여 자동 재시도 없음 | 통과 | +| 감사 Event | `MANUAL_RETRY` 1건, Claim Token 미포함 | 통과 | +| 이전 `INDEXED` Version 존재 | 문서 상태·현재 포인터 보존 | 통과 | +| Job 없음 | `EMBEDDING_JOB_NOT_FOUND` | 통과 | +| `PENDING` Job | `EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED` | 통과 | +| `PROCESSING` Job | `EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED` | 통과 | +| `INDEXED` Job | `EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED` | 통과 | +| 최신 Version 아님 | `EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID`, 상태 변경 없음 | 통과 | +| 삭제된 문서 | `EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID` | 통과 | +| 같은 Version에 살아 있는 Job | `EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID` | 통과 | +| Version이 `FAILED`가 아님 | `DOCUMENT_INDEXING_FAILURE_INCONSISTENT` | 통과 | + +## 4. Controller 계약 검증 + +`IndexingJobAdminControllerTest`의 수동 재처리 Test 7개 통과. + +| 시나리오 | 기대 | 결과 | +|---|---|---| +| ADMIN 정상 요청 | 200, 재개 지점 포함, `claimToken`·`errorMessage` 미노출 | 통과 | +| `jobId = 0` | 400 `COMMON-002` | 통과 | +| Job 없음 | 404 `EMBEDDING-JOB-001` | 통과 | +| 최종 실패 Job 아님 | 409 `EMBEDDING-JOB-008` | 통과 | +| 대상 조건 불충족 | 409 `EMBEDDING-JOB-009` | 통과 | +| 종료 데이터 불일치 | 500 `DOCUMENT-INDEXING-004` | 통과 | +| USER 권한·미인증 | 403 | 통과 | + +## 5. PostgreSQL 통합 검증 + +실행: + +```bash +DB_PORT='' \ +DB_SSLMODE=disable \ +JWT_SECRET='' \ +MINIO_ENDPOINT='' \ +MINIO_ACCESS_KEY='' \ +MINIO_SECRET_KEY='' \ +MINIO_BUCKET='' \ +./gradlew test \ + --tests 'com.opensource.docgrid.domain.embedding.integration.EmbeddingJobManualRetryIntegrationTest' +``` + +결과: + +```text +tests=8 failures=0 errors=0 skipped=0 +``` + +### 5.1 재현 절차 + +각 Test는 격리 Schema에 다음 상태를 직접 구성한 뒤 실제 Service Transaction을 호출한다. + +```text +embedding_jobs status=FAILED, retry_count=3, max_retry_count=3, + locked_by_worker_id / claim_token / lock_expires_at / failed_at 존재 +embedding_job_attempts status=FAILED (attempt_no=1) +indexing_events FAILED 1건 +document_versions status=FAILED +embeddings status=STALE (대상 Version) +documents status=FAILED 또는 이전 INDEXED Version 보유 +``` + +### 5.2 시나리오별 결과 + +| 시나리오 | 확인 항목 | 결과 | +|---|---|---| +| 최종 실패 Job 재처리 | `status=PENDING`, 소유권 4개 Column과 `failed_at`·`next_retry_at` `null`, `findNextPendingForUpdate`가 즉시 해당 Job 반환 | 통과 | +| Chunk 유지·Embedding 정리 | Version `CHUNKED`, `document_chunks` 1건 유지, 대상 Version `embeddings` 0건, 문서 `INDEXING` | 통과 | +| Chunk 없는 Job | Version `UPLOADED` | 통과 | +| 감사 이력 | `retry_count=3`, Attempt 1건 `FAILED` 유지, `FAILED` Event 1건 유지, `MANUAL_RETRY` Event 1건 추가, metadata에 Claim Token 없음 | 통과 | +| 이전 검색 Version 보호 | 문서 `INDEXED` 유지, `current_version_id`가 이전 Version, 이전 Version Embedding `ACTIVE`, 재처리 전후 Vector 검색 결과가 모두 `이전 검색 본문` | 통과 | +| 동시 요청 2건 | 커밋 1건, 나머지 1건 `EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED`, `MANUAL_RETRY` Event 정확히 1건, 대상 Embedding 0건 | 통과 | +| 자동 재시도 예정 Job | 409로 거부, `next_retry_at` 유지, Version `FAILED` 유지, `MANUAL_RETRY` Event 0건 | 통과 | +| 최신 Version 아님 | 409로 거부, Job `FAILED` 유지, 대상 Embedding 1건 그대로 유지 | 통과 | + +## 6. Swagger 수동 검증 + +이번 작업에서는 Swagger 수동 검증을 수행하지 않았다. 수동 재처리 API는 `hasRole("ADMIN")`으로 보호되며 +검증하려면 ADMIN 계정 생성과 JWT 발급, 그리고 자동 재시도를 모두 소진한 최종 실패 Job을 실제로 +만들어야 한다. 이 사전 상태는 현재 로컬에서 실제 문서 업로드부터 Worker 실행까지 전 구간을 돌려야 +재현할 수 있어, 예정된 로컬 전체 관통 E2E 작업에서 함께 수행하는 것이 적절하다. + +대체 검증으로 다음 두 계층을 사용했다. + +- HTTP 계약: `@WebMvcTest` 기반 Controller Test로 상태 코드, 응답 필드, 민감 정보 미노출, ADMIN 권한 + 차단을 확인 +- 실제 DB 동작: 실제 PostgreSQL 통합 Test로 상태 전이 원자성, 검색 보호, 동시 요청 수렴을 확인 + +## 7. 결론 + +- 최종 실패 Job만 수동 재처리 가능하고 나머지 상태는 모두 거부한다. +- 현재 검색 가능한 Version은 재처리 전후 동일한 검색 결과를 유지한다. +- Attempt와 재시도 감사 이력을 삭제하지 않는다. +- 소유권 정보가 초기화되어 과거 Claim Token으로는 후속 단계를 수행할 수 없다. +- 동시 재처리 요청은 하나의 상태 전이로 수렴한다. +- 전체 653개 Test가 실패 없이 통과한다. diff --git a/src/main/java/com/opensource/docgrid/domain/document/entity/DocumentVersion.java b/src/main/java/com/opensource/docgrid/domain/document/entity/DocumentVersion.java index 418b2a3..1db0798 100644 --- a/src/main/java/com/opensource/docgrid/domain/document/entity/DocumentVersion.java +++ b/src/main/java/com/opensource/docgrid/domain/document/entity/DocumentVersion.java @@ -37,7 +37,8 @@ * index: document_id, file_object_id, status에 대한 조회 인덱스. * *

주의사항: 파이프라인 상태는 UPLOADED -> PARSING -> CHUNKED -> EMBEDDING -> INDEXED 순으로 전이되며, - * 실패 시 FAILED로 전이된다. + * 실패 시 FAILED로 전이된다. FAILED에서 벗어나는 유일한 경로는 관리자 수동 재처리이며, 이때만 + * UPLOADED 또는 CHUNKED 재개 지점으로 되돌아간다. */ @Getter @Entity @@ -154,6 +155,28 @@ public void markIndexed(LocalDateTime indexedAt) { this.indexedAt = indexedAt; } + /** + * 최종 실패한 Version을 수동 재처리가 다시 진행할 수 있는 재개 지점으로 되돌린다. + * + *

파이프라인 각 단계는 Version 상태로 재개 지점을 판단하므로, 이미 저장된 Chunk Set이 있으면 + * CHUNKED로 되돌려 파싱을 생략하고 없으면 UPLOADED로 되돌려 파싱부터 다시 수행한다. + * 실제 Chunk 존재 여부 판단은 호출 Service가 담당한다. + * + * @param resumeStatus 재개 지점이 될 UPLOADED 또는 CHUNKED 상태 + */ + public void reopenFailedForRetry(DocumentVersionStatus resumeStatus) { + // 1. 검색 중이거나 처리 중인 Version이 재처리로 이전 단계로 되돌아가지 않게 한다. + if (status != DocumentVersionStatus.FAILED) { + throw new IllegalStateException("FAILED 상태의 문서 버전만 재처리로 되돌릴 수 있습니다."); + } + // 2. 저장된 Chunk·Embedding Set과 어긋나는 중간 단계로는 재개할 수 없다. + if (resumeStatus != DocumentVersionStatus.UPLOADED + && resumeStatus != DocumentVersionStatus.CHUNKED) { + throw new IllegalArgumentException("재처리 재개 지점은 UPLOADED 또는 CHUNKED만 가능합니다."); + } + this.status = resumeStatus; + } + public void markFailed() { // 처리 중인 Version만 실패할 수 있고 완료되거나 이미 실패한 결과는 덮어쓰지 않는다. if (status == DocumentVersionStatus.INDEXED || status == DocumentVersionStatus.FAILED) { diff --git a/src/main/java/com/opensource/docgrid/domain/embedding/controller/IndexingJobAdminController.java b/src/main/java/com/opensource/docgrid/domain/embedding/controller/IndexingJobAdminController.java index 948124e..8e4ce36 100644 --- a/src/main/java/com/opensource/docgrid/domain/embedding/controller/IndexingJobAdminController.java +++ b/src/main/java/com/opensource/docgrid/domain/embedding/controller/IndexingJobAdminController.java @@ -23,6 +23,7 @@ import com.opensource.docgrid.domain.embedding.dto.response.DocumentEmbeddingsResponse; import com.opensource.docgrid.domain.embedding.dto.response.DocumentIndexingCompletionResponse; import com.opensource.docgrid.domain.embedding.dto.response.DocumentIndexingFailureResponse; +import com.opensource.docgrid.domain.embedding.dto.response.ManualRetriedIndexingJobResponse; import com.opensource.docgrid.domain.embedding.dto.response.RenewedEmbeddingJobLeaseResponse; import com.opensource.docgrid.domain.embedding.dto.response.StartedEmbeddingJobAttemptResponse; import com.opensource.docgrid.domain.document.service.DocumentParsingService; @@ -35,6 +36,7 @@ import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobAttemptService.StartResult; import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobClaimService; import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobLeaseService; +import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobManualRetryService; import com.opensource.docgrid.global.common.response.ApiResponse; import com.opensource.docgrid.global.common.response.ErrorResponse; import com.opensource.docgrid.global.common.response.ResponseUtils; @@ -50,10 +52,10 @@ /** * 관리자용 Embedding Job Claim·Lease 갱신, Attempt 시작과 문서 Chunk·Embedding·인덱싱 완료·실패 - * 실행을 HTTP API로 제공한다. + * 실행 및 최종 실패 Job 수동 재처리를 HTTP API로 제공한다. * - *

HTTP 입력 검증과 성공 상태 변환만 담당한다. Job Claim 및 현재 소유권 기반 파이프라인 단계의 - * Transaction·외부 호출·동시성 규칙은 각 Service에 위임한다. + *

HTTP 입력 검증과 성공 상태 변환만 담당한다. Job Claim 및 현재 소유권 기반 파이프라인 단계와 + * 수동 재처리의 Transaction·외부 호출·동시성 규칙은 각 Service에 위임한다. */ @Tag(name = "Admin - Indexing Job", description = "관리자 전용 인덱싱 Job 제어 API") @Validated @@ -69,6 +71,7 @@ public class IndexingJobAdminController { private final DocumentEmbeddingService documentEmbeddingService; private final DocumentIndexingCompletionService documentIndexingCompletionService; private final DocumentIndexingFailureService documentIndexingFailureService; + private final EmbeddingJobManualRetryService embeddingJobManualRetryService; @Operation( summary = "PENDING Job Claim", @@ -465,4 +468,51 @@ public ResponseEntity> failIndexing // 최초 실패와 멱등 재생 모두 같은 Attempt 기반 실패 응답을 200 OK로 반환한다. return ResponseUtils.ok(documentIndexingFailureService.fail(jobId, attemptId, request)); } + + @Operation( + summary = "최종 실패 Job 수동 재처리", + description = "자동 재시도를 모두 마치고 최종 실패한 Job만 즉시 Claim 가능한 PENDING 상태로 되돌립니다. " + + "처리 중이거나 자동 재시도가 예정된 Job과 이미 재처리된 Job의 중복 요청은 409로 거부합니다. " + + "이미 저장된 Chunk가 있으면 파싱을 생략하고 임베딩 단계부터 다시 시작하며, " + + "현재 검색 가능한 이전 Version과 기존 Attempt 이력, 재시도 횟수는 그대로 유지합니다. " + + "재시도 횟수는 초기화하지 않으므로 이번 재처리가 다시 실패하면 곧바로 최종 실패로 종료됩니다." + ) + @ApiResponses({ + @io.swagger.v3.oas.annotations.responses.ApiResponse( + responseCode = "200", + description = "수동 재처리 성공" + ), + @io.swagger.v3.oas.annotations.responses.ApiResponse( + responseCode = "400", + description = "Job ID 형식 오류", + content = @Content(schema = @Schema(implementation = ErrorResponse.class)) + ), + @io.swagger.v3.oas.annotations.responses.ApiResponse( + responseCode = "403", + description = "인증되지 않았거나 ADMIN 권한 없음", + content = @Content(schema = @Schema(implementation = ErrorResponse.class)) + ), + @io.swagger.v3.oas.annotations.responses.ApiResponse( + responseCode = "404", + description = "Embedding Job 없음", + content = @Content(schema = @Schema(implementation = ErrorResponse.class)) + ), + @io.swagger.v3.oas.annotations.responses.ApiResponse( + responseCode = "409", + description = "최종 실패 Job이 아니거나 최신 Version·문서 상태가 재처리 조건을 만족하지 않음", + content = @Content(schema = @Schema(implementation = ErrorResponse.class)) + ), + @io.swagger.v3.oas.annotations.responses.ApiResponse( + responseCode = "500", + description = "Version 또는 Document 종료 데이터 불일치", + content = @Content(schema = @Schema(implementation = ErrorResponse.class)) + ) + }) + @PostMapping(value = "/{jobId}/retry", produces = MediaType.APPLICATION_JSON_VALUE) + public ResponseEntity> retryIndexingJob( + @PathVariable @Positive Long jobId + ) { + // Service가 Job → Version → Document 잠금과 대상 검증 및 Queue 복귀를 한 Transaction으로 처리한다. + return ResponseUtils.ok(embeddingJobManualRetryService.retry(jobId)); + } } diff --git a/src/main/java/com/opensource/docgrid/domain/embedding/converter/EmbeddingJobConverter.java b/src/main/java/com/opensource/docgrid/domain/embedding/converter/EmbeddingJobConverter.java index d79e1d6..6653cf1 100644 --- a/src/main/java/com/opensource/docgrid/domain/embedding/converter/EmbeddingJobConverter.java +++ b/src/main/java/com/opensource/docgrid/domain/embedding/converter/EmbeddingJobConverter.java @@ -4,7 +4,9 @@ import org.springframework.stereotype.Component; +import com.opensource.docgrid.domain.document.entity.DocumentVersion; import com.opensource.docgrid.domain.embedding.dto.response.ClaimedEmbeddingJobResponse; +import com.opensource.docgrid.domain.embedding.dto.response.ManualRetriedIndexingJobResponse; import com.opensource.docgrid.domain.embedding.dto.response.RenewedEmbeddingJobLeaseResponse; import com.opensource.docgrid.domain.embedding.entity.EmbeddingJob; @@ -12,7 +14,7 @@ * Embedding Job Entity를 API 전용 응답 DTO로 변환하는 Converter. * *

Controller에 Entity와 연관 Entity를 직접 노출하지 않고 Claim과 Lease 갱신 이후 Worker가 - * 필요한 식별자와 소유권 시각만 전달한다. + * 필요한 식별자와 소유권 시각, 관리자 수동 재처리 결과만 전달한다. */ @Component public class EmbeddingJobConverter { @@ -55,4 +57,32 @@ public RenewedEmbeddingJobLeaseResponse toRenewedLeaseResponse( embeddingJob.getLockExpiresAt() ); } + + /** + * 수동 재처리로 Queue에 복귀한 Job과 재개 지점을 응답 DTO로 변환한다. + * + *

재처리 직후에는 소유 Worker가 없으므로 Claim 관련 값은 담지 않고, 운영자가 확인해야 하는 + * 재개 대상과 보존된 재시도 횟수만 전달한다. + * + * @param embeddingJob PENDING으로 되돌아가고 소유권이 초기화된 Job + * @param documentVersion 재개 지점 상태로 되돌아간 대상 Version + * @param requeuedAt 재처리 Transaction이 사용한 기준 시각 + * @return 관리자에게 노출해도 안전한 수동 재처리 결과 + */ + public ManualRetriedIndexingJobResponse toManualRetriedResponse( + EmbeddingJob embeddingJob, + DocumentVersion documentVersion, + LocalDateTime requeuedAt + ) { + return new ManualRetriedIndexingJobResponse( + embeddingJob.getId(), + embeddingJob.getStatus(), + documentVersion.getDocument().getId(), + documentVersion.getId(), + documentVersion.getStatus(), + embeddingJob.getRetryCount(), + embeddingJob.getMaxRetryCount(), + requeuedAt + ); + } } diff --git a/src/main/java/com/opensource/docgrid/domain/embedding/dto/response/ManualRetriedIndexingJobResponse.java b/src/main/java/com/opensource/docgrid/domain/embedding/dto/response/ManualRetriedIndexingJobResponse.java new file mode 100644 index 0000000..d78e7ae --- /dev/null +++ b/src/main/java/com/opensource/docgrid/domain/embedding/dto/response/ManualRetriedIndexingJobResponse.java @@ -0,0 +1,44 @@ +package com.opensource.docgrid.domain.embedding.dto.response; + +import java.time.LocalDateTime; + +import com.opensource.docgrid.domain.document.enums.DocumentVersionStatus; +import com.opensource.docgrid.domain.embedding.enums.EmbeddingJobStatus; + +import io.swagger.v3.oas.annotations.media.Schema; + +/** + * 최종 실패한 Embedding Job을 관리자가 수동 재처리했을 때 반환되는 응답 DTO. + * + *

재처리 후 Queue 상태와 실제 재개 지점을 확인할 수 있는 값만 전달한다. Claim Token, 실패 원인 + * 상세와 내부 오류 정보는 포함하지 않는다. + */ +public record ManualRetriedIndexingJobResponse( + @Schema(description = "수동 재처리한 Embedding Job 식별자", example = "10") + Long jobId, + + @Schema(description = "재처리 후 Job 상태", example = "PENDING") + EmbeddingJobStatus status, + + @Schema(description = "재처리 대상 문서 식별자", example = "3") + Long documentId, + + @Schema(description = "재처리 대상 문서 버전 식별자", example = "5") + Long documentVersionId, + + @Schema( + description = "재처리가 다시 시작할 문서 버전 상태. Chunk가 이미 있으면 CHUNKED, 없으면 UPLOADED", + example = "CHUNKED" + ) + DocumentVersionStatus documentVersionStatus, + + @Schema(description = "보존된 기존 자동 재시도 횟수", example = "3") + int retryCount, + + @Schema(description = "Job에 설정된 최대 자동 재시도 횟수", example = "3") + int maxRetryCount, + + @Schema(description = "Queue 복귀 시각", example = "2026-08-06T15:00:00") + LocalDateTime requeuedAt +) { +} diff --git a/src/main/java/com/opensource/docgrid/domain/embedding/entity/EmbeddingJob.java b/src/main/java/com/opensource/docgrid/domain/embedding/entity/EmbeddingJob.java index 49ca5c8..f95022a 100644 --- a/src/main/java/com/opensource/docgrid/domain/embedding/entity/EmbeddingJob.java +++ b/src/main/java/com/opensource/docgrid/domain/embedding/entity/EmbeddingJob.java @@ -254,6 +254,34 @@ public void renewLease( this.lockExpiresAt = renewedLockExpiresAt; } + /** + * 최종 실패로 종결된 Job을 관리자 요청으로 다시 Queue에 복귀시킨다. + * + *

자동 재시도는 남은 횟수가 있는 PROCESSING Job만 되돌리므로 FAILED Job을 다시 처리할 수 있는 + * 경로는 이 메서드뿐이다. 호출 Service가 Job 행 잠금과 대상 Version·Document 조건을 먼저 + * 검증해야 하며, Entity는 종결되지 않은 Job이 수동 재처리로 소유권을 잃는 것을 마지막으로 방어한다. + * + *

retryCount와 마지막 오류 Snapshot은 그대로 두므로 수동 재처리는 추가 실행 1회만 부여하고, + * 다시 실패하면 자동 재시도 없이 최종 실패로 종결된다. + */ + public void requeueForManualRetry() { + // 1. 처리 중이거나 이미 Queue에 있는 Job의 소유권과 진행 상태를 덮어쓰지 않는다. + if (status != EmbeddingJobStatus.FAILED) { + throw new IllegalStateException("FAILED 상태의 Job만 수동으로 재처리할 수 있습니다."); + } + + // 2. 지연 없이 다음 Claim 후보가 되도록 Queue 상태로 되돌리고 종결 시각을 제거한다. + this.status = EmbeddingJobStatus.PENDING; + this.nextRetryAt = null; + this.failedAt = null; + + // 3. 실패 당시 감사용으로 남아 있던 소유권을 해제해 과거 Token이 다시 사용되지 않게 한다. + this.lockedByWorker = null; + this.claimToken = null; + this.lockedAt = null; + this.lockExpiresAt = null; + } + public void markFailed(String errorCode, String errorMessage, LocalDateTime failedAt) { // 현재 Claim을 보유한 처리 중 Job만 최종 실패로 종결할 수 있다. if (status != EmbeddingJobStatus.PROCESSING) { diff --git a/src/main/java/com/opensource/docgrid/domain/embedding/repository/EmbeddingRepository.java b/src/main/java/com/opensource/docgrid/domain/embedding/repository/EmbeddingRepository.java index b3257fd..dd3727f 100644 --- a/src/main/java/com/opensource/docgrid/domain/embedding/repository/EmbeddingRepository.java +++ b/src/main/java/com/opensource/docgrid/domain/embedding/repository/EmbeddingRepository.java @@ -14,6 +14,7 @@ *

생성 Transaction은 Version·Model 단위 저장 개수로 부분 저장을 구분한다. 완료 Transaction은 * Vector를 Java Heap으로 역직렬화하지 않고 DB 집계로 관계·차원·Hash 불변식을 검증하고, * 이전 현재 Version 또는 최종 실패한 Version의 ACTIVE Set을 STALE로 일괄 전환한다. + * 수동 재처리 Transaction은 다시 생성할 대상 Version의 Set만 삭제한다. */ public interface EmbeddingRepository extends JpaRepository { @@ -47,6 +48,21 @@ int markActiveAsStaleByDocumentVersionId( @Param("documentVersionId") Long documentVersionId ); + /** + * 수동 재처리로 다시 인덱싱할 Version의 Embedding 행을 한 SQL로 제거한다. + * + *

최종 실패 시 이미 STALE로 전환돼 검색에서 제외된 행만 대상이 되며, 남겨두면 재처리가 + * Version·Model 단위 Embedding 개수 불변식 검증에서 차단된다. Chunk Set은 삭제하지 않는다. + * + * @return 실제 삭제된 행 수 + */ + @Modifying(flushAutomatically = true) + @Query(value = """ + DELETE FROM embeddings + WHERE document_version_id = :documentVersionId + """, nativeQuery = true) + int deleteByDocumentVersionId(@Param("documentVersionId") Long documentVersionId); + /** * 대상 Chunk Set과 연결됐거나 대상 Version으로 역정규화된 Model 행 중 완료 불변식 위반 수를 계산한다. * diff --git a/src/main/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobManualRetryService.java b/src/main/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobManualRetryService.java new file mode 100644 index 0000000..ca53eea --- /dev/null +++ b/src/main/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobManualRetryService.java @@ -0,0 +1,230 @@ +package com.opensource.docgrid.domain.embedding.service.command; + +import java.time.Clock; +import java.time.LocalDateTime; +import java.time.temporal.ChronoUnit; +import java.util.EnumSet; +import java.util.Objects; +import java.util.Set; + +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.converter.EmbeddingJobConverter; +import com.opensource.docgrid.domain.embedding.dto.response.ManualRetriedIndexingJobResponse; +import com.opensource.docgrid.domain.embedding.entity.EmbeddingJob; +import com.opensource.docgrid.domain.embedding.enums.EmbeddingJobStatus; +import com.opensource.docgrid.domain.embedding.repository.EmbeddingJobRepository; +import com.opensource.docgrid.domain.embedding.repository.EmbeddingRepository; +import com.opensource.docgrid.domain.worker.entity.IndexingEvent; +import com.opensource.docgrid.domain.worker.enums.IndexingEventType; +import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository; +import com.opensource.docgrid.global.exception.DocGridException; +import com.opensource.docgrid.global.exception.ErrorCode; + +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + +/** + * 자동 재시도를 마치고 최종 실패로 종결된 Embedding Job을 관리자 요청으로 다시 Queue에 넣는 Command Service. + * + *

Claim은 PENDING만, Lease 만료 복구는 PROCESSING만 대상으로 하므로 FAILED Job을 다시 처리할 수 + * 있는 경로는 이 Service뿐이다. Job 행을 먼저 잠가 Claim·완료·실패·복구와의 경쟁을 직렬화하고, + * 기존 경로와 같은 Job → Version → Document 잠금 순서로 대상 조건을 검증한 뒤 Job, Version, Document + * 상태와 감사 Event를 하나의 Transaction에서 확정한다. 외부 I/O는 수행하지 않는다. + * + *

현재 검색 가능한 이전 Version은 그대로 보존하고, 이미 STALE로 전환된 대상 Version의 Embedding만 + * 삭제한다. Chunk Set과 Attempt 이력, retryCount는 삭제하거나 초기화하지 않는다. + */ +@Slf4j +@Service +@RequiredArgsConstructor +@Transactional +public class EmbeddingJobManualRetryService { + + private static final String MANUAL_RETRY_MESSAGE = + "관리자 요청으로 최종 실패한 Embedding Job을 다시 Queue에 넣었습니다."; + // 같은 Version에 이미 살아 있는 Job이 있으면 같은 대상이 두 번 처리되므로 재처리를 거부한다. + private static final Set LIVE_JOB_STATUSES = EnumSet.of( + EmbeddingJobStatus.PENDING, + EmbeddingJobStatus.PROCESSING + ); + // 재처리 후 완료 Transaction이 다시 확정할 수 있는 문서 상태만 허용한다. + private static final Set RETRYABLE_DOCUMENT_STATUSES = EnumSet.of( + DocumentStatus.UPLOADED, + DocumentStatus.INDEXING, + DocumentStatus.INDEXED, + DocumentStatus.FAILED + ); + + private final EmbeddingJobRepository embeddingJobRepository; + private final DocumentVersionRepository documentVersionRepository; + private final DocumentRepository documentRepository; + private final DocumentChunkRepository documentChunkRepository; + private final EmbeddingRepository embeddingRepository; + private final IndexingEventRepository indexingEventRepository; + private final EmbeddingJobConverter embeddingJobConverter; + private final Clock clock; + + /** + * 최종 실패한 Job 한 건을 즉시 Claim 가능한 PENDING 상태로 되돌린다. + * + * @param jobId 수동 재처리할 Embedding Job 식별자 + * @return 재처리 후 Queue 상태와 실제 재개 지점 + */ + public ManualRetriedIndexingJobResponse retry(Long jobId) { + // 1. Claim·완료·실패·복구 경쟁을 Job 행에서 직렬화하고 재처리 기준 시각을 고정한다. + EmbeddingJob embeddingJob = findLockedJob(jobId); + // PostgreSQL TIMESTAMP 정밀도와 맞춰 응답 시각과 Event 기록 시각을 동일하게 유지한다. + LocalDateTime requeuedAt = LocalDateTime.now(clock).truncatedTo(ChronoUnit.MICROS); + + // 2. 처리 중이거나 자동 재시도가 예정된 Job과 이미 재처리된 중복 요청을 함께 거부한다. + if (embeddingJob.getStatus() != EmbeddingJobStatus.FAILED) { + throw new DocGridException(ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED); + } + + // 3. 기존 완료·실패 경로와 같은 잠금 순서를 유지한 뒤 재처리 대상 조건을 검증한다. + DocumentVersion documentVersion = findLockedVersion(embeddingJob); + Document document = findLockedDocument(documentVersion); + validateRetryTarget(documentVersion, document); + + // 4. 저장된 Chunk Set 유무로 재개 지점을 정하고 검색에서 제외된 대상 Embedding만 제거한다. + DocumentVersionStatus resumeStatus = resolveResumeStatus(documentVersion); + int deletedEmbeddingCount = embeddingRepository + .deleteByDocumentVersionId(documentVersion.getId()); + + // 5. Job, Version, Document 상태와 감사 Event를 같은 Transaction과 시각으로 확정한다. + embeddingJob.requeueForManualRetry(); + documentVersion.reopenFailedForRetry(resumeStatus); + restoreDocumentStatus(document, documentVersion); + indexingEventRepository.save(IndexingEvent.builder() + .embeddingJob(embeddingJob) + .eventType(IndexingEventType.MANUAL_RETRY) + .fromStatus(EmbeddingJobStatus.FAILED.name()) + .toStatus(EmbeddingJobStatus.PENDING.name()) + .message(MANUAL_RETRY_MESSAGE) + .metadataJson(manualRetryMetadata(embeddingJob, resumeStatus, deletedEmbeddingCount)) + .occurredAt(requeuedAt) + .build()); + + log.info( + "인덱싱 Job 수동 재처리: jobId={}, documentId={}, versionId={}, resumeStatus={}, " + + "retryCount={}, deletedEmbeddingCount={}", + embeddingJob.getId(), + document.getId(), + documentVersion.getId(), + resumeStatus, + embeddingJob.getRetryCount(), + deletedEmbeddingCount + ); + return embeddingJobConverter.toManualRetriedResponse( + embeddingJob, + documentVersion, + requeuedAt + ); + } + + private EmbeddingJob findLockedJob(Long jobId) { + return embeddingJobRepository.findByIdForUpdate(jobId) + .orElseThrow(() -> new DocGridException(ErrorCode.EMBEDDING_JOB_NOT_FOUND)); + } + + private DocumentVersion findLockedVersion(EmbeddingJob embeddingJob) { + if (embeddingJob.getDocumentVersion() == null + || embeddingJob.getDocumentVersion().getId() == null) { + throw new DocGridException(ErrorCode.DOCUMENT_INDEXING_FAILURE_INCONSISTENT); + } + return documentVersionRepository.findByIdForUpdate(embeddingJob.getDocumentVersion().getId()) + .orElseThrow(() -> new DocGridException(ErrorCode.DOCUMENT_INDEXING_FAILURE_INCONSISTENT)); + } + + private Document findLockedDocument(DocumentVersion documentVersion) { + if (documentVersion.getDocument() == null + || documentVersion.getDocument().getId() == null) { + throw new DocGridException(ErrorCode.DOCUMENT_INDEXING_FAILURE_INCONSISTENT); + } + return documentRepository.findByIdForUpdate(documentVersion.getDocument().getId()) + .orElseThrow(() -> new DocGridException(ErrorCode.DOCUMENT_INDEXING_FAILURE_INCONSISTENT)); + } + + private void validateRetryTarget(DocumentVersion documentVersion, Document document) { + // 1. 최종 실패 Job의 Version은 반드시 FAILED여야 하며 그렇지 않으면 종료 데이터가 깨진 상태다. + if (documentVersion.getStatus() != DocumentVersionStatus.FAILED) { + throw new DocGridException(ErrorCode.DOCUMENT_INDEXING_FAILURE_INCONSISTENT); + } + + // 2. 삭제됐거나 다시 인덱싱을 확정할 수 없는 문서는 재처리 대상이 아니다. + if (document.getDeletedAt() != null + || !RETRYABLE_DOCUMENT_STATUSES.contains(document.getStatus())) { + throw new DocGridException(ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID); + } + + // 3. 현재 Version 포인터가 다른 문서를 가리키면 검색 보호 판단 자체를 신뢰할 수 없다. + DocumentVersion currentVersion = document.getCurrentVersion(); + if (currentVersion == null + || currentVersion.getId() == null + || currentVersion.getDocument() == null + || !Objects.equals(currentVersion.getDocument().getId(), document.getId())) { + throw new DocGridException(ErrorCode.DOCUMENT_INDEXING_FAILURE_INCONSISTENT); + } + + // 4. 더 새로운 Version이 올라온 뒤라면 과거 Version을 다시 인덱싱해도 완료할 수 없다. + DocumentVersion latestVersion = documentVersionRepository + .findTopByDocumentIdOrderByVersionNoDesc(document.getId()) + .orElseThrow(() -> new DocGridException(ErrorCode.DOCUMENT_INDEXING_FAILURE_INCONSISTENT)); + if (!Objects.equals(latestVersion.getId(), documentVersion.getId())) { + throw new DocGridException(ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID); + } + + // 5. 같은 Version을 처리 중이거나 대기 중인 Job이 있으면 중복 처리가 되므로 거부한다. + if (embeddingJobRepository.countByDocumentVersionIdAndStatusIn( + documentVersion.getId(), + LIVE_JOB_STATUSES + ) > 0) { + throw new DocGridException(ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID); + } + } + + /** + * 이미 저장된 Chunk Set이 있으면 파싱 단계를 생략하도록 재개 지점을 결정한다. + * + *

Chunk Set은 저장과 CHUNKED 전이가 같은 Transaction에서 일어나므로 존재하면 항상 완전하다. + */ + private DocumentVersionStatus resolveResumeStatus(DocumentVersion documentVersion) { + return documentChunkRepository.existsByDocumentVersionId(documentVersion.getId()) + ? DocumentVersionStatus.CHUNKED + : DocumentVersionStatus.UPLOADED; + } + + private void restoreDocumentStatus(Document document, DocumentVersion retriedVersion) { + // 이전 INDEXED Version이 현재 검색 대상이면 포인터와 문서 상태를 그대로 보존한다. + if (document.getStatus() == DocumentStatus.INDEXED + && !Objects.equals(document.getCurrentVersion().getId(), retriedVersion.getId())) { + return; + } + // 그 밖의 경우 검색 가능한 Version이 없으므로 완료 Transaction이 확정할 수 있는 처리 중 상태로 되돌린다. + document.markIndexing(); + } + + private String manualRetryMetadata( + EmbeddingJob embeddingJob, + DocumentVersionStatus resumeStatus, + int deletedEmbeddingCount + ) { + return """ + {"resumeVersionStatus":"%s","retryCount":%d,"maxRetryCount":%d,"deletedEmbeddingCount":%d} + """.formatted( + resumeStatus, + embeddingJob.getRetryCount(), + embeddingJob.getMaxRetryCount(), + deletedEmbeddingCount + ).strip(); + } +} diff --git a/src/main/java/com/opensource/docgrid/domain/worker/enums/IndexingEventType.java b/src/main/java/com/opensource/docgrid/domain/worker/enums/IndexingEventType.java index ace7dae..e05130e 100644 --- a/src/main/java/com/opensource/docgrid/domain/worker/enums/IndexingEventType.java +++ b/src/main/java/com/opensource/docgrid/domain/worker/enums/IndexingEventType.java @@ -14,5 +14,6 @@ public enum IndexingEventType { INDEXED, LEASE_EXPIRED, FAILED, - RETRY + RETRY, + MANUAL_RETRY } diff --git a/src/main/java/com/opensource/docgrid/global/exception/ErrorCode.java b/src/main/java/com/opensource/docgrid/global/exception/ErrorCode.java index f994315..6baa90c 100644 --- a/src/main/java/com/opensource/docgrid/global/exception/ErrorCode.java +++ b/src/main/java/com/opensource/docgrid/global/exception/ErrorCode.java @@ -210,6 +210,18 @@ public enum ErrorCode { "EMBEDDING-JOB-007", "동일한 Embedding Job Attempt에 다른 실패 내용이 이미 기록되었습니다." ), + // 처리 중이거나 자동 재시도가 예정된 Job과 중복 수동 재처리 요청을 함께 거부할 때 사용한다. + EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED( + HttpStatus.CONFLICT, + "EMBEDDING-JOB-008", + "최종 실패한 Embedding Job만 수동으로 재처리할 수 있습니다." + ), + // Job 상태는 최종 실패지만 대상 문서나 Version이 재처리 조건을 만족하지 않는 경우 사용한다. + EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID( + HttpStatus.CONFLICT, + "EMBEDDING-JOB-009", + "현재 문서 상태에서는 Embedding Job을 수동으로 재처리할 수 없습니다." + ), // EMBEDDING MODEL EMBEDDING_MODEL_NOT_CONFIGURED( diff --git a/src/test/java/com/opensource/docgrid/domain/embedding/controller/IndexingJobAdminControllerTest.java b/src/test/java/com/opensource/docgrid/domain/embedding/controller/IndexingJobAdminControllerTest.java index 3d0c137..17e23a1 100644 --- a/src/test/java/com/opensource/docgrid/domain/embedding/controller/IndexingJobAdminControllerTest.java +++ b/src/test/java/com/opensource/docgrid/domain/embedding/controller/IndexingJobAdminControllerTest.java @@ -36,6 +36,7 @@ import com.opensource.docgrid.domain.embedding.dto.response.DocumentEmbeddingsResponse; import com.opensource.docgrid.domain.embedding.dto.response.DocumentIndexingCompletionResponse; import com.opensource.docgrid.domain.embedding.dto.response.DocumentIndexingFailureResponse; +import com.opensource.docgrid.domain.embedding.dto.response.ManualRetriedIndexingJobResponse; import com.opensource.docgrid.domain.embedding.dto.response.RenewedEmbeddingJobLeaseResponse; import com.opensource.docgrid.domain.embedding.dto.response.StartedEmbeddingJobAttemptResponse; import com.opensource.docgrid.domain.embedding.enums.EmbeddingJobStatus; @@ -48,13 +49,15 @@ import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobAttemptService.StartResult; import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobClaimService; import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobLeaseService; +import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobManualRetryService; import com.opensource.docgrid.domain.worker.enums.AttemptStatus; import com.opensource.docgrid.global.config.SecurityConfig; import com.opensource.docgrid.global.exception.DocGridException; import com.opensource.docgrid.global.exception.ErrorCode; /** - * 관리자용 Job Claim, Attempt 시작과 Document Chunk·Embedding 생성·인덱싱 완료·실패 API 계약을 검증한다. + * 관리자용 Job Claim, Attempt 시작과 Document Chunk·Embedding 생성·인덱싱 완료·실패 및 최종 실패 Job + * 수동 재처리 API 계약을 검증한다. * *

각 API의 최초 생성·멱등 재생·Validation·비즈니스 오류 및 ADMIN Security 동작을 * 실제 Service 실행 없이 Controller 경계에서 확인한다. @@ -71,6 +74,7 @@ class IndexingJobAdminControllerTest { private static final String EMBEDDINGS_URL = "/admin/indexing-jobs/10/attempts/100/embeddings"; private static final String COMPLETE_URL = "/admin/indexing-jobs/10/attempts/100/complete"; private static final String FAIL_URL = "/admin/indexing-jobs/10/attempts/100/fail"; + private static final String RETRY_URL = "/admin/indexing-jobs/10/retry"; private static final Long JOB_ID = 10L; private static final Long ATTEMPT_ID = 100L; private static final Long WORKER_ID = 1L; @@ -105,6 +109,7 @@ class IndexingJobAdminControllerTest { @MockitoBean private DocumentEmbeddingService documentEmbeddingService; @MockitoBean private DocumentIndexingCompletionService documentIndexingCompletionService; @MockitoBean private DocumentIndexingFailureService documentIndexingFailureService; + @MockitoBean private EmbeddingJobManualRetryService embeddingJobManualRetryService; @MockitoBean private JpaMetamodelMappingContext jpaMetamodelMappingContext; @MockitoBean private JwtProvider jwtProvider; @MockitoBean private McpAccessTokenCommandService mcpAccessTokenCommandService; @@ -679,6 +684,77 @@ void failIndexing_returnsForbidden_withoutAdminRole() throws Exception { .andExpect(status().isForbidden()); } + @Test + @DisplayName("ADMIN 사용자의 수동 재처리 요청은 재개 지점을 담은 200을 반환한다") + void retryIndexingJob_returnsOkWithoutSensitiveFields() throws Exception { + given(embeddingJobManualRetryService.retry(JOB_ID)).willReturn(createManualRetryResponse()); + + mockMvc.perform(post(RETRY_URL) + .with(user("admin").roles("ADMIN"))) + .andExpect(status().isOk()) + .andExpect(jsonPath("$.success").value(true)) + .andExpect(jsonPath("$.data.jobId").value(JOB_ID)) + .andExpect(jsonPath("$.data.status").value("PENDING")) + .andExpect(jsonPath("$.data.documentId").value(3)) + .andExpect(jsonPath("$.data.documentVersionId").value(5)) + .andExpect(jsonPath("$.data.documentVersionStatus").value("CHUNKED")) + .andExpect(jsonPath("$.data.retryCount").value(3)) + .andExpect(jsonPath("$.data.maxRetryCount").value(3)) + .andExpect(jsonPath("$.data.requeuedAt").value("2026-08-06T15:00:00")) + .andExpect(jsonPath("$.data.claimToken").doesNotExist()) + .andExpect(jsonPath("$.data.errorMessage").doesNotExist()); + } + + @Test + @DisplayName("Job ID가 양수가 아닌 수동 재처리 요청은 400을 반환한다") + void retryIndexingJob_returnsBadRequest_when_jobIdIsNotPositive() throws Exception { + mockMvc.perform(post("/admin/indexing-jobs/0/retry") + .with(user("admin").roles("ADMIN"))) + .andExpect(status().isBadRequest()) + .andExpect(jsonPath("$.code").value("COMMON-002")); + } + + @ParameterizedTest + @MethodSource("manualRetryBusinessErrors") + @DisplayName("수동 재처리 비즈니스 오류를 정의된 HTTP 상태와 코드로 반환한다") + void retryIndexingJob_returnsDefinedError( + ErrorCode errorCode, + int expectedStatus, + String expectedCode + ) throws Exception { + given(embeddingJobManualRetryService.retry(JOB_ID)) + .willThrow(new DocGridException(errorCode)); + + mockMvc.perform(post(RETRY_URL) + .with(user("admin").roles("ADMIN"))) + .andExpect(status().is(expectedStatus)) + .andExpect(jsonPath("$.code").value(expectedCode)); + } + + @Test + @DisplayName("일반 사용자와 미인증 사용자는 수동 재처리를 요청할 수 없다") + void retryIndexingJob_returnsForbidden_withoutAdminRole() throws Exception { + mockMvc.perform(post(RETRY_URL) + .with(user("user").roles("USER"))) + .andExpect(status().isForbidden()); + + mockMvc.perform(post(RETRY_URL)) + .andExpect(status().isForbidden()); + } + + private static Stream manualRetryBusinessErrors() { + return Stream.of( + Arguments.of(ErrorCode.EMBEDDING_JOB_NOT_FOUND, 404, "EMBEDDING-JOB-001"), + Arguments.of(ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED, 409, "EMBEDDING-JOB-008"), + Arguments.of(ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID, 409, "EMBEDDING-JOB-009"), + Arguments.of( + ErrorCode.DOCUMENT_INDEXING_FAILURE_INCONSISTENT, + 500, + "DOCUMENT-INDEXING-004" + ) + ); + } + private static Stream invalidAttemptRequests() { return Stream.of( Arguments.of("Job ID가 양수가 아님", "/admin/indexing-jobs/0/attempts", VALID_ATTEMPT_BODY), @@ -983,6 +1059,19 @@ private DocumentIndexingCompletionResponse createCompletionResponse() { ); } + private ManualRetriedIndexingJobResponse createManualRetryResponse() { + return new ManualRetriedIndexingJobResponse( + JOB_ID, + EmbeddingJobStatus.PENDING, + 3L, + 5L, + DocumentVersionStatus.CHUNKED, + 3, + 3, + LocalDateTime.of(2026, 8, 6, 15, 0) + ); + } + private DocumentIndexingFailureResponse createFailureResponse() { return new DocumentIndexingFailureResponse( JOB_ID, diff --git a/src/test/java/com/opensource/docgrid/domain/embedding/integration/EmbeddingJobManualRetryIntegrationTest.java b/src/test/java/com/opensource/docgrid/domain/embedding/integration/EmbeddingJobManualRetryIntegrationTest.java new file mode 100644 index 0000000..f63e8e5 --- /dev/null +++ b/src/test/java/com/opensource/docgrid/domain/embedding/integration/EmbeddingJobManualRetryIntegrationTest.java @@ -0,0 +1,595 @@ +package com.opensource.docgrid.domain.embedding.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 java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import org.junit.jupiter.api.AfterAll; +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.test.annotation.DirtiesContext; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.test.context.DynamicPropertyRegistry; +import org.springframework.test.context.DynamicPropertySource; +import org.springframework.transaction.support.TransactionTemplate; + +import com.opensource.docgrid.domain.embedding.dto.response.ManualRetriedIndexingJobResponse; +import com.opensource.docgrid.domain.embedding.entity.EmbeddingJob; +import com.opensource.docgrid.domain.embedding.repository.EmbeddingJobRepository; +import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobManualRetryService; +import com.opensource.docgrid.domain.search.dto.VectorSearchCandidate; +import com.opensource.docgrid.domain.search.service.query.VectorSearchQueryService; +import com.opensource.docgrid.global.exception.DocGridException; +import com.opensource.docgrid.global.exception.ErrorCode; + +/** + * 실제 PostgreSQL에서 최종 실패 Job 수동 재처리의 상태 전이, 검색 보호와 동시 요청 수렴을 검증한다. + * + *

격리 Schema에 최종 실패로 종결된 실행 상태를 직접 구성한 뒤 실제 Service Transaction과 행 잠금을 + * 사용해, Queue 복귀가 부분 상태 없이 원자 커밋되고 현재 검색 가능한 이전 Version이 그대로 유지되는지 + * 확인한다. + */ +@Tag("integration") +@ActiveProfiles("test") +@SpringBootTest +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_CLASS) +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +@DisplayName("Embedding Job 수동 재처리 PostgreSQL 통합 테스트") +class EmbeddingJobManualRetryIntegrationTest { + + private static final String TEST_SCHEMA = "docgrid_manual_retry_integration_test"; + private static final int VECTOR_DIMENSION = 1024; + private static final int CONCURRENT_REQUESTS = 2; + private static final long TIMEOUT_SECONDS = 10; + private static final String CLAIM_TOKEN = "34c19d16-6ae1-4f6a-a35d-0123456789ab"; + private static final String CONTENT_HASH = + "26e4a23eec4241e034f1b4631f0222f1895847637c35e77687d5945f75edb42c"; + + @Autowired private JdbcTemplate jdbcTemplate; + @Autowired private TransactionTemplate transactionTemplate; + @Autowired private EmbeddingJobRepository embeddingJobRepository; + @Autowired private EmbeddingJobManualRetryService manualRetryService; + @Autowired private VectorSearchQueryService vectorSearchQueryService; + + @DynamicPropertySource + static void configureDatabase(DynamicPropertyRegistry registry) { + registry.add("TEST_DB_SCHEMA", () -> TEST_SCHEMA); + registry.add("jwt.secret", () -> "docgrid-manual-retry-integration-test-secret-key-2026"); + } + + @BeforeEach + void resetState() { + jdbcTemplate.execute(""" + TRUNCATE TABLE + search_results, + search_queries, + embeddings, + indexing_events, + document_chunks, + embedding_job_attempts, + embedding_jobs, + document_versions, + documents, + worker_nodes, + users + RESTART IDENTITY CASCADE + """); + } + + @AfterAll + void dropSchema() { + jdbcTemplate.execute("DROP SCHEMA IF EXISTS " + TEST_SCHEMA + " CASCADE"); + } + + @Test + @DisplayName("최종 실패 Job은 소유권을 비우고 즉시 Claim 가능한 PENDING으로 복귀한다") + void manualRetry_requeuesTerminalFailedJobWithoutOwnership() { + ExecutionContext context = insertTerminalFailedExecution(true); + + ManualRetriedIndexingJobResponse response = manualRetryService.retry(context.jobId()); + + assertThat(response.documentVersionStatus().name()).isEqualTo("CHUNKED"); + assertThat(queryString("SELECT status FROM embedding_jobs WHERE id = ?", context.jobId())) + .isEqualTo("PENDING"); + assertThat(queryLong( + "SELECT locked_by_worker_id FROM embedding_jobs WHERE id = ?", + context.jobId() + )).isNull(); + assertThat(queryString("SELECT claim_token FROM embedding_jobs WHERE id = ?", context.jobId())) + .isNull(); + assertThat(queryDateTime( + "SELECT lock_expires_at FROM embedding_jobs WHERE id = ?", + context.jobId() + )).isNull(); + assertThat(queryDateTime("SELECT failed_at FROM embedding_jobs WHERE id = ?", context.jobId())) + .isNull(); + assertThat(queryDateTime( + "SELECT next_retry_at FROM embedding_jobs WHERE id = ?", + context.jobId() + )).isNull(); + // 지연 없이 다음 Claim 후보가 되어야 Worker Polling이 곧바로 재처리를 시작할 수 있다. + assertThat(findPendingJobAt(LocalDateTime.now())).isEqualTo(context.jobId()); + } + + @Test + @DisplayName("이미 저장된 Chunk는 유지하고 STALE Embedding만 삭제해 임베딩 단계부터 재개한다") + void manualRetry_keepsChunksAndDeletesStaleEmbeddings() { + ExecutionContext context = insertTerminalFailedExecution(true); + + manualRetryService.retry(context.jobId()); + + assertThat(queryString( + "SELECT status FROM document_versions WHERE id = ?", + context.targetVersionId() + )).isEqualTo("CHUNKED"); + assertThat(countBy( + "SELECT COUNT(*) FROM document_chunks WHERE document_version_id = ?", + context.targetVersionId() + )).isOne(); + assertThat(countBy( + "SELECT COUNT(*) FROM embeddings WHERE document_version_id = ?", + context.targetVersionId() + )).isZero(); + assertThat(queryString("SELECT status FROM documents WHERE id = ?", context.documentId())) + .isEqualTo("INDEXING"); + } + + @Test + @DisplayName("Chunk가 없는 최종 실패 Job은 파싱부터 다시 시작하도록 UPLOADED로 재개한다") + void manualRetry_resumesFromUploaded_when_chunksDoNotExist() { + ExecutionContext context = insertTerminalFailedExecution(false); + + ManualRetriedIndexingJobResponse response = manualRetryService.retry(context.jobId()); + + assertThat(response.documentVersionStatus().name()).isEqualTo("UPLOADED"); + assertThat(queryString( + "SELECT status FROM document_versions WHERE id = ?", + context.targetVersionId() + )).isEqualTo("UPLOADED"); + } + + @Test + @DisplayName("재처리는 기존 Attempt 이력과 재시도 횟수를 그대로 두고 MANUAL_RETRY 이벤트만 추가한다") + void manualRetry_preservesAuditHistory() { + ExecutionContext context = insertTerminalFailedExecution(true); + + manualRetryService.retry(context.jobId()); + + assertThat(queryInteger( + "SELECT retry_count FROM embedding_jobs WHERE id = ?", + context.jobId() + )).isEqualTo(3); + assertThat(queryString( + "SELECT status FROM embedding_job_attempts WHERE id = ?", + context.attemptId() + )).isEqualTo("FAILED"); + assertThat(countBy( + "SELECT COUNT(*) FROM embedding_job_attempts WHERE embedding_job_id = ?", + context.jobId() + )).isOne(); + assertThat(eventCount(context.jobId(), "FAILED")).isOne(); + assertThat(eventCount(context.jobId(), "MANUAL_RETRY")).isOne(); + assertThat(jdbcTemplate.queryForObject(""" + SELECT metadata_json + FROM indexing_events + WHERE embedding_job_id = ? AND event_type = 'MANUAL_RETRY' + """, String.class, context.jobId())) + .contains("\"resumeVersionStatus\":\"CHUNKED\"", "\"retryCount\":3") + .doesNotContain(CLAIM_TOKEN); + } + + @Test + @DisplayName("이전 INDEXED Version이 검색 중이면 재처리 후에도 검색 결과와 현재 포인터를 보존한다") + void manualRetry_preservesPreviousSearchableVersion() { + ExecutionContext context = insertReplacementVersionFailedExecution(); + assertThat(search(context)) + .extracting(VectorSearchCandidate::chunkText) + .containsExactly("이전 검색 본문"); + + manualRetryService.retry(context.jobId()); + + assertThat(queryString("SELECT status FROM documents WHERE id = ?", context.documentId())) + .isEqualTo("INDEXED"); + assertThat(queryLong( + "SELECT current_version_id FROM documents WHERE id = ?", + context.documentId() + )).isEqualTo(context.previousVersionId()); + assertThat(queryString( + "SELECT status FROM embeddings WHERE document_version_id = ?", + context.previousVersionId() + )).isEqualTo("ACTIVE"); + assertThat(search(context)) + .extracting(VectorSearchCandidate::chunkText) + .containsExactly("이전 검색 본문"); + } + + @Test + @DisplayName("동시 수동 재처리 요청은 하나의 상태 전이로 수렴하고 나머지는 충돌로 거부된다") + void manualRetryConcurrently_commitsSingleTransition() throws Exception { + ExecutionContext context = insertTerminalFailedExecution(true); + CyclicBarrier startBarrier = new CyclicBarrier(CONCURRENT_REQUESTS); + ExecutorService executor = Executors.newFixedThreadPool(CONCURRENT_REQUESTS); + + List outcomes; + try { + List> futures = List.of( + executor.submit(() -> retryAfterBarrier(context, startBarrier)), + executor.submit(() -> retryAfterBarrier(context, startBarrier)) + ); + outcomes = List.of( + futures.get(0).get(TIMEOUT_SECONDS, TimeUnit.SECONDS), + futures.get(1).get(TIMEOUT_SECONDS, TimeUnit.SECONDS) + ); + } finally { + executor.shutdownNow(); + assertThat(executor.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS)).isTrue(); + } + + assertThat(outcomes).filteredOn(RetryOutcome::committed).hasSize(1); + assertThat(outcomes).filteredOn(outcome -> !outcome.committed()) + .extracting(RetryOutcome::errorCode) + .containsExactly(ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED.name()); + assertThat(queryString("SELECT status FROM embedding_jobs WHERE id = ?", context.jobId())) + .isEqualTo("PENDING"); + assertThat(eventCount(context.jobId(), "MANUAL_RETRY")).isOne(); + assertThat(countBy( + "SELECT COUNT(*) FROM embeddings WHERE document_version_id = ?", + context.targetVersionId() + )).isZero(); + } + + @Test + @DisplayName("자동 재시도가 예정된 PENDING Job은 수동 재처리를 거부하고 예약을 유지한다") + void manualRetry_rejectsRetryScheduledJob() { + ExecutionContext context = insertTerminalFailedExecution(true); + jdbcTemplate.update(""" + UPDATE embedding_jobs + SET status = 'PENDING', + failed_at = NULL, + next_retry_at = CURRENT_TIMESTAMP + INTERVAL '5 minutes' + WHERE id = ? + """, context.jobId()); + + assertThatThrownBy(() -> manualRetryService.retry(context.jobId())) + .isInstanceOf(DocGridException.class) + .hasFieldOrPropertyWithValue( + "errorCode", + ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED + ); + assertThat(queryDateTime( + "SELECT next_retry_at FROM embedding_jobs WHERE id = ?", + context.jobId() + )).isNotNull(); + assertThat(queryString( + "SELECT status FROM document_versions WHERE id = ?", + context.targetVersionId() + )).isEqualTo("FAILED"); + assertThat(eventCount(context.jobId(), "MANUAL_RETRY")).isZero(); + } + + @Test + @DisplayName("더 새로운 Version이 올라온 뒤에는 과거 Version 재처리를 거부한다") + void manualRetry_rejectsStaleVersionTarget() { + ExecutionContext context = insertTerminalFailedExecution(true); + insertVersion(context.documentId(), context.userId(), 2, "UPLOADED"); + + assertThatThrownBy(() -> manualRetryService.retry(context.jobId())) + .isInstanceOf(DocGridException.class) + .hasFieldOrPropertyWithValue( + "errorCode", + ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID + ); + assertThat(queryString("SELECT status FROM embedding_jobs WHERE id = ?", context.jobId())) + .isEqualTo("FAILED"); + assertThat(countBy( + "SELECT COUNT(*) FROM embeddings WHERE document_version_id = ?", + context.targetVersionId() + )).isOne(); + } + + private ExecutionContext insertTerminalFailedExecution(boolean withChunk) { + BaseContext base = insertBase("FAILED"); + Long versionId = insertVersion(base.documentId(), base.userId(), 1, "FAILED"); + setCurrentVersion(base.documentId(), versionId); + if (withChunk) { + insertChunkAndEmbedding( + base.documentId(), + versionId, + base.embeddingModelId(), + "실패 대상 본문", + 1.0f, + "STALE" + ); + } + JobContext job = insertFailedJob(versionId, base.embeddingModelId(), base.workerId()); + return new ExecutionContext( + base.userId(), + base.workerId(), + base.documentId(), + null, + versionId, + base.embeddingModelId(), + job.jobId(), + job.attemptId() + ); + } + + private ExecutionContext insertReplacementVersionFailedExecution() { + BaseContext base = insertBase("INDEXED"); + Long previousVersionId = insertVersion(base.documentId(), base.userId(), 1, "INDEXED"); + setCurrentVersion(base.documentId(), previousVersionId); + insertChunkAndEmbedding( + base.documentId(), + previousVersionId, + base.embeddingModelId(), + "이전 검색 본문", + 0.8f, + "ACTIVE" + ); + + Long targetVersionId = insertVersion(base.documentId(), base.userId(), 2, "FAILED"); + insertChunkAndEmbedding( + base.documentId(), + targetVersionId, + base.embeddingModelId(), + "실패 대상 본문", + 1.0f, + "STALE" + ); + JobContext job = insertFailedJob(targetVersionId, base.embeddingModelId(), base.workerId()); + return new ExecutionContext( + base.userId(), + base.workerId(), + base.documentId(), + previousVersionId, + targetVersionId, + base.embeddingModelId(), + job.jobId(), + job.attemptId() + ); + } + + private BaseContext insertBase(String documentStatus) { + String suffix = UUID.randomUUID().toString(); + Long userId = jdbcTemplate.queryForObject(""" + INSERT INTO users (email, password_hash, name, status, created_at, updated_at) + VALUES (?, 'password-hash', 'Manual Retry Test User', 'ACTIVE', + CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, "manual-retry-" + suffix + "@example.com"); + Long workerId = jdbcTemplate.queryForObject(""" + INSERT INTO worker_nodes ( + worker_name, instance_id, status, last_heartbeat_at, started_at, + created_at, updated_at + ) + VALUES ('manual-retry-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 (?, 'Manual Retry Test Document', 'TXT', 'UPLOAD', ?, 'PRIVATE', + CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, userId, documentStatus); + Long embeddingModelId = jdbcTemplate.queryForObject(""" + SELECT id + FROM embedding_models + WHERE is_active = TRUE AND is_searchable = TRUE + """, Long.class); + return new BaseContext(userId, workerId, documentId, embeddingModelId); + } + + private Long insertVersion(Long documentId, Long userId, int versionNo, String status) { + return jdbcTemplate.queryForObject(""" + INSERT INTO document_versions ( + document_id, version_no, title_snapshot, content_type, status, + indexed_at, created_by, created_at, updated_at + ) + VALUES (?, ?, 'Manual Retry Test Version', 'text/plain', ?, + CASE WHEN ? = 'INDEXED' THEN CURRENT_TIMESTAMP ELSE NULL END, + ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, documentId, versionNo, status, status, userId); + } + + private void setCurrentVersion(Long documentId, Long versionId) { + jdbcTemplate.update( + "UPDATE documents SET current_version_id = ? WHERE id = ?", + versionId, + documentId + ); + } + + private void insertChunkAndEmbedding( + Long documentId, + Long versionId, + Long embeddingModelId, + String chunkText, + float firstVectorValue, + String embeddingStatus + ) { + Long chunkId = jdbcTemplate.queryForObject(""" + 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, ?, 3, 0, ?, ?, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, versionId, chunkText, chunkText.length(), CONTENT_HASH); + jdbcTemplate.update(""" + INSERT INTO embeddings ( + chunk_id, document_id, document_version_id, embedding_model_id, + vector, dimension, vector_hash, status, created_at, updated_at + ) + VALUES (?, ?, ?, ?, CAST(? AS vector), ?, ?, ?, + CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + """, + chunkId, + documentId, + versionId, + embeddingModelId, + vector(firstVectorValue), + VECTOR_DIMENSION, + CONTENT_HASH, + embeddingStatus + ); + } + + /** + * 자동 재시도를 모두 소진하고 소유권 정보가 남아 있는 최종 실패 Job과 FAILED Attempt를 만든다. + */ + private JobContext insertFailedJob(Long versionId, Long embeddingModelId, Long workerId) { + 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, failed_at, error_code, error_message, + created_at, updated_at + ) + VALUES (?, ?, 'FAILED', 0, 3, 3, ?, CURRENT_TIMESTAMP - INTERVAL '10 seconds', + CURRENT_TIMESTAMP - INTERVAL '5 seconds', ?, + CURRENT_TIMESTAMP - INTERVAL '30 seconds', + CURRENT_TIMESTAMP - INTERVAL '5 seconds', + 'EMBEDDING_PROVIDER_UNAVAILABLE', 'Provider timeout', + CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, versionId, embeddingModelId, workerId, CLAIM_TOKEN); + Long attemptId = jdbcTemplate.queryForObject(""" + INSERT INTO embedding_job_attempts ( + embedding_job_id, worker_node_id, attempt_no, claim_token, status, + started_at, ended_at, duration_ms, error_code, error_message, + created_at, updated_at + ) + VALUES (?, ?, 1, ?, 'FAILED', CURRENT_TIMESTAMP - INTERVAL '30 seconds', + CURRENT_TIMESTAMP - INTERVAL '5 seconds', 25000, + 'EMBEDDING_PROVIDER_UNAVAILABLE', 'Provider timeout', + CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + RETURNING id + """, Long.class, jobId, workerId, CLAIM_TOKEN); + jdbcTemplate.update(""" + INSERT INTO indexing_events ( + embedding_job_id, event_type, from_status, to_status, message, + occurred_at, created_at, updated_at + ) + VALUES (?, 'FAILED', 'PROCESSING', 'FAILED', + 'Embedding Job을 최종 실패로 종료했습니다.', + CURRENT_TIMESTAMP - INTERVAL '5 seconds', + CURRENT_TIMESTAMP, CURRENT_TIMESTAMP) + """, jobId); + return new JobContext(jobId, attemptId); + } + + private RetryOutcome retryAfterBarrier( + ExecutionContext context, + CyclicBarrier startBarrier + ) throws Exception { + startBarrier.await(TIMEOUT_SECONDS, TimeUnit.SECONDS); + try { + manualRetryService.retry(context.jobId()); + return new RetryOutcome(true, null); + } catch (DocGridException exception) { + return new RetryOutcome(false, exception.getErrorCode().name()); + } + } + + private Long findPendingJobAt(LocalDateTime claimedAt) { + return transactionTemplate.execute(status -> embeddingJobRepository + .findNextPendingForUpdate(claimedAt) + .map(EmbeddingJob::getId) + .orElse(null)); + } + + private List search(ExecutionContext context) { + float[] queryVector = new float[VECTOR_DIMENSION]; + queryVector[0] = 1.0f; + return vectorSearchQueryService.search( + queryVector, + context.embeddingModelId(), + List.of(context.documentId()), + 5 + ); + } + + private String vector(float firstValue) { + return "[" + firstValue + "," + "0,".repeat(VECTOR_DIMENSION - 2) + "0]"; + } + + private String queryString(String sql, Long id) { + return jdbcTemplate.queryForObject(sql, String.class, id); + } + + private Long queryLong(String sql, Long id) { + return jdbcTemplate.queryForObject(sql, Long.class, id); + } + + private Integer queryInteger(String sql, Long id) { + return jdbcTemplate.queryForObject(sql, Integer.class, id); + } + + private LocalDateTime queryDateTime(String sql, Long id) { + return jdbcTemplate.queryForObject(sql, LocalDateTime.class, id); + } + + private Integer countBy(String sql, Long id) { + return jdbcTemplate.queryForObject(sql, Integer.class, id); + } + + private int eventCount(Long jobId, String eventType) { + return jdbcTemplate.queryForObject(""" + SELECT COUNT(*) + FROM indexing_events + WHERE embedding_job_id = ? AND event_type = ? + """, Integer.class, jobId, eventType); + } + + /** + * 한 테스트가 사용하는 사용자, Worker, 문서와 임베딩 모델 식별자 묶음. + */ + private record BaseContext(Long userId, Long workerId, Long documentId, Long embeddingModelId) { + } + + /** + * 최종 실패로 종결된 Job과 그 Attempt 식별자 묶음. + */ + private record JobContext(Long jobId, Long attemptId) { + } + + /** + * 수동 재처리 검증에 필요한 전체 실행 Context 식별자 묶음. + */ + private record ExecutionContext( + Long userId, + Long workerId, + Long documentId, + Long previousVersionId, + Long targetVersionId, + Long embeddingModelId, + Long jobId, + Long attemptId + ) { + } + + /** + * 동시 재처리 요청 한 건의 커밋 여부와 거부 사유 코드. + */ + private record RetryOutcome(boolean committed, String errorCode) { + } +} diff --git a/src/test/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobManualRetryServiceTest.java b/src/test/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobManualRetryServiceTest.java new file mode 100644 index 0000000..e87415d --- /dev/null +++ b/src/test/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobManualRetryServiceTest.java @@ -0,0 +1,437 @@ +package com.opensource.docgrid.domain.embedding.service.command; + +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.anyCollection; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.then; +import static org.mockito.Mockito.never; + +import java.time.Clock; +import java.time.LocalDateTime; +import java.time.ZoneId; +import java.util.Optional; + +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.ArgumentCaptor; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.test.util.ReflectionTestUtils; + +import com.opensource.docgrid.domain.document.entity.Document; +import com.opensource.docgrid.domain.document.entity.DocumentVersion; +import com.opensource.docgrid.domain.document.enums.DocumentSourceType; +import com.opensource.docgrid.domain.document.enums.DocumentStatus; +import com.opensource.docgrid.domain.document.enums.DocumentType; +import com.opensource.docgrid.domain.document.enums.DocumentVersionStatus; +import com.opensource.docgrid.domain.document.enums.VisibilityType; +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.converter.EmbeddingJobConverter; +import com.opensource.docgrid.domain.embedding.dto.response.ManualRetriedIndexingJobResponse; +import com.opensource.docgrid.domain.embedding.entity.EmbeddingJob; +import com.opensource.docgrid.domain.embedding.enums.EmbeddingJobStatus; +import com.opensource.docgrid.domain.embedding.repository.EmbeddingJobRepository; +import com.opensource.docgrid.domain.embedding.repository.EmbeddingRepository; +import com.opensource.docgrid.domain.worker.entity.IndexingEvent; +import com.opensource.docgrid.domain.worker.entity.WorkerNode; +import com.opensource.docgrid.domain.worker.enums.IndexingEventType; +import com.opensource.docgrid.domain.worker.enums.WorkerStatus; +import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository; +import com.opensource.docgrid.global.exception.DocGridException; +import com.opensource.docgrid.global.exception.ErrorCode; + +/** + * EmbeddingJobManualRetryService의 대상 검증, 재개 지점 결정과 소유권 초기화를 검증한다. + * + *

Repository는 이미 잠긴 Job, Version, Document를 반환한다고 가정하고, 최종 실패가 아닌 Job과 + * 재처리 조건을 만족하지 않는 문서가 어떤 상태 변경이나 Embedding 삭제도 실행하지 않는지 확인한다. + */ +@ExtendWith(MockitoExtension.class) +@DisplayName("EmbeddingJobManualRetryService 테스트") +class EmbeddingJobManualRetryServiceTest { + + private static final Long JOB_ID = 10L; + private static final Long DOCUMENT_ID = 3L; + private static final Long VERSION_ID = 5L; + private static final Long WORKER_ID = 1L; + private static final String CLAIM_TOKEN = "34c19d16-6ae1-4f6a-a35d-0123456789ab"; + private static final LocalDateTime REQUEUED_AT = LocalDateTime.of(2026, 8, 6, 15, 0); + private static final LocalDateTime FAILED_AT = REQUEUED_AT.minusHours(1); + + @Mock private EmbeddingJobRepository embeddingJobRepository; + @Mock private DocumentVersionRepository documentVersionRepository; + @Mock private DocumentRepository documentRepository; + @Mock private DocumentChunkRepository documentChunkRepository; + @Mock private EmbeddingRepository embeddingRepository; + @Mock private IndexingEventRepository indexingEventRepository; + + private EmbeddingJobManualRetryService manualRetryService; + + @BeforeEach + void setUp() { + Clock clock = Clock.fixed( + REQUEUED_AT.atZone(ZoneId.systemDefault()).toInstant(), + ZoneId.systemDefault() + ); + manualRetryService = new EmbeddingJobManualRetryService( + embeddingJobRepository, + documentVersionRepository, + documentRepository, + documentChunkRepository, + embeddingRepository, + indexingEventRepository, + new EmbeddingJobConverter(), + clock + ); + } + + @Test + @DisplayName("Chunk가 남아 있는 최종 실패 Job은 파싱을 생략하는 CHUNKED 지점으로 재개한다") + void retry_requeuesJobAndResumesFromChunked_when_chunksExist() { + Document document = createDocument(DocumentStatus.FAILED); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + givenLockedTarget(embeddingJob, documentVersion, document); + given(documentChunkRepository.existsByDocumentVersionId(VERSION_ID)).willReturn(true); + given(embeddingRepository.deleteByDocumentVersionId(VERSION_ID)).willReturn(4); + + ManualRetriedIndexingJobResponse response = manualRetryService.retry(JOB_ID); + + assertThat(response.jobId()).isEqualTo(JOB_ID); + assertThat(response.status()).isEqualTo(EmbeddingJobStatus.PENDING); + assertThat(response.documentId()).isEqualTo(DOCUMENT_ID); + assertThat(response.documentVersionId()).isEqualTo(VERSION_ID); + assertThat(response.documentVersionStatus()).isEqualTo(DocumentVersionStatus.CHUNKED); + assertThat(response.retryCount()).isEqualTo(3); + assertThat(response.maxRetryCount()).isEqualTo(3); + assertThat(response.requeuedAt()).isEqualTo(REQUEUED_AT); + assertThat(documentVersion.getStatus()).isEqualTo(DocumentVersionStatus.CHUNKED); + assertThat(document.getStatus()).isEqualTo(DocumentStatus.INDEXING); + } + + @Test + @DisplayName("Chunk가 없는 최종 실패 Job은 파싱부터 다시 시작하는 UPLOADED 지점으로 재개한다") + void retry_resumesFromUploaded_when_chunksDoNotExist() { + Document document = createDocument(DocumentStatus.FAILED); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + givenLockedTarget(embeddingJob, documentVersion, document); + given(documentChunkRepository.existsByDocumentVersionId(VERSION_ID)).willReturn(false); + given(embeddingRepository.deleteByDocumentVersionId(VERSION_ID)).willReturn(0); + + ManualRetriedIndexingJobResponse response = manualRetryService.retry(JOB_ID); + + assertThat(response.documentVersionStatus()).isEqualTo(DocumentVersionStatus.UPLOADED); + assertThat(documentVersion.getStatus()).isEqualTo(DocumentVersionStatus.UPLOADED); + } + + @Test + @DisplayName("재처리한 Job은 즉시 Claim 가능하도록 소유권과 종료 시각을 모두 초기화한다") + void retry_clearsOwnershipAndKeepsRetryHistory() { + Document document = createDocument(DocumentStatus.FAILED); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + givenLockedTarget(embeddingJob, documentVersion, document); + given(documentChunkRepository.existsByDocumentVersionId(VERSION_ID)).willReturn(true); + + manualRetryService.retry(JOB_ID); + + assertThat(embeddingJob.getStatus()).isEqualTo(EmbeddingJobStatus.PENDING); + assertThat(embeddingJob.getLockedByWorker()).isNull(); + assertThat(embeddingJob.getClaimToken()).isNull(); + assertThat(embeddingJob.getLockedAt()).isNull(); + assertThat(embeddingJob.getLockExpiresAt()).isNull(); + assertThat(embeddingJob.getFailedAt()).isNull(); + assertThat(embeddingJob.getNextRetryAt()).isNull(); + // Retry 이력은 감사 대상이므로 삭제하지 않는다. 자동 재시도 여유를 남기지 않으므로 이번 + // 재처리 실행이 다시 실패하면 재예약 없이 곧바로 최종 실패로 종결된다. + assertThat(embeddingJob.getRetryCount()).isEqualTo(3); + assertThat(embeddingJob.hasRemainingRetries()).isFalse(); + } + + @Test + @DisplayName("수동 재처리는 Claim Token 없는 감사 Event를 같은 시각으로 기록한다") + void retry_savesManualRetryEventWithoutClaimToken() { + Document document = createDocument(DocumentStatus.FAILED); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + givenLockedTarget(embeddingJob, documentVersion, document); + given(documentChunkRepository.existsByDocumentVersionId(VERSION_ID)).willReturn(true); + given(embeddingRepository.deleteByDocumentVersionId(VERSION_ID)).willReturn(4); + + manualRetryService.retry(JOB_ID); + + ArgumentCaptor eventCaptor = ArgumentCaptor.forClass(IndexingEvent.class); + then(indexingEventRepository).should().save(eventCaptor.capture()); + IndexingEvent event = eventCaptor.getValue(); + assertThat(event.getEventType()).isEqualTo(IndexingEventType.MANUAL_RETRY); + assertThat(event.getFromStatus()).isEqualTo(EmbeddingJobStatus.FAILED.name()); + assertThat(event.getToStatus()).isEqualTo(EmbeddingJobStatus.PENDING.name()); + assertThat(event.getOccurredAt()).isEqualTo(REQUEUED_AT); + assertThat(event.getMetadataJson()) + .contains("\"resumeVersionStatus\":\"CHUNKED\"") + .contains("\"retryCount\":3", "\"maxRetryCount\":3", "\"deletedEmbeddingCount\":4") + .doesNotContain(CLAIM_TOKEN); + } + + @Test + @DisplayName("현재 검색 대상인 이전 INDEXED Version이 있으면 문서 상태와 포인터를 보존한다") + void retry_keepsIndexedDocument_when_previousSearchableVersionExists() { + Document document = createDocument(DocumentStatus.INDEXED); + DocumentVersion previousVersion = DocumentVersion.builder() + .document(document) + .versionNo(1) + .status(DocumentVersionStatus.INDEXED) + .build(); + ReflectionTestUtils.setField(previousVersion, "id", 4L); + document.updateCurrentVersion(previousVersion); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + givenLockedTarget(embeddingJob, documentVersion, document); + given(documentChunkRepository.existsByDocumentVersionId(VERSION_ID)).willReturn(true); + + manualRetryService.retry(JOB_ID); + + assertThat(document.getStatus()).isEqualTo(DocumentStatus.INDEXED); + assertThat(document.getCurrentVersion().getId()).isEqualTo(4L); + assertThat(embeddingJob.getStatus()).isEqualTo(EmbeddingJobStatus.PENDING); + } + + @Test + @DisplayName("존재하지 않는 Job은 Not Found 예외가 발생한다") + void retry_throws_when_jobDoesNotExist() { + given(embeddingJobRepository.findByIdForUpdate(JOB_ID)).willReturn(Optional.empty()); + + assertThatThrownBy(() -> manualRetryService.retry(JOB_ID)) + .isInstanceOf(DocGridException.class) + .hasFieldOrPropertyWithValue("errorCode", ErrorCode.EMBEDDING_JOB_NOT_FOUND); + then(embeddingRepository).shouldHaveNoInteractions(); + then(indexingEventRepository).shouldHaveNoInteractions(); + } + + @Test + @DisplayName("자동 재시도가 예정된 PENDING Job은 수동 재처리를 거부한다") + void retry_throws_when_jobIsPendingRetry() { + Document document = createDocument(DocumentStatus.INDEXING); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + ReflectionTestUtils.setField(embeddingJob, "status", EmbeddingJobStatus.PENDING); + given(embeddingJobRepository.findByIdForUpdate(JOB_ID)).willReturn(Optional.of(embeddingJob)); + + assertThatThrownBy(() -> manualRetryService.retry(JOB_ID)) + .isInstanceOf(DocGridException.class) + .hasFieldOrPropertyWithValue( + "errorCode", + ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED + ); + then(documentVersionRepository).shouldHaveNoInteractions(); + then(embeddingRepository).shouldHaveNoInteractions(); + then(indexingEventRepository).shouldHaveNoInteractions(); + } + + @Test + @DisplayName("처리 중인 PROCESSING Job은 수동 재처리를 거부한다") + void retry_throws_when_jobIsProcessing() { + Document document = createDocument(DocumentStatus.INDEXING); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + ReflectionTestUtils.setField(embeddingJob, "status", EmbeddingJobStatus.PROCESSING); + given(embeddingJobRepository.findByIdForUpdate(JOB_ID)).willReturn(Optional.of(embeddingJob)); + + assertThatThrownBy(() -> manualRetryService.retry(JOB_ID)) + .isInstanceOf(DocGridException.class) + .hasFieldOrPropertyWithValue( + "errorCode", + ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED + ); + then(embeddingRepository).shouldHaveNoInteractions(); + } + + @Test + @DisplayName("이미 완료된 INDEXED Job은 수동 재처리를 거부한다") + void retry_throws_when_jobIsIndexed() { + Document document = createDocument(DocumentStatus.INDEXED); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + ReflectionTestUtils.setField(embeddingJob, "status", EmbeddingJobStatus.INDEXED); + given(embeddingJobRepository.findByIdForUpdate(JOB_ID)).willReturn(Optional.of(embeddingJob)); + + assertThatThrownBy(() -> manualRetryService.retry(JOB_ID)) + .isInstanceOf(DocGridException.class) + .hasFieldOrPropertyWithValue( + "errorCode", + ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_NOT_ALLOWED + ); + then(embeddingRepository).shouldHaveNoInteractions(); + } + + @Test + @DisplayName("더 새로운 Version이 올라온 문서는 과거 Version 재처리를 거부한다") + void retry_throws_when_targetIsNotLatestVersion() { + Document document = createDocument(DocumentStatus.FAILED); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + DocumentVersion newerVersion = DocumentVersion.builder() + .document(document) + .versionNo(3) + .status(DocumentVersionStatus.UPLOADED) + .build(); + ReflectionTestUtils.setField(newerVersion, "id", 6L); + given(embeddingJobRepository.findByIdForUpdate(JOB_ID)).willReturn(Optional.of(embeddingJob)); + given(documentVersionRepository.findByIdForUpdate(VERSION_ID)) + .willReturn(Optional.of(documentVersion)); + given(documentRepository.findByIdForUpdate(DOCUMENT_ID)).willReturn(Optional.of(document)); + given(documentVersionRepository.findTopByDocumentIdOrderByVersionNoDesc(DOCUMENT_ID)) + .willReturn(Optional.of(newerVersion)); + + assertThatThrownBy(() -> manualRetryService.retry(JOB_ID)) + .isInstanceOf(DocGridException.class) + .hasFieldOrPropertyWithValue( + "errorCode", + ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID + ); + assertThat(embeddingJob.getStatus()).isEqualTo(EmbeddingJobStatus.FAILED); + then(embeddingRepository).should(never()).deleteByDocumentVersionId(anyLong()); + then(indexingEventRepository).should(never()).save(any()); + } + + @Test + @DisplayName("삭제된 문서의 Job은 수동 재처리를 거부한다") + void retry_throws_when_documentIsDeleted() { + Document document = createDocument(DocumentStatus.FAILED); + ReflectionTestUtils.setField(document, "deletedAt", FAILED_AT); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + given(embeddingJobRepository.findByIdForUpdate(JOB_ID)).willReturn(Optional.of(embeddingJob)); + given(documentVersionRepository.findByIdForUpdate(VERSION_ID)) + .willReturn(Optional.of(documentVersion)); + given(documentRepository.findByIdForUpdate(DOCUMENT_ID)).willReturn(Optional.of(document)); + + assertThatThrownBy(() -> manualRetryService.retry(JOB_ID)) + .isInstanceOf(DocGridException.class) + .hasFieldOrPropertyWithValue( + "errorCode", + ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID + ); + then(embeddingRepository).should(never()).deleteByDocumentVersionId(anyLong()); + } + + @Test + @DisplayName("같은 Version에 살아 있는 Job이 있으면 중복 처리를 막기 위해 거부한다") + void retry_throws_when_liveJobExistsForSameVersion() { + Document document = createDocument(DocumentStatus.FAILED); + DocumentVersion documentVersion = createFailedVersion(document); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + given(embeddingJobRepository.findByIdForUpdate(JOB_ID)).willReturn(Optional.of(embeddingJob)); + given(documentVersionRepository.findByIdForUpdate(VERSION_ID)) + .willReturn(Optional.of(documentVersion)); + given(documentRepository.findByIdForUpdate(DOCUMENT_ID)).willReturn(Optional.of(document)); + given(documentVersionRepository.findTopByDocumentIdOrderByVersionNoDesc(DOCUMENT_ID)) + .willReturn(Optional.of(documentVersion)); + given(embeddingJobRepository.countByDocumentVersionIdAndStatusIn( + eq(VERSION_ID), + anyCollection() + )).willReturn(1L); + + assertThatThrownBy(() -> manualRetryService.retry(JOB_ID)) + .isInstanceOf(DocGridException.class) + .hasFieldOrPropertyWithValue( + "errorCode", + ErrorCode.EMBEDDING_JOB_MANUAL_RETRY_TARGET_INVALID + ); + then(embeddingRepository).should(never()).deleteByDocumentVersionId(anyLong()); + then(indexingEventRepository).should(never()).save(any()); + } + + @Test + @DisplayName("최종 실패 Job의 Version이 FAILED가 아니면 종료 데이터 불변식 오류로 중단한다") + void retry_throws_when_versionIsNotFailed() { + Document document = createDocument(DocumentStatus.FAILED); + DocumentVersion documentVersion = createFailedVersion(document); + ReflectionTestUtils.setField(documentVersion, "status", DocumentVersionStatus.EMBEDDING); + EmbeddingJob embeddingJob = createFailedJob(documentVersion); + given(embeddingJobRepository.findByIdForUpdate(JOB_ID)).willReturn(Optional.of(embeddingJob)); + given(documentVersionRepository.findByIdForUpdate(VERSION_ID)) + .willReturn(Optional.of(documentVersion)); + given(documentRepository.findByIdForUpdate(DOCUMENT_ID)).willReturn(Optional.of(document)); + + assertThatThrownBy(() -> manualRetryService.retry(JOB_ID)) + .isInstanceOf(DocGridException.class) + .hasFieldOrPropertyWithValue( + "errorCode", + ErrorCode.DOCUMENT_INDEXING_FAILURE_INCONSISTENT + ); + then(embeddingRepository).should(never()).deleteByDocumentVersionId(anyLong()); + } + + private void givenLockedTarget( + EmbeddingJob embeddingJob, + DocumentVersion documentVersion, + Document document + ) { + given(embeddingJobRepository.findByIdForUpdate(JOB_ID)).willReturn(Optional.of(embeddingJob)); + given(documentVersionRepository.findByIdForUpdate(VERSION_ID)) + .willReturn(Optional.of(documentVersion)); + given(documentRepository.findByIdForUpdate(DOCUMENT_ID)).willReturn(Optional.of(document)); + given(documentVersionRepository.findTopByDocumentIdOrderByVersionNoDesc(DOCUMENT_ID)) + .willReturn(Optional.of(documentVersion)); + } + + private Document createDocument(DocumentStatus status) { + Document document = Document.builder() + .title("수동 재처리 테스트 문서") + .documentType(DocumentType.PDF) + .sourceType(DocumentSourceType.UPLOAD) + .status(status) + .visibility(VisibilityType.PRIVATE) + .build(); + ReflectionTestUtils.setField(document, "id", DOCUMENT_ID); + return document; + } + + private DocumentVersion createFailedVersion(Document document) { + DocumentVersion documentVersion = DocumentVersion.builder() + .document(document) + .versionNo(2) + .status(DocumentVersionStatus.FAILED) + .build(); + ReflectionTestUtils.setField(documentVersion, "id", VERSION_ID); + if (document.getCurrentVersion() == null) { + document.updateCurrentVersion(documentVersion); + } + return documentVersion; + } + + /** + * 자동 재시도를 모두 소진하고 소유권 정보가 남아 있는 최종 실패 Job을 만든다. + */ + private EmbeddingJob createFailedJob(DocumentVersion documentVersion) { + WorkerNode workerNode = WorkerNode.builder() + .workerName("manual-retry-worker") + .instanceId("manual-retry-worker-instance") + .status(WorkerStatus.ACTIVE) + .lastHeartbeatAt(FAILED_AT.minusMinutes(1)) + .startedAt(FAILED_AT.minusMinutes(10)) + .build(); + ReflectionTestUtils.setField(workerNode, "id", WORKER_ID); + EmbeddingJob embeddingJob = EmbeddingJob.builder() + .documentVersion(documentVersion) + .status(EmbeddingJobStatus.PENDING) + .priority(0) + .maxRetryCount(3) + .build(); + ReflectionTestUtils.setField(embeddingJob, "id", JOB_ID); + ReflectionTestUtils.setField(embeddingJob, "retryCount", 3); + embeddingJob.claim(workerNode, CLAIM_TOKEN, FAILED_AT.minusMinutes(5), FAILED_AT); + embeddingJob.markFailed("EMBEDDING_SERVER_UNAVAILABLE", "임베딩 서버 장애", FAILED_AT); + return embeddingJob; + } +}