From 3784b044bfe1c7e96bc24184a310a6f42408a8ae Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Tue, 7 Jul 2026 14:36:07 +0900 Subject: [PATCH 1/7] =?UTF-8?q?[Refactor]=20=EA=B2=B0=EC=A0=9C=20UNKNOWN?= =?UTF-8?q?=20=EC=83=81=ED=83=9C=20=EB=8F=84=EC=9E=85=20=EB=B0=8F=20analys?= =?UTF-8?q?is=20=ED=81=AC=EB=A0=88=EB=94=A7=20=EB=A9=94=EC=84=9C=EB=93=9C?= =?UTF-8?q?=EB=AA=85=20=EC=A0=95=EB=A6=AC=20(#62)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../service/AnalysisAdminDebugService.java | 1 + .../analysis/service/AnalysisAiClient.java | 1 + .../service/AnalysisAsyncFacadeService.java | 1 + .../service/AnalysisAsyncProcessor.java | 1 + .../service/AnalysisAsyncTaskService.java | 1 + .../service/AnalysisExecutionPayload.java | 1 + .../service/AnalysisImprovementRules.java | 1 + .../service/AnalysisQueueProperties.java | 1 + .../analysis/service/AnalysisService.java | 17 +++++------------ .../service/AnalysisTaskMessagePublisher.java | 1 + .../service/AnalysisWorkerBridgeService.java | 7 +++---- .../analysis/service/QuestionService.java | 1 + .../domain/payment/entity/Payment.java | 5 ++--- .../domain/payment/entity/PaymentStatus.java | 1 + .../domain/payment/service/PaymentService.java | 2 +- .../service/PaymentTransactionService.java | 11 +++++++++-- .../service/AnalysisAsyncFacadeServiceTest.java | 4 ++-- .../AnalysisWorkerBridgeServiceTest.java | 4 ++-- .../payment/service/PaymentServiceTest.java | 4 ++-- 19 files changed, 37 insertions(+), 28 deletions(-) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAdminDebugService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAdminDebugService.java index f7bb3f2..d073b10 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAdminDebugService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAdminDebugService.java @@ -19,6 +19,7 @@ @Service @RequiredArgsConstructor @Transactional(readOnly = true) +// 관리자용으로 분석 참조 데이터와 retrieval 결과를 미리 점검하는 서비스다. public class AnalysisAdminDebugService { private final MockApplyRepository mockApplyRepository; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAiClient.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAiClient.java index 9f8e4ed..d58015a 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAiClient.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAiClient.java @@ -24,6 +24,7 @@ @Component @Slf4j @RequiredArgsConstructor +// 자소서 분석에 필요한 프롬프트를 만들고 LLM 호출을 수행하는 클라이언트다. public class AnalysisAiClient { private static final int MAX_REFERENCE_SECTION_LENGTH = 3000; private static final int MAX_REFERENCE_FIELD_LENGTH = 300; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeService.java index 3fa23e0..53d9678 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeService.java @@ -15,6 +15,7 @@ @Service @RequiredArgsConstructor +// 분석 비동기 작업의 접수와 상태 조회를 외부 API 관점에서 조율하는 서비스다. public class AnalysisAsyncFacadeService { private static final String ACTIVE_TASK_UNIQUE_CONSTRAINT = "uk_analysis_async_tasks_active_user_mock_apply"; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncProcessor.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncProcessor.java index 3b632c0..e113255 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncProcessor.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncProcessor.java @@ -6,6 +6,7 @@ @Service @RequiredArgsConstructor +// 분석 비동기 작업을 MQ 메시지로 변환해 워커 실행 경로로 넘기는 서비스다. public class AnalysisAsyncProcessor { private final AnalysisTaskMessagePublisher analysisTaskMessagePublisher; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java index 816cf5d..97d270c 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java @@ -19,6 +19,7 @@ @Service @RequiredArgsConstructor +// 분석 비동기 task 엔티티의 생성, 상태 전이, 조회를 전담하는 서비스다. public class AnalysisAsyncTaskService { private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisExecutionPayload.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisExecutionPayload.java index 5461bf8..078c996 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisExecutionPayload.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisExecutionPayload.java @@ -5,6 +5,7 @@ import java.util.List; +// 실제 분석 실행에 필요한 공고, 문항, 답변 데이터를 묶어 전달하는 payload다. public record AnalysisExecutionPayload( Long userId, Long mockApplyId, diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisImprovementRules.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisImprovementRules.java index 6f4c012..cee9ba6 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisImprovementRules.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisImprovementRules.java @@ -2,6 +2,7 @@ import java.util.List; +// 개선 문구를 후처리할 때 사용하는 내부 규칙 모음이다. final class AnalysisImprovementRules { private static final List BANNED_PHRASES = List.of( "추가하세요", diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisQueueProperties.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisQueueProperties.java index 6c155eb..f3462f8 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisQueueProperties.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisQueueProperties.java @@ -4,6 +4,7 @@ import org.springframework.stereotype.Component; @Component +// 분석 워커 큐 관련 설정값을 주입받아 묶어두는 설정 객체다. public record AnalysisQueueProperties( @Value("${app.worker.analysis.routing-key:analysis.execute}") String routingKey ) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java index 0385b48..5242f1c 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java @@ -47,6 +47,7 @@ @Slf4j @RequiredArgsConstructor @Transactional(readOnly = true) +// 자소서 분석의 핵심 비즈니스 로직과 결과 저장을 담당하는 메인 서비스다. public class AnalysisService { private static final int MIN_SCORE = 0; private static final int MAX_SCORE = 100; @@ -73,16 +74,15 @@ public class AnalysisService { public AnalysisResponse analyze(User user, Long mockApplyId) { validateAnalysisRequest(user, mockApplyId); String referenceId = "mockApplyId=" + mockApplyId; - reserveAnalysisCredit(user, referenceId); + deductAnalysisCredit(user, referenceId); try { AnalysisExecutionPayload payload = prepareAnalysisExecution(user, mockApplyId); AnalysisLlmResponse llmResponse = executeAnalysis(payload); AnalysisResponse response = finalizeAnalysis(user, mockApplyId, payload, llmResponse); - confirmAnalysisCredit(user, referenceId); return response; } catch (RuntimeException e) { - releaseAnalysisCredit(user, referenceId); + refundAnalysisCredit(user, referenceId); throw e; } } @@ -103,19 +103,12 @@ public void validateAnalysisRequest(User user, Long mockApplyId) { } @Transactional - public void reserveAnalysisCredit(User user, String referenceId) { + public void deductAnalysisCredit(User user, String referenceId) { creditService.use(user, 1, "자소서 분석 크레딧 예약", referenceId); } @Transactional - public void confirmAnalysisCredit(User user, String referenceId) { - if (user == null || referenceId == null) { - return; - } - } - - @Transactional - public void releaseAnalysisCredit(User user, String referenceId) { + public void refundAnalysisCredit(User user, String referenceId) { creditService.refund(user, 1, "자소서 분석 크레딧 예약 해제", referenceId); } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisTaskMessagePublisher.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisTaskMessagePublisher.java index 6a8e781..2104ad2 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisTaskMessagePublisher.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisTaskMessagePublisher.java @@ -9,6 +9,7 @@ @Service @RequiredArgsConstructor +// 분석 작업 메시지를 공통 worker exchange로 발행하는 publisher다. public class AnalysisTaskMessagePublisher { private final RabbitPublishSupport rabbitPublishSupport; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java index 1d28b78..d2a2121 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeService.java @@ -23,6 +23,7 @@ @Service @RequiredArgsConstructor +// 외부 분석 워커와 내부 분석 도메인 상태를 연결해 주는 브리지 서비스다. public class AnalysisWorkerBridgeService { private final AnalysisAsyncTaskService analysisAsyncTaskService; @@ -156,7 +157,7 @@ private void reserveCreditIfNeeded(AnalysisAsyncTask task) { User user = userService.getUser(task.getUserId()); String creditReferenceId = "analysisTaskId=" + task.getTaskId(); - analysisService.reserveAnalysisCredit(user, creditReferenceId); + analysisService.deductAnalysisCredit(user, creditReferenceId); analysisAsyncTaskService.markCreditReserved(task.getTaskId(), creditReferenceId); } @@ -164,8 +165,6 @@ private void confirmCreditIfNeeded(AnalysisAsyncTask task, User user) { if (task.getCreditStatus() != CreditStatus.RESERVED || task.getCreditReferenceId() == null) { return; } - - analysisService.confirmAnalysisCredit(user, task.getCreditReferenceId()); analysisAsyncTaskService.markCreditConfirmed(task.getTaskId()); } @@ -174,7 +173,7 @@ private void releaseCreditIfNeeded(AnalysisAsyncTask task) { return; } User user = userService.getUser(task.getUserId()); - analysisService.releaseAnalysisCredit(user, task.getCreditReferenceId()); + analysisService.refundAnalysisCredit(user, task.getCreditReferenceId()); analysisAsyncTaskService.markCreditReleased(task.getTaskId()); } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/QuestionService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/QuestionService.java index c355f83..f72e28f 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/QuestionService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/QuestionService.java @@ -34,6 +34,7 @@ @Service @RequiredArgsConstructor @Transactional(readOnly = true) +// 분석 전 단계에서 문항 선택, 추가, 답변 저장을 담당하는 서비스다. public class QuestionService { private static final int MIN_SELECTION_COUNT = 1; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/Payment.java b/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/Payment.java index 62b99f7..de03f3b 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/Payment.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/Payment.java @@ -81,9 +81,8 @@ public void fail() { this.status = PaymentStatus.FAILED; } - public void resetToPending() { - this.paymentKey = null; - this.status = PaymentStatus.PENDING; + public void markUnknown() { + this.status = PaymentStatus.UNKNOWN; } public boolean belongsTo(Long userId) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/PaymentStatus.java b/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/PaymentStatus.java index c274f83..03fbb0e 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/PaymentStatus.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/payment/entity/PaymentStatus.java @@ -6,6 +6,7 @@ public enum PaymentStatus { PENDING, PROCESSING, + UNKNOWN, FAILED, COMPLETED } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentService.java b/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentService.java index 47de7c1..96662a4 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentService.java @@ -83,7 +83,7 @@ public PaymentConfirmResponse confirm(User user, PaymentConfirmRequest request) validateTossResponse(request, tossResponse); } catch (GeneralException e) { if (e.getCode() == GeneralErrorCode.EXTERNAL_SERVICE_TIMEOUT) { - paymentTransactionService.rollbackConfirmationToPending( + paymentTransactionService.markConfirmationUnknown( validatedUser.getId(), request.orderId(), request.paymentKey() diff --git a/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java b/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java index 4ed0e4c..64c4edc 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/payment/service/PaymentTransactionService.java @@ -32,6 +32,13 @@ public PaymentConfirmationStart startConfirmation(Long userId, PaymentConfirmReq } return new PaymentConfirmationStart(payment, true); } + if (payment.getStatus() == PaymentStatus.UNKNOWN) { + if (!payment.hasPaymentKey(request.paymentKey())) { + throw new GeneralException(GeneralErrorCode.PAYMENT_ALREADY_PROCESSED, "이미 처리된 결제입니다."); + } + payment.markProcessing(request.paymentKey()); + return new PaymentConfirmationStart(payment, false); + } if (payment.getStatus() == PaymentStatus.PROCESSING || payment.getStatus() == PaymentStatus.FAILED) { throw new GeneralException(GeneralErrorCode.PAYMENT_ALREADY_PROCESSED, "이미 처리된 결제입니다."); } @@ -77,13 +84,13 @@ public void failConfirmation(Long userId, String orderId, String paymentKey) { } @Transactional - public void rollbackConfirmationToPending(Long userId, String orderId, String paymentKey) { + public void markConfirmationUnknown(Long userId, String orderId, String paymentKey) { Payment payment = getOwnedPaymentForUpdate(userId, orderId); if (!payment.hasPaymentKey(paymentKey)) { return; } if (payment.getStatus() == PaymentStatus.PROCESSING) { - payment.resetToPending(); + payment.markUnknown(); } } diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeServiceTest.java index 443b93d..a7c69be 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeServiceTest.java @@ -60,7 +60,7 @@ void submitReturnsExistingTaskWhenCreateConflicts() { assertThat(response.taskId()).isEqualTo(existingTask.getTaskId()); assertThat(response.status()).isEqualTo("PENDING"); - verify(analysisService, never()).reserveAnalysisCredit(eq(user), anyString()); + verify(analysisService, never()).deductAnalysisCredit(eq(user), anyString()); verify(analysisAsyncProcessor, never()).process(eq(existingTask.getTaskId()), eq(1L), eq(10L), eq(3)); } @@ -98,7 +98,7 @@ void submitDoesNotReserveCreditBeforeWorkerStarts() { assertThat(response.taskId()).isEqualTo(createdTask.getTaskId()); assertThat(response.status()).isEqualTo("PENDING"); verify(analysisAsyncProcessor, times(1)).process(createdTask.getTaskId(), 1L, 10L, 3); - verify(analysisService, never()).reserveAnalysisCredit(eq(user), anyString()); + verify(analysisService, never()).deductAnalysisCredit(eq(user), anyString()); } @Test diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeServiceTest.java index 84b6857..1656527 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisWorkerBridgeServiceTest.java @@ -79,7 +79,7 @@ void getContextReservesCreditBeforePreparingExecution() { analysisWorkerBridgeService.getContext(task.getTaskId(), 1L, 10L); - verify(analysisService).reserveAnalysisCredit(user, "analysisTaskId=" + task.getTaskId()); + verify(analysisService).deductAnalysisCredit(user, "analysisTaskId=" + task.getTaskId()); verify(analysisAsyncTaskService).markCreditReserved(task.getTaskId(), "analysisTaskId=" + task.getTaskId()); verify(analysisService).prepareAnalysisExecution(user, 10L); } @@ -117,7 +117,7 @@ void getContextDoesNotReserveCreditTwice() { analysisWorkerBridgeService.getContext(task.getTaskId(), 1L, 10L); - verify(analysisService, never()).reserveAnalysisCredit(eq(user), anyString()); + verify(analysisService, never()).deductAnalysisCredit(eq(user), anyString()); verify(analysisAsyncTaskService, never()).markCreditReserved(eq(task.getTaskId()), anyString()); verify(analysisService).prepareAnalysisExecution(user, 10L); } diff --git a/src/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java index bb0e9fa..21238a8 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/payment/service/PaymentServiceTest.java @@ -184,7 +184,7 @@ void confirmMarksPaymentAsFailedWhenTossConfirmFails() { } @Test - @DisplayName("토스 승인 타임아웃이면 결제를 PENDING으로 되돌리고 재시도를 허용한다") + @DisplayName("토스 승인 타임아웃이면 결제를 UNKNOWN으로 남기고 재시도를 허용한다") void confirmAllowsRetryWhenTossConfirmTimesOut() { User user = saveUser("payment-confirm-timeout@example.com"); PaymentPrepareResponse prepared = paymentService.prepare(user, new PaymentPrepareRequest("ONE_TIME")); @@ -208,7 +208,7 @@ void confirmAllowsRetryWhenTossConfirmTimesOut() { .isEqualTo(GeneralErrorCode.EXTERNAL_SERVICE_TIMEOUT); Payment timedOutPayment = paymentRepository.findByOrderId(prepared.orderId()).orElseThrow(); - assertThat(timedOutPayment.getStatus()).isEqualTo(PaymentStatus.PENDING); + assertThat(timedOutPayment.getStatus()).isEqualTo(PaymentStatus.UNKNOWN); PaymentConfirmResponse retryResponse = paymentService.confirm(user, request); From abb0a4b66cd23a6c5ffb4d796167eee8a29cdb86 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Tue, 7 Jul 2026 14:46:25 +0900 Subject: [PATCH 2/7] =?UTF-8?q?Fix:=20=EC=B1=84=EC=9A=A9=EA=B3=B5=EA=B3=A0?= =?UTF-8?q?=20async=20retry=20=EC=83=81=ED=83=9C=EB=A8=B8=EC=8B=A0=20?= =?UTF-8?q?=EB=B0=8F=20worker=20=EB=A9=94=ED=83=80=EB=8D=B0=EC=9D=B4?= =?UTF-8?q?=ED=84=B0=20=EB=B3=B4=EA=B0=95=20(#110)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../JobPostingWorkerInternalController.java | 18 ++++++++-- .../JobPostingWorkerFailureRequest.java | 4 ++- .../worker/JobPostingWorkerRetryRequest.java | 4 ++- .../entity/JobPostingAsyncTask.java | 23 ++++++++++++ .../service/JobPostingAsyncTaskService.java | 5 +++ .../JobPostingWorkerBridgeService.java | 36 +++++++++++++++++-- .../JobPostingAsyncTaskServiceTest.java | 32 +++++++++++++++++ 7 files changed, 116 insertions(+), 6 deletions(-) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingWorkerInternalController.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingWorkerInternalController.java index a5cfa8a..1d22206 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingWorkerInternalController.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingWorkerInternalController.java @@ -55,7 +55,14 @@ public ApiResponse markRetry( @Valid @RequestBody JobPostingWorkerRetryRequest request ) { internalApiKeyValidator.validate(internalApiKey); - jobPostingWorkerBridgeService.markRetry(taskId, request.failureReason(), request.errorMessage(), request.retryCount()); + jobPostingWorkerBridgeService.markRetry( + taskId, + request.failureReason(), + request.errorMessage(), + request.retryCount(), + request.workerId(), + request.queueLatencyMillis() + ); return ApiResponse.onSuccess("채용 공고 worker 작업 재시도 상태를 반영했습니다."); } @@ -79,7 +86,14 @@ public ApiResponse failTask( @Valid @RequestBody JobPostingWorkerFailureRequest request ) { internalApiKeyValidator.validate(internalApiKey); - jobPostingWorkerBridgeService.failTask(taskId, request.failureReason(), request.errorMessage(), request.retryCount()); + jobPostingWorkerBridgeService.failTask( + taskId, + request.failureReason(), + request.errorMessage(), + request.retryCount(), + request.workerId(), + request.queueLatencyMillis() + ); return ApiResponse.onSuccess("채용 공고 worker 작업 실패 상태를 반영했습니다."); } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java index 3305fef..db9f8e3 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java @@ -8,6 +8,8 @@ public record JobPostingWorkerFailureRequest( @NotNull FailureReason failureReason, @NotBlank String errorMessage, - @Min(0) int retryCount + @Min(0) int retryCount, + @NotBlank String workerId, + Long queueLatencyMillis ) { } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java index 345b49f..ca17138 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java @@ -8,6 +8,8 @@ public record JobPostingWorkerRetryRequest( @NotNull FailureReason failureReason, @NotBlank String errorMessage, - @Min(0) int retryCount + @Min(0) int retryCount, + @NotBlank String workerId, + Long queueLatencyMillis ) { } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java index a5d811e..c670b55 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java @@ -108,6 +108,13 @@ public void markSuccess(String resultPayload) { } public void markRetryScheduled(FailureReason failureReason, String errorMessage, int retryCount) { + if (isTerminal()) { + return; + } + if (retryCount >= maxRetryCount) { + markFailed(failureReason, errorMessage, retryCount); + return; + } this.status = TaskStatus.PENDING; this.message = "채용 공고 비동기 재시도를 대기 중입니다."; this.failureReason = failureReason; @@ -117,6 +124,9 @@ public void markRetryScheduled(FailureReason failureReason, String errorMessage, } public void markFailed(FailureReason failureReason, String errorMessage, int retryCount) { + if (status == TaskStatus.SUCCEEDED) { + return; + } this.status = TaskStatus.FAILED; this.message = "채용 공고 비동기 처리에 실패했습니다."; this.failureReason = failureReason; @@ -125,6 +135,19 @@ public void markFailed(FailureReason failureReason, String errorMessage, int ret this.completedAt = LocalDateTime.now(); } + public void updateWorkerMetadata(String workerId, Long queueLatencyMillis) { + if (workerId != null && !workerId.isBlank()) { + this.workerId = workerId; + } + if (queueLatencyMillis != null) { + this.queueLatencyMillis = Math.max(0L, queueLatencyMillis); + } + } + + private boolean isTerminal() { + return status == TaskStatus.SUCCEEDED || status == TaskStatus.FAILED; + } + public enum FailureReason { RATE_LIMIT, QUEUE_TIMEOUT, diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java index 4a4fd79..3135870 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java @@ -90,6 +90,11 @@ public void markFailed(String taskId, FailureReason failureReason, String errorM task.markFailed(failureReason, errorMessage, retryCount); } + @Transactional + public void updateWorkerMetadata(String taskId, String workerId, Long queueLatencyMillis) { + getTaskState(taskId).updateWorkerMetadata(workerId, queueLatencyMillis); + } + @Transactional public JobPostingAsyncStatusResponse getTask(String taskId) { JobPostingAsyncTask taskState = getTaskState(taskId); diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java index 768bf97..63c5cff 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java @@ -39,11 +39,35 @@ public JobPostingIngestResponse completeTask(String taskId, JobPostingIngestResp return jobPostingAsyncTaskService.markSuccess(taskId, result); } - public void markRetry(String taskId, FailureReason failureReason, String errorMessage, int retryCount) { + public void markRetry( + String taskId, + FailureReason failureReason, + String errorMessage, + int retryCount, + String workerId, + Long queueLatencyMillis + ) { + JobPostingAsyncTask task = getTask(taskId); + if (task.getStatus() == TaskStatus.SUCCEEDED || task.getStatus() == TaskStatus.FAILED) { + return; + } + jobPostingAsyncTaskService.updateWorkerMetadata(taskId, workerId, queueLatencyMillis); jobPostingAsyncTaskService.markRetryScheduled(taskId, failureReason, errorMessage, retryCount); } - public void failTask(String taskId, FailureReason failureReason, String errorMessage, int retryCount) { + public void failTask( + String taskId, + FailureReason failureReason, + String errorMessage, + int retryCount, + String workerId, + Long queueLatencyMillis + ) { + JobPostingAsyncTask task = getTask(taskId); + if (task.getStatus() == TaskStatus.SUCCEEDED || task.getStatus() == TaskStatus.FAILED) { + return; + } + jobPostingAsyncTaskService.updateWorkerMetadata(taskId, workerId, queueLatencyMillis); jobPostingAsyncTaskService.markFailed(taskId, failureReason, errorMessage, retryCount); } @@ -104,6 +128,14 @@ public JobPostingIngestResponse finalizeAndComplete( return jobPostingAsyncTaskService.markSuccess(taskId, result); } + private JobPostingAsyncTask getTask(String taskId) { + return jobPostingAsyncTaskRepository.findById(taskId) + .orElseThrow(() -> new GeneralException( + GeneralErrorCode.JOB_POSTING_ASYNC_TASK_NOT_FOUND, + "해당 비동기 작업을 찾을 수 없습니다. taskId=" + taskId + )); + } + private String fallbackCompanyName(String companyName) { if (companyName == null || companyName.isBlank()) { return "미분류 회사"; diff --git a/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java index 844754d..ecf3a5e 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java @@ -108,4 +108,36 @@ void getTaskMarksRunningTimeout() { assertThat(response.getStatus()).isEqualTo("FAILED"); assertThat(response.getFailureReason()).isEqualTo(FailureReason.WORKER_TIMEOUT.name()); } + + @Test + @DisplayName("재시도 횟수가 최대값에 도달하면 FAILED로 전이한다") + void markRetryScheduledFailsWhenMaxRetryReached() { + JobPostingAsyncTask task = JobPostingAsyncTask.pending(7L, 3); + + when(jobPostingAsyncTaskRepository.findById(task.getTaskId())).thenReturn(Optional.of(task)); + + jobPostingAsyncTaskService.markRetryScheduled( + task.getTaskId(), + FailureReason.INTERNAL_ERROR, + "retry exhausted", + 3 + ); + + assertThat(task.getStatus()).isEqualTo(JobPostingAsyncTask.TaskStatus.FAILED); + assertThat(task.getFailureReason()).isEqualTo(FailureReason.INTERNAL_ERROR); + assertThat(task.getRetryCount()).isEqualTo(3); + } + + @Test + @DisplayName("worker 메타데이터를 별도로 갱신할 수 있다") + void updateWorkerMetadataUpdatesWorkerFields() { + JobPostingAsyncTask task = JobPostingAsyncTask.pending(7L, 3); + + when(jobPostingAsyncTaskRepository.findById(task.getTaskId())).thenReturn(Optional.of(task)); + + jobPostingAsyncTaskService.updateWorkerMetadata(task.getTaskId(), "worker-2", 1234L); + + assertThat(task.getWorkerId()).isEqualTo("worker-2"); + assertThat(task.getQueueLatencyMillis()).isEqualTo(1234L); + } } From 4977edd836a61b05e41fe78f9a935e4c5dea6f71 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Tue, 7 Jul 2026 14:59:32 +0900 Subject: [PATCH 3/7] =?UTF-8?q?[Feat]=20async=20task=20SSE=20=EC=83=81?= =?UTF-8?q?=ED=83=9C=20=EC=8A=A4=ED=8A=B8=EB=A6=BC=20=EC=B6=94=EA=B0=80=20?= =?UTF-8?q?(#109)=20=EC=B1=84=EC=9A=A9=EA=B3=B5=EA=B3=A0=20=EB=B0=8F=20?= =?UTF-8?q?=EC=9E=90=EC=86=8C=EC=84=9C=20=EB=B6=84=EC=84=9D=20async=20?= =?UTF-8?q?=EC=83=81=ED=83=9C=20=EC=A1=B0=ED=9A=8C=EB=A5=BC=20=ED=8F=B4?= =?UTF-8?q?=EB=A7=81=20=EB=B0=A9=EC=8B=9D=EC=97=90=EC=84=9C=20SSE=20?= =?UTF-8?q?=EA=B8=B0=EB=B0=98=20=EC=8A=A4=ED=8A=B8=EB=A6=BC=20=EB=B0=A9?= =?UTF-8?q?=EC=8B=9D=EC=9C=BC=EB=A1=9C=20=ED=99=95=EC=9E=A5=ED=96=88?= =?UTF-8?q?=EC=8A=B5=EB=8B=88=EB=8B=A4.=20=EA=B3=B5=ED=86=B5=20SSE=20emitt?= =?UTF-8?q?er=20=EB=A0=88=EC=A7=80=EC=8A=A4=ED=8A=B8=EB=A6=AC=EB=A5=BC=20?= =?UTF-8?q?=EC=B6=94=EA=B0=80=ED=95=98=EA=B3=A0,=20async=20task=20?= =?UTF-8?q?=EC=83=81=ED=83=9C=20=EB=B3=80=EA=B2=BD=20=EC=8B=9C=EC=A0=90?= =?UTF-8?q?=EB=A7=88=EB=8B=A4=20=EC=A6=89=EC=8B=9C=20=EC=9D=B4=EB=B2=A4?= =?UTF-8?q?=ED=8A=B8=EB=A5=BC=20=EB=B0=9C=ED=96=89=ED=95=98=EB=8F=84?= =?UTF-8?q?=EB=A1=9D=20=EC=97=B0=EA=B2=B0=ED=95=B4=20=ED=94=84=EB=A1=A0?= =?UTF-8?q?=ED=8A=B8=EA=B0=80=20=EC=8B=A4=EC=8B=9C=EA=B0=84=EC=9C=BC?= =?UTF-8?q?=EB=A1=9C=20=EC=83=81=ED=83=9C=EB=A5=BC=20=EB=B0=98=EC=98=81?= =?UTF-8?q?=ED=95=A0=20=EC=88=98=20=EC=9E=88=EA=B2=8C=20=ED=96=88=EC=8A=B5?= =?UTF-8?q?=EB=8B=88=EB=8B=A4.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/AnalysisController.java | 16 ++++ .../service/AnalysisAsyncSseService.java | 42 ++++++++++ .../service/AnalysisAsyncTaskService.java | 17 +++- .../controller/JobPostingAiController.java | 17 ++++ .../service/JobPostingAsyncSseService.java | 42 ++++++++++ .../service/JobPostingAsyncTaskService.java | 5 ++ .../global/sse/SseSubscriptionRegistry.java | 79 +++++++++++++++++++ .../JobPostingAsyncTaskServiceTest.java | 9 ++- 8 files changed, 222 insertions(+), 5 deletions(-) create mode 100644 src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java create mode 100644 src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java create mode 100644 src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java index 23b0585..b202304 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java @@ -4,6 +4,7 @@ import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncSubmitResponse; import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisResponse; import com.jobdri.jobdri_api.domain.analysis.service.AnalysisAsyncFacadeService; +import com.jobdri.jobdri_api.domain.analysis.service.AnalysisAsyncSseService; import com.jobdri.jobdri_api.domain.analysis.service.AnalysisService; import com.jobdri.jobdri_api.global.apiPayload.ApiResponse; import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode; @@ -13,11 +14,13 @@ import io.swagger.v3.oas.annotations.tags.Tag; import lombok.RequiredArgsConstructor; import org.springframework.security.core.annotation.AuthenticationPrincipal; +import org.springframework.http.MediaType; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.PathVariable; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; @RestController @RequiredArgsConstructor @@ -27,6 +30,7 @@ public class AnalysisController { private final AnalysisService analysisService; private final AnalysisAsyncFacadeService analysisAsyncFacadeService; + private final AnalysisAsyncSseService analysisAsyncSseService; @Operation(summary = "자소서 분석 비동기 실행", description = "저장된 문항 답변과 공고 정보를 기반으로 자소서 분석 작업을 접수합니다.") @PostMapping @@ -53,6 +57,18 @@ public ApiResponse getAnalysisTask( ); } + @Operation(summary = "자소서 분석 비동기 작업 상태 SSE 구독", description = "taskId로 자소서 분석 비동기 작업 상태를 SSE 스트림으로 구독합니다.") + @GetMapping(value = "/async/{taskId}/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter streamAnalysisTask( + @AuthenticationPrincipal UserDetailsImpl userDetails, + @PathVariable Long mockApplyId, + @PathVariable String taskId + ) { + AnalysisAsyncStatusResponse initialStatus = + analysisAsyncFacadeService.getTask(getAuthenticatedUser(userDetails), mockApplyId, taskId); + return analysisAsyncSseService.subscribe(initialStatus); + } + @Operation(summary = "자소서 분석 결과 조회", description = "저장된 자소서 분석 결과를 조회합니다.") @GetMapping public ApiResponse getAnalysis( diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java new file mode 100644 index 0000000..6324635 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java @@ -0,0 +1,42 @@ +package com.jobdri.jobdri_api.domain.analysis.service; + +import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncStatusResponse; +import com.jobdri.jobdri_api.global.sse.SseSubscriptionRegistry; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +@Service +@RequiredArgsConstructor +public class AnalysisAsyncSseService { + + private static final String EVENT_NAME = "analysis-status"; + + private final SseSubscriptionRegistry sseSubscriptionRegistry; + + public SseEmitter subscribe(AnalysisAsyncStatusResponse initialStatus) { + return sseSubscriptionRegistry.subscribe( + channelKey(initialStatus.taskId()), + EVENT_NAME, + initialStatus, + isTerminal(initialStatus) + ); + } + + public void publish(AnalysisAsyncStatusResponse statusResponse) { + sseSubscriptionRegistry.publish( + channelKey(statusResponse.taskId()), + EVENT_NAME, + statusResponse, + isTerminal(statusResponse) + ); + } + + private String channelKey(String taskId) { + return "analysis:" + taskId; + } + + private boolean isTerminal(AnalysisAsyncStatusResponse statusResponse) { + return "SUCCEEDED".equals(statusResponse.status()) || "FAILED".equals(statusResponse.status()); + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java index 97d270c..d21f276 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java @@ -23,6 +23,7 @@ public class AnalysisAsyncTaskService { private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository; + private final AnalysisAsyncSseService analysisAsyncSseService; @Value("${app.worker.analysis.max-retry-count:3}") private int maxRetryCount; @@ -50,22 +51,30 @@ public Optional findActiveTask(Long userId, Long mockApplyId) @Transactional public void markRunning(String taskId, String workerId, int retryCount, Instant submittedAt) { - getTask(taskId).markRunning(workerId, retryCount, submittedAt); + AnalysisAsyncTask task = getTask(taskId); + task.markRunning(workerId, retryCount, submittedAt); + analysisAsyncSseService.publish(toStatusResponse(task)); } @Transactional public void markSuccess(String taskId) { - getTask(taskId).markSuccess(); + AnalysisAsyncTask task = getTask(taskId); + task.markSuccess(); + analysisAsyncSseService.publish(toStatusResponse(task)); } @Transactional public void markRetryScheduled(String taskId, FailureReason failureReason, String errorMessage, int retryCount) { - getTask(taskId).markRetryScheduled(failureReason, errorMessage, retryCount); + AnalysisAsyncTask task = getTask(taskId); + task.markRetryScheduled(failureReason, errorMessage, retryCount); + analysisAsyncSseService.publish(toStatusResponse(task)); } @Transactional public void markFailed(String taskId, FailureReason failureReason, String errorMessage, int retryCount) { - getTask(taskId).markFailed(failureReason, errorMessage, retryCount); + AnalysisAsyncTask task = getTask(taskId); + task.markFailed(failureReason, errorMessage, retryCount); + analysisAsyncSseService.publish(toStatusResponse(task)); } @Transactional diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java index 3e3fedb..c3137b3 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java @@ -7,6 +7,7 @@ import com.jobdri.jobdri_api.domain.jobposting.dto.response.JobPostingExtractResponse; import com.jobdri.jobdri_api.domain.jobposting.service.JobPostingAiService; import com.jobdri.jobdri_api.domain.jobposting.service.JobPostingAsyncFacadeService; +import com.jobdri.jobdri_api.domain.jobposting.service.JobPostingAsyncSseService; import com.jobdri.jobdri_api.domain.user.service.UserService; import com.jobdri.jobdri_api.global.apiPayload.ApiResponse; import com.jobdri.jobdri_api.global.security.UserDetailsImpl; @@ -20,6 +21,7 @@ import lombok.RequiredArgsConstructor; import org.springframework.http.MediaType; import org.springframework.security.core.annotation.AuthenticationPrincipal; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; @@ -35,6 +37,7 @@ public class JobPostingAiController { private final JobPostingAiService jobPostingAiService; private final JobPostingAsyncFacadeService jobPostingAsyncFacadeService; + private final JobPostingAsyncSseService jobPostingAsyncSseService; private final UserService userService; @Operation( @@ -108,6 +111,20 @@ public ApiResponse getIngestJobPostingAsyncStatus ); } + @Operation( + summary = "채용 공고 비동기 작업 상태 SSE 구독", + description = "taskId로 비동기 작업 상태를 SSE 스트림으로 구독합니다." + ) + @GetMapping(value = "/ingest/async/{taskId}/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) + public SseEmitter streamIngestJobPostingAsyncStatus( + @AuthenticationPrincipal UserDetailsImpl userDetails, + @PathVariable String taskId + ) { + var user = validateAuthenticatedUser(userDetails); + JobPostingAsyncStatusResponse initialStatus = jobPostingAsyncFacadeService.getTask(user, taskId); + return jobPostingAsyncSseService.subscribe(initialStatus); + } + private com.jobdri.jobdri_api.domain.user.entity.User validateAuthenticatedUser(UserDetailsImpl userDetails) { return userService.validateUser(userDetails == null ? null : userDetails.getUser()); } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java new file mode 100644 index 0000000..94dbc0e --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java @@ -0,0 +1,42 @@ +package com.jobdri.jobdri_api.domain.jobposting.service; + +import com.jobdri.jobdri_api.domain.jobposting.dto.response.JobPostingAsyncStatusResponse; +import com.jobdri.jobdri_api.global.sse.SseSubscriptionRegistry; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +@Service +@RequiredArgsConstructor +public class JobPostingAsyncSseService { + + private static final String EVENT_NAME = "job-posting-status"; + + private final SseSubscriptionRegistry sseSubscriptionRegistry; + + public SseEmitter subscribe(JobPostingAsyncStatusResponse initialStatus) { + return sseSubscriptionRegistry.subscribe( + channelKey(initialStatus.getTaskId()), + EVENT_NAME, + initialStatus, + isTerminal(initialStatus) + ); + } + + public void publish(JobPostingAsyncStatusResponse statusResponse) { + sseSubscriptionRegistry.publish( + channelKey(statusResponse.getTaskId()), + EVENT_NAME, + statusResponse, + isTerminal(statusResponse) + ); + } + + private String channelKey(String taskId) { + return "job-posting:" + taskId; + } + + private boolean isTerminal(JobPostingAsyncStatusResponse statusResponse) { + return "SUCCEEDED".equals(statusResponse.getStatus()) || "FAILED".equals(statusResponse.getStatus()); + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java index 3135870..aa8c6d4 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java @@ -27,6 +27,7 @@ public class JobPostingAsyncTaskService { private final JobPostingAsyncTaskRepository jobPostingAsyncTaskRepository; private final ObjectMapper objectMapper; + private final JobPostingAsyncSseService jobPostingAsyncSseService; @Value("${app.worker.job-posting.max-retry-count:3}") private int maxRetryCount; @@ -54,6 +55,7 @@ public void markRunning(String taskId, String workerId, int retryCount, Instant return; } task.markRunning(workerId, retryCount, submittedAt); + jobPostingAsyncSseService.publish(toStatusResponse(task)); } @Transactional @@ -69,6 +71,7 @@ public JobPostingIngestResponse markSuccess(String taskId, JobPostingIngestRespo ); } task.markSuccess(serializeResult(result)); + jobPostingAsyncSseService.publish(toStatusResponse(task)); return result; } @@ -79,6 +82,7 @@ public void markRetryScheduled(String taskId, FailureReason failureReason, Strin return; } task.markRetryScheduled(failureReason, errorMessage, retryCount); + jobPostingAsyncSseService.publish(toStatusResponse(task)); } @Transactional @@ -88,6 +92,7 @@ public void markFailed(String taskId, FailureReason failureReason, String errorM return; } task.markFailed(failureReason, errorMessage, retryCount); + jobPostingAsyncSseService.publish(toStatusResponse(task)); } @Transactional diff --git a/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java b/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java new file mode 100644 index 0000000..0b35047 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java @@ -0,0 +1,79 @@ +package com.jobdri.jobdri_api.global.sse; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; + +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.CopyOnWriteArrayList; + +@Component +public class SseSubscriptionRegistry { + + private final Map> emitters = new ConcurrentHashMap<>(); + private final long timeoutMillis; + + public SseSubscriptionRegistry(@Value("${app.sse.timeout-millis:1800000}") long timeoutMillis) { + this.timeoutMillis = timeoutMillis; + } + + public SseEmitter subscribe(String channelKey, String eventName, Object initialPayload, boolean completeAfterInitial) { + SseEmitter emitter = new SseEmitter(timeoutMillis); + emitters.computeIfAbsent(channelKey, ignored -> new CopyOnWriteArrayList<>()).add(emitter); + + emitter.onCompletion(() -> remove(channelKey, emitter)); + emitter.onTimeout(() -> { + emitter.complete(); + remove(channelKey, emitter); + }); + emitter.onError(ignored -> remove(channelKey, emitter)); + + try { + emitter.send(SseEmitter.event().name("connected").data("connected")); + emitter.send(SseEmitter.event().name(eventName).data(initialPayload)); + if (completeAfterInitial) { + emitter.complete(); + remove(channelKey, emitter); + } + } catch (IOException e) { + emitter.completeWithError(e); + remove(channelKey, emitter); + } + + return emitter; + } + + public void publish(String channelKey, String eventName, Object payload, boolean completeAfterPublish) { + List subscribers = emitters.get(channelKey); + if (subscribers == null || subscribers.isEmpty()) { + return; + } + + for (SseEmitter emitter : subscribers) { + try { + emitter.send(SseEmitter.event().name(eventName).data(payload)); + if (completeAfterPublish) { + emitter.complete(); + remove(channelKey, emitter); + } + } catch (IOException e) { + emitter.completeWithError(e); + remove(channelKey, emitter); + } + } + } + + private void remove(String channelKey, SseEmitter emitter) { + List subscribers = emitters.get(channelKey); + if (subscribers == null) { + return; + } + subscribers.remove(emitter); + if (subscribers.isEmpty()) { + emitters.remove(channelKey); + } + } +} diff --git a/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java index ecf3a5e..a01e644 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java @@ -28,11 +28,18 @@ class JobPostingAsyncTaskServiceTest { @Mock private JobPostingAsyncTaskRepository jobPostingAsyncTaskRepository; + @Mock + private JobPostingAsyncSseService jobPostingAsyncSseService; + private JobPostingAsyncTaskService jobPostingAsyncTaskService; @BeforeEach void setUp() { - jobPostingAsyncTaskService = new JobPostingAsyncTaskService(jobPostingAsyncTaskRepository, new ObjectMapper()); + jobPostingAsyncTaskService = new JobPostingAsyncTaskService( + jobPostingAsyncTaskRepository, + new ObjectMapper(), + jobPostingAsyncSseService + ); } @Test From f387d0e6b112d641766d2c8cdc669a176ef24ea5 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Tue, 7 Jul 2026 15:02:06 +0900 Subject: [PATCH 4/7] =?UTF-8?q?[Feat]=20SSE=20=ED=95=98=ED=8A=B8=EB=B9=84?= =?UTF-8?q?=ED=8A=B8=20=EC=9D=B4=EB=B2=A4=ED=8A=B8=20=EC=B6=94=EA=B0=80=20?= =?UTF-8?q?(#109)=20SSE=20=EA=B8=B0=EB=B0=98=20async=20=EC=83=81=ED=83=9C?= =?UTF-8?q?=20=EC=8A=A4=ED=8A=B8=EB=A6=BC=EC=97=90=20=EC=A3=BC=EA=B8=B0?= =?UTF-8?q?=EC=A0=81=EC=9D=B8=20heartbeat=20=EC=9D=B4=EB=B2=A4=ED=8A=B8?= =?UTF-8?q?=EB=A5=BC=20=EC=B6=94=EA=B0=80=ED=95=B4=20=EC=9E=A5=EC=8B=9C?= =?UTF-8?q?=EA=B0=84=20=EC=9C=A0=EC=A7=80=EB=90=98=EB=8A=94=20=EC=97=B0?= =?UTF-8?q?=EA=B2=B0=EC=9D=B4=20=ED=94=84=EB=A1=9D=EC=8B=9C=EB=82=98=20?= =?UTF-8?q?=EB=B8=8C=EB=9D=BC=EC=9A=B0=EC=A0=80=EC=97=90=EC=84=9C=20?= =?UTF-8?q?=EC=9C=A0=ED=9C=B4=20=EC=97=B0=EA=B2=B0=EB=A1=9C=20=EB=81=8A?= =?UTF-8?q?=EA=B8=B0=EC=A7=80=20=EC=95=8A=EB=8F=84=EB=A1=9D=20=EB=B3=B4?= =?UTF-8?q?=EA=B0=95=ED=96=88=EC=8A=B5=EB=8B=88=EB=8B=A4.=20=EA=B3=B5?= =?UTF-8?q?=ED=86=B5=20SSE=20=EB=A0=88=EC=A7=80=EC=8A=A4=ED=8A=B8=EB=A6=AC?= =?UTF-8?q?=EC=97=90=EC=84=9C=20heartbeat=EB=A5=BC=20=EA=B4=80=EB=A6=AC?= =?UTF-8?q?=ED=95=98=EB=8F=84=EB=A1=9D=20=EA=B5=AC=EC=84=B1=ED=95=B4=20?= =?UTF-8?q?=EC=B1=84=EC=9A=A9=EA=B3=B5=EA=B3=A0=20=EB=B0=8F=20=EC=9E=90?= =?UTF-8?q?=EC=86=8C=EC=84=9C=20=EB=B6=84=EC=84=9D=20=EC=8A=A4=ED=8A=B8?= =?UTF-8?q?=EB=A6=BC=EC=97=90=20=EB=8F=99=EC=9D=BC=ED=95=98=EA=B2=8C=20?= =?UTF-8?q?=EC=A0=81=EC=9A=A9=ED=96=88=EC=8A=B5=EB=8B=88=EB=8B=A4.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../global/sse/SseSubscriptionRegistry.java | 50 ++++++++++++++++++- 1 file changed, 49 insertions(+), 1 deletion(-) diff --git a/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java b/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java index 0b35047..170d049 100644 --- a/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java +++ b/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java @@ -1,5 +1,6 @@ package com.jobdri.jobdri_api.global.sse; +import jakarta.annotation.PreDestroy; import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; @@ -9,15 +10,35 @@ import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; @Component public class SseSubscriptionRegistry { private final Map> emitters = new ConcurrentHashMap<>(); private final long timeoutMillis; + private final ScheduledExecutorService heartbeatExecutor; - public SseSubscriptionRegistry(@Value("${app.sse.timeout-millis:1800000}") long timeoutMillis) { + public SseSubscriptionRegistry( + @Value("${app.sse.timeout-millis:1800000}") long timeoutMillis, + @Value("${app.sse.heartbeat-interval-millis:15000}") long heartbeatIntervalMillis + ) { this.timeoutMillis = timeoutMillis; + this.heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> { + Thread thread = new Thread(runnable, "sse-heartbeat"); + thread.setDaemon(true); + return thread; + }); + if (heartbeatIntervalMillis > 0) { + this.heartbeatExecutor.scheduleAtFixedRate( + this::sendHeartbeatSafely, + heartbeatIntervalMillis, + heartbeatIntervalMillis, + TimeUnit.MILLISECONDS + ); + } } public SseEmitter subscribe(String channelKey, String eventName, Object initialPayload, boolean completeAfterInitial) { @@ -66,6 +87,33 @@ public void publish(String channelKey, String eventName, Object payload, boolean } } + @PreDestroy + void shutdown() { + heartbeatExecutor.shutdownNow(); + } + + private void sendHeartbeatSafely() { + try { + sendHeartbeat(); + } catch (Exception ignored) { + // Heartbeat failures should not affect the rest of the application. + } + } + + private void sendHeartbeat() { + for (Map.Entry> entry : emitters.entrySet()) { + String channelKey = entry.getKey(); + for (SseEmitter emitter : entry.getValue()) { + try { + emitter.send(SseEmitter.event().name("heartbeat").data("ping")); + } catch (IOException e) { + emitter.completeWithError(e); + remove(channelKey, emitter); + } + } + } + } + private void remove(String channelKey, SseEmitter emitter) { List subscribers = emitters.get(channelKey); if (subscribers == null) { From 0faa17a734ca3a5c363e9bbfa2782dc441d34df4 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Tue, 7 Jul 2026 15:14:31 +0900 Subject: [PATCH 5/7] =?UTF-8?q?[Fix]=20async=20stuck=20task=20sweep=20?= =?UTF-8?q?=EB=B0=8F=20timeout=20=EB=B3=B5=EA=B5=AC=20=EB=A1=9C=EC=A7=81?= =?UTF-8?q?=20=EC=B6=94=EA=B0=80=20(#109)=20=EB=B9=84=EB=8F=99=EA=B8=B0=20?= =?UTF-8?q?task=EA=B0=80=20PENDING=20=EB=98=90=EB=8A=94=20RUNNING=20?= =?UTF-8?q?=EC=83=81=ED=83=9C=EC=97=90=EC=84=9C=20=EC=9E=A5=EC=8B=9C?= =?UTF-8?q?=EA=B0=84=20=EB=A9=88=EC=B6=B0=20=EC=9E=88=EC=9D=84=20=EB=95=8C?= =?UTF-8?q?=20=EC=A3=BC=EA=B8=B0=EC=A0=81=EC=9C=BC=EB=A1=9C=20timeout=20?= =?UTF-8?q?=EC=97=AC=EB=B6=80=EB=A5=BC=20=EC=A0=90=EA=B2=80=ED=95=B4=20FAI?= =?UTF-8?q?LED=EB=A1=9C=20=EC=A0=95=EB=A6=AC=ED=95=98=EB=8A=94=20stuck-tas?= =?UTF-8?q?k=20sweep=EC=9D=84=20=EC=B6=94=EA=B0=80=ED=96=88=EC=8A=B5?= =?UTF-8?q?=EB=8B=88=EB=8B=A4.=20=EC=B1=84=EC=9A=A9=EA=B3=B5=EA=B3=A0?= =?UTF-8?q?=EB=8A=94=20=EC=83=81=ED=83=9C=EB=A7=8C=20=EC=A0=95=EB=A6=AC?= =?UTF-8?q?=ED=95=98=EA=B3=A0,=20=EC=9E=90=EC=86=8C=EC=84=9C=20=EB=B6=84?= =?UTF-8?q?=EC=84=9D=EC=9D=80=20timeout=20=EC=8B=A4=ED=8C=A8=20=EC=8B=9C?= =?UTF-8?q?=20=EC=98=88=EC=95=BD=EB=90=9C=20=ED=81=AC=EB=A0=88=EB=94=A7?= =?UTF-8?q?=EC=9D=84=20=ED=95=A8=EA=BB=98=20=ED=99=98=EB=B6=88=ED=95=98?= =?UTF-8?q?=EB=8F=84=EB=A1=9D=20=EA=B5=AC=EC=84=B1=ED=95=B4=20=EB=B9=84?= =?UTF-8?q?=EB=8F=99=EA=B8=B0=20=EB=B3=B5=EA=B5=AC=EC=99=80=20SSE=20?= =?UTF-8?q?=EC=83=81=ED=83=9C=20=EC=A0=84=ED=8C=8C=EC=9D=98=20=EC=9A=B4?= =?UTF-8?q?=EC=98=81=20=EC=9D=BC=EA=B4=80=EC=84=B1=EC=9D=84=20=EB=86=92?= =?UTF-8?q?=EC=98=80=EC=8A=B5=EB=8B=88=EB=8B=A4.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../jobdri_api/JobdriApiApplication.java | 2 + .../AnalysisAsyncTaskRepository.java | 3 + .../service/AnalysisAsyncSweepService.java | 94 +++++++++++++++++++ .../JobPostingAsyncTaskRepository.java | 5 + .../service/JobPostingAsyncTaskService.java | 15 +++ .../scheduling/AsyncTaskSweepScheduler.java | 34 +++++++ 6 files changed, 153 insertions(+) create mode 100644 src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java create mode 100644 src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java diff --git a/src/main/java/com/jobdri/jobdri_api/JobdriApiApplication.java b/src/main/java/com/jobdri/jobdri_api/JobdriApiApplication.java index 897d2d4..b96a502 100644 --- a/src/main/java/com/jobdri/jobdri_api/JobdriApiApplication.java +++ b/src/main/java/com/jobdri/jobdri_api/JobdriApiApplication.java @@ -4,10 +4,12 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.retry.annotation.EnableRetry; import org.springframework.scheduling.annotation.EnableAsync; +import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication @EnableAsync @EnableRetry +@EnableScheduling public class JobdriApiApplication { public static void main(String[] args) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/repository/AnalysisAsyncTaskRepository.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/repository/AnalysisAsyncTaskRepository.java index 045c904..4426619 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/repository/AnalysisAsyncTaskRepository.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/repository/AnalysisAsyncTaskRepository.java @@ -5,6 +5,7 @@ import org.springframework.data.jpa.repository.JpaRepository; import java.util.Collection; +import java.util.List; import java.util.Optional; public interface AnalysisAsyncTaskRepository extends JpaRepository { @@ -15,4 +16,6 @@ Optional findFirstByUserIdAndMockApplyIdAndStatusInOrderByCre Long mockApplyId, Collection statuses ); + + List findByStatusIn(Collection statuses); } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java new file mode 100644 index 0000000..1e59dc3 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java @@ -0,0 +1,94 @@ +package com.jobdri.jobdri_api.domain.analysis.service; + +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.CreditStatus; +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.FailureReason; +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.TaskStatus; +import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; +import com.jobdri.jobdri_api.domain.user.entity.User; +import com.jobdri.jobdri_api.domain.user.service.UserService; +import lombok.RequiredArgsConstructor; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import java.time.Duration; +import java.time.LocalDateTime; +import java.util.EnumSet; + +@Service +@RequiredArgsConstructor +public class AnalysisAsyncSweepService { + + private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository; + private final AnalysisAsyncTaskService analysisAsyncTaskService; + private final AnalysisService analysisService; + private final UserService userService; + + @Value("${app.worker.analysis.queue-timeout-minutes:10}") + private long queueTimeoutMinutes; + + @Value("${app.worker.analysis.processing-timeout-minutes:20}") + private long processingTimeoutMinutes; + + @Transactional + public int sweepTimedOutTasks() { + int expiredCount = 0; + for (AnalysisAsyncTask task : analysisAsyncTaskRepository.findByStatusIn(EnumSet.of(TaskStatus.PENDING, TaskStatus.RUNNING))) { + ExpirationDecision expirationDecision = resolveExpiration(task); + if (expirationDecision == null) { + continue; + } + + releaseCreditIfNeeded(task); + analysisAsyncTaskService.markFailed( + task.getTaskId(), + expirationDecision.failureReason(), + expirationDecision.errorMessage(), + task.getRetryCount() + ); + expiredCount++; + } + return expiredCount; + } + + private ExpirationDecision resolveExpiration(AnalysisAsyncTask task) { + LocalDateTime now = LocalDateTime.now(); + if (task.getStatus() == TaskStatus.PENDING && isExpired(task.getSubmittedAt(), now, queueTimeoutMinutes)) { + return new ExpirationDecision( + FailureReason.QUEUE_TIMEOUT, + "자소서 분석 작업이 대기열에서 시간 내 처리되지 않았습니다." + ); + } + + LocalDateTime lastActivityAt = task.getLastAttemptAt() != null ? task.getLastAttemptAt() : task.getStartedAt(); + if (task.getStatus() == TaskStatus.RUNNING && isExpired(lastActivityAt, now, processingTimeoutMinutes)) { + return new ExpirationDecision( + FailureReason.INTERNAL_ERROR, + "자소서 분석 작업이 처리 제한 시간을 초과했습니다." + ); + } + + return null; + } + + private boolean isExpired(LocalDateTime baseTime, LocalDateTime now, long timeoutMinutes) { + if (baseTime == null || timeoutMinutes <= 0) { + return false; + } + return Duration.between(baseTime, now).toMinutes() >= timeoutMinutes; + } + + private void releaseCreditIfNeeded(AnalysisAsyncTask task) { + if (task.getCreditStatus() != CreditStatus.RESERVED || task.getCreditReferenceId() == null) { + return; + } + + User user = userService.getUser(task.getUserId()); + analysisService.refundAnalysisCredit(user, task.getCreditReferenceId()); + analysisAsyncTaskService.markCreditReleased(task.getTaskId()); + } + + private record ExpirationDecision(FailureReason failureReason, String errorMessage) { + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/repository/JobPostingAsyncTaskRepository.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/repository/JobPostingAsyncTaskRepository.java index 4841c0c..5deefb5 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/repository/JobPostingAsyncTaskRepository.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/repository/JobPostingAsyncTaskRepository.java @@ -1,10 +1,15 @@ package com.jobdri.jobdri_api.domain.jobposting.repository; import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask; +import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask.TaskStatus; import org.springframework.data.jpa.repository.JpaRepository; +import java.util.Collection; +import java.util.List; import java.util.Optional; public interface JobPostingAsyncTaskRepository extends JpaRepository { Optional findByTaskIdAndUserId(String taskId, Long userId); + + List findByStatusIn(Collection statuses); } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java index aa8c6d4..596a73e 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java @@ -20,6 +20,7 @@ import java.time.Duration; import java.time.Instant; import java.time.LocalDateTime; +import java.util.EnumSet; @Service @RequiredArgsConstructor @@ -122,6 +123,20 @@ public JobPostingAsyncStatusResponse getTask(User user, String taskId) { return toStatusResponse(taskState); } + @Transactional + public int sweepTimedOutTasks() { + int expiredCount = 0; + for (JobPostingAsyncTask task : jobPostingAsyncTaskRepository.findByStatusIn(EnumSet.of(TaskStatus.PENDING, TaskStatus.RUNNING))) { + TaskStatus beforeStatus = task.getStatus(); + expireTimedOutTaskIfNeeded(task); + if (beforeStatus != task.getStatus() && task.getStatus() == TaskStatus.FAILED) { + jobPostingAsyncSseService.publish(toStatusResponse(task)); + expiredCount++; + } + } + return expiredCount; + } + private JobPostingAsyncStatusResponse toStatusResponse(JobPostingAsyncTask taskState) { return JobPostingAsyncStatusResponse.builder() .taskId(taskState.getTaskId()) diff --git a/src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java b/src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java new file mode 100644 index 0000000..5e60c04 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java @@ -0,0 +1,34 @@ +package com.jobdri.jobdri_api.global.scheduling; + +import com.jobdri.jobdri_api.domain.analysis.service.AnalysisAsyncSweepService; +import com.jobdri.jobdri_api.domain.jobposting.service.JobPostingAsyncTaskService; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Component; + +@Slf4j +@Component +@RequiredArgsConstructor +public class AsyncTaskSweepScheduler { + + private final JobPostingAsyncTaskService jobPostingAsyncTaskService; + private final AnalysisAsyncSweepService analysisAsyncSweepService; + + @Scheduled( + fixedDelayString = "${app.async-task-sweep.fixed-delay-millis:60000}", + initialDelayString = "${app.async-task-sweep.initial-delay-millis:30000}" + ) + public void sweepTimedOutTasks() { + int expiredJobPostingTasks = jobPostingAsyncTaskService.sweepTimedOutTasks(); + int expiredAnalysisTasks = analysisAsyncSweepService.sweepTimedOutTasks(); + + if (expiredJobPostingTasks > 0 || expiredAnalysisTasks > 0) { + log.info( + "Async task sweep expired tasks. jobPostingExpired={} analysisExpired={}", + expiredJobPostingTasks, + expiredAnalysisTasks + ); + } + } +} From fa683baa287db6d2eed06d152a4abb2c364532f1 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Tue, 7 Jul 2026 15:44:16 +0900 Subject: [PATCH 6/7] =?UTF-8?q?[Fix]=20async=20SSE=20=EB=B0=9C=ED=96=89=20?= =?UTF-8?q?=EC=8B=9C=EC=A0=90=20=EB=B0=8F=20sweep=20=EC=95=88=EC=A0=95?= =?UTF-8?q?=EC=84=B1=20=EB=B3=B4=EA=B0=95=20(#109)=20async=20SSE=20?= =?UTF-8?q?=EB=B0=9C=ED=96=89=EC=9D=84=20=ED=8A=B8=EB=9E=9C=EC=9E=AD?= =?UTF-8?q?=EC=85=98=20=EC=BB=A4=EB=B0=8B=20=EC=9D=B4=ED=9B=84=EB=A1=9C=20?= =?UTF-8?q?=EC=A7=80=EC=97=B0=ED=95=B4=20=ED=81=B4=EB=9D=BC=EC=9D=B4?= =?UTF-8?q?=EC=96=B8=ED=8A=B8=EA=B0=80=20=EB=AF=B8=EC=BB=A4=EB=B0=8B=20?= =?UTF-8?q?=EC=83=81=ED=83=9C=EB=A5=BC=20=EA=B4=80=EC=B8=A1=ED=95=98?= =?UTF-8?q?=EC=A7=80=20=EC=95=8A=EB=8F=84=EB=A1=9D=20=EC=88=98=EC=A0=95?= =?UTF-8?q?=ED=95=98=EA=B3=A0,=20SSE=20=EA=B5=AC=EB=8F=85/=ED=95=98?= =?UTF-8?q?=ED=8A=B8=EB=B9=84=ED=8A=B8=20=EC=98=88=EC=99=B8=20=EC=B2=98?= =?UTF-8?q?=EB=A6=AC=EC=99=80=20=EC=B1=84=EB=84=90=EB=B3=84=20=EB=8F=99?= =?UTF-8?q?=EA=B8=B0=ED=99=94=EB=A5=BC=20=EB=B3=B4=EA=B0=95=ED=96=88?= =?UTF-8?q?=EC=8A=B5=EB=8B=88=EB=8B=A4.=20=EB=98=90=ED=95=9C=20analysis=20?= =?UTF-8?q?sweep=EB=A5=BC=20per-task=20=ED=8A=B8=EB=9E=9C=EC=9E=AD?= =?UTF-8?q?=EC=85=98=EC=9C=BC=EB=A1=9C=20=EB=B6=84=EB=A6=AC=ED=95=98?= =?UTF-8?q?=EA=B3=A0=20scheduler=20=EC=98=88=EC=99=B8=20=EA=B2=A9=EB=A6=AC?= =?UTF-8?q?=EB=A5=BC=20=EC=B6=94=EA=B0=80=ED=96=88=EC=9C=BC=EB=A9=B0,=20?= =?UTF-8?q?=EC=B1=84=EC=9A=A9=EA=B3=B5=EA=B3=A0=20worker=20retry/failed=20?= =?UTF-8?q?=ED=8A=B8=EB=9E=9C=EC=9E=AD=EC=85=98=20=EC=9D=BC=EA=B4=80?= =?UTF-8?q?=EC=84=B1=20=EB=B0=8F=20DTO=20=EA=B2=80=EC=A6=9D=EC=9D=84=20?= =?UTF-8?q?=EA=B0=95=ED=99=94=ED=95=B4=20=EB=B9=84=EB=8F=99=EA=B8=B0=20?= =?UTF-8?q?=EC=83=81=ED=83=9C=20=EB=B3=B5=EA=B5=AC=20=EA=B2=BD=EB=A1=9C?= =?UTF-8?q?=EC=9D=98=20=EC=95=88=EC=A0=95=EC=84=B1=EC=9D=84=20=EB=86=92?= =?UTF-8?q?=EC=98=80=EC=8A=B5=EB=8B=88=EB=8B=A4.:wq?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../controller/AnalysisController.java | 8 +- .../service/AnalysisAsyncSseService.java | 14 ++- .../service/AnalysisAsyncSweepService.java | 44 +++++-- .../service/AnalysisAsyncTaskService.java | 23 +++- .../analysis/service/AnalysisService.java | 4 +- .../controller/JobPostingAiController.java | 6 +- .../JobPostingWorkerFailureRequest.java | 3 +- .../worker/JobPostingWorkerRetryRequest.java | 3 +- .../entity/JobPostingAsyncTask.java | 2 +- .../service/JobPostingAsyncSseService.java | 14 ++- .../service/JobPostingAsyncTaskService.java | 25 +++- .../JobPostingWorkerBridgeService.java | 2 + .../scheduling/AsyncTaskSweepScheduler.java | 22 +++- .../global/sse/SseSubscriptionRegistry.java | 114 ++++++++++++------ .../JobPostingAsyncTaskServiceTest.java | 26 ++++ 15 files changed, 231 insertions(+), 79 deletions(-) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java index b202304..c9985a3 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/controller/AnalysisController.java @@ -64,9 +64,11 @@ public SseEmitter streamAnalysisTask( @PathVariable Long mockApplyId, @PathVariable String taskId ) { - AnalysisAsyncStatusResponse initialStatus = - analysisAsyncFacadeService.getTask(getAuthenticatedUser(userDetails), mockApplyId, taskId); - return analysisAsyncSseService.subscribe(initialStatus); + var user = getAuthenticatedUser(userDetails); + return analysisAsyncSseService.subscribe( + taskId, + () -> analysisAsyncFacadeService.getTask(user, mockApplyId, taskId) + ); } @Operation(summary = "자소서 분석 결과 조회", description = "저장된 자소서 분석 결과를 조회합니다.") diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java index 6324635..895359a 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSseService.java @@ -1,11 +1,14 @@ package com.jobdri.jobdri_api.domain.analysis.service; import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncStatusResponse; +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask.TaskStatus; import com.jobdri.jobdri_api.global.sse.SseSubscriptionRegistry; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Service; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import java.util.function.Supplier; + @Service @RequiredArgsConstructor public class AnalysisAsyncSseService { @@ -14,12 +17,12 @@ public class AnalysisAsyncSseService { private final SseSubscriptionRegistry sseSubscriptionRegistry; - public SseEmitter subscribe(AnalysisAsyncStatusResponse initialStatus) { + public SseEmitter subscribe(String taskId, Supplier initialStatusSupplier) { return sseSubscriptionRegistry.subscribe( - channelKey(initialStatus.taskId()), + channelKey(taskId), EVENT_NAME, - initialStatus, - isTerminal(initialStatus) + initialStatusSupplier, + this::isTerminal ); } @@ -37,6 +40,7 @@ private String channelKey(String taskId) { } private boolean isTerminal(AnalysisAsyncStatusResponse statusResponse) { - return "SUCCEEDED".equals(statusResponse.status()) || "FAILED".equals(statusResponse.status()); + TaskStatus status = TaskStatus.valueOf(statusResponse.status()); + return status == TaskStatus.SUCCEEDED || status == TaskStatus.FAILED; } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java index 1e59dc3..4214588 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java @@ -8,7 +8,9 @@ import com.jobdri.jobdri_api.domain.user.entity.User; import com.jobdri.jobdri_api.domain.user.service.UserService; import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; @@ -16,6 +18,7 @@ import java.time.LocalDateTime; import java.util.EnumSet; +@Slf4j @Service @RequiredArgsConstructor public class AnalysisAsyncSweepService { @@ -24,6 +27,8 @@ public class AnalysisAsyncSweepService { private final AnalysisAsyncTaskService analysisAsyncTaskService; private final AnalysisService analysisService; private final UserService userService; + @Lazy + private final AnalysisAsyncSweepService self; @Value("${app.worker.analysis.queue-timeout-minutes:10}") private long queueTimeoutMinutes; @@ -31,27 +36,40 @@ public class AnalysisAsyncSweepService { @Value("${app.worker.analysis.processing-timeout-minutes:20}") private long processingTimeoutMinutes; - @Transactional public int sweepTimedOutTasks() { int expiredCount = 0; for (AnalysisAsyncTask task : analysisAsyncTaskRepository.findByStatusIn(EnumSet.of(TaskStatus.PENDING, TaskStatus.RUNNING))) { - ExpirationDecision expirationDecision = resolveExpiration(task); - if (expirationDecision == null) { - continue; + try { + expiredCount += self.sweepTimedOutTask(task.getTaskId()); + } catch (RuntimeException e) { + log.error("Analysis async task sweep failed for taskId={}", task.getTaskId(), e); } - - releaseCreditIfNeeded(task); - analysisAsyncTaskService.markFailed( - task.getTaskId(), - expirationDecision.failureReason(), - expirationDecision.errorMessage(), - task.getRetryCount() - ); - expiredCount++; } return expiredCount; } + @Transactional + public int sweepTimedOutTask(String taskId) { + AnalysisAsyncTask task = analysisAsyncTaskRepository.findById(taskId).orElse(null); + if (task == null || task.getStatus() == TaskStatus.SUCCEEDED || task.getStatus() == TaskStatus.FAILED) { + return 0; + } + + ExpirationDecision expirationDecision = resolveExpiration(task); + if (expirationDecision == null) { + return 0; + } + + releaseCreditIfNeeded(task); + analysisAsyncTaskService.markFailed( + task.getTaskId(), + expirationDecision.failureReason(), + expirationDecision.errorMessage(), + task.getRetryCount() + ); + return 1; + } + private ExpirationDecision resolveExpiration(AnalysisAsyncTask task) { LocalDateTime now = LocalDateTime.now(); if (task.getStatus() == TaskStatus.PENDING && isExpired(task.getSubmittedAt(), now, queueTimeoutMinutes)) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java index d21f276..c9970d1 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java @@ -12,6 +12,8 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; import java.time.Instant; import java.util.EnumSet; @@ -53,28 +55,28 @@ public Optional findActiveTask(Long userId, Long mockApplyId) public void markRunning(String taskId, String workerId, int retryCount, Instant submittedAt) { AnalysisAsyncTask task = getTask(taskId); task.markRunning(workerId, retryCount, submittedAt); - analysisAsyncSseService.publish(toStatusResponse(task)); + publishAfterCommit(toStatusResponse(task)); } @Transactional public void markSuccess(String taskId) { AnalysisAsyncTask task = getTask(taskId); task.markSuccess(); - analysisAsyncSseService.publish(toStatusResponse(task)); + publishAfterCommit(toStatusResponse(task)); } @Transactional public void markRetryScheduled(String taskId, FailureReason failureReason, String errorMessage, int retryCount) { AnalysisAsyncTask task = getTask(taskId); task.markRetryScheduled(failureReason, errorMessage, retryCount); - analysisAsyncSseService.publish(toStatusResponse(task)); + publishAfterCommit(toStatusResponse(task)); } @Transactional public void markFailed(String taskId, FailureReason failureReason, String errorMessage, int retryCount) { AnalysisAsyncTask task = getTask(taskId); task.markFailed(failureReason, errorMessage, retryCount); - analysisAsyncSseService.publish(toStatusResponse(task)); + publishAfterCommit(toStatusResponse(task)); } @Transactional @@ -145,4 +147,17 @@ private AnalysisAsyncStatusResponse toStatusResponse(AnalysisAsyncTask task) { .result(null) .build(); } + + private void publishAfterCommit(AnalysisAsyncStatusResponse statusResponse) { + if (!TransactionSynchronizationManager.isSynchronizationActive()) { + analysisAsyncSseService.publish(statusResponse); + return; + } + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + @Override + public void afterCommit() { + analysisAsyncSseService.publish(statusResponse); + } + }); + } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java index 5242f1c..01b7650 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisService.java @@ -104,12 +104,12 @@ public void validateAnalysisRequest(User user, Long mockApplyId) { @Transactional public void deductAnalysisCredit(User user, String referenceId) { - creditService.use(user, 1, "자소서 분석 크레딧 예약", referenceId); + creditService.use(user, 1, "자소서 분석 크레딧 차감", referenceId); } @Transactional public void refundAnalysisCredit(User user, String referenceId) { - creditService.refund(user, 1, "자소서 분석 크레딧 예약 해제", referenceId); + creditService.refund(user, 1, "자소서 분석 크레딧 환불", referenceId); } @Transactional(readOnly = true) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java index c3137b3..103a98e 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/controller/JobPostingAiController.java @@ -121,8 +121,10 @@ public SseEmitter streamIngestJobPostingAsyncStatus( @PathVariable String taskId ) { var user = validateAuthenticatedUser(userDetails); - JobPostingAsyncStatusResponse initialStatus = jobPostingAsyncFacadeService.getTask(user, taskId); - return jobPostingAsyncSseService.subscribe(initialStatus); + return jobPostingAsyncSseService.subscribe( + taskId, + () -> jobPostingAsyncFacadeService.getTask(user, taskId) + ); } private com.jobdri.jobdri_api.domain.user.entity.User validateAuthenticatedUser(UserDetailsImpl userDetails) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java index db9f8e3..3dc9a3c 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerFailureRequest.java @@ -4,12 +4,13 @@ import jakarta.validation.constraints.Min; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Size; public record JobPostingWorkerFailureRequest( @NotNull FailureReason failureReason, @NotBlank String errorMessage, @Min(0) int retryCount, - @NotBlank String workerId, + @NotBlank @Size(max = 100) String workerId, Long queueLatencyMillis ) { } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java index ca17138..99e3ec1 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/dto/worker/JobPostingWorkerRetryRequest.java @@ -4,12 +4,13 @@ import jakarta.validation.constraints.Min; import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Size; public record JobPostingWorkerRetryRequest( @NotNull FailureReason failureReason, @NotBlank String errorMessage, @Min(0) int retryCount, - @NotBlank String workerId, + @NotBlank @Size(max = 100) String workerId, Long queueLatencyMillis ) { } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java index c670b55..c2fc1c0 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/entity/JobPostingAsyncTask.java @@ -124,7 +124,7 @@ public void markRetryScheduled(FailureReason failureReason, String errorMessage, } public void markFailed(FailureReason failureReason, String errorMessage, int retryCount) { - if (status == TaskStatus.SUCCEEDED) { + if (isTerminal()) { return; } this.status = TaskStatus.FAILED; diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java index 94dbc0e..9a1aebc 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncSseService.java @@ -1,11 +1,14 @@ package com.jobdri.jobdri_api.domain.jobposting.service; import com.jobdri.jobdri_api.domain.jobposting.dto.response.JobPostingAsyncStatusResponse; +import com.jobdri.jobdri_api.domain.jobposting.entity.JobPostingAsyncTask.TaskStatus; import com.jobdri.jobdri_api.global.sse.SseSubscriptionRegistry; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Service; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; +import java.util.function.Supplier; + @Service @RequiredArgsConstructor public class JobPostingAsyncSseService { @@ -14,12 +17,12 @@ public class JobPostingAsyncSseService { private final SseSubscriptionRegistry sseSubscriptionRegistry; - public SseEmitter subscribe(JobPostingAsyncStatusResponse initialStatus) { + public SseEmitter subscribe(String taskId, Supplier initialStatusSupplier) { return sseSubscriptionRegistry.subscribe( - channelKey(initialStatus.getTaskId()), + channelKey(taskId), EVENT_NAME, - initialStatus, - isTerminal(initialStatus) + initialStatusSupplier, + this::isTerminal ); } @@ -37,6 +40,7 @@ private String channelKey(String taskId) { } private boolean isTerminal(JobPostingAsyncStatusResponse statusResponse) { - return "SUCCEEDED".equals(statusResponse.getStatus()) || "FAILED".equals(statusResponse.getStatus()); + TaskStatus status = TaskStatus.valueOf(statusResponse.getStatus()); + return status == TaskStatus.SUCCEEDED || status == TaskStatus.FAILED; } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java index 596a73e..266b8fc 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskService.java @@ -16,6 +16,8 @@ import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionSynchronization; +import org.springframework.transaction.support.TransactionSynchronizationManager; import java.time.Duration; import java.time.Instant; @@ -56,7 +58,7 @@ public void markRunning(String taskId, String workerId, int retryCount, Instant return; } task.markRunning(workerId, retryCount, submittedAt); - jobPostingAsyncSseService.publish(toStatusResponse(task)); + publishAfterCommit(toStatusResponse(task)); } @Transactional @@ -72,7 +74,7 @@ public JobPostingIngestResponse markSuccess(String taskId, JobPostingIngestRespo ); } task.markSuccess(serializeResult(result)); - jobPostingAsyncSseService.publish(toStatusResponse(task)); + publishAfterCommit(toStatusResponse(task)); return result; } @@ -83,7 +85,7 @@ public void markRetryScheduled(String taskId, FailureReason failureReason, Strin return; } task.markRetryScheduled(failureReason, errorMessage, retryCount); - jobPostingAsyncSseService.publish(toStatusResponse(task)); + publishAfterCommit(toStatusResponse(task)); } @Transactional @@ -93,7 +95,7 @@ public void markFailed(String taskId, FailureReason failureReason, String errorM return; } task.markFailed(failureReason, errorMessage, retryCount); - jobPostingAsyncSseService.publish(toStatusResponse(task)); + publishAfterCommit(toStatusResponse(task)); } @Transactional @@ -130,7 +132,7 @@ public int sweepTimedOutTasks() { TaskStatus beforeStatus = task.getStatus(); expireTimedOutTaskIfNeeded(task); if (beforeStatus != task.getStatus() && task.getStatus() == TaskStatus.FAILED) { - jobPostingAsyncSseService.publish(toStatusResponse(task)); + publishAfterCommit(toStatusResponse(task)); expiredCount++; } } @@ -227,4 +229,17 @@ private JobPostingIngestResponse deserializeResult(String resultPayload) { ); } } + + private void publishAfterCommit(JobPostingAsyncStatusResponse statusResponse) { + if (!TransactionSynchronizationManager.isSynchronizationActive()) { + jobPostingAsyncSseService.publish(statusResponse); + return; + } + TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { + @Override + public void afterCommit() { + jobPostingAsyncSseService.publish(statusResponse); + } + }); + } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java index 63c5cff..72497ed 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingWorkerBridgeService.java @@ -39,6 +39,7 @@ public JobPostingIngestResponse completeTask(String taskId, JobPostingIngestResp return jobPostingAsyncTaskService.markSuccess(taskId, result); } + @Transactional public void markRetry( String taskId, FailureReason failureReason, @@ -55,6 +56,7 @@ public void markRetry( jobPostingAsyncTaskService.markRetryScheduled(taskId, failureReason, errorMessage, retryCount); } + @Transactional public void failTask( String taskId, FailureReason failureReason, diff --git a/src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java b/src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java index 5e60c04..40cc55d 100644 --- a/src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java +++ b/src/main/java/com/jobdri/jobdri_api/global/scheduling/AsyncTaskSweepScheduler.java @@ -20,8 +20,8 @@ public class AsyncTaskSweepScheduler { initialDelayString = "${app.async-task-sweep.initial-delay-millis:30000}" ) public void sweepTimedOutTasks() { - int expiredJobPostingTasks = jobPostingAsyncTaskService.sweepTimedOutTasks(); - int expiredAnalysisTasks = analysisAsyncSweepService.sweepTimedOutTasks(); + int expiredJobPostingTasks = sweepJobPostingTasks(); + int expiredAnalysisTasks = sweepAnalysisTasks(); if (expiredJobPostingTasks > 0 || expiredAnalysisTasks > 0) { log.info( @@ -31,4 +31,22 @@ public void sweepTimedOutTasks() { ); } } + + private int sweepJobPostingTasks() { + try { + return jobPostingAsyncTaskService.sweepTimedOutTasks(); + } catch (RuntimeException e) { + log.error("Job posting async task sweep failed.", e); + return 0; + } + } + + private int sweepAnalysisTasks() { + try { + return analysisAsyncSweepService.sweepTimedOutTasks(); + } catch (RuntimeException e) { + log.error("Analysis async task sweep failed.", e); + return 0; + } + } } diff --git a/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java b/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java index 170d049..3f5f6dc 100644 --- a/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java +++ b/src/main/java/com/jobdri/jobdri_api/global/sse/SseSubscriptionRegistry.java @@ -5,25 +5,30 @@ import org.springframework.stereotype.Component; import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; -import java.io.IOException; import java.util.List; import java.util.Map; +import java.util.concurrent.ExecutorService; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.function.Predicate; +import java.util.function.Supplier; @Component public class SseSubscriptionRegistry { private final Map> emitters = new ConcurrentHashMap<>(); + private final Map channelMonitors = new ConcurrentHashMap<>(); private final long timeoutMillis; private final ScheduledExecutorService heartbeatExecutor; + private final ExecutorService heartbeatDispatchExecutor; public SseSubscriptionRegistry( @Value("${app.sse.timeout-millis:1800000}") long timeoutMillis, - @Value("${app.sse.heartbeat-interval-millis:15000}") long heartbeatIntervalMillis + @Value("${app.sse.heartbeat-interval-millis:15000}") long heartbeatIntervalMillis, + @Value("${app.sse.heartbeat-dispatch-thread-count:4}") int heartbeatDispatchThreadCount ) { this.timeoutMillis = timeoutMillis; this.heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> { @@ -31,6 +36,14 @@ public SseSubscriptionRegistry( thread.setDaemon(true); return thread; }); + this.heartbeatDispatchExecutor = Executors.newFixedThreadPool( + Math.max(1, heartbeatDispatchThreadCount), + runnable -> { + Thread thread = new Thread(runnable, "sse-heartbeat-dispatch"); + thread.setDaemon(true); + return thread; + } + ); if (heartbeatIntervalMillis > 0) { this.heartbeatExecutor.scheduleAtFixedRate( this::sendHeartbeatSafely, @@ -42,47 +55,63 @@ public SseSubscriptionRegistry( } public SseEmitter subscribe(String channelKey, String eventName, Object initialPayload, boolean completeAfterInitial) { - SseEmitter emitter = new SseEmitter(timeoutMillis); - emitters.computeIfAbsent(channelKey, ignored -> new CopyOnWriteArrayList<>()).add(emitter); + return subscribe(channelKey, eventName, () -> initialPayload, ignored -> completeAfterInitial); + } - emitter.onCompletion(() -> remove(channelKey, emitter)); - emitter.onTimeout(() -> { - emitter.complete(); - remove(channelKey, emitter); - }); - emitter.onError(ignored -> remove(channelKey, emitter)); + public SseEmitter subscribe( + String channelKey, + String eventName, + Supplier initialPayloadSupplier, + Predicate completeAfterInitialPredicate + ) { + SseEmitter emitter = new SseEmitter(timeoutMillis); + synchronized (channelMonitor(channelKey)) { + emitters.computeIfAbsent(channelKey, ignored -> new CopyOnWriteArrayList<>()).add(emitter); - try { - emitter.send(SseEmitter.event().name("connected").data("connected")); - emitter.send(SseEmitter.event().name(eventName).data(initialPayload)); - if (completeAfterInitial) { + emitter.onCompletion(() -> remove(channelKey, emitter)); + emitter.onTimeout(() -> { emitter.complete(); remove(channelKey, emitter); + }); + emitter.onError(ignored -> remove(channelKey, emitter)); + + T initialPayload = initialPayloadSupplier.get(); + boolean completeAfterInitial = completeAfterInitialPredicate.test(initialPayload); + + try { + emitter.send(SseEmitter.event().name("connected").data("connected")); + emitter.send(SseEmitter.event().name(eventName).data(initialPayload)); + if (completeAfterInitial) { + emitter.complete(); + remove(channelKey, emitter); + } + } catch (Exception e) { + emitter.completeWithError(e); + remove(channelKey, emitter); } - } catch (IOException e) { - emitter.completeWithError(e); - remove(channelKey, emitter); } return emitter; } public void publish(String channelKey, String eventName, Object payload, boolean completeAfterPublish) { - List subscribers = emitters.get(channelKey); - if (subscribers == null || subscribers.isEmpty()) { - return; - } + synchronized (channelMonitor(channelKey)) { + List subscribers = emitters.get(channelKey); + if (subscribers == null || subscribers.isEmpty()) { + return; + } - for (SseEmitter emitter : subscribers) { - try { - emitter.send(SseEmitter.event().name(eventName).data(payload)); - if (completeAfterPublish) { - emitter.complete(); + for (SseEmitter emitter : subscribers) { + try { + emitter.send(SseEmitter.event().name(eventName).data(payload)); + if (completeAfterPublish) { + emitter.complete(); + remove(channelKey, emitter); + } + } catch (Exception e) { + emitter.completeWithError(e); remove(channelKey, emitter); } - } catch (IOException e) { - emitter.completeWithError(e); - remove(channelKey, emitter); } } } @@ -90,6 +119,7 @@ public void publish(String channelKey, String eventName, Object payload, boolean @PreDestroy void shutdown() { heartbeatExecutor.shutdownNow(); + heartbeatDispatchExecutor.shutdownNow(); } private void sendHeartbeatSafely() { @@ -104,12 +134,21 @@ private void sendHeartbeat() { for (Map.Entry> entry : emitters.entrySet()) { String channelKey = entry.getKey(); for (SseEmitter emitter : entry.getValue()) { - try { - emitter.send(SseEmitter.event().name("heartbeat").data("ping")); - } catch (IOException e) { - emitter.completeWithError(e); - remove(channelKey, emitter); - } + heartbeatDispatchExecutor.execute(() -> sendHeartbeat(channelKey, emitter)); + } + } + } + + private void sendHeartbeat(String channelKey, SseEmitter emitter) { + synchronized (channelMonitor(channelKey)) { + if (!emitters.getOrDefault(channelKey, new CopyOnWriteArrayList<>()).contains(emitter)) { + return; + } + try { + emitter.send(SseEmitter.event().name("heartbeat").data("ping")); + } catch (Exception e) { + emitter.completeWithError(e); + remove(channelKey, emitter); } } } @@ -122,6 +161,11 @@ private void remove(String channelKey, SseEmitter emitter) { subscribers.remove(emitter); if (subscribers.isEmpty()) { emitters.remove(channelKey); + channelMonitors.remove(channelKey); } } + + private Object channelMonitor(String channelKey) { + return channelMonitors.computeIfAbsent(channelKey, ignored -> new Object()); + } } diff --git a/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java index a01e644..523ac62 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAsyncTaskServiceTest.java @@ -19,6 +19,8 @@ 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.Mockito.never; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -133,6 +135,7 @@ void markRetryScheduledFailsWhenMaxRetryReached() { assertThat(task.getStatus()).isEqualTo(JobPostingAsyncTask.TaskStatus.FAILED); assertThat(task.getFailureReason()).isEqualTo(FailureReason.INTERNAL_ERROR); assertThat(task.getRetryCount()).isEqualTo(3); + verify(jobPostingAsyncSseService).publish(any()); } @Test @@ -147,4 +150,27 @@ void updateWorkerMetadataUpdatesWorkerFields() { assertThat(task.getWorkerId()).isEqualTo("worker-2"); assertThat(task.getQueueLatencyMillis()).isEqualTo(1234L); } + + @Test + @DisplayName("이미 종료된 task에 대한 재시도 예약은 무시된다") + void markRetryScheduledDoesNothingWhenTaskIsTerminal() { + JobPostingAsyncTask task = JobPostingAsyncTask.pending(7L, 3); + task.markFailed(FailureReason.INTERNAL_ERROR, "failed", 1); + String originalError = task.getError(); + int originalRetryCount = task.getRetryCount(); + + when(jobPostingAsyncTaskRepository.findById(task.getTaskId())).thenReturn(Optional.of(task)); + + jobPostingAsyncTaskService.markRetryScheduled( + task.getTaskId(), + FailureReason.QUEUE_TIMEOUT, + "should be ignored", + 2 + ); + + assertThat(task.getFailureReason()).isEqualTo(FailureReason.INTERNAL_ERROR); + assertThat(task.getError()).isEqualTo(originalError); + assertThat(task.getRetryCount()).isEqualTo(originalRetryCount); + verify(jobPostingAsyncSseService, never()).publish(any()); + } } From 06df34657c3b740f4d1e53a0360982844a251525 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Tue, 7 Jul 2026 16:00:01 +0900 Subject: [PATCH 7/7] =?UTF-8?q?[Fix]=20analysis=20async=20sweep=20?= =?UTF-8?q?=EC=88=9C=ED=99=98=20=EC=9D=98=EC=A1=B4=EC=84=B1=20=EC=A0=9C?= =?UTF-8?q?=EA=B1=B0=20(#110)=20analysis=20async=20stuck-task=20sweep?= =?UTF-8?q?=EC=97=90=20per-task=20=ED=8A=B8=EB=9E=9C=EC=9E=AD=EC=85=98?= =?UTF-8?q?=EC=9D=84=20=EC=A0=81=EC=9A=A9=ED=95=98=EB=8A=94=20=EA=B3=BC?= =?UTF-8?q?=EC=A0=95=EC=97=90=EC=84=9C=20=EB=B0=9C=EC=83=9D=ED=95=9C=20?= =?UTF-8?q?=EC=88=9C=ED=99=98=20=EC=9D=98=EC=A1=B4=EC=84=B1=EC=9D=84=20?= =?UTF-8?q?=EC=A0=9C=EA=B1=B0=ED=96=88=EC=8A=B5=EB=8B=88=EB=8B=A4.=20self-?= =?UTF-8?q?injection=20=EB=8C=80=EC=8B=A0=20TransactionTemplate=20?= =?UTF-8?q?=EA=B8=B0=EB=B0=98=EC=9C=BC=EB=A1=9C=20=EA=B0=81=20task?= =?UTF-8?q?=EB=A5=BC=20=EA=B0=9C=EB=B3=84=20=ED=8A=B8=EB=9E=9C=EC=9E=AD?= =?UTF-8?q?=EC=85=98=EC=97=90=EC=84=9C=20=EC=B2=98=EB=A6=AC=ED=95=98?= =?UTF-8?q?=EB=8F=84=EB=A1=9D=20=EB=B3=80=EA=B2=BD=ED=95=B4=20=EC=8A=A4?= =?UTF-8?q?=ED=94=84=EB=A7=81=20=EC=BB=A8=ED=85=8D=EC=8A=A4=ED=8A=B8=20?= =?UTF-8?q?=EC=B4=88=EA=B8=B0=ED=99=94=20=EC=8B=A4=ED=8C=A8=EC=99=80=20?= =?UTF-8?q?=ED=85=8C=EC=8A=A4=ED=8A=B8=20=EC=A0=84=EB=B0=98=EC=9D=98=20?= =?UTF-8?q?=EC=97=B0=EC=87=84=20=EC=8B=A4=ED=8C=A8=EB=A5=BC=20=ED=95=B4?= =?UTF-8?q?=EC=86=8C=ED=96=88=EC=8A=B5=EB=8B=88=EB=8B=A4.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../analysis/service/AnalysisAsyncSweepService.java | 10 ++++------ 1 file changed, 4 insertions(+), 6 deletions(-) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java index 4214588..50b7b85 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncSweepService.java @@ -10,9 +10,9 @@ import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.springframework.beans.factory.annotation.Value; -import org.springframework.context.annotation.Lazy; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; import java.time.Duration; import java.time.LocalDateTime; @@ -27,8 +27,7 @@ public class AnalysisAsyncSweepService { private final AnalysisAsyncTaskService analysisAsyncTaskService; private final AnalysisService analysisService; private final UserService userService; - @Lazy - private final AnalysisAsyncSweepService self; + private final TransactionTemplate transactionTemplate; @Value("${app.worker.analysis.queue-timeout-minutes:10}") private long queueTimeoutMinutes; @@ -40,7 +39,7 @@ public int sweepTimedOutTasks() { int expiredCount = 0; for (AnalysisAsyncTask task : analysisAsyncTaskRepository.findByStatusIn(EnumSet.of(TaskStatus.PENDING, TaskStatus.RUNNING))) { try { - expiredCount += self.sweepTimedOutTask(task.getTaskId()); + expiredCount += transactionTemplate.execute(status -> sweepTimedOutTask(task.getTaskId())); } catch (RuntimeException e) { log.error("Analysis async task sweep failed for taskId={}", task.getTaskId(), e); } @@ -48,8 +47,7 @@ public int sweepTimedOutTasks() { return expiredCount; } - @Transactional - public int sweepTimedOutTask(String taskId) { + private int sweepTimedOutTask(String taskId) { AnalysisAsyncTask task = analysisAsyncTaskRepository.findById(taskId).orElse(null); if (task == null || task.getStatus() == TaskStatus.SUCCEEDED || task.getStatus() == TaskStatus.FAILED) { return 0;