Merge pull request #94 from DocGrid/feature/92

[Feat] Worker 자동 Polling 및 인덱싱 실행 오케스트레이션

Author: Gimini-3

docs/design/Gimini-3-#92-worker-polling-indexing-orchestration.md

@@ -0,0 +1,116 @@
+# #92 Worker Polling 및 인덱싱 실행 오케스트레이션 검증 결과
+
+## 1. 검증 정보
+
+- 실행일: 2026-08-03 (Asia/Seoul)
+- 대상 브랜치: `feature/92`
+- 애플리케이션: Spring Boot 3.5.16, Java 17
+- 데이터베이스: Docker Desktop의 격리된 PostgreSQL 14.6(OpenSQL 호환) + pgvector
+- 스키마: Flyway V1~V35 적용, 통합 테스트별 격리 스키마 사용
+- 최종 결과: 정규 테스트 576개와 동시성 테스트 10개 통과, 실패·오류·Skip 0개
+
+검증에는 localhost에만 노출한 작업 전용 컨테이너와 데이터 볼륨을 사용했다. 기존
+`local-opensql` 컨테이너와 `opensql_data` 공유 볼륨은 변경하지 않았다. 검증 종료 후 작업 전용
+컨테이너와 볼륨은 삭제했으며, 운영 Secret은 사용하거나 기록하지 않았다.
+
+## 2. Swagger/OpenAPI 수동 검증
+
+- 결과: 해당 없음
+- 근거: 이번 변경은 Worker 내부 Scheduler, 실행 Service와 설정만 추가하며 Controller, 요청·응답 DTO,
+  Endpoint 및 OpenAPI 계약을 변경하지 않는다.
+
+따라서 Swagger에서 호출할 신규·변경 API가 없으며, Worker 내부 실행 계약은 아래 자동 테스트와 실제
+PostgreSQL 동시성 테스트로 검증했다.
+
+## 3. 전체 정규 테스트
+
+실행 명령의 환경 값은 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 test
+```
+
+결과:
+
+```text
+BUILD SUCCESSFUL
+test suites=87 tests=576 failures=0 errors=0 skipped=0
+```
+
+기본 `test` Task에서 Worker 등록·Heartbeat·상태 관리, Polling Scheduler, 실행 슬롯, 파이프라인,
+Lease 갱신, 실패 보고, 종료 절차와 기존 도메인 회귀 테스트를 함께 검증했다.
+
+## 4. PostgreSQL 동시성 검증
+
+실행:
+
+```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 claimConcurrencyTest
+```
+
+결과:
+
+```text
+BUILD SUCCESSFUL
+test suites=3 tests=10 failures=0 errors=0 skipped=0
+```
+
+| 테스트 클래스 | 테스트 수 | 결과 |
+| --- | ---: | --- |
+| `EmbeddingJobClaimConcurrencyIntegrationTest` | 2 | 통과 |
+| `EmbeddingJobLeaseRecoveryIntegrationTest` | 4 | 통과 |
+| `WorkerOrchestrationIntegrationTest` | 4 | 통과 |
+
+새 Worker 오케스트레이션 통합 테스트는 다음 경쟁 조건을 실제 PostgreSQL 잠금과 트랜잭션으로
+검증했다.
+
+| 검증 항목 | 결과 |
+| --- | --- |
+| 두 Poller가 한 Job에 경쟁할 때 Claim과 파이프라인 제출이 한 번만 발생 | 통과 |
+| 실행 슬롯 2개인 Worker가 5개 Job 중 2개만 PROCESSING으로 Claim | 통과 |
+| 활성 실행의 Lease 갱신이 원래 만료 시각의 복구를 차단 | 통과 |
+| Lease 갱신 중단 후 동시 복구가 Job을 정확히 한 번만 재예약 | 통과 |
+
+## 5. 표준 빌드
+
+동일한 격리 DB와 테스트 전용 인증 설정에서 다음 명령을 실행했다.
+
+```bash
+./gradlew build
+```
+
+결과: `BUILD SUCCESSFUL`. Compile, Test, Check, Boot JAR 및 JAR 생성 단계가 모두 성공했다.
+
+## 6. 확인된 불변식
+
+- 여러 Worker가 같은 대기 Job을 조회해도 PostgreSQL Claim은 한 Worker에만 귀속된다.
+- 한 Worker가 소유하는 PROCESSING Job 수는 로컬 실행 슬롯 수를 넘지 않는다.
+- 슬롯이 없을 때 Poller는 추가 Job을 Claim하지 않고 다음 주기를 기다린다.
+- 활성 파이프라인은 완료 전까지 Lease를 갱신하며, 갱신된 Job은 이전 만료 시각에 복구되지 않는다.
+- Lease 갱신이 멈춘 만료 Job에 여러 복구 실행이 경쟁해도 Retry 전이는 한 번만 적용된다.
+- 성공·재시도·최종 실패 경로에서 Lease 갱신과 실행 슬롯은 정리된다.
+- 종료 요청 후 새 Polling은 시작되지 않고, 대기 중인 실행은 제한 시간 정책에 따라 정리된다.
+
+## 7. 환경 진단 기록과 제한 사항
+
+- 기존 공유 OpenSQL 볼륨은 이미지가 기대하는 내부 Role과 초기화 상태가 달라 사용할 수 없었다.
+  공유 데이터를 수정하지 않고 별도 작업 전용 컨테이너와 볼륨으로 전환했다.
+- 최초 일부 기존 통합 테스트 실행은 테스트용 JWT 환경 값 누락으로 Spring Context 구성 단계에서
+  실패했다. 테스트 전용 값을 제공한 재실행과 최종 전체 실행은 모두 통과했다.
+- 프로젝트 기본 `test` Task가 제외하는 `claim-concurrency`는 전용 Task로 별도 실행했다.
+- 외부 MinIO 의존 테스트, 실제 임베딩 서버 네트워크 동작과 Benchmark는 이번 검증 범위가 아니다.
+- 테스트용 컨테이너와 볼륨은 종료 시 삭제했으므로 그 안의 데이터는 복구하지 않는다.

docs/test-results/Gimini-3-#92-worker-polling-indexing-orchestration.md

@@ -0,0 +1,34 @@
+package com.opensource.docgrid.domain.document.service.query;
+
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import com.opensource.docgrid.domain.document.enums.DocumentVersionStatus;
+import com.opensource.docgrid.domain.document.repository.DocumentVersionRepository;
+import com.opensource.docgrid.global.exception.DocGridException;
+import com.opensource.docgrid.global.exception.ErrorCode;
+
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Worker 파이프라인의 시작 단계를 결정할 문서 버전 상태 Snapshot을 조회한다.
+ *
+ * <p>조회 결과는 경로 선택에만 사용하며 Entity를 Worker 계층에 전달하지 않는다. 실제 상태 변경 가능
+ * 여부와 소유권은 각 Command Service가 Job과 Version을 잠근 뒤 다시 검증한다.
+ */
+@Service
+@RequiredArgsConstructor
+@Transactional(readOnly = true)
+public class DocumentIndexingStageQueryService {
+
+    private final DocumentVersionRepository documentVersionRepository;
+
+    /**
+     * Claim 응답이 가리키는 문서 버전의 현재 파이프라인 상태를 반환한다.
+     */
+    public DocumentVersionStatus getStatus(Long documentVersionId) {
+        return documentVersionRepository.findById(documentVersionId)
+            .map(documentVersion -> documentVersion.getStatus())
+            .orElseThrow(() -> new DocGridException(ErrorCode.INDEXING_STATUS_INCONSISTENT));
+    }
+}

src/main/java/com/opensource/docgrid/domain/document/service/query/DocumentIndexingStageQueryService.java

@@ -14,7 +14,7 @@
 import lombok.Setter;
 
 /**
- * 인덱싱 Worker의 실행 여부, Heartbeat, DEAD 판정, Job Lease와 만료 복구를 바인딩하는 설정 클래스.
+ * 인덱싱 Worker의 실행 여부, Polling, 동시 실행, Heartbeat와 Job Lease 생명주기를 바인딩하는 설정 클래스.
  *
  * <p>{@code indexing.worker} 환경 설정을 타입 안전한 {@link Duration}으로 제공하고, 애플리케이션 시작
  * 단계에서 서로 모순되거나 0 이하인 시간 설정을 차단한다.
@@ -38,10 +38,21 @@ public class IndexingWorkerProperties {
     @NotNull
     private Duration deadThreshold = Duration.ofSeconds(30);
 
+    // 등록된 Worker가 실행 슬롯을 확인하고 새 Job을 찾는 주기다.
+    @NotNull
+    private Duration pollingInterval = Duration.ofSeconds(1);
+
+    @Min(1)
+    private int maxConcurrency = 2;
+
     // Claim 후 Worker가 소유권을 유지하는 기본 시간이다. 만료 복구는 후속 처리에서 사용한다.
     @NotNull
     private Duration leaseDuration = Duration.ofMinutes(5);
 
+    // 활성 실행은 Lease 만료 전에 이 주기로 소유권을 갱신한다.
+    @NotNull
+    private Duration leaseRenewalInterval = Duration.ofMinutes(1);
+
     // 만료 Lease 복구 작업의 실행 주기와 한 번에 조회할 최대 Job 수다.
     @NotNull
     private Duration leaseRecoveryInterval = Duration.ofSeconds(30);
@@ -56,6 +67,10 @@ public class IndexingWorkerProperties {
     @NotNull
     private Duration retryMaxDelay = Duration.ofMinutes(5);
 
+    // 종료 시 신규 Claim을 막은 뒤 활성 실행이 스스로 끝나기를 기다리는 최대 시간이다.
+    @NotNull
+    private Duration shutdownGracePeriod = Duration.ofSeconds(30);
+
     /**
      * Heartbeat가 양수이고 DEAD 기준보다 짧은지 검증한다.
      */
@@ -68,6 +83,16 @@ public boolean isTimingValid() {
             && deadThreshold.compareTo(heartbeatInterval) > 0;
     }
 
+    /**
+     * 빈 작업 조회가 Busy Loop가 되지 않도록 Polling 주기가 양수인지 검증한다.
+     */
+    @AssertTrue(message = "Job Polling 주기는 0보다 커야 합니다.")
+    public boolean isPollingIntervalValid() {
+        return pollingInterval != null
+            && !pollingInterval.isZero()
+            && !pollingInterval.isNegative();
+    }
+
     /**
      * 발급 즉시 만료되는 Lease가 만들어지지 않도록 Lease 기간이 양수인지 검증한다.
      */
@@ -78,6 +103,18 @@ public boolean isLeaseDurationValid() {
             && !leaseDuration.isNegative();
     }
 
+    /**
+     * 활성 Job이 만료 전에 갱신될 수 있도록 갱신 주기가 양수이고 Lease 기간보다 짧은지 검증한다.
+     */
+    @AssertTrue(message = "Lease 갱신 주기는 0보다 크고 Lease 기간보다 짧아야 합니다.")
+    public boolean isLeaseRenewalIntervalValid() {
+        return leaseRenewalInterval != null
+            && leaseDuration != null
+            && !leaseRenewalInterval.isZero()
+            && !leaseRenewalInterval.isNegative()
+            && leaseRenewalInterval.compareTo(leaseDuration) < 0;
+    }
+
     /**
      * 만료 Lease 복구 Scheduler가 과도하게 반복되지 않도록 실행 주기가 양수인지 검증한다.
      */
@@ -99,4 +136,13 @@ public boolean isRetryDelayValid() {
             && !retryInitialDelay.isNegative()
             && retryMaxDelay.compareTo(retryInitialDelay) >= 0;
     }
+
+    /**
+     * 즉시 종료는 허용하되 음수 대기 시간은 Executor 종료 계약으로 사용할 수 없으므로 차단한다.
+     */
+    @AssertTrue(message = "Worker 종료 유예 시간은 0보다 작을 수 없습니다.")
+    public boolean isShutdownGracePeriodValid() {
+        return shutdownGracePeriod != null
+            && !shutdownGracePeriod.isNegative();
+    }
 }

src/main/java/com/opensource/docgrid/domain/worker/config/IndexingWorkerProperties.java

@@ -0,0 +1,68 @@
+package com.opensource.docgrid.domain.worker.config;
+
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.SynchronousQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+
+import com.opensource.docgrid.domain.worker.execution.WorkerExecutionSlotPool;
+
+/**
+ * Worker Job 실행과 Lease 갱신에 사용하는 제한된 Thread 자원을 구성한다.
+ *
+ * <p>Job Executor는 Queue에 Claim을 쌓지 않고 설정된 동시성만 즉시 실행한다. Lease Scheduler는 활성
+ * 실행의 짧은 갱신 호출만 담당하며, Worker가 비활성화된 API 전용 실행에는 어떤 Thread도 만들지 않는다.
+ */
+@Configuration
+@ConditionalOnProperty(prefix = "indexing.worker", name = "enabled", havingValue = "true")
+public class WorkerExecutionConfig {
+
+    public static final String WORKER_JOB_EXECUTOR = "workerJobExecutor";
+    public static final String WORKER_LEASE_SCHEDULER = "workerLeaseScheduler";
+
+    /**
+     * 최대 동시 실행 수와 같은 크기의 무대기 Job Executor를 만든다.
+     */
+    @Bean(name = WORKER_JOB_EXECUTOR, destroyMethod = "shutdownNow")
+    public ThreadPoolExecutor workerJobExecutor(IndexingWorkerProperties properties) {
+        int maxConcurrency = properties.getMaxConcurrency();
+        return new ThreadPoolExecutor(
+            maxConcurrency,
+            maxConcurrency,
+            0L,
+            TimeUnit.MILLISECONDS,
+            new SynchronousQueue<>(),
+            new CustomizableThreadFactory("indexing-worker-job-"),
+            new ThreadPoolExecutor.AbortPolicy()
+        );
+    }
+
+    /**
+     * 모든 활성 실행의 Lease 갱신을 직렬로 예약하는 단일 Thread Scheduler를 만든다.
+     */
+    @Bean(name = WORKER_LEASE_SCHEDULER, destroyMethod = "shutdownNow")
+    public ScheduledThreadPoolExecutor workerLeaseScheduler() {
+        ScheduledThreadPoolExecutor scheduler = new ScheduledThreadPoolExecutor(
+            1,
+            new CustomizableThreadFactory("indexing-worker-lease-")
+        );
+        // 취소된 실행별 갱신 작업이 Scheduler Queue에 남아 종료와 메모리 회수를 늦추지 않게 한다.
+        scheduler.setRemoveOnCancelPolicy(true);
+        scheduler.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);
+        scheduler.setContinueExistingPeriodicTasksAfterShutdownPolicy(false);
+        return scheduler;
+    }
+
+    /**
+     * Claim 전에 실행 가능 여부를 예약하는 프로세스 로컬 슬롯 풀을 만든다.
+     */
+    @Bean
+    public WorkerExecutionSlotPool workerExecutionSlotPool(IndexingWorkerProperties properties) {
+        return new WorkerExecutionSlotPool(properties.getMaxConcurrency());
+    }
+}