Feat/#109 polling to sse#120
Conversation
채용공고 및 자소서 분석 async 상태 조회를 폴링 방식에서 SSE 기반 스트림 방식으로 확장했습니다. 공통 SSE emitter 레지스트리를 추가하고, async task 상태 변경 시점마다 즉시 이벤트를 발행하도록 연결해 프론트가 실시간으로 상태를 반영할 수 있게 했습니다.
SSE 기반 async 상태 스트림에 주기적인 heartbeat 이벤트를 추가해 장시간 유지되는 연결이 프록시나 브라우저에서 유휴 연결로 끊기지 않도록 보강했습니다. 공통 SSE 레지스트리에서 heartbeat를 관리하도록 구성해 채용공고 및 자소서 분석 스트림에 동일하게 적용했습니다.
비동기 task가 PENDING 또는 RUNNING 상태에서 장시간 멈춰 있을 때 주기적으로 timeout 여부를 점검해 FAILED로 정리하는 stuck-task sweep을 추가했습니다. 채용공고는 상태만 정리하고, 자소서 분석은 timeout 실패 시 예약된 크레딧을 함께 환불하도록 구성해 비동기 복구와 SSE 상태 전파의 운영 일관성을 높였습니다.
|
Warning Review limit reached
Next review available in: 15 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. 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: defaults Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (15)
📝 WalkthroughWalkthroughThis PR adds SSE-based streaming of async task status for analysis and job-posting domains via a shared SseSubscriptionRegistry, renames analysis credit handling from reserve/confirm/release to deduct/refund, adds sweep services for timed-out tasks with a scheduled trigger, tracks job-posting worker metadata, and introduces an UNKNOWN payment status for confirmation timeouts. ChangesAsync Task SSE Streaming and Sweeping
Payment Confirmation Timeout Handling
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant Controller
participant SseSubscriptionRegistry
Client->>Controller: GET .../async/{taskId}/stream
Controller->>SseSubscriptionRegistry: subscribe(channelKey, initialStatus)
SseSubscriptionRegistry-->>Client: connected + status event
Note over Controller,SseSubscriptionRegistry: On status transition
Controller->>SseSubscriptionRegistry: publish(channelKey, updatedStatus)
SseSubscriptionRegistry-->>Client: status update event
sequenceDiagram
participant Scheduler
participant AnalysisAsyncSweepService
participant JobPostingAsyncTaskService
participant AnalysisService
Scheduler->>AnalysisAsyncSweepService: sweepTimedOutTasks()
AnalysisAsyncSweepService->>AnalysisService: refundAnalysisCredit(user, referenceId)
Scheduler->>JobPostingAsyncTaskService: sweepTimedOutTasks()
JobPostingAsyncTaskService-->>Scheduler: expired task counts
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 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: 7
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java (1)
52-97: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftSSE
publish()fires before the transaction commits.In
markRunning(Line 59),markSuccess(Line 75),markRetryScheduled(Line 86),markFailed(Line 96), andsweepTimedOutTasks(Line 133),jobPostingAsyncSseService.publish(...)is called on the in-memory, not-yet-committed entity state, before the@Transactionalmethod returns and the transaction actually commits. This means a subscribed client can receive an SSE event (e.g. terminalSUCCEEDED/FAILED, which also closes their SSE stream) for a state that hasn't been durably persisted yet. If the flush/commit subsequently fails (constraint violation, connection issue, etc.), the client would have already been notified — and its stream closed — for a transition that never actually took effect, while a fresh poll ofGET /ingest/async/{taskId}could show different data.Consider publishing only after the transaction successfully commits, e.g. via
TransactionSynchronizationManager.registerSynchronization(...)'safterCommit()hook, so notifications are only ever sent for durably-committed state.♻️ Sketch of a post-commit publish helper
private void publishAfterCommit(JobPostingAsyncStatusResponse response) { if (TransactionSynchronizationManager.isSynchronizationActive()) { TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { `@Override` public void afterCommit() { jobPostingAsyncSseService.publish(response); } }); } else { jobPostingAsyncSseService.publish(response); } }Replace each
jobPostingAsyncSseService.publish(toStatusResponse(task))call withpublishAfterCommit(toStatusResponse(task)).Also applies to: 126-138
🤖 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/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java` around lines 52 - 97, `JobPostingAsyncTaskService` is publishing SSE updates before the enclosing transaction commits, so clients can observe uncommitted or later-rolled-back state. Update `markRunning`, `markSuccess`, `markRetryScheduled`, `markFailed`, and `sweepTimedOutTasks` to publish via a post-commit hook instead of calling `jobPostingAsyncSseService.publish(toStatusResponse(task))` directly. Add a helper in this service (or equivalent) that registers `afterCommit()` with `TransactionSynchronizationManager` and only publishes the `toStatusResponse(task)` result once the transaction has successfully committed.src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java (1)
72-88: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftMove the credit reservation out of
analyze()'s transaction.
deductAnalysisCredit()andrefundAnalysisCredit()both run inside the outeranalyze()transaction, so the credit change isn’t committed beforeexecuteAnalysis()starts. If the AI call or later finalize step throws, the refund is rolled back with the original deduction, and the DB work stays open for the whole remote request. Use an independent transaction for the reservation/refund so the balance change commits separately.🤖 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/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java` around lines 72 - 88, The credit reservation logic in AnalysisService.analyze() is running inside the outer transaction, so deductAnalysisCredit() and refundAnalysisCredit() are not committed independently before executeAnalysis() starts. Move the reservation/refund handling into its own transactional boundary (for example via a separate method on AnalysisService or a dedicated helper) so the balance change commits separately, and keep analyze() focused on validateAnalysisRequest(), prepareAnalysisExecution(), executeAnalysis(), and finalizeAnalysis().
🧹 Nitpick comments (12)
src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java (1)
86-95: 🗄️ Data Integrity & Integration | 🔵 TrivialNo automated reconciliation path for stuck
UNKNOWNpayments.
markConfirmationUnknownonly flips status toUNKNOWN; recovery relies entirely on the same user re-issuing a confirm request with the samepaymentKey. If the user abandons the flow after a timeout (browser closed, app killed), the payment staysUNKNOWNindefinitely even if Toss actually completed the charge — resulting in a charged customer with no credited balance, discoverable only via manual investigation.Toss exposes a payment inquiry API by paymentKey/orderId that could be polled by a scheduled reconciliation job (similar in spirit to the async-task sweep added elsewhere in this PR) to resolve
UNKNOWNpayments automatically.🤖 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/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java` around lines 86 - 95, The markConfirmationUnknown flow in PaymentTransactionService only moves a payment to UNKNOWN and leaves no path to recover stuck payments later. Update the payment lifecycle so UNKNOWN entries can be reconciled automatically, ideally by adding a scheduled reconciliation job that uses Toss inquiry by paymentKey/orderId and reuses the payment state transitions in PaymentTransactionService. Make the fix centered around markConfirmationUnknown and the existing payment status handling so abandoned flows are eventually resolved without relying on the user to retry.src/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java (1)
186-221: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider adding a test for
UNKNOWNretry with a mismatchedpaymentKey.Current tests cover the happy-path retry (same
paymentKeyafter timeout → success), but not the branch instartConfirmationwhere anUNKNOWNpayment receives a confirm request with a differentpaymentKeyand should throwPAYMENT_ALREADY_PROCESSED. Adding this case would close coverage on the new branch.🤖 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/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java` around lines 186 - 221, Add a test to cover the UNKNOWN retry path when the confirm request uses a different paymentKey than the stored payment. Extend PaymentServiceTest around confirmAllowsRetryWhenTossConfirmTimesOut() to first force a timeout, then call paymentService.confirm with a mismatched PaymentConfirmRequest and assert GeneralException with GeneralErrorCode.PAYMENT_ALREADY_PROCESSED. Use the existing paymentService.confirm, TossPaymentConfirmResponse, and paymentRepository.findByOrderId setup to exercise the startConfirmation branch for UNKNOWN payments.src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java (1)
39-41: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winTerminal-status check via hardcoded strings duplicated across domains.
isTerminalcomparesstatusResponse.status()against the literal strings"SUCCEEDED"/"FAILED", and the identical logic is duplicated verbatim inJobPostingAsyncSseService. If the underlyingTaskStatusenum values ever change, these string literals won't get a compile-time signal. Consider exposing/using the domainTaskStatusenum (or a shared helper) instead of string literals, and/or extracting the shared subscribe/publish/isTerminal boilerplate into a small generic helper reused by both domains.🤖 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/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java` around lines 39 - 41, The terminal-status check in AnalysisAsyncSseService is hardcoded with duplicated string literals, and the same logic exists in JobPostingAsyncSseService. Replace the literal comparisons in isTerminal with the shared domain TaskStatus enum (or a shared helper based on it) so status checks are compile-time safe. If possible, extract the repeated subscribe/publish/isTerminal flow into a small reusable helper used by both services to remove the duplicated boilerplate.src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java (1)
126-138: 🚀 Performance & Scalability | 🔵 TrivialConsider batching the sweep query for scalability.
sweepTimedOutTasks()loads and processes everyPENDING/RUNNINGtask in a single transaction/loop with no pagination. This is fine at current scale but could become a long-running transaction (and hold row locks) if the number of stuck tasks grows. Worth keeping in mind as usage scales — e.g. paginate viaPageableor cap the batch size per scheduler tick.🤖 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/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java` around lines 126 - 138, The sweep in JobPostingAsyncTaskService.sweepTimedOutTasks() processes all PENDING/RUNNING tasks in one transaction, which can become a long-running lock-heavy operation as data grows. Update the repository call and loop to process tasks in bounded batches, using a Pageable/limit or similar batch size per scheduler tick, and keep the existing expireTimedOutTaskIfNeeded and publish(toStatusResponse(task)) flow unchanged for each processed task.src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java (1)
153-169: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winNaming still implies a hold/reservation, though the underlying call now performs a direct deduction.
reserveCreditIfNeeded/markCreditReserved/CreditStatus.RESERVED/confirmCreditIfNeededretain reserve/confirm terminology, but they now wrapanalysisService.deductAnalysisCredit(...), which charges credit immediately rather than placing a hold. This is functionally fine (release/refund still checksRESERVEDcorrectly), but the mismatch between naming and actual semantics could confuse future maintainers reasoning about credit lifecycle guarantees.Consider renaming to align with the
AnalysisServicededuct/refund terminology (e.g.,deductCreditIfNeeded/markCreditDeducted) for consistency.🤖 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/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java` around lines 153 - 169, The credit lifecycle names in AnalysisWorkerBridgeService still imply a reservation/hold even though the flow now performs an immediate deduction via analysisService.deductAnalysisCredit(...). Update the helper names and related status/marking methods to match the actual semantics (for example, reserveCreditIfNeeded, markCreditReserved, and confirmCreditIfNeeded should align with deduction/confirmation terminology), while keeping the existing release/refund behavior consistent with CreditStatus and the task reference ID handling.src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java (2)
138-149: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winLGTM on this test itself.
Separately, consider adding a test for the new
isTerminal()guard inmarkRetryScheduled(e.g., calling it on an already-SUCCEEDED/FAILED task should be a no-op), mirroringAnalysisWorkerBridgeServiceTest.terminalTaskDoesNotReopen.🤖 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/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java` around lines 138 - 149, The new isTerminal() guard in markRetryScheduled is not covered by tests; add a test in JobPostingAsyncTaskServiceTest that calls markRetryScheduled on an already terminal JobPostingAsyncTask (SUCCEEDED or FAILED) and verifies it remains unchanged, similar to AnalysisWorkerBridgeServiceTest.terminalTaskDoesNotReopen. Use the existing markRetryScheduled and isTerminal behavior to keep the assertion focused on the no-op path.
118-136: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winConsider also asserting the SSE publish on this transition.
This test verifies the entity-level state transition but not that
jobPostingAsyncSseService.publish(...)fires when a task is auto-failed via max-retry exhaustion — the core behavior this PR cohort introduces.🤖 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/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java` around lines 118 - 136, The max-retry failure test in JobPostingAsyncTaskServiceTest only checks the entity state transition; update it to also verify that JobPostingAsyncSseService.publish(...) is invoked when markRetryScheduled(...) auto-transitions a task to FAILED. Use the existing task setup and add a verification for the publish call on the max-retry exhaustion path so the test covers the new SSE behavior introduced by JobPostingAsyncTaskService.markRetryScheduled.src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java (1)
126-136: 🎯 Functional Correctness | 🔵 Trivial | ⚡ Quick winInconsistent terminal-state guard between
markFailedandmarkRetryScheduled.
markRetryScheduledusesisTerminal()(SUCCEEDED or FAILED) to no-op, butmarkFailedonly guards againstSUCCEEDED. A secondmarkFailed(...)call on an already-FAILED task will still overwritecompletedAt/error/failureReason/retryCountand can trigger a duplicate SSE publish downstream. Current callers happen to guard before invoking this, but the entity's own invariant should be self-consistent to avoid this becoming a bug when a new caller is added.♻️ Proposed fix
public void markFailed(FailureReason failureReason, String errorMessage, int retryCount) { - if (status == TaskStatus.SUCCEEDED) { + if (isTerminal()) { return; }🤖 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/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java` around lines 126 - 136, `JobPostingAsyncTask.markFailed` has a weaker terminal-state check than `markRetryScheduled`, so a second failure can overwrite an already-FAILED task. Update `markFailed(FailureReason, String, int)` to use the same terminal guard as `isTerminal()` (or equivalent) so it no-ops for both SUCCEEDED and FAILED states, keeping `completedAt`, `error`, `failureReason`, and `retryCount` stable on repeat calls.src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java (1)
11-13: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winAdd
@Size(max = 100)toworkerIdto match the entity column length.Same as the
JobPostingWorkerFailureRequestDTO —JobPostingAsyncTask.workerIdis capped at 100 chars, so it's worth enforcing that at validation time here too.♻️ Proposed fix
`@Min`(0) int retryCount, - `@NotBlank` String workerId, + `@NotBlank` `@Size`(max = 100) String workerId, Long queueLatencyMillis🤖 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/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java` around lines 11 - 13, Add a size constraint to JobPostingWorkerRetryRequest.workerId so it matches the 100-character limit used by JobPostingAsyncTask and the similar JobPostingWorkerFailureRequest DTO. Update the workerId field annotation in JobPostingWorkerRetryRequest to include `@Size`(max = 100) alongside the existing non-blank validation, keeping validation aligned with the entity column length.src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java (1)
11-13: 🗄️ Data Integrity & Integration | 🔵 Trivial | ⚡ Quick winAdd
@Size(max = 100)toworkerIdto match the entity column length.
JobPostingAsyncTask.workerIdis@Column(name = "worker_id", length = 100). Without a matching@Sizeconstraint here, an oversized value fails at the DB layer (or gets silently truncated depending on DB config) instead of failing validation with a clear error.♻️ Proposed fix
`@Min`(0) int retryCount, - `@NotBlank` String workerId, + `@NotBlank` `@Size`(max = 100) String workerId, Long queueLatencyMillis🤖 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/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java` around lines 11 - 13, The workerId field in JobPostingWorkerFailureRequest only has `@NotBlank`, so oversized values can slip past validation and fail later against the JobPostingAsyncTask.workerId column length. Update the JobPostingWorkerFailureRequest record to add `@Size`(max = 100) on workerId alongside the existing validation annotations so the DTO matches the entity constraint and fails fast during validation.src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java (1)
22-33: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winA throwing job-posting sweep permanently starves the analysis sweep each cycle.
jobPostingAsyncTaskService.sweepTimedOutTasks()andanalysisAsyncSweepService.sweepTimedOutTasks()run sequentially with no error isolation. If the job-posting sweep throws (persistently, e.g. due to a malformed task or a bug), the analysis sweep never executes for that invocation — and since the same failure would recur every scheduled cycle, analysis tasks would never get swept until the underlying bug is fixed.Wrap each call independently so a failure in one domain doesn't block the other:
🛡️ Suggested isolation
public void sweepTimedOutTasks() { - int expiredJobPostingTasks = jobPostingAsyncTaskService.sweepTimedOutTasks(); - int expiredAnalysisTasks = analysisAsyncSweepService.sweepTimedOutTasks(); + int expiredJobPostingTasks = safeSweep("jobPosting", jobPostingAsyncTaskService::sweepTimedOutTasks); + int expiredAnalysisTasks = safeSweep("analysis", analysisAsyncSweepService::sweepTimedOutTasks); if (expiredJobPostingTasks > 0 || expiredAnalysisTasks > 0) { log.info( "Async task sweep expired tasks. jobPostingExpired={} analysisExpired={}", expiredJobPostingTasks, expiredAnalysisTasks ); } } + + private int safeSweep(String name, java.util.function.IntSupplier sweep) { + try { + return sweep.getAsInt(); + } catch (Exception e) { + log.error("Async task sweep failed for {}", name, e); + return 0; + } + }🤖 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/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java` around lines 22 - 33, The sweepTimedOutTasks method in AsyncTaskSweepScheduler runs jobPostingAsyncTaskService.sweepTimedOutTasks() and analysisAsyncSweepService.sweepTimedOutTasks() sequentially without isolation, so a failure in one can prevent the other from running. Update sweepTimedOutTasks to wrap each service call independently, handling/logging errors per call so a thrown exception from job posting does not block the analysis sweep (and vice versa). Keep the logging around expired counts in AsyncTaskSweepScheduler, but ensure both sweeps always get a chance to execute each cycle.src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java (1)
34-53: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winOne failing task rolls back the entire sweep batch.
sweepTimedOutTasks()processes all timed-out tasks inside a single@Transactionalmethod. If any single iteration throws (e.g.userService.getUserfails withUSER_NOT_FOUNDfor a staleuserId, or a DB constraint issue), the whole transaction rolls back — including themarkFailed/credit-release work already done for other, unrelated tasks in that same sweep cycle. This delays recovery of all currently-stuck tasks until the next scheduled run succeeds end-to-end.Consider processing each task in its own transaction (e.g. call a
@Transactionalper-task method from a non-transactional loop) and logging+skipping failures for individual tasks instead of failing the batch.🤖 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/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java` around lines 34 - 53, The sweep in AnalysisAsyncSweepService.sweepTimedOutTasks currently runs all expired-task handling inside one transaction, so a single exception can roll back credit release and markFailed work for every task in the batch. Refactor the loop so sweepTimedOutTasks is non-transactional and delegates each task to a separate `@Transactional` helper (for example, a per-task method that performs releaseCreditIfNeeded and analysisAsyncTaskService.markFailed), and wrap each iteration with try/catch to log the failure and continue processing the remaining AnalysisAsyncTask entries.
🤖 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
`@src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java`:
- Around line 17-24: The fetch and SSE subscription in
AnalysisAsyncSseService.subscribe are not atomic, so a task can finish after
AnalysisController.streamAnalysisTask reads initialStatus but before the emitter
is registered. Move the status lookup into the subscription flow or synchronize
the read/subscribe/publish path on the same channel key so the initial snapshot
and emitter registration happen under one lock. Use the existing
AnalysisAsyncSseService.subscribe, channelKey, and sseSubscriptionRegistry
interactions as the place to enforce this.
In
`@src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java`:
- Around line 34-53: The timeout sweep in
AnalysisAsyncSweepService.sweepTimedOutTasks() can race with concurrent task
completion because it scans PENDING/RUNNING tasks without any lock or version
check. Update the sweep so the mark-failed-and-refund path is atomic with
completion handling, using the AnalysisAsyncTask / analysisAsyncTaskRepository
flow to re-read or lock the task before calling releaseCreditIfNeeded() and
analysisAsyncTaskService.markFailed(). Ensure a task that has already been
completed by completeTask() cannot be refunded or overwritten as FAILED.
In
`@src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java`:
- Around line 52-78: The async status updates are being published too early in
AnalysisAsyncTaskService, so clients can receive a status change before the
transaction is actually committed. Update markRunning, markSuccess,
markRetryScheduled, and markFailed to defer analysisAsyncSseService.publish(...)
until after commit, using an afterCommit hook or a `@TransactionalEventListener`
with phase AFTER_COMMIT. Also apply the same pattern used by
sweepTimedOutTasks() so all SSE emissions only happen after durable state
changes.
In
`@src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java`:
- Around line 106-113: The credit history descriptions in AnalysisService are
still using reservation-era wording, so update the strings passed to
creditService.use in deductAnalysisCredit and creditService.refund in
refundAnalysisCredit to reflect deduct/refund semantics instead of “예약” and “예약
해제”. Keep the method names as-is and only change the transaction descriptions so
the logged history matches the current behavior.
In
`@src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java`:
- Around line 42-72: Add a transactional boundary to the bridge methods so the
worker metadata update and status transition commit together. In
JobPostingWorkerBridgeService, annotate markRetry and failTask with
`@Transactional` so the calls to jobPostingAsyncTaskService.updateWorkerMetadata,
markRetryScheduled, and markFailed participate in the same transaction instead
of separate ones.
In `@src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java`:
- Around line 70-88: The exception handling in SseSubscriptionRegistry.publish()
and sendHeartbeat() is too narrow because SseEmitter.send() can also throw
IllegalStateException, which can escape and disrupt callers. Broaden the catch
in publish(), sendHeartbeat(), and the related subscribe() cleanup path to
handle any send failure, then always complete and remove the affected emitter
from the registry so stale subscriptions do not keep failing on later publishes
or heartbeats. Ensure the fix is applied in the SseSubscriptionRegistry methods
that iterate emitters and call SseEmitter.send().
- Around line 24-42: The heartbeat dispatch in SseSubscriptionRegistry is
single-threaded, so one slow SseEmitter.send() can stall heartbeats for every
other subscriber. Update SseSubscriptionRegistry so sendHeartbeatSafely (and the
channel/emitter iteration it drives) does not run all blocking writes on the
lone heartbeatExecutor thread; dispatch each channel or emitter heartbeat to a
bounded worker pool, or otherwise enforce a strict timeout around the send. Keep
the heartbeat scheduling in the constructor, but make the actual heartbeat
delivery in sendHeartbeatSafely non-blocking for unrelated subscriptions.
---
Outside diff comments:
In
`@src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java`:
- Around line 72-88: The credit reservation logic in AnalysisService.analyze()
is running inside the outer transaction, so deductAnalysisCredit() and
refundAnalysisCredit() are not committed independently before executeAnalysis()
starts. Move the reservation/refund handling into its own transactional boundary
(for example via a separate method on AnalysisService or a dedicated helper) so
the balance change commits separately, and keep analyze() focused on
validateAnalysisRequest(), prepareAnalysisExecution(), executeAnalysis(), and
finalizeAnalysis().
In
`@src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java`:
- Around line 52-97: `JobPostingAsyncTaskService` is publishing SSE updates
before the enclosing transaction commits, so clients can observe uncommitted or
later-rolled-back state. Update `markRunning`, `markSuccess`,
`markRetryScheduled`, `markFailed`, and `sweepTimedOutTasks` to publish via a
post-commit hook instead of calling
`jobPostingAsyncSseService.publish(toStatusResponse(task))` directly. Add a
helper in this service (or equivalent) that registers `afterCommit()` with
`TransactionSynchronizationManager` and only publishes the
`toStatusResponse(task)` result once the transaction has successfully committed.
---
Nitpick comments:
In
`@src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java`:
- Around line 39-41: The terminal-status check in AnalysisAsyncSseService is
hardcoded with duplicated string literals, and the same logic exists in
JobPostingAsyncSseService. Replace the literal comparisons in isTerminal with
the shared domain TaskStatus enum (or a shared helper based on it) so status
checks are compile-time safe. If possible, extract the repeated
subscribe/publish/isTerminal flow into a small reusable helper used by both
services to remove the duplicated boilerplate.
In
`@src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java`:
- Around line 34-53: The sweep in AnalysisAsyncSweepService.sweepTimedOutTasks
currently runs all expired-task handling inside one transaction, so a single
exception can roll back credit release and markFailed work for every task in the
batch. Refactor the loop so sweepTimedOutTasks is non-transactional and
delegates each task to a separate `@Transactional` helper (for example, a per-task
method that performs releaseCreditIfNeeded and
analysisAsyncTaskService.markFailed), and wrap each iteration with try/catch to
log the failure and continue processing the remaining AnalysisAsyncTask entries.
In
`@src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java`:
- Around line 153-169: The credit lifecycle names in AnalysisWorkerBridgeService
still imply a reservation/hold even though the flow now performs an immediate
deduction via analysisService.deductAnalysisCredit(...). Update the helper names
and related status/marking methods to match the actual semantics (for example,
reserveCreditIfNeeded, markCreditReserved, and confirmCreditIfNeeded should
align with deduction/confirmation terminology), while keeping the existing
release/refund behavior consistent with CreditStatus and the task reference ID
handling.
In
`@src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java`:
- Around line 11-13: The workerId field in JobPostingWorkerFailureRequest only
has `@NotBlank`, so oversized values can slip past validation and fail later
against the JobPostingAsyncTask.workerId column length. Update the
JobPostingWorkerFailureRequest record to add `@Size`(max = 100) on workerId
alongside the existing validation annotations so the DTO matches the entity
constraint and fails fast during validation.
In
`@src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java`:
- Around line 11-13: Add a size constraint to
JobPostingWorkerRetryRequest.workerId so it matches the 100-character limit used
by JobPostingAsyncTask and the similar JobPostingWorkerFailureRequest DTO.
Update the workerId field annotation in JobPostingWorkerRetryRequest to include
`@Size`(max = 100) alongside the existing non-blank validation, keeping validation
aligned with the entity column length.
In
`@src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java`:
- Around line 126-136: `JobPostingAsyncTask.markFailed` has a weaker
terminal-state check than `markRetryScheduled`, so a second failure can
overwrite an already-FAILED task. Update `markFailed(FailureReason, String,
int)` to use the same terminal guard as `isTerminal()` (or equivalent) so it
no-ops for both SUCCEEDED and FAILED states, keeping `completedAt`, `error`,
`failureReason`, and `retryCount` stable on repeat calls.
In
`@src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java`:
- Around line 126-138: The sweep in
JobPostingAsyncTaskService.sweepTimedOutTasks() processes all PENDING/RUNNING
tasks in one transaction, which can become a long-running lock-heavy operation
as data grows. Update the repository call and loop to process tasks in bounded
batches, using a Pageable/limit or similar batch size per scheduler tick, and
keep the existing expireTimedOutTaskIfNeeded and publish(toStatusResponse(task))
flow unchanged for each processed task.
In
`@src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java`:
- Around line 86-95: The markConfirmationUnknown flow in
PaymentTransactionService only moves a payment to UNKNOWN and leaves no path to
recover stuck payments later. Update the payment lifecycle so UNKNOWN entries
can be reconciled automatically, ideally by adding a scheduled reconciliation
job that uses Toss inquiry by paymentKey/orderId and reuses the payment state
transitions in PaymentTransactionService. Make the fix centered around
markConfirmationUnknown and the existing payment status handling so abandoned
flows are eventually resolved without relying on the user to retry.
In
`@src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java`:
- Around line 22-33: The sweepTimedOutTasks method in AsyncTaskSweepScheduler
runs jobPostingAsyncTaskService.sweepTimedOutTasks() and
analysisAsyncSweepService.sweepTimedOutTasks() sequentially without isolation,
so a failure in one can prevent the other from running. Update
sweepTimedOutTasks to wrap each service call independently, handling/logging
errors per call so a thrown exception from job posting does not block the
analysis sweep (and vice versa). Keep the logging around expired counts in
AsyncTaskSweepScheduler, but ensure both sweeps always get a chance to execute
each cycle.
In
`@src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java`:
- Around line 138-149: The new isTerminal() guard in markRetryScheduled is not
covered by tests; add a test in JobPostingAsyncTaskServiceTest that calls
markRetryScheduled on an already terminal JobPostingAsyncTask (SUCCEEDED or
FAILED) and verifies it remains unchanged, similar to
AnalysisWorkerBridgeServiceTest.terminalTaskDoesNotReopen. Use the existing
markRetryScheduled and isTerminal behavior to keep the assertion focused on the
no-op path.
- Around line 118-136: The max-retry failure test in
JobPostingAsyncTaskServiceTest only checks the entity state transition; update
it to also verify that JobPostingAsyncSseService.publish(...) is invoked when
markRetryScheduled(...) auto-transitions a task to FAILED. Use the existing task
setup and add a verification for the publish call on the max-retry exhaustion
path so the test covers the new SSE behavior introduced by
JobPostingAsyncTaskService.markRetryScheduled.
In
`@src/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java`:
- Around line 186-221: Add a test to cover the UNKNOWN retry path when the
confirm request uses a different paymentKey than the stored payment. Extend
PaymentServiceTest around confirmAllowsRetryWhenTossConfirmTimesOut() to first
force a timeout, then call paymentService.confirm with a mismatched
PaymentConfirmRequest and assert GeneralException with
GeneralErrorCode.PAYMENT_ALREADY_PROCESSED. Use the existing
paymentService.confirm, TossPaymentConfirmResponse, and
paymentRepository.findByOrderId setup to exercise the startConfirmation branch
for UNKNOWN payments.
🪄 Autofix (Beta)
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: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: f62b624e-3010-4ee1-b143-dc56f4aff7ef
📒 Files selected for processing (36)
src/main/java/com/jobdri/jobdri_api/JobdriApiApplication.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/repository/AnalysisAsyncTaskRepository.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAdminDebugService.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAiClient.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeService.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncProcessor.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisExecutionPayload.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisImprovementRules.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisQueueProperties.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisTaskMessagePublisher.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.javasrc/main/java/com/jobdri/jobdri_api/domain/analysis/service/QuestionService.javasrc/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.javasrc/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingWorkerInternalController.javasrc/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.javasrc/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.javasrc/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.javasrc/main/java/com/jobdri/jobdri_api/domain/jobposting/repository/JobPostingAsyncTaskRepository.javasrc/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.javasrc/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.javasrc/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.javasrc/main/java/com/jobdri/jobdri_api/domain/payment/entity/Payment.javasrc/main/java/com/jobdri/jobdri_api/domain/payment/entity/PaymentStatus.javasrc/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentService.javasrc/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.javasrc/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.javasrc/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.javasrc/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeServiceTest.javasrc/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeServiceTest.javasrc/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.javasrc/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java
async SSE 발행을 트랜잭션 커밋 이후로 지연해 클라이언트가 미커밋 상태를 관측하지 않도록 수정하고, SSE 구독/하트비트 예외 처리와 채널별 동기화를 보강했습니다. 또한 analysis sweep를 per-task 트랜잭션으로 분리하고 scheduler 예외 격리를 추가했으며, 채용공고 worker retry/failed 트랜잭션 일관성 및 DTO 검증을 강화해 비동기 상태 복구 경로의 안정성을 높였습니다.:wq
analysis async stuck-task sweep에 per-task 트랜잭션을 적용하는 과정에서 발생한 순환 의존성을 제거했습니다. self-injection 대신 TransactionTemplate 기반으로 각 task를 개별 트랜잭션에서 처리하도록 변경해 스프링 컨텍스트 초기화 실패와 테스트 전반의 연쇄 실패를 해소했습니다.
[Feat] SSE 기반 알림 및 읽음 처리 기능 추가 (#120)
1 participant
✨ 어떤 이유로 PR를 하셨나요?
📋 세부 내용 - 왜 해당 PR이 필요한지 작업 내용을 자세하게 설명해주세요
기존 async 처리 구조는 채용공고와 자소서 분석이 모두 MQ 기반으로 동작하지만, 상태 전이와 복구 방식이 서로 달라 운영 일관성이 떨어졌습니다. 특히 채용공고 async는 retry 상태머신과 worker 메타데이터 반영이 부족했고, 프론트는 task 완료 여부를 폴링으로만 확인하고 있었습니다.
이번 PR에서는 다음 내용을 반영했습니다.
기대 효과:
📸 작업 화면 스크린샷
🚨 관련 이슈 번호 [ #109 ]