[Feat] 인덱싱 Job Lease 갱신 및 만료 복구

[Feat] 인덱싱 Job Lease 갱신 및 만료 복구

Author: Gimini-3

docs/design/Gimini-3-#90-indexing-job-lease-recovery.md

@@ -0,0 +1,98 @@
+# #90 인덱싱 Job Lease 갱신 및 만료 복구 검증 결과
+
+## 1. 검증 정보
+
+- 실행일: 2026-08-03 (Asia/Seoul)
+- 대상 브랜치: `feature/90`
+- 애플리케이션: Spring Boot 3.5.16, Java 17
+- 데이터베이스: 격리된 PostgreSQL 14(OpenSQL 호환) + pgvector
+- 검증 범위: 전체 기본 Build와 Lease 복구 전용 동시성 Test
+- 최종 결과: 기본 Test 547개와 동시성 Test 4개 통과, 실패·오류·Skip 0개
+
+기존 개발 데이터와 영구 Volume은 사용하거나 변경하지 않았다. 검증에는 localhost에만 노출한 일회용
+PostgreSQL 컨테이너를 사용했고, Flyway Migration과 Test Class별 격리 Schema를 적용했다. 인증 관련
+환경 값은 테스트 전용 설정을 사용했으며 실제 값은 기록하지 않는다.
+
+## 2. 전체 회귀 검증
+
+실행 명령의 환경 값은 Placeholder로 대체한다.
+
+```bash
+DB_HOST=localhost \
+DB_PORT='<isolated-test-port>' \
+DB_NAME=docgrid \
+DB_USER='<local-test-user>' \
+DB_PASSWORD='<local-test-password>' \
+DB_SSLMODE=disable \
+JWT_SECRET='<test-only-secret>' \
+./gradlew clean build
+```
+
+결과:
+
+```text
+BUILD SUCCESSFUL
+tests=547 failures=0 errors=0 skipped=0
+```
+
+기본 `test` Task에서 다음 영역을 함께 회귀 검증했다.
+
+- Lease 갱신 API의 정상·Token 불일치·만료·상태 불일치 응답
+- `EmbeddingJob`의 Lease 갱신과 Retry·최종 실패 전이
+- DEAD Worker 확정과 오래 만료된 Job 후보 조회
+- 복구 Service의 소유권 재확인, Attempt 처리와 실패 이벤트 기록
+- Scheduler의 단계별 실패 격리와 후보별 계속 진행
+
+## 3. Lease 복구 동시성 검증
+
+실행:
+
+```bash
+DB_HOST=localhost \
+DB_PORT='<isolated-test-port>' \
+DB_NAME=docgrid \
+DB_USER='<local-test-user>' \
+DB_PASSWORD='<local-test-password>' \
+DB_SSLMODE=disable \
+./gradlew claimConcurrencyTest \
+  --tests 'com.opensource.docgrid.domain.embedding.integration.EmbeddingJobLeaseRecoveryIntegrationTest'
+```
+
+결과:
+
+```text
+BUILD SUCCESSFUL
+tests=4 failures=0 errors=0 skipped=0
+```
+
+| 검증 항목 | 결과 |
+| --- | --- |
+| 만료 후보를 오래된 Lease 순서로 제한 조회 | 통과 |
+| 두 복구 실행이 같은 Job에 경쟁할 때 단일 상태 전이 | 통과 |
+| Attempt가 없는 Job 복구 시 새 Attempt를 합성하지 않음 | 통과 |
+| Retry 소진 Job을 최종 실패 상태로 전이 | 통과 |
+
+PostgreSQL `timestamp` 정밀도로 나노초 경계가 절삭되는 문제를 확인해, 경계 Fixture를 1초 뒤로
+조정했다. 이후 같은 조건에서 전용 동시성 Test 4개가 모두 통과했다.
+
+## 4. 확인된 불변식
+
+- Lease 갱신은 `PROCESSING` 상태, 일치하는 Worker·Claim Token, 만료 전 Lease에서만 성공한다.
+- 복구 후보 Snapshot 조회와 실제 복구 Transaction을 분리한다.
+- 실제 복구는 Job 행 잠금 뒤 Lease 만료와 상태를 다시 확인한다.
+- 같은 만료 Job에 여러 복구 실행이 경쟁해도 하나만 Retry 또는 최종 실패를 적용한다.
+- Retry 복구는 Worker·Claim Token·Lease를 제거하고 다음 실행을 `PENDING`으로 돌린다.
+- Retry를 모두 소진하면 Job과 관련 상태를 최종 실패 정책으로 전이한다.
+- 기존 Attempt가 없는 Job에는 복구 과정에서 임의의 Attempt를 만들지 않는다.
+- 한 후보의 복구 실패가 같은 Batch의 나머지 후보 처리를 중단하지 않는다.
+
+## 5. 환경 진단 기록과 제한 사항
+
+- 기존 Compose 데이터 Volume은 컨테이너 Image의 기대 Role과 달라 초기 접속에 실패했다. 기존
+  Volume을 수정하지 않고 별도 일회용 데이터베이스로 전환했다.
+- Image 기본 Bootstrap 경로에서는 SSL·Host 인증 설정이 Test 접속 조건과 맞지 않았다. localhost
+  전용 Test Instance로 재구성한 뒤 Flyway와 Test를 실행했다.
+- 최초 전체 Test의 Application Context 실패 11건은 테스트용 JWT 설정 누락 때문이었다. 테스트 전용
+  설정을 주입한 재실행에서는 547개가 모두 통과했다.
+- 프로젝트 기본 Build가 제외하는 외부 MinIO 의존 태그와 Benchmark는 이번 검증 범위가 아니다.
+- 운영 부하에서의 복구 처리량, Scheduler 주기와 Batch 크기 적정성은 측정하지 않았다.

docs/test-results/Gimini-3-#90-indexing-job-lease-recovery.md

@@ -13,15 +13,17 @@
 import org.springframework.web.bind.annotation.RestController;
 
 import com.opensource.docgrid.domain.embedding.dto.request.CompleteDocumentIndexingRequest;
-import com.opensource.docgrid.domain.embedding.dto.request.FailDocumentIndexingRequest;
 import com.opensource.docgrid.domain.embedding.dto.request.CreateDocumentChunksRequest;
 import com.opensource.docgrid.domain.embedding.dto.request.CreateDocumentEmbeddingsRequest;
+import com.opensource.docgrid.domain.embedding.dto.request.FailDocumentIndexingRequest;
+import com.opensource.docgrid.domain.embedding.dto.request.RenewEmbeddingJobLeaseRequest;
 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.DocumentChunksResponse;
 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.RenewedEmbeddingJobLeaseResponse;
 import com.opensource.docgrid.domain.embedding.dto.response.StartedEmbeddingJobAttemptResponse;
 import com.opensource.docgrid.domain.document.service.DocumentParsingService;
 import com.opensource.docgrid.domain.document.service.command.DocumentChunkTransactionService.ChunkResult;
@@ -32,6 +34,7 @@
 import com.opensource.docgrid.domain.embedding.service.command.EmbeddingJobAttemptService;
 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.global.common.response.ApiResponse;
 import com.opensource.docgrid.global.common.response.ErrorResponse;
 import com.opensource.docgrid.global.common.response.ResponseUtils;
@@ -46,7 +49,8 @@
 import lombok.RequiredArgsConstructor;
 
 /**
- * 관리자용 Embedding Job Claim, Attempt 시작과 문서 Chunk·Embedding·인덱싱 완료·실패 실행을 HTTP API로 제공한다.
+ * 관리자용 Embedding Job Claim·Lease 갱신, Attempt 시작과 문서 Chunk·Embedding·인덱싱 완료·실패
+ * 실행을 HTTP API로 제공한다.
  *
  * <p>HTTP 입력 검증과 성공 상태 변환만 담당한다. Job Claim 및 현재 소유권 기반 파이프라인 단계의
  * Transaction·외부 호출·동시성 규칙은 각 Service에 위임한다.
@@ -59,6 +63,7 @@
 public class IndexingJobAdminController {
 
     private final EmbeddingJobClaimService embeddingJobClaimService;
+    private final EmbeddingJobLeaseService embeddingJobLeaseService;
     private final EmbeddingJobAttemptService embeddingJobAttemptService;
     private final DocumentParsingService documentParsingService;
     private final DocumentEmbeddingService documentEmbeddingService;
@@ -111,6 +116,56 @@ public ResponseEntity<ApiResponse<ClaimedEmbeddingJobResponse>> claim(
         return ResponseUtils.ok(claimedJob.get());
     }
 
+    @Operation(
+        summary = "PROCESSING Job Lease 갱신",
+        description = "현재 Job의 Worker ID와 Claim Token 및 만료 전 Lease를 검증하고, "
+            + "Heartbeat가 유효한 ACTIVE 또는 IDLE Worker의 Lease 만료 시각만 연장합니다. "
+            + "Claim Token은 응답에 포함하지 않습니다."
+    )
+    @ApiResponses({
+        @io.swagger.v3.oas.annotations.responses.ApiResponse(
+            responseCode = "200",
+            description = "Job Lease 갱신 성공"
+        ),
+        @io.swagger.v3.oas.annotations.responses.ApiResponse(
+            responseCode = "400",
+            description = "Job ID, Worker ID 또는 Claim Token 형식 오류",
+            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 또는 Worker 없음",
+            content = @Content(schema = @Schema(implementation = ErrorResponse.class))
+        ),
+        @io.swagger.v3.oas.annotations.responses.ApiResponse(
+            responseCode = "409",
+            description = "Job 상태, 현재 소유권, Lease 또는 Worker 생존 상태 오류",
+            content = @Content(schema = @Schema(implementation = ErrorResponse.class))
+        ),
+        @io.swagger.v3.oas.annotations.responses.ApiResponse(
+            responseCode = "500",
+            description = "PROCESSING Job의 소유권 데이터 불일치",
+            content = @Content(schema = @Schema(implementation = ErrorResponse.class))
+        )
+    })
+    @PostMapping(
+        value = "/{jobId}/lease/renew",
+        consumes = MediaType.APPLICATION_JSON_VALUE,
+        produces = MediaType.APPLICATION_JSON_VALUE
+    )
+    public ResponseEntity<ApiResponse<RenewedEmbeddingJobLeaseResponse>> renewLease(
+        @PathVariable @Positive Long jobId,
+        @Valid @RequestBody RenewEmbeddingJobLeaseRequest request
+    ) {
+        // Service가 Job → Worker 잠금과 소유권·생존 검증 및 Lease 갱신을 한 Transaction으로 처리한다.
+        return ResponseUtils.ok(embeddingJobLeaseService.renew(jobId, request));
+    }
+
     @Operation(
         summary = "Embedding Job Attempt 시작",
         description = "현재 PROCESSING Job의 Worker ID와 Claim Token 및 유효한 Lease를 검증한 뒤 "

src/main/java/com/opensource/docgrid/domain/embedding/controller/IndexingJobAdminController.java

@@ -1,15 +1,18 @@
 package com.opensource.docgrid.domain.embedding.converter;
 
+import java.time.LocalDateTime;
+
 import org.springframework.stereotype.Component;
 
 import com.opensource.docgrid.domain.embedding.dto.response.ClaimedEmbeddingJobResponse;
+import com.opensource.docgrid.domain.embedding.dto.response.RenewedEmbeddingJobLeaseResponse;
 import com.opensource.docgrid.domain.embedding.entity.EmbeddingJob;
 
 /**
  * Embedding Job Entity를 API 전용 응답 DTO로 변환하는 Converter.
  *
- * <p>Controller에 Entity와 연관 Entity를 직접 노출하지 않고 Claim 이후 Worker가 필요한 식별자와
- * Lease 정보만 전달한다.
+ * <p>Controller에 Entity와 연관 Entity를 직접 노출하지 않고 Claim과 Lease 갱신 이후 Worker가
+ * 필요한 식별자와 소유권 시각만 전달한다.
  */
 @Component
 public class EmbeddingJobConverter {
@@ -33,4 +36,23 @@ public ClaimedEmbeddingJobResponse toClaimedResponse(EmbeddingJob embeddingJob)
             embeddingJob.getLockExpiresAt()
         );
     }
+
+    /**
+     * 갱신된 현재 Lease를 Claim Token 없이 API 응답으로 변환한다.
+     *
+     * @param embeddingJob PROCESSING 상태와 현재 Worker를 유지한 갱신 대상 Job
+     * @param renewedAt 갱신 Transaction이 사용한 기준 시각
+     * @return Worker가 다음 갱신 시점을 결정할 수 있는 안전한 Lease 응답
+     */
+    public RenewedEmbeddingJobLeaseResponse toRenewedLeaseResponse(
+        EmbeddingJob embeddingJob,
+        LocalDateTime renewedAt
+    ) {
+        return new RenewedEmbeddingJobLeaseResponse(
+            embeddingJob.getId(),
+            embeddingJob.getLockedByWorker().getId(),
+            renewedAt,
+            embeddingJob.getLockExpiresAt()
+        );
+    }
 }

src/main/java/com/opensource/docgrid/domain/embedding/converter/EmbeddingJobConverter.java

@@ -0,0 +1,32 @@
+package com.opensource.docgrid.domain.embedding.dto.request;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+import jakarta.validation.constraints.NotBlank;
+import jakarta.validation.constraints.NotNull;
+import jakarta.validation.constraints.Pattern;
+import jakarta.validation.constraints.Positive;
+import jakarta.validation.constraints.Size;
+
+/**
+ * 현재 Embedding Job Claim의 Lease를 갱신하기 위한 요청 DTO.
+ *
+ * <p>Worker ID와 canonical UUID Claim Token은 현재 소유권 검증에만 사용하며 성공 응답, Event와 일반
+ * 로그에는 Token을 다시 노출하지 않는다.
+ */
+public record RenewEmbeddingJobLeaseRequest(
+    @Schema(description = "현재 Job을 소유한 Worker 식별자", example = "7")
+    @NotNull
+    @Positive
+    Long workerId,
+
+    @Schema(description = "현재 Claim의 canonical UUID Token",
+        example = "34c19d16-6ae1-4f6a-a35d-0123456789ab")
+    @NotBlank
+    @Size(max = 36)
+    @Pattern(
+        regexp = "^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$",
+        message = "canonical UUID 형식이어야 합니다."
+    )
+    String claimToken
+) {
+}

src/main/java/com/opensource/docgrid/domain/embedding/dto/request/RenewEmbeddingJobLeaseRequest.java

@@ -0,0 +1,26 @@
+package com.opensource.docgrid.domain.embedding.dto.response;
+
+import java.time.LocalDateTime;
+
+import io.swagger.v3.oas.annotations.media.Schema;
+
+/**
+ * 현재 Embedding Job Claim의 갱신된 Lease 결과를 반환하는 응답 DTO.
+ *
+ * <p>Worker가 다음 갱신 시점을 결정하는 데 필요한 식별자와 시각만 제공한다. 소유권 증명 값인 Claim
+ * Token과 내부 Job·Worker 상태는 응답 경계 밖에 유지한다.
+ */
+public record RenewedEmbeddingJobLeaseResponse(
+    @Schema(description = "갱신된 Embedding Job 식별자", example = "101")
+    Long jobId,
+
+    @Schema(description = "현재 Job을 소유한 Worker 식별자", example = "7")
+    Long workerId,
+
+    @Schema(description = "Lease 갱신 기준 시각", example = "2026-08-03T15:00:00")
+    LocalDateTime renewedAt,
+
+    @Schema(description = "갱신된 Lease 만료 시각", example = "2026-08-03T15:05:00")
+    LocalDateTime lockExpiresAt
+) {
+}

src/main/java/com/opensource/docgrid/domain/embedding/dto/response/RenewedEmbeddingJobLeaseResponse.java

@@ -219,6 +219,41 @@ public boolean hasRemainingRetries() {
         return retryCount < maxRetryCount;
     }
 
+    /**
+     * 현재 PROCESSING Claim의 최초 잠금 시각과 소유권은 유지하고 Lease 만료 시각만 연장한다.
+     *
+     * <p>Service가 Job 행 잠금, 현재 Worker·Token과 기존 Lease 유효성을 먼저 검증해야 한다. Entity는
+     * 갱신이 기존 만료 시각을 줄이거나 이미 끝난 Job에 새 소유권처럼 적용되는 것을 마지막으로 방어한다.
+     *
+     * @param renewedAt Lease 갱신 기준 시각
+     * @param renewedLockExpiresAt 새 Lease 만료 시각
+     */
+    public void renewLease(
+        LocalDateTime renewedAt,
+        LocalDateTime renewedLockExpiresAt
+    ) {
+        // 1. 현재 소유권을 가진 처리 중 Job 이외의 종료·대기 상태는 갱신하지 않는다.
+        if (status != EmbeddingJobStatus.PROCESSING) {
+            throw new IllegalStateException("PROCESSING 상태의 Job Lease만 갱신할 수 있습니다.");
+        }
+        // 2. 현재 Lease가 이미 만료됐거나 소유권 시간이 누락된 모순 상태를 갱신으로 숨기지 않는다.
+        if (renewedAt == null
+            || lockedAt == null
+            || lockExpiresAt == null
+            || !lockExpiresAt.isAfter(renewedAt)) {
+            throw new IllegalStateException("유효한 현재 Lease만 갱신할 수 있습니다.");
+        }
+        // 3. 새 만료 시각은 갱신 기준 이후이며 기존 만료 시각을 실제로 연장해야 한다.
+        if (renewedLockExpiresAt == null
+            || !renewedLockExpiresAt.isAfter(renewedAt)
+            || !renewedLockExpiresAt.isAfter(lockExpiresAt)) {
+            throw new IllegalArgumentException("새 Lease 만료 시각은 현재 Lease보다 늦어야 합니다.");
+        }
+
+        // 4. lockedAt, Worker와 Claim Token은 같은 Claim 세대의 감사·소유권 정보이므로 보존한다.
+        this.lockExpiresAt = renewedLockExpiresAt;
+    }
+
     public void markFailed(String errorCode, String errorMessage, LocalDateTime failedAt) {
         // 현재 Claim을 보유한 처리 중 Job만 최종 실패로 종결할 수 있다.
         if (status != EmbeddingJobStatus.PROCESSING) {