[Feat] Worker 자동 Polling 및 인덱싱 실행 오케스트레이션 - #94
Merged
Merged
Conversation
|
Warning Review limit reached
Next review available in: 46 minutes You've used all free OSS reviews for now. Wait for the free limit to reset to keep reviewing this public repository. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
📝 WalkthroughWalkthroughWorker 자동 Polling과 인덱싱 오케스트레이션을 추가했다. 실행 슬롯, Job Claim, Pipeline, Lease 갱신, 실패 보고, Graceful Shutdown을 구현했다. 단위·PostgreSQL 동시성 테스트와 빌드 결과를 문서화했다. ChangesWorker 오케스트레이션
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant WorkerJobPollingScheduler
participant WorkerExecutionSlotPool
participant EmbeddingJobClaimService
participant WorkerIndexingPipeline
participant WorkerLeaseRenewalManager
WorkerJobPollingScheduler->>WorkerExecutionSlotPool: 실행 슬롯 확보
WorkerJobPollingScheduler->>EmbeddingJobClaimService: PENDING Job Claim
EmbeddingJobClaimService-->>WorkerJobPollingScheduler: ClaimedEmbeddingJobResponse
WorkerJobPollingScheduler->>WorkerIndexingPipeline: Pipeline 실행 제출
WorkerIndexingPipeline->>WorkerLeaseRenewalManager: Lease 갱신 등록
WorkerIndexingPipeline->>WorkerIndexingPipeline: 상태별 Chunk·Embedding 처리
WorkerIndexingPipeline->>WorkerLeaseRenewalManager: Lease 소유권 확인
WorkerIndexingPipeline->>WorkerExecutionSlotPool: 실행 슬롯 반환
Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (3)
src/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingPipeline.java (1)
36-36: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚖️ Poor tradeoff상태 변경 Service를
command패키지로 분리하세요.두 클래스는 Job 상태를 변경하는 Service를 직접 호출합니다. 현재 루트
service패키지에 있으므로 command/query 책임 분리 규칙과 맞지 않습니다.
src/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingPipeline.java#L36-L36:WorkerIndexingPipeline을domain.worker.service.command로 이동하고 참조를 갱신하세요.src/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingFailureReporter.java#L22-L22:WorkerIndexingFailureReporter를domain.worker.service.command로 이동하고 참조를 갱신하세요.As per coding guidelines, "Separate service responsibilities into
commandpackages for state changes andquerypackages for read-only operations."🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingPipeline.java` at line 36, The state-changing services are currently in the root service package instead of the command package. Move WorkerIndexingPipeline in src/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingPipeline.java:36-36 and WorkerIndexingFailureReporter in src/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingFailureReporter.java:22-22 to domain.worker.service.command, then update all package declarations, imports, and references.Source: Coding guidelines
src/test/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerExecutionLifecycleManagerTest.java (1)
62-71: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value중복 검증을 제거해 주세요.
라인 62의
should()는should(times(1))과 같습니다. 라인 71이 같은 내용을 다시 검증합니다. 이 테스트는shutdown()을 두 번 호출해도 한 번만 실행되는 것을 확인하므로 라인 71만 남기는 편이 의도를 더 잘 드러냅니다.♻️ 제안 변경
- then(pollingScheduler).should().stopPolling(); then(jobExecutor).should().shutdown();🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/test/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerExecutionLifecycleManagerTest.java` around lines 62 - 71, Remove the redundant pollingScheduler.stopPolling() verification from the first interaction block in WorkerExecutionLifecycleManagerTest, keeping the later should(times(1)) assertion as the sole verification that stopPolling() is called exactly once.src/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerExecutionLifecycleManager.java (1)
80-88: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low value
leaseScheduler종료 대기를 추가하면 종료 로그가 더 정확해집니다.
shutdown()은 큐에 남은 작업만 막고 실행 중인 갱신 Task는 기다리지 않습니다. 라인 81의stopAll()이 예약을 취소하므로 남는 것은 이미 시작된 갱신 1건뿐입니다. 그 갱신은 컨텍스트가 닫힌 뒤 DataSource 종료로 실패할 수 있습니다.renew가 예외를 삼키므로 기능 영향은 없지만 종료 로그에 오류가 섞입니다.짧은 대기를 추가하면 종료 시점이 결정적이 됩니다.
♻️ 제안 변경
leaseRenewalManager.stopAll(); leaseScheduler.shutdown(); + boolean leaseTerminated = awaitLeaseTermination(); log.info( - "Worker 실행 종료를 완료했습니다. graceful={}, cancelledTaskCount={}, activeLeaseCount={}", + "Worker 실행 종료를 완료했습니다. graceful={}, leaseTerminated={}, cancelledTaskCount={}, activeLeaseCount={}", terminated, + leaseTerminated, cancelledTasks.size(), leaseRenewalManager.getActiveHandleCount() );
awaitLeaseTermination()은awaitJobTermination()과 같은 형태로 추가합니다.// 남은 갱신 Task가 끝날 때까지 짧게 기다려 종료 시점을 결정적으로 만든다. private boolean awaitLeaseTermination() { try { return leaseScheduler.awaitTermination(1, TimeUnit.SECONDS); } catch (InterruptedException exception) { Thread.currentThread().interrupt(); return false; } }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerExecutionLifecycleManager.java` around lines 80 - 88, Update WorkerExecutionLifecycleManager’s shutdown flow after leaseScheduler.shutdown() to await lease-task termination for up to one second via a new awaitLeaseTermination() helper, preserving the existing interruption behavior by restoring the thread interrupt flag and returning false. Invoke this wait before the completion log so shutdown timing is deterministic.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@docs/design/Gimini-3-`#92-worker-polling-indexing-orchestration.md:
- Around line 179-185: Update the documented failure-exclusion contract to match
WorkerIndexingFailureClassifier: either add EMBEDDING_JOB_NOT_FOUND to the
listed no-report ownership-loss errors if that behavior is intended, or remove
it from the classifier’s exclusion handling if it is not. Keep the documentation
and classifier behavior consistent.
In `@docs/test-results/Gimini-3-`#92-worker-polling-indexing-orchestration.md:
- Around line 16-18: 문서의 테스트 결과 섹션에 Swagger 수동 검증 절차와 결과를 추가하세요. Swagger 대상 API가
없는 Worker 내부 변경이라면 기존 자동 테스트 및 PostgreSQL 동시성 검증과 함께 `해당 없음`을 명시하고 그 근거를 기록하세요.
In
`@src/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerJobPollingScheduler.java`:
- Around line 59-93: Separate the scheduled poll/claim work in
WorkerJobPollingScheduler from updateHeartbeat() and recoverExpiredLeases() by
configuring a dedicated TaskScheduler or increasing the scheduling pool through
WorkerSchedulingConfig. Ensure poll() can hold its database lock without
blocking heartbeat and lease-recovery executions, while preserving the existing
scheduling behavior.
---
Nitpick comments:
In
`@src/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerExecutionLifecycleManager.java`:
- Around line 80-88: Update WorkerExecutionLifecycleManager’s shutdown flow
after leaseScheduler.shutdown() to await lease-task termination for up to one
second via a new awaitLeaseTermination() helper, preserving the existing
interruption behavior by restoring the thread interrupt flag and returning
false. Invoke this wait before the completion log so shutdown timing is
deterministic.
In
`@src/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingPipeline.java`:
- Line 36: The state-changing services are currently in the root service package
instead of the command package. Move WorkerIndexingPipeline in
src/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingPipeline.java:36-36
and WorkerIndexingFailureReporter in
src/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingFailureReporter.java:22-22
to domain.worker.service.command, then update all package declarations, imports,
and references.
In
`@src/test/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerExecutionLifecycleManagerTest.java`:
- Around line 62-71: Remove the redundant pollingScheduler.stopPolling()
verification from the first interaction block in
WorkerExecutionLifecycleManagerTest, keeping the later should(times(1))
assertion as the sole verification that stopPolling() is called exactly once.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: e34aec4c-6f6e-4bce-8b9f-9a84453fb496
📒 Files selected for processing (25)
docs/design/Gimini-3-#92-worker-polling-indexing-orchestration.mddocs/test-results/Gimini-3-#92-worker-polling-indexing-orchestration.mdsrc/main/java/com/opensource/docgrid/domain/document/service/query/DocumentIndexingStageQueryService.javasrc/main/java/com/opensource/docgrid/domain/worker/config/IndexingWorkerProperties.javasrc/main/java/com/opensource/docgrid/domain/worker/config/WorkerExecutionConfig.javasrc/main/java/com/opensource/docgrid/domain/worker/execution/WorkerExecutionSlotPool.javasrc/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerExecutionLifecycleManager.javasrc/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerJobPollingScheduler.javasrc/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerLifecycleManager.javasrc/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingFailure.javasrc/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingFailureClassifier.javasrc/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingFailureReporter.javasrc/main/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingPipeline.javasrc/main/java/com/opensource/docgrid/domain/worker/service/WorkerLeaseRenewalHandle.javasrc/main/java/com/opensource/docgrid/domain/worker/service/WorkerLeaseRenewalManager.javasrc/main/resources/application.ymlsrc/test/java/com/opensource/docgrid/domain/worker/config/IndexingWorkerPropertiesTest.javasrc/test/java/com/opensource/docgrid/domain/worker/execution/WorkerExecutionSlotPoolTest.javasrc/test/java/com/opensource/docgrid/domain/worker/integration/WorkerOrchestrationIntegrationTest.javasrc/test/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerExecutionLifecycleManagerTest.javasrc/test/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerJobPollingSchedulerTest.javasrc/test/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingFailureClassifierTest.javasrc/test/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingFailureReporterTest.javasrc/test/java/com/opensource/docgrid/domain/worker/service/WorkerIndexingPipelineTest.javasrc/test/java/com/opensource/docgrid/domain/worker/service/WorkerLeaseRenewalManagerTest.java
Comment on lines
+59
to
+93
| @Scheduled( | ||
| fixedDelayString = "${indexing.worker.polling-interval:1s}", | ||
| initialDelayString = "${indexing.worker.polling-interval:1s}" | ||
| ) | ||
| public void poll() { | ||
| if (!slotPool.isAccepting() || !polling.compareAndSet(false, true)) { | ||
| return; | ||
| } | ||
|
|
||
| try { | ||
| Optional<Long> registeredWorkerId = workerLifecycleManager.getWorkerId(); | ||
| if (registeredWorkerId.isEmpty()) { | ||
| return; | ||
| } | ||
|
|
||
| // 1. 한 주기에는 설정된 전체 슬롯 수까지만 Claim을 시도하고 사용 중 슬롯은 즉시 건너뛴다. | ||
| for (int index = 0; index < slotPool.getCapacity(); index++) { | ||
| Optional<WorkerExecutionSlot> acquiredSlot = slotPool.tryAcquire(); | ||
| if (acquiredSlot.isEmpty()) { | ||
| return; | ||
| } | ||
|
|
||
| // 2. 슬롯 획득 뒤 시작된 종료와 Claim 오류는 이 Slot만 반환하고 현재 주기를 끝낸다. | ||
| WorkerExecutionSlot executionSlot = acquiredSlot.get(); | ||
| if (!slotPool.isAccepting()) { | ||
| executionSlot.close(); | ||
| return; | ||
| } | ||
| Optional<ClaimedEmbeddingJobResponse> claimedJob = claim( | ||
| registeredWorkerId.get(), | ||
| executionSlot | ||
| ); | ||
| if (claimedJob.isEmpty()) { | ||
| return; | ||
| } |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# 1. 스케줄러 풀 설정을 확인한다.
fd -e yml -e yaml -e properties . src/main/resources --exec rg -n 'scheduling|pool' {} \;
# 2. 커스텀 TaskScheduler 및 다른 `@Scheduled` 사용처를 확인한다.
rg -nP --type=java -C3 '`@Scheduled`|TaskScheduler|SchedulingConfigurer' src/main/javaRepository: DocGrid/backend
Length of output: 3039
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '## Candidate files matching worker lifecycle schedulers\n'
fd -a 'Worker(JobPolling|Heartbeat|LeaseRecovery)Scheduler\.java' src/main/java || true
printf '\n## Scheduling configuration files\n'
fd -a 'application.*|application-local.*|application-test.*' src/main/resources src/test/resources || true
printf '\n## WorkerJobPollingScheduler relevant section\n'
FILE="src/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerJobPollingScheduler.java"
if [ -f "$FILE" ]; then
sed -n '1,180p' "$FILE" | cat -n
fi
printf '\n## WorkerHeartbeatScheduler relevant section\n'
FILE="src/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerHeartbeatScheduler.java"
if [ -f "$FILE" ]; then
sed -n '1,140p' "$FILE" | cat -n
fi
printf '\n## WorkerLeaseRecoveryScheduler relevant section\n'
FILE="src/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerLeaseRecoveryScheduler.java"
if [ -f "$FILE" ]; then
sed -n '1,140p' "$FILE" | cat -n
fi
printf '\n## Scheduling/TaskScheduler references in Java\n'
rg -n --type=java -C2 '`@Scheduled`|TaskScheduler|SchedulingConfigurer|ThreadPoolTaskScheduler|`@EnableScheduling`|spring\.task\.scheduling|task\.scheduling' . || true
printf '\n## Config references to scheduling/pool\n'
fd -e yml -e yaml -e properties . src/main/resources src/test/resources --exec rg -n -C2 -i 'scheduling|task\.scheduling|pool-name|pool-size|scheduler' {} \; || true
printf '\n## claim/findNextPendingForUpdate references\n'
rg -n --type=java -C4 'claim\(|findNextPendingForUpdate|PendingForUpdate|LockModeType\.PESSIMISTIC|FOR UPDATE' src/main/java src/test/java || trueRepository: DocGrid/backend
Length of output: 50372
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '## WorkerSchedulingConfig\n'
FILE="src/main/java/com/opensource/docgrid/domain/worker/config/WorkerSchedulingConfig.java"
if [ -f "$FILE" ]; then
sed -n '1,240p' "$FILE" | cat -n
fi
printf '\n## All `@Scheduled` symbols with file names\n'
python3 - <<'PY'
import pathlib
for p in pathlib.Path('src/main/java').rglob('*.java'):
text = p.read_text(errors='ignore')
if '`@Scheduled`' in text:
print(f'FILE {p}')
for i,line in enumerate(text.splitlines(),1):
if '`@Scheduled`' in line:
start=max(1,i-2); end=min(len(text.splitlines()),i+8)
lines=text.splitlines()
for n in range(start,end+1):
print(f'{n}: {lines[n-1]}')
PY
printf '\n## TaskScheduler and pool-size references in tracked files\n'
git grep -n -i 'TaskScheduler\|ThreadPoolTaskScheduler\|SchedulingConfigurer\|spring\.task\.scheduling\.pool\.size\|task\.scheduling\.pool\.size\|pool\.size' -- '*.java' '**/*.yml' '**/*.yaml' '**/*.properties' || trueRepository: DocGrid/backend
Length of output: 2555
Polling과 Heartbeat를 전용 스케줄러로 분리하세요.
WorkerSchedulingConfig는 기본 @EnableScheduling만并提供 TaskScheduler 설정이 없습니다. 따라서 poll(), updateHeartbeat(), recoverExpiredLeases()가 Spring 기본 스케줄러를 공유하고, poll() 내부 DB 잠금을 가진 claim() 호출이 길어지면 Heartbeat도 기다립니다. 전용 TaskScheduler 또는 spring.task.scheduling.pool.size 설정으로 Claim 경로를 별도의 스레드로 분리해 주세요.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In
`@src/main/java/com/opensource/docgrid/domain/worker/lifecycle/WorkerJobPollingScheduler.java`
around lines 59 - 93, Separate the scheduled poll/claim work in
WorkerJobPollingScheduler from updateHeartbeat() and recoverExpiredLeases() by
configuring a dedicated TaskScheduler or increasing the scheduling pool through
WorkerSchedulingConfig. Ensure poll() can hold its database lock without
blocking heartbeat and lease-recovery executions, while preserving the existing
scheduling behavior.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
변경 내용
변경 이유
기존에는 Worker 등록, Claim, 파싱, Embedding, 완료·실패 기능이 개별적으로 존재했지만 이를 자동으로 연결하는 실행 Loop가 없어 관리자 API를 수동 호출해야 했습니다. 이 변경으로 Worker가 업로드된 문서를 제한된 동시성 안에서 자동으로 인덱싱 완료까지 처리합니다.
영향
indexing.worker.enabled=false는 유지되어 API 전용 실행에는 영향이 없습니다.검증
./gradlew test: 576개 통과, 실패·오류·Skip 0./gradlew claimConcurrencyTest: 10개 통과, 실패·오류·Skip 0./gradlew build: 성공Closes #92
Summary by CodeRabbit
새로운 기능
문서화