From 4539229f5f3a30d53b115fafc5cc093828f8c190 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Wed, 12 Aug 2026 10:32:09 +0900 Subject: [PATCH 1/7] =?UTF-8?q?[Fix]=20async=20sweep=20=EB=8C=80=EC=83=81?= =?UTF-8?q?=20=EC=A0=95=ED=95=A9=EC=84=B1=20=EB=B3=B4=EC=99=84=20(#271)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../usecase/async/AnalysisAsyncUseCase.java | 17 ++++++- .../analysis/entity/AnalysisAsyncTask.java | 19 +++++++ .../AnalysisAsyncTaskRepository.java | 8 +++ .../async/AnalysisAsyncTaskService.java | 25 +++++++++ .../async/AnalysisAsyncFacadeServiceTest.java | 51 +++++++++++++++++++ 5 files changed, 119 insertions(+), 1 deletion(-) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/application/usecase/async/AnalysisAsyncUseCase.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/application/usecase/async/AnalysisAsyncUseCase.java index c5f9a4a..602b9db 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/application/usecase/async/AnalysisAsyncUseCase.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/application/usecase/async/AnalysisAsyncUseCase.java @@ -15,6 +15,8 @@ import org.hibernate.exception.ConstraintViolationException; import org.springframework.dao.DataIntegrityViolationException; +import java.util.Optional; + // 분석 비동기 작업의 접수/조회/취소 유스케이스를 조율한다. public class AnalysisAsyncUseCase { private static final String ACTIVE_TASK_UNIQUE_CONSTRAINT = "uk_analysis_async_tasks_active_user_mock_apply"; @@ -64,6 +66,11 @@ public AnalysisAsyncCancelResponse cancel(User user, Long mockApplyId, String ta } private AnalysisAsyncSubmitResponse createCachedOrProcessTask(User user, Long mockApplyId) { + Optional recoverableTask = + analysisAsyncTaskService.findRecoverablePublishFailureTask(user.getId(), mockApplyId); + if (recoverableTask.isPresent()) { + return reopenAndProcessTask(user, mockApplyId, recoverableTask.get()); + } if (analysisService.hasReusableAnalysis(user, mockApplyId)) { return toCachedResponse(); } @@ -77,8 +84,16 @@ private AnalysisAsyncSubmitResponse createAndProcessTask(User user, Long mockApp } AnalysisAsyncTask task = pendingTaskResult.task(); - String taskId = task.getTaskId(); + return processTask(user, mockApplyId, task); + } + private AnalysisAsyncSubmitResponse reopenAndProcessTask(User user, Long mockApplyId, AnalysisAsyncTask task) { + AnalysisAsyncTask reopenedTask = analysisAsyncTaskService.reopenPublishFailureTask(task.getTaskId()); + return processTask(user, mockApplyId, reopenedTask); + } + + private AnalysisAsyncSubmitResponse processTask(User user, Long mockApplyId, AnalysisAsyncTask task) { + String taskId = task.getTaskId(); try { analysisAsyncProcessor.process( taskId, diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/entity/AnalysisAsyncTask.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/entity/AnalysisAsyncTask.java index 3dac990..99d5c4e 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/entity/AnalysisAsyncTask.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/entity/AnalysisAsyncTask.java @@ -169,6 +169,25 @@ public boolean isRecoverablePublishFailure() { && failureReason == AnalysisAsyncFailureReason.PUBLISH_FAILED; } + public void reopenForRepublish() { + if (!isRecoverablePublishFailure()) { + return; + } + this.status = AnalysisAsyncTaskStatus.PENDING; + this.message = "자소서 분석 비동기 작업이 다시 접수되었습니다."; + this.error = null; + this.failureReason = null; + this.workerId = null; + this.submittedAt = LocalDateTime.now(); + this.lastAttemptAt = null; + this.queueLatencyMillis = null; + this.startedAt = null; + this.completedAt = null; + this.currentStep = "VALIDATING_INPUT"; + this.progressPercent = 0; + this.estimatedRemainingSeconds = null; + } + public void markRetryScheduled(AnalysisAsyncFailureReason failureReason, String errorMessage, int retryCount) { if (isTerminal()) { return; 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 de6ea0d..cfc1180 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 @@ -1,6 +1,7 @@ package com.jobdri.jobdri_api.domain.analysis.repository; import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; +import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncFailureReason; import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncTaskStatus; import jakarta.persistence.LockModeType; import org.springframework.data.jpa.repository.JpaRepository; @@ -21,6 +22,13 @@ Optional findFirstByUserIdAndMockApplyIdAndStatusInOrderByCre Collection statuses ); + Optional findFirstByUserIdAndMockApplyIdAndStatusAndFailureReasonOrderByCreatedAtDesc( + Long userId, + Long mockApplyId, + AnalysisAsyncTaskStatus status, + AnalysisAsyncFailureReason failureReason + ); + List findByUserIdAndMockApplyIdInAndStatusIn( Long userId, Collection mockApplyIds, diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java index 9696c62..ea312e6 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java @@ -82,6 +82,17 @@ public Optional findActiveTask(Long userId, Long mockApplyId) ); } + @Transactional(readOnly = true) + public Optional findRecoverablePublishFailureTask(Long userId, Long mockApplyId) { + return analysisAsyncTaskRepository + .findFirstByUserIdAndMockApplyIdAndStatusAndFailureReasonOrderByCreatedAtDesc( + userId, + mockApplyId, + AnalysisAsyncTaskStatus.FAILED, + AnalysisAsyncFailureReason.PUBLISH_FAILED + ); + } + @Transactional public void markRunning(String taskId, String workerId, int retryCount, Instant submittedAt) { AnalysisAsyncTask task = getTask(taskId); @@ -185,6 +196,20 @@ public void markCreditReleased(String taskId) { getTask(taskId).markCreditReleased(); } + @Transactional + public AnalysisAsyncTask reopenPublishFailureTask(String taskId) { + AnalysisAsyncTask task = getTask(taskId); + if (!task.isRecoverablePublishFailure()) { + throw new GeneralException( + GeneralErrorCode.INVALID_PARAMETER, + "재접수할 수 없는 자소서 분석 비동기 작업입니다. taskId=" + taskId + ); + } + task.reopenForRepublish(); + publishAfterCommit(toStatusResponse(task)); + return task; + } + @Transactional(readOnly = true) public AnalysisAsyncCreditStatus getCreditStatus(String taskId) { return getTask(taskId).getCreditStatus(); diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncFacadeServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncFacadeServiceTest.java index 37eb34c..c54c894 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncFacadeServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncFacadeServiceTest.java @@ -117,6 +117,31 @@ void submitDoesNotReserveCreditBeforeWorkerStarts() { verify(analysisAsyncProcessor, times(1)).process(createdTask.getTaskId(), 1L, 10L, 3); } + @Test + @DisplayName("이전 PUBLISH_FAILED task가 있으면 새 task를 만들지 않고 같은 task를 재접수한다") + void submitReopensRecoverablePublishFailureTask() { + User user = User.signup("테스트 사용자", "analysis-async-recoverable@example.com", "encoded-password"); + ReflectionTestUtils.setField(user, "id", 1L); + AnalysisAsyncTask failedTask = AnalysisAsyncTask.pending(1L, 10L, 3); + failedTask.markFailed(AnalysisAsyncFailureReason.PUBLISH_FAILED, "publish failed", 0); + + when(userService.validateUser(user)).thenReturn(user); + when(analysisAsyncTaskService.findActiveTask(1L, 10L)).thenReturn(Optional.empty()); + when(analysisAsyncTaskService.findRecoverablePublishFailureTask(1L, 10L)).thenReturn(Optional.of(failedTask)); + when(analysisAsyncTaskService.reopenPublishFailureTask(failedTask.getTaskId())).thenReturn(failedTask); + + AnalysisAsyncSubmitResponse response = analysisAsyncFacadeService.submit(user, 10L); + + assertThat(response.taskId()).isEqualTo(failedTask.getTaskId()); + assertThat(response.status()).isEqualTo("PENDING"); + assertThat(response.cached()).isFalse(); + assertThat(response.resultAvailable()).isFalse(); + verify(analysisAsyncTaskService, never()).createPendingTask(1L, 10L); + verify(analysisService, never()).hasReusableAnalysis(user, 10L); + verify(analysisAsyncTaskService).reopenPublishFailureTask(failedTask.getTaskId()); + verify(analysisAsyncProcessor).process(failedTask.getTaskId(), 1L, 10L, 3); + } + @Test @DisplayName("메시지 발행 실패 시 task를 삭제하지 않고 PUBLISH_FAILED로 실패 처리한다") void submitMarksTaskFailedWhenPublishFails() { @@ -143,6 +168,32 @@ void submitMarksTaskFailedWhenPublishFails() { verify(analysisAsyncTaskService, never()).deleteTask(createdTask.getTaskId()); } + @Test + @DisplayName("재접수한 PUBLISH_FAILED task의 발행이 다시 실패하면 같은 task를 다시 실패 처리한다") + void submitMarksReopenedTaskFailedWhenPublishFailsAgain() { + User user = User.signup("테스트 사용자", "analysis-async-republish-fail@example.com", "encoded-password"); + ReflectionTestUtils.setField(user, "id", 1L); + AnalysisAsyncTask failedTask = AnalysisAsyncTask.pending(1L, 10L, 3); + failedTask.markFailed(AnalysisAsyncFailureReason.PUBLISH_FAILED, "publish failed", 0); + RuntimeException publishException = new RuntimeException("publish failed again"); + + when(userService.validateUser(user)).thenReturn(user); + when(analysisAsyncTaskService.findActiveTask(1L, 10L)).thenReturn(Optional.empty()); + when(analysisAsyncTaskService.findRecoverablePublishFailureTask(1L, 10L)).thenReturn(Optional.of(failedTask)); + when(analysisAsyncTaskService.reopenPublishFailureTask(failedTask.getTaskId())).thenReturn(failedTask); + doThrow(publishException).when(analysisAsyncProcessor).process(failedTask.getTaskId(), 1L, 10L, 3); + + assertThatThrownBy(() -> analysisAsyncFacadeService.submit(user, 10L)) + .isSameAs(publishException); + + verify(analysisAsyncTaskService).markFailed( + failedTask.getTaskId(), + AnalysisAsyncFailureReason.PUBLISH_FAILED, + "자소서 분석 비동기 작업 발행에 실패했습니다.", + failedTask.getRetryCount() + ); + } + @Test @DisplayName("제약 조건명이 없으면 메시지 기반으로 중복 충돌로 분류하지 않는다") void submitDoesNotTreatMessageOnlyConflictAsDuplicate() { From 78ea1c26a1fa9af49344af2adcba64c70bdaed94 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Wed, 12 Aug 2026 10:36:19 +0900 Subject: [PATCH 2/7] =?UTF-8?q?[Refactor]=20async=20timeout=20=EC=A1=B0?= =?UTF-8?q?=ED=9A=8C=EB=A5=BC=20batch=20query=EB=A1=9C=20=EB=B6=84?= =?UTF-8?q?=EB=A6=AC=20(#271)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../AnalysisAsyncTaskSweepCoordinator.java | 62 ++++++++- .../AnalysisAsyncTaskRepository.java | 23 +++- ...AnalysisAsyncTaskSweepCoordinatorTest.java | 121 ++++++++++++++++++ 3 files changed, 198 insertions(+), 8 deletions(-) create mode 100644 src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java index 7723c01..65d1589 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java @@ -11,15 +11,18 @@ import com.jobdri.jobdri_api.domain.user.entity.User; import com.jobdri.jobdri_api.domain.user.service.UserService; import lombok.extern.slf4j.Slf4j; +import org.springframework.data.domain.PageRequest; import org.springframework.transaction.support.TransactionTemplate; import java.time.Duration; import java.time.LocalDateTime; -import java.util.EnumSet; +import java.util.List; @Slf4j // timeout/retry 기준으로 만료된 분석 async task를 정리한다. public class AnalysisAsyncTaskSweepCoordinator { + private static final int SWEEP_BATCH_SIZE = 100; + private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository; private final AnalysisAsyncTaskService analysisAsyncTaskService; private final AnalysisCreditService analysisCreditService; @@ -44,18 +47,58 @@ public AnalysisAsyncTaskSweepCoordinator( } public int sweepTimedOutTasks() { + LocalDateTime now = LocalDateTime.now(); + int expiredCount = sweepTimedOutPendingTasks(now); + expiredCount += sweepTimedOutRunningTasks(now); + return expiredCount; + } + + private int sweepTimedOutPendingTasks(LocalDateTime now) { + LocalDateTime deadline = now.minusSeconds(analysisQueueProperties.getQueueTimeoutSeconds()); + return sweepTimedOutTaskIds( + () -> analysisAsyncTaskRepository.findTimedOutPendingTaskIds( + deadline, + PageRequest.of(0, SWEEP_BATCH_SIZE) + ) + ); + } + + private int sweepTimedOutRunningTasks(LocalDateTime now) { + LocalDateTime deadline = now.minusSeconds(analysisQueueProperties.getProcessingTimeoutSeconds()); + return sweepTimedOutTaskIds( + () -> analysisAsyncTaskRepository.findTimedOutRunningTaskIds( + deadline, + PageRequest.of(0, SWEEP_BATCH_SIZE) + ) + ); + } + + private int sweepTimedOutTaskIds(TaskIdBatchLoader taskIdBatchLoader) { int expiredCount = 0; - for (AnalysisAsyncTask task : analysisAsyncTaskRepository.findByStatusIn(EnumSet.of(AnalysisAsyncTaskStatus.PENDING, AnalysisAsyncTaskStatus.RUNNING))) { - try { - expiredCount += transactionTemplate.execute(status -> sweepTimedOutTask(task.getTaskId())); - } catch (RuntimeException e) { - log.error("Analysis async task sweep failed for taskId={}", task.getTaskId(), e); + while (true) { + List taskIds = taskIdBatchLoader.load(); + if (taskIds.isEmpty()) { + return expiredCount; + } + for (String taskId : taskIds) { + expiredCount += sweepTimedOutTask(taskId); + } + if (taskIds.size() < SWEEP_BATCH_SIZE) { + return expiredCount; } } - return expiredCount; } private int sweepTimedOutTask(String taskId) { + try { + return transactionTemplate.execute(status -> sweepTimedOutTaskInTransaction(taskId)); + } catch (RuntimeException e) { + log.error("Analysis async task sweep failed for taskId={}", taskId, e); + return 0; + } + } + + private int sweepTimedOutTaskInTransaction(String taskId) { AnalysisAsyncTask task = analysisAsyncTaskRepository.findByIdForUpdate(taskId).orElse(null); if (task == null || task.getStatus() == AnalysisAsyncTaskStatus.SUCCEEDED || task.getStatus() == AnalysisAsyncTaskStatus.FAILED) { return 0; @@ -117,4 +160,9 @@ private void releaseCreditIfNeeded(AnalysisAsyncTask task) { private record ExpirationDecision(AnalysisAsyncFailureReason failureReason, String errorMessage) { } + + @FunctionalInterface + private interface TaskIdBatchLoader { + List load(); + } } 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 cfc1180..bd407c6 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 @@ -6,6 +6,7 @@ import jakarta.persistence.LockModeType; import org.springframework.data.jpa.repository.JpaRepository; import org.springframework.data.jpa.repository.Lock; +import org.springframework.data.domain.Pageable; import org.springframework.data.jpa.repository.Query; import org.springframework.data.repository.query.Param; @@ -35,7 +36,27 @@ List findByUserIdAndMockApplyIdInAndStatusIn( Collection statuses ); - List findByStatusIn(Collection statuses); + @Query(""" + select task.taskId + from AnalysisAsyncTask task + where task.status = 'PENDING' + and task.submittedAt is not null + and task.submittedAt <= :deadline + order by task.submittedAt asc + """) + List findTimedOutPendingTaskIds(@Param("deadline") java.time.LocalDateTime deadline, Pageable pageable); + + @Query(""" + select task.taskId + from AnalysisAsyncTask task + where task.status = 'RUNNING' + and ( + (task.lastAttemptAt is not null and task.lastAttemptAt <= :deadline) + or (task.lastAttemptAt is null and task.startedAt is not null and task.startedAt <= :deadline) + ) + order by coalesce(task.lastAttemptAt, task.startedAt) asc + """) + List findTimedOutRunningTaskIds(@Param("deadline") java.time.LocalDateTime deadline, Pageable pageable); @Lock(LockModeType.PESSIMISTIC_WRITE) @Query("select task from AnalysisAsyncTask task where task.taskId = :taskId") diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java new file mode 100644 index 0000000..672c47c --- /dev/null +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java @@ -0,0 +1,121 @@ +package com.jobdri.jobdri_api.domain.analysis.infrastructure.async; + +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; +import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; +import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisAsyncTaskService; +import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisQueueProperties; +import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisCreditService; +import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncFailureReason; +import com.jobdri.jobdri_api.domain.user.service.UserService; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.data.domain.Pageable; +import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.TransactionStatus; +import org.springframework.transaction.support.TransactionCallback; +import org.springframework.transaction.support.TransactionTemplate; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Optional; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class AnalysisAsyncTaskSweepCoordinatorTest { + + @Mock + private AnalysisAsyncTaskRepository analysisAsyncTaskRepository; + + @Mock + private AnalysisAsyncTaskService analysisAsyncTaskService; + + @Mock + private AnalysisCreditService analysisCreditService; + + @Mock + private UserService userService; + + @Mock + private TransactionTemplate transactionTemplate; + + private AnalysisAsyncTaskSweepCoordinator analysisAsyncTaskSweepCoordinator; + private AnalysisQueueProperties analysisQueueProperties; + + @BeforeEach + void setUp() { + analysisQueueProperties = new AnalysisQueueProperties(); + analysisAsyncTaskSweepCoordinator = new AnalysisAsyncTaskSweepCoordinator( + analysisAsyncTaskRepository, + analysisAsyncTaskService, + analysisCreditService, + userService, + transactionTemplate, + analysisQueueProperties + ); + lenient().doAnswer(invocation -> { + @SuppressWarnings("unchecked") + TransactionCallback callback = invocation.getArgument(0); + return callback.doInTransaction(org.mockito.Mockito.mock(TransactionStatus.class)); + }).when(transactionTemplate).execute(any()); + } + + @Test + @DisplayName("sweep는 timeout 대상 task id만 batch 조회해 처리한다") + void sweepTimedOutTasksLoadsOnlyTimedOutTaskIdsInBatches() { + AnalysisAsyncTask pendingTask = AnalysisAsyncTask.pending(1L, 10L, 3); + ReflectionTestUtils.setField(pendingTask, "submittedAt", LocalDateTime.now().minusMinutes(10)); + AnalysisAsyncTask runningTask = AnalysisAsyncTask.pending(2L, 20L, 3); + runningTask.markRunning("worker-1", 0, null); + ReflectionTestUtils.setField(runningTask, "lastAttemptAt", LocalDateTime.now().minusMinutes(20)); + + when(analysisAsyncTaskRepository.findTimedOutPendingTaskIds(any(LocalDateTime.class), any(Pageable.class))) + .thenReturn(List.of(pendingTask.getTaskId()), List.of()); + when(analysisAsyncTaskRepository.findTimedOutRunningTaskIds(any(LocalDateTime.class), any(Pageable.class))) + .thenReturn(List.of(runningTask.getTaskId()), List.of()); + when(analysisAsyncTaskRepository.findByIdForUpdate(pendingTask.getTaskId())).thenReturn(Optional.of(pendingTask)); + when(analysisAsyncTaskRepository.findByIdForUpdate(runningTask.getTaskId())).thenReturn(Optional.of(runningTask)); + + int expiredCount = analysisAsyncTaskSweepCoordinator.sweepTimedOutTasks(); + + assertThat(expiredCount).isEqualTo(2); + verify(analysisAsyncTaskService).markFailed( + pendingTask.getTaskId(), + AnalysisAsyncFailureReason.QUEUE_TIMEOUT, + "자소서 분석 작업이 대기열에서 시간 내 처리되지 않았습니다.", + pendingTask.getRetryCount() + ); + verify(analysisAsyncTaskService).markFailed( + runningTask.getTaskId(), + AnalysisAsyncFailureReason.INTERNAL_ERROR, + "자소서 분석 작업이 처리 제한 시간을 초과했습니다.", + runningTask.getRetryCount() + ); + } + + @Test + @DisplayName("batch가 비어 있으면 추가 처리 없이 종료한다") + void sweepTimedOutTasksStopsWhenNoTimedOutTaskIdsExist() { + when(analysisAsyncTaskRepository.findTimedOutPendingTaskIds(any(LocalDateTime.class), any(Pageable.class))) + .thenReturn(List.of()); + when(analysisAsyncTaskRepository.findTimedOutRunningTaskIds(any(LocalDateTime.class), any(Pageable.class))) + .thenReturn(List.of()); + + int expiredCount = analysisAsyncTaskSweepCoordinator.sweepTimedOutTasks(); + + assertThat(expiredCount).isZero(); + verify(analysisAsyncTaskRepository, never()).findByIdForUpdate(anyString()); + verify(analysisAsyncTaskService, never()).markFailed(anyString(), any(), anyString(), org.mockito.ArgumentMatchers.anyInt()); + } +} From 8e5574adcd56c8d8eb2492a5269c0d5004814ad4 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Wed, 12 Aug 2026 10:37:43 +0900 Subject: [PATCH 3/7] =?UTF-8?q?[Refactor]=20async=20sweep=20=EC=8B=9C?= =?UTF-8?q?=EA=B0=84=20=EC=9D=98=EC=A1=B4=EC=84=B1=EC=9D=84=20Clock?= =?UTF-8?q?=EC=9C=BC=EB=A1=9C=20=EC=B9=98=ED=99=98=20(#271)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../AnalysisAsyncTaskSweepCoordinator.java | 10 ++++-- .../async/AnalysisAsyncSweepService.java | 8 +++-- .../jobdri_api/global/config/ClockConfig.java | 15 +++++++++ ...AnalysisAsyncTaskSweepCoordinatorTest.java | 32 +++++++++++++++---- 4 files changed, 53 insertions(+), 12 deletions(-) create mode 100644 src/main/java/com/jobdri/jobdri_api/global/config/ClockConfig.java diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java index 65d1589..39a028b 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java @@ -14,6 +14,7 @@ import org.springframework.data.domain.PageRequest; import org.springframework.transaction.support.TransactionTemplate; +import java.time.Clock; import java.time.Duration; import java.time.LocalDateTime; import java.util.List; @@ -29,6 +30,7 @@ public class AnalysisAsyncTaskSweepCoordinator { private final UserService userService; private final TransactionTemplate transactionTemplate; private final AnalysisQueueProperties analysisQueueProperties; + private final Clock clock; public AnalysisAsyncTaskSweepCoordinator( AnalysisAsyncTaskRepository analysisAsyncTaskRepository, @@ -36,7 +38,8 @@ public AnalysisAsyncTaskSweepCoordinator( AnalysisCreditService analysisCreditService, UserService userService, TransactionTemplate transactionTemplate, - AnalysisQueueProperties analysisQueueProperties + AnalysisQueueProperties analysisQueueProperties, + Clock clock ) { this.analysisAsyncTaskRepository = analysisAsyncTaskRepository; this.analysisAsyncTaskService = analysisAsyncTaskService; @@ -44,10 +47,11 @@ public AnalysisAsyncTaskSweepCoordinator( this.userService = userService; this.transactionTemplate = transactionTemplate; this.analysisQueueProperties = analysisQueueProperties; + this.clock = clock; } public int sweepTimedOutTasks() { - LocalDateTime now = LocalDateTime.now(); + LocalDateTime now = LocalDateTime.now(clock); int expiredCount = sweepTimedOutPendingTasks(now); expiredCount += sweepTimedOutRunningTasks(now); return expiredCount; @@ -120,7 +124,7 @@ private int sweepTimedOutTaskInTransaction(String taskId) { } private ExpirationDecision resolveExpiration(AnalysisAsyncTask task) { - LocalDateTime now = LocalDateTime.now(); + LocalDateTime now = LocalDateTime.now(clock); if (task.getStatus() == AnalysisAsyncTaskStatus.PENDING && isExpired(task.getSubmittedAt(), now, analysisQueueProperties.getQueueTimeoutSeconds())) { return new ExpirationDecision( diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncSweepService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncSweepService.java index af83518..da93ed3 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncSweepService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncSweepService.java @@ -7,6 +7,8 @@ import org.springframework.stereotype.Service; import org.springframework.transaction.support.TransactionTemplate; +import java.time.Clock; + @Service public class AnalysisAsyncSweepService extends AnalysisAsyncTaskSweepCoordinator { @@ -16,7 +18,8 @@ public AnalysisAsyncSweepService( AnalysisCreditService analysisCreditService, UserService userService, TransactionTemplate transactionTemplate, - AnalysisQueueProperties analysisQueueProperties + AnalysisQueueProperties analysisQueueProperties, + Clock clock ) { super( analysisAsyncTaskRepository, @@ -24,7 +27,8 @@ public AnalysisAsyncSweepService( analysisCreditService, userService, transactionTemplate, - analysisQueueProperties + analysisQueueProperties, + clock ); } } diff --git a/src/main/java/com/jobdri/jobdri_api/global/config/ClockConfig.java b/src/main/java/com/jobdri/jobdri_api/global/config/ClockConfig.java new file mode 100644 index 0000000..c4c8217 --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/global/config/ClockConfig.java @@ -0,0 +1,15 @@ +package com.jobdri.jobdri_api.global.config; + +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; + +import java.time.Clock; + +@Configuration +public class ClockConfig { + + @Bean + public Clock systemClock() { + return Clock.systemDefaultZone(); + } +} diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java index 672c47c..4d2e005 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java @@ -19,13 +19,17 @@ import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionTemplate; +import java.time.Clock; +import java.time.Instant; import java.time.LocalDateTime; +import java.time.ZoneId; import java.util.List; import java.util.Optional; import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.lenient; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.never; @@ -52,17 +56,20 @@ class AnalysisAsyncTaskSweepCoordinatorTest { private AnalysisAsyncTaskSweepCoordinator analysisAsyncTaskSweepCoordinator; private AnalysisQueueProperties analysisQueueProperties; + private Clock clock; @BeforeEach void setUp() { analysisQueueProperties = new AnalysisQueueProperties(); + clock = Clock.fixed(Instant.parse("2026-08-12T00:00:00Z"), ZoneId.of("Asia/Seoul")); analysisAsyncTaskSweepCoordinator = new AnalysisAsyncTaskSweepCoordinator( analysisAsyncTaskRepository, analysisAsyncTaskService, analysisCreditService, userService, transactionTemplate, - analysisQueueProperties + analysisQueueProperties, + clock ); lenient().doAnswer(invocation -> { @SuppressWarnings("unchecked") @@ -74,15 +81,19 @@ void setUp() { @Test @DisplayName("sweep는 timeout 대상 task id만 batch 조회해 처리한다") void sweepTimedOutTasksLoadsOnlyTimedOutTaskIdsInBatches() { + LocalDateTime now = LocalDateTime.now(clock); AnalysisAsyncTask pendingTask = AnalysisAsyncTask.pending(1L, 10L, 3); - ReflectionTestUtils.setField(pendingTask, "submittedAt", LocalDateTime.now().minusMinutes(10)); + ReflectionTestUtils.setField(pendingTask, "submittedAt", now.minusMinutes(10)); AnalysisAsyncTask runningTask = AnalysisAsyncTask.pending(2L, 20L, 3); runningTask.markRunning("worker-1", 0, null); - ReflectionTestUtils.setField(runningTask, "lastAttemptAt", LocalDateTime.now().minusMinutes(20)); + ReflectionTestUtils.setField(runningTask, "lastAttemptAt", now.minusMinutes(20)); - when(analysisAsyncTaskRepository.findTimedOutPendingTaskIds(any(LocalDateTime.class), any(Pageable.class))) + LocalDateTime pendingDeadline = now.minusSeconds(analysisQueueProperties.getQueueTimeoutSeconds()); + LocalDateTime runningDeadline = now.minusSeconds(analysisQueueProperties.getProcessingTimeoutSeconds()); + + when(analysisAsyncTaskRepository.findTimedOutPendingTaskIds(eq(pendingDeadline), any(Pageable.class))) .thenReturn(List.of(pendingTask.getTaskId()), List.of()); - when(analysisAsyncTaskRepository.findTimedOutRunningTaskIds(any(LocalDateTime.class), any(Pageable.class))) + when(analysisAsyncTaskRepository.findTimedOutRunningTaskIds(eq(runningDeadline), any(Pageable.class))) .thenReturn(List.of(runningTask.getTaskId()), List.of()); when(analysisAsyncTaskRepository.findByIdForUpdate(pendingTask.getTaskId())).thenReturn(Optional.of(pendingTask)); when(analysisAsyncTaskRepository.findByIdForUpdate(runningTask.getTaskId())).thenReturn(Optional.of(runningTask)); @@ -107,9 +118,16 @@ void sweepTimedOutTasksLoadsOnlyTimedOutTaskIdsInBatches() { @Test @DisplayName("batch가 비어 있으면 추가 처리 없이 종료한다") void sweepTimedOutTasksStopsWhenNoTimedOutTaskIdsExist() { - when(analysisAsyncTaskRepository.findTimedOutPendingTaskIds(any(LocalDateTime.class), any(Pageable.class))) + LocalDateTime now = LocalDateTime.now(clock); + when(analysisAsyncTaskRepository.findTimedOutPendingTaskIds( + eq(now.minusSeconds(analysisQueueProperties.getQueueTimeoutSeconds())), + any(Pageable.class) + )) .thenReturn(List.of()); - when(analysisAsyncTaskRepository.findTimedOutRunningTaskIds(any(LocalDateTime.class), any(Pageable.class))) + when(analysisAsyncTaskRepository.findTimedOutRunningTaskIds( + eq(now.minusSeconds(analysisQueueProperties.getProcessingTimeoutSeconds())), + any(Pageable.class) + )) .thenReturn(List.of()); int expiredCount = analysisAsyncTaskSweepCoordinator.sweepTimedOutTasks(); From 2fcd0ca0b95e3e6680f603f56d4666434212bf72 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Wed, 12 Aug 2026 10:45:59 +0900 Subject: [PATCH 4/7] =?UTF-8?q?[Refactor]=20async=20credit=20refund=20?= =?UTF-8?q?=EC=B1=85=EC=9E=84=20=EA=B3=B5=EC=9A=A9=ED=99=94=20(#271)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../AnalysisAsyncTaskSweepCoordinator.java | 26 ++------ .../async/AnalysisAsyncWorkerBridge.java | 17 ++--- .../async/AnalysisAsyncCreditCoordinator.java | 33 ++++++++++ .../async/AnalysisAsyncSweepService.java | 8 +-- .../async/AnalysisAsyncTaskService.java | 17 +---- .../async/AnalysisWorkerBridgeService.java | 2 + ...AnalysisAsyncTaskSweepCoordinatorTest.java | 11 +--- .../AnalysisAsyncCreditCoordinatorTest.java | 64 +++++++++++++++++++ .../AnalysisWorkerBridgeServiceTest.java | 13 ++-- 9 files changed, 124 insertions(+), 67 deletions(-) create mode 100644 src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinator.java create mode 100644 src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinatorTest.java diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java index 39a028b..90b0f9d 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java @@ -1,15 +1,12 @@ package com.jobdri.jobdri_api.domain.analysis.infrastructure.async; import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; -import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncCreditStatus; import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncFailureReason; import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncTaskStatus; import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; +import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisAsyncCreditCoordinator; import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisAsyncTaskService; import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisQueueProperties; -import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisCreditService; -import com.jobdri.jobdri_api.domain.user.entity.User; -import com.jobdri.jobdri_api.domain.user.service.UserService; import lombok.extern.slf4j.Slf4j; import org.springframework.data.domain.PageRequest; import org.springframework.transaction.support.TransactionTemplate; @@ -26,8 +23,7 @@ public class AnalysisAsyncTaskSweepCoordinator { private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository; private final AnalysisAsyncTaskService analysisAsyncTaskService; - private final AnalysisCreditService analysisCreditService; - private final UserService userService; + private final AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; private final TransactionTemplate transactionTemplate; private final AnalysisQueueProperties analysisQueueProperties; private final Clock clock; @@ -35,16 +31,14 @@ public class AnalysisAsyncTaskSweepCoordinator { public AnalysisAsyncTaskSweepCoordinator( AnalysisAsyncTaskRepository analysisAsyncTaskRepository, AnalysisAsyncTaskService analysisAsyncTaskService, - AnalysisCreditService analysisCreditService, - UserService userService, + AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator, TransactionTemplate transactionTemplate, AnalysisQueueProperties analysisQueueProperties, Clock clock ) { this.analysisAsyncTaskRepository = analysisAsyncTaskRepository; this.analysisAsyncTaskService = analysisAsyncTaskService; - this.analysisCreditService = analysisCreditService; - this.userService = userService; + this.analysisAsyncCreditCoordinator = analysisAsyncCreditCoordinator; this.transactionTemplate = transactionTemplate; this.analysisQueueProperties = analysisQueueProperties; this.clock = clock; @@ -113,7 +107,7 @@ private int sweepTimedOutTaskInTransaction(String taskId) { return 0; } - releaseCreditIfNeeded(task); + analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task); analysisAsyncTaskService.markFailed( task.getTaskId(), expirationDecision.failureReason(), @@ -152,16 +146,6 @@ private boolean isExpired(LocalDateTime baseTime, LocalDateTime now, long timeou return Duration.between(baseTime, now).getSeconds() >= timeoutSeconds; } - private void releaseCreditIfNeeded(AnalysisAsyncTask task) { - if (task.getCreditStatus() != AnalysisAsyncCreditStatus.RESERVED || task.getCreditReferenceId() == null) { - return; - } - - User user = userService.getUser(task.getUserId()); - analysisCreditService.refund(user, task.getCreditReferenceId()); - analysisAsyncTaskService.markCreditReleased(task.getTaskId()); - } - private record ExpirationDecision(AnalysisAsyncFailureReason failureReason, String errorMessage) { } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncWorkerBridge.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncWorkerBridge.java index b742e80..c25796b 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncWorkerBridge.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncWorkerBridge.java @@ -14,6 +14,7 @@ import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncTaskStatus; import com.jobdri.jobdri_api.domain.analysis.entity.Question; import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; +import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisAsyncCreditCoordinator; import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisAsyncTaskService; import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisCreditService; import com.jobdri.jobdri_api.domain.analysis.application.model.AnalysisExecutionPayload; @@ -49,6 +50,7 @@ public class AnalysisAsyncWorkerBridge { private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository; private final AnalysisService analysisService; private final AnalysisCreditService analysisCreditService; + private final AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; private final UserService userService; private final WorkerTaskResultService workerTaskResultService; private final AnalysisInputFingerprintProvider analysisInputFingerprintProvider; @@ -60,6 +62,7 @@ public AnalysisAsyncWorkerBridge( AnalysisAsyncTaskRepository analysisAsyncTaskRepository, AnalysisService analysisService, AnalysisCreditService analysisCreditService, + AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator, UserService userService, WorkerTaskResultService workerTaskResultService, AnalysisInputFingerprintProvider analysisInputFingerprintProvider, @@ -70,6 +73,7 @@ public AnalysisAsyncWorkerBridge( this.analysisAsyncTaskRepository = analysisAsyncTaskRepository; this.analysisService = analysisService; this.analysisCreditService = analysisCreditService; + this.analysisAsyncCreditCoordinator = analysisAsyncCreditCoordinator; this.userService = userService; this.workerTaskResultService = workerTaskResultService; this.analysisInputFingerprintProvider = analysisInputFingerprintProvider; @@ -124,7 +128,7 @@ public void failTask( } analysisAsyncTaskService.updateWorkerMetadata(taskId, workerId, queueLatencyMillis); - releaseCreditIfNeeded(task); + analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task); analysisAsyncTaskService.markFailed(taskId, failureReason, errorMessage, retryCount); try (var ignored = LoggingContext.with("worker.task.failed", null, workerContext(taskId, "ANALYSIS", workerId, retryCount, queueLatencyMillis))) { log.warn("Analysis worker failed task: failureReason={}", failureReason); @@ -414,22 +418,13 @@ private void confirmCreditIfNeeded(AnalysisAsyncTask task) { analysisAsyncTaskService.markCreditConfirmed(task.getTaskId()); } - private void releaseCreditIfNeeded(AnalysisAsyncTask task) { - if (task.getCreditStatus() != AnalysisAsyncCreditStatus.RESERVED || task.getCreditReferenceId() == null) { - return; - } - User user = userService.getUser(task.getUserId()); - analysisCreditService.refund(user, task.getCreditReferenceId()); - analysisAsyncTaskService.markCreditReleased(task.getTaskId()); - } - private void releaseCreditAfterContextFailure(String taskId) { transactionTemplate.execute(status -> { AnalysisAsyncTask task = getTaskForUpdate(taskId); if (task.getExecutionContextSnapshot() != null) { return null; } - releaseCreditIfNeeded(task); + analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task); return null; }); } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinator.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinator.java new file mode 100644 index 0000000..a3585af --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinator.java @@ -0,0 +1,33 @@ +package com.jobdri.jobdri_api.domain.analysis.service.async; + +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; +import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisCreditService; +import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncCreditStatus; +import com.jobdri.jobdri_api.domain.user.entity.User; +import com.jobdri.jobdri_api.domain.user.service.UserService; +import org.springframework.stereotype.Service; + +@Service +public class AnalysisAsyncCreditCoordinator { + private final AnalysisCreditService analysisCreditService; + private final UserService userService; + + public AnalysisAsyncCreditCoordinator( + AnalysisCreditService analysisCreditService, + UserService userService + ) { + this.analysisCreditService = analysisCreditService; + this.userService = userService; + } + + public boolean releaseReservedCreditIfNeeded(AnalysisAsyncTask task) { + if (task.getCreditStatus() != AnalysisAsyncCreditStatus.RESERVED || task.getCreditReferenceId() == null) { + return false; + } + + User user = userService.getUser(task.getUserId()); + analysisCreditService.refund(user, task.getCreditReferenceId()); + task.markCreditReleased(); + return true; + } +} diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncSweepService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncSweepService.java index da93ed3..abee116 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncSweepService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncSweepService.java @@ -2,8 +2,6 @@ import com.jobdri.jobdri_api.domain.analysis.infrastructure.async.AnalysisAsyncTaskSweepCoordinator; import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; -import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisCreditService; -import com.jobdri.jobdri_api.domain.user.service.UserService; import org.springframework.stereotype.Service; import org.springframework.transaction.support.TransactionTemplate; @@ -15,8 +13,7 @@ public class AnalysisAsyncSweepService extends AnalysisAsyncTaskSweepCoordinator public AnalysisAsyncSweepService( AnalysisAsyncTaskRepository analysisAsyncTaskRepository, AnalysisAsyncTaskService analysisAsyncTaskService, - AnalysisCreditService analysisCreditService, - UserService userService, + AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator, TransactionTemplate transactionTemplate, AnalysisQueueProperties analysisQueueProperties, Clock clock @@ -24,8 +21,7 @@ public AnalysisAsyncSweepService( super( analysisAsyncTaskRepository, analysisAsyncTaskService, - analysisCreditService, - userService, + analysisAsyncCreditCoordinator, transactionTemplate, analysisQueueProperties, clock diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java index ea312e6..38c54b0 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java @@ -12,9 +12,6 @@ import com.jobdri.jobdri_api.domain.notification.entity.NotificationType; import com.jobdri.jobdri_api.domain.notification.service.NotificationService; import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; -import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisCreditService; -import com.jobdri.jobdri_api.domain.user.entity.User; -import com.jobdri.jobdri_api.domain.user.service.UserService; import com.jobdri.jobdri_api.global.metrics.AsyncMetricsRecorder; import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode; import com.jobdri.jobdri_api.global.apiPayload.exception.GeneralException; @@ -57,8 +54,7 @@ public class AnalysisAsyncTaskService { private final NotificationService notificationService; private final AsyncMetricsRecorder asyncMetricsRecorder; private final AnalysisQueueProperties analysisQueueProperties; - private final AnalysisCreditService analysisCreditService; - private final UserService userService; + private final AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; private final AsyncProgressCalculator asyncProgressCalculator; @Transactional @@ -163,7 +159,7 @@ public AnalysisAsyncCancelResponse cancelTask(Long userId, Long mockApplyId, Str boolean cancelled = task.getStatus() == AnalysisAsyncTaskStatus.CANCELLED; boolean newlyCancelled = previousStatus != AnalysisAsyncTaskStatus.CANCELLED && cancelled; if (newlyCancelled) { - releaseCreditIfNeeded(task); + analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task); recordProcessingMetric(task, "cancelled"); } if (newlyCancelled || previousCancelledAt == null && cancelled) { @@ -282,15 +278,6 @@ private AnalysisAsyncStatusResponse toStatusResponse(AnalysisAsyncTask task, Ana .build(); } - private void releaseCreditIfNeeded(AnalysisAsyncTask task) { - if (task.getCreditStatus() != AnalysisAsyncCreditStatus.RESERVED || task.getCreditReferenceId() == null) { - return; - } - User user = userService.getUser(task.getUserId()); - analysisCreditService.refund(user, task.getCreditReferenceId()); - task.markCreditReleased(); - } - private String resolveCurrentStep(AnalysisAsyncTask task) { return asyncProgressCalculator.resolveCurrentStep( toProgressStatus(task.getStatus()), diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeService.java index b21ccb9..77f188e 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeService.java @@ -21,6 +21,7 @@ public AnalysisWorkerBridgeService( AnalysisAsyncTaskRepository analysisAsyncTaskRepository, AnalysisService analysisService, AnalysisCreditService analysisCreditService, + AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator, UserService userService, WorkerTaskResultService workerTaskResultService, AnalysisInputFingerprintProvider analysisInputFingerprintProvider, @@ -32,6 +33,7 @@ public AnalysisWorkerBridgeService( analysisAsyncTaskRepository, analysisService, analysisCreditService, + analysisAsyncCreditCoordinator, userService, workerTaskResultService, analysisInputFingerprintProvider, diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java index 4d2e005..9712b2e 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java @@ -2,11 +2,10 @@ import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; +import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisAsyncCreditCoordinator; import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisAsyncTaskService; import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisQueueProperties; -import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisCreditService; import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncFailureReason; -import com.jobdri.jobdri_api.domain.user.service.UserService; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.DisplayName; import org.junit.jupiter.api.Test; @@ -46,10 +45,7 @@ class AnalysisAsyncTaskSweepCoordinatorTest { private AnalysisAsyncTaskService analysisAsyncTaskService; @Mock - private AnalysisCreditService analysisCreditService; - - @Mock - private UserService userService; + private AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; @Mock private TransactionTemplate transactionTemplate; @@ -65,8 +61,7 @@ void setUp() { analysisAsyncTaskSweepCoordinator = new AnalysisAsyncTaskSweepCoordinator( analysisAsyncTaskRepository, analysisAsyncTaskService, - analysisCreditService, - userService, + analysisAsyncCreditCoordinator, transactionTemplate, analysisQueueProperties, clock diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinatorTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinatorTest.java new file mode 100644 index 0000000..339874d --- /dev/null +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinatorTest.java @@ -0,0 +1,64 @@ +package com.jobdri.jobdri_api.domain.analysis.service.async; + +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; +import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisCreditService; +import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncCreditStatus; +import com.jobdri.jobdri_api.domain.user.entity.User; +import com.jobdri.jobdri_api.domain.user.service.UserService; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.InjectMocks; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import org.springframework.test.util.ReflectionTestUtils; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class AnalysisAsyncCreditCoordinatorTest { + + @Mock + private AnalysisCreditService analysisCreditService; + + @Mock + private UserService userService; + + @InjectMocks + private AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; + + @Test + @DisplayName("예약된 크레딧이 있으면 환불 후 RELEASED 상태로 전이한다") + void releaseReservedCreditIfNeededRefundsReservedCredit() { + AnalysisAsyncTask task = AnalysisAsyncTask.pending(1L, 10L, 3); + ReflectionTestUtils.invokeMethod(task, "markCreditReserved", "analysisTaskId=" + task.getTaskId()); + User user = User.signup("테스트 사용자", "analysis-credit-release@example.com", "encoded-password"); + ReflectionTestUtils.setField(user, "id", 1L); + + when(userService.getUser(1L)).thenReturn(user); + + boolean released = analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task); + + assertThat(released).isTrue(); + assertThat(task.getCreditStatus()).isEqualTo(AnalysisAsyncCreditStatus.RELEASED); + verify(userService).getUser(1L); + verify(analysisCreditService).refund(user, "analysisTaskId=" + task.getTaskId()); + } + + @Test + @DisplayName("예약된 크레딧이 없으면 환불을 수행하지 않는다") + void releaseReservedCreditIfNeededSkipsWhenNoReservedCredit() { + AnalysisAsyncTask task = AnalysisAsyncTask.pending(1L, 10L, 3); + + boolean released = analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task); + + assertThat(released).isFalse(); + verify(userService, never()).getUser(any()); + verify(analysisCreditService, never()).refund(any(), anyString()); + } +} diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java index b748aee..a8a928b 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java @@ -69,6 +69,9 @@ class AnalysisWorkerBridgeServiceTest { @Mock private AnalysisCreditService analysisCreditService; + @Mock + private AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; + @Mock private UserService userService; @@ -275,11 +278,10 @@ void failTaskReleasesReservedCreditOnlyOnce() { ReflectionTestUtils.setField(user, "id", 1L); when(analysisAsyncTaskRepository.findByIdForUpdate(task.getTaskId())).thenReturn(Optional.of(task)); - when(userService.getUser(1L)).thenReturn(user); - doAnswer(invocation -> { + when(analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task)).thenAnswer(invocation -> { task.markCreditReleased(); - return null; - }).when(analysisAsyncTaskService).markCreditReleased(task.getTaskId()); + return true; + }); doAnswer(invocation -> { task.markFailed(AnalysisAsyncFailureReason.INTERNAL_ERROR, "error", 1); return null; @@ -288,8 +290,7 @@ void failTaskReleasesReservedCreditOnlyOnce() { analysisWorkerBridgeService.failTask(task.getTaskId(), AnalysisAsyncFailureReason.INTERNAL_ERROR, "error", 1, "worker-1", 10L); analysisWorkerBridgeService.failTask(task.getTaskId(), AnalysisAsyncFailureReason.INTERNAL_ERROR, "error", 1, "worker-1", 10L); - verify(analysisCreditService, times(1)).refund(user, "analysisTaskId=" + task.getTaskId()); - verify(analysisAsyncTaskService, times(1)).markCreditReleased(task.getTaskId()); + verify(analysisAsyncCreditCoordinator, times(1)).releaseReservedCreditIfNeeded(task); } @Test From 6b8c6c53af14cb2fb2eea7b6fee1d234dc640706 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Wed, 12 Aug 2026 10:53:51 +0900 Subject: [PATCH 5/7] =?UTF-8?q?[Fix]=20async=20credit=20=EC=83=81=ED=83=9C?= =?UTF-8?q?=20=EC=A0=84=EC=9D=B4=20=EB=A9=B1=EB=93=B1=EC=84=B1=20=EB=B3=B4?= =?UTF-8?q?=EA=B0=95=20(#271)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../analysis/entity/AnalysisAsyncTask.java | 32 ++++++++++- .../async/AnalysisAsyncWorkerBridge.java | 26 +-------- .../async/AnalysisAsyncCreditCoordinator.java | 21 ++++++- .../async/AnalysisWorkerBridgeService.java | 3 - .../service/core/AnalysisCreditService.java | 4 +- src/main/resources/schema.sql | 7 +++ .../AnalysisAsyncCreditCoordinatorTest.java | 57 ++++++++++++++++++- .../AnalysisWorkerBridgeServiceTest.java | 42 +++++++------- 8 files changed, 137 insertions(+), 55 deletions(-) diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/entity/AnalysisAsyncTask.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/entity/AnalysisAsyncTask.java index 99d5c4e..248f47d 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/entity/AnalysisAsyncTask.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/entity/AnalysisAsyncTask.java @@ -38,6 +38,9 @@ public class AnalysisAsyncTask extends CreatedAtEntity { @Column(name = "credit_reference_id", length = 100) private String creditReferenceId; + @Column(name = "credit_reference_version", nullable = false) + private int creditReferenceVersion; + @Enumerated(EnumType.STRING) @Column(name = "credit_status", nullable = false, length = 20) private AnalysisAsyncCreditStatus creditStatus; @@ -107,6 +110,7 @@ public static AnalysisAsyncTask pending(Long userId, Long mockApplyId, int maxRe task.userId = userId; task.mockApplyId = mockApplyId; task.creditStatus = AnalysisAsyncCreditStatus.NONE; + task.creditReferenceVersion = 0; task.status = AnalysisAsyncTaskStatus.PENDING; task.message = "자소서 분석 비동기 작업이 접수되었습니다."; task.retryCount = 0; @@ -118,17 +122,39 @@ public static AnalysisAsyncTask pending(Long userId, Long mockApplyId, int maxRe return task; } - public void markCreditReserved(String creditReferenceId) { + public boolean markCreditReserved(String creditReferenceId) { + if (!canReserveCredit() || creditReferenceId == null || creditReferenceId.isBlank()) { + return false; + } this.creditReferenceId = creditReferenceId; + this.creditReferenceVersion += 1; this.creditStatus = AnalysisAsyncCreditStatus.RESERVED; + return true; } - public void markCreditConfirmed() { + public boolean markCreditConfirmed() { + if (creditStatus != AnalysisAsyncCreditStatus.RESERVED || creditReferenceId == null) { + return false; + } this.creditStatus = AnalysisAsyncCreditStatus.CONFIRMED; + return true; } - public void markCreditReleased() { + public boolean markCreditReleased() { + if (creditStatus != AnalysisAsyncCreditStatus.RESERVED || creditReferenceId == null) { + return false; + } this.creditStatus = AnalysisAsyncCreditStatus.RELEASED; + return true; + } + + public boolean canReserveCredit() { + return creditStatus == AnalysisAsyncCreditStatus.NONE + || creditStatus == AnalysisAsyncCreditStatus.RELEASED; + } + + public int nextCreditReferenceVersion() { + return creditReferenceVersion + 1; } public void markRunning(String workerId, int retryCount, Instant messageSubmittedAt) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncWorkerBridge.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncWorkerBridge.java index c25796b..081e3ee 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncWorkerBridge.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncWorkerBridge.java @@ -9,14 +9,12 @@ import com.jobdri.jobdri_api.domain.analysis.dto.internal.worker.AnalysisWorkerResultStoreRequest; import com.jobdri.jobdri_api.domain.analysis.dto.internal.worker.CorpusReferenceContext; import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; -import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncCreditStatus; import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncFailureReason; import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncTaskStatus; import com.jobdri.jobdri_api.domain.analysis.entity.Question; import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisAsyncCreditCoordinator; import com.jobdri.jobdri_api.domain.analysis.service.async.AnalysisAsyncTaskService; -import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisCreditService; import com.jobdri.jobdri_api.domain.analysis.application.model.AnalysisExecutionPayload; import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisInputFingerprintProvider; import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisService; @@ -49,7 +47,6 @@ public class AnalysisAsyncWorkerBridge { private final AnalysisAsyncTaskService analysisAsyncTaskService; private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository; private final AnalysisService analysisService; - private final AnalysisCreditService analysisCreditService; private final AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; private final UserService userService; private final WorkerTaskResultService workerTaskResultService; @@ -61,7 +58,6 @@ public AnalysisAsyncWorkerBridge( AnalysisAsyncTaskService analysisAsyncTaskService, AnalysisAsyncTaskRepository analysisAsyncTaskRepository, AnalysisService analysisService, - AnalysisCreditService analysisCreditService, AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator, UserService userService, WorkerTaskResultService workerTaskResultService, @@ -72,7 +68,6 @@ public AnalysisAsyncWorkerBridge( this.analysisAsyncTaskService = analysisAsyncTaskService; this.analysisAsyncTaskRepository = analysisAsyncTaskRepository; this.analysisService = analysisService; - this.analysisCreditService = analysisCreditService; this.analysisAsyncCreditCoordinator = analysisAsyncCreditCoordinator; this.userService = userService; this.workerTaskResultService = workerTaskResultService; @@ -287,7 +282,7 @@ private ContextAccess loadContextAccess(String taskId, Long userId, Long mockApp if (task.getExecutionContextSnapshot() != null) { return new ContextAccess(readContextSnapshot(task)); } - reserveCreditIfNeeded(task); + analysisAsyncCreditCoordinator.reserveCreditIfNeeded(task); return new ContextAccess(null); } @@ -304,7 +299,7 @@ private AnalysisWorkerContextResponse persistContextSnapshot( if (task.getExecutionContextSnapshot() != null) { return readContextSnapshot(task); } - reserveCreditIfNeeded(task); + analysisAsyncCreditCoordinator.reserveCreditIfNeeded(task); task.captureExecutionSnapshot(contextSnapshot, inputFingerprint); return context; } @@ -399,23 +394,8 @@ private void rejectIfCancelled(AnalysisAsyncTask task, String message) { } } - private void reserveCreditIfNeeded(AnalysisAsyncTask task) { - if (task.getCreditStatus() == AnalysisAsyncCreditStatus.RESERVED - || task.getCreditStatus() == AnalysisAsyncCreditStatus.CONFIRMED) { - return; - } - - User user = userService.getUser(task.getUserId()); - String creditReferenceId = analysisCreditService.createAsyncReferenceId(task.getTaskId()); - analysisCreditService.deduct(user, creditReferenceId); - analysisAsyncTaskService.markCreditReserved(task.getTaskId(), creditReferenceId); - } - private void confirmCreditIfNeeded(AnalysisAsyncTask task) { - if (task.getCreditStatus() != AnalysisAsyncCreditStatus.RESERVED || task.getCreditReferenceId() == null) { - return; - } - analysisAsyncTaskService.markCreditConfirmed(task.getTaskId()); + analysisAsyncCreditCoordinator.confirmReservedCreditIfNeeded(task); } private void releaseCreditAfterContextFailure(String taskId) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinator.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinator.java index a3585af..88a78c5 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinator.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinator.java @@ -27,7 +27,24 @@ public boolean releaseReservedCreditIfNeeded(AnalysisAsyncTask task) { User user = userService.getUser(task.getUserId()); analysisCreditService.refund(user, task.getCreditReferenceId()); - task.markCreditReleased(); - return true; + return task.markCreditReleased(); + } + + public boolean reserveCreditIfNeeded(AnalysisAsyncTask task) { + if (!task.canReserveCredit()) { + return false; + } + + User user = userService.getUser(task.getUserId()); + String creditReferenceId = analysisCreditService.createAsyncReferenceId( + task.getTaskId(), + task.nextCreditReferenceVersion() + ); + analysisCreditService.deduct(user, creditReferenceId); + return task.markCreditReserved(creditReferenceId); + } + + public boolean confirmReservedCreditIfNeeded(AnalysisAsyncTask task) { + return task.markCreditConfirmed(); } } diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeService.java index 77f188e..89c55cd 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeService.java @@ -3,7 +3,6 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.jobdri.jobdri_api.domain.analysis.infrastructure.async.AnalysisAsyncWorkerBridge; import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; -import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisCreditService; import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisInputFingerprintProvider; import com.jobdri.jobdri_api.domain.analysis.service.core.AnalysisService; import com.jobdri.jobdri_api.domain.user.service.UserService; @@ -20,7 +19,6 @@ public AnalysisWorkerBridgeService( AnalysisAsyncTaskService analysisAsyncTaskService, AnalysisAsyncTaskRepository analysisAsyncTaskRepository, AnalysisService analysisService, - AnalysisCreditService analysisCreditService, AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator, UserService userService, WorkerTaskResultService workerTaskResultService, @@ -32,7 +30,6 @@ public AnalysisWorkerBridgeService( analysisAsyncTaskService, analysisAsyncTaskRepository, analysisService, - analysisCreditService, analysisAsyncCreditCoordinator, userService, workerTaskResultService, diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/core/AnalysisCreditService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/core/AnalysisCreditService.java index 2b14be7..6799a26 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/core/AnalysisCreditService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/core/AnalysisCreditService.java @@ -16,8 +16,8 @@ public String createSyncReferenceId(Long mockApplyId, String inputFingerprint) { return "mockApplyId=" + mockApplyId + ":fingerprint=" + inputFingerprint; } - public String createAsyncReferenceId(String taskId) { - return "analysisTaskId=" + taskId; + public String createAsyncReferenceId(String taskId, int creditReferenceVersion) { + return "analysisTaskId=" + taskId + ":creditVersion=" + creditReferenceVersion; } @Transactional diff --git a/src/main/resources/schema.sql b/src/main/resources/schema.sql index f5d2c80..5da7422 100644 --- a/src/main/resources/schema.sql +++ b/src/main/resources/schema.sql @@ -58,6 +58,10 @@ CREATE INDEX IF NOT EXISTS idx_job_posting_embeddings_hnsw CREATE INDEX IF NOT EXISTS idx_analysis_async_tasks_user_mock_apply_status ON analysis_async_tasks (user_id, mock_apply_id, status); +CREATE UNIQUE INDEX IF NOT EXISTS idx_analysis_async_tasks_credit_reference_id_unique + ON analysis_async_tasks (credit_reference_id) + WHERE credit_reference_id IS NOT NULL; + ALTER TABLE IF EXISTS analyses ADD COLUMN IF NOT EXISTS missing_keywords TEXT NOT NULL DEFAULT '[]'; @@ -109,6 +113,9 @@ ALTER TABLE IF EXISTS analysis_async_tasks ALTER TABLE IF EXISTS analysis_async_tasks ADD COLUMN IF NOT EXISTS input_fingerprint_snapshot VARCHAR(64); +ALTER TABLE IF EXISTS analysis_async_tasks + ADD COLUMN IF NOT EXISTS credit_reference_version INTEGER NOT NULL DEFAULT 0; + ALTER TABLE IF EXISTS job_posting_async_tasks ADD COLUMN IF NOT EXISTS cancel_requested BOOLEAN NOT NULL DEFAULT FALSE; diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinatorTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinatorTest.java index 339874d..13f9d7d 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinatorTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinatorTest.java @@ -36,18 +36,20 @@ class AnalysisAsyncCreditCoordinatorTest { @DisplayName("예약된 크레딧이 있으면 환불 후 RELEASED 상태로 전이한다") void releaseReservedCreditIfNeededRefundsReservedCredit() { AnalysisAsyncTask task = AnalysisAsyncTask.pending(1L, 10L, 3); - ReflectionTestUtils.invokeMethod(task, "markCreditReserved", "analysisTaskId=" + task.getTaskId()); User user = User.signup("테스트 사용자", "analysis-credit-release@example.com", "encoded-password"); ReflectionTestUtils.setField(user, "id", 1L); when(userService.getUser(1L)).thenReturn(user); + when(analysisCreditService.createAsyncReferenceId(task.getTaskId(), 1)) + .thenReturn("analysisTaskId=" + task.getTaskId() + ":creditVersion=1"); + analysisAsyncCreditCoordinator.reserveCreditIfNeeded(task); boolean released = analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task); assertThat(released).isTrue(); assertThat(task.getCreditStatus()).isEqualTo(AnalysisAsyncCreditStatus.RELEASED); - verify(userService).getUser(1L); - verify(analysisCreditService).refund(user, "analysisTaskId=" + task.getTaskId()); + verify(userService, org.mockito.Mockito.times(2)).getUser(1L); + verify(analysisCreditService).refund(user, "analysisTaskId=" + task.getTaskId() + ":creditVersion=1"); } @Test @@ -61,4 +63,53 @@ void releaseReservedCreditIfNeededSkipsWhenNoReservedCredit() { verify(userService, never()).getUser(any()); verify(analysisCreditService, never()).refund(any(), anyString()); } + + @Test + @DisplayName("크레딧 재예약은 version을 올린 새 reference로 한 번만 수행한다") + void reserveCreditIfNeededUsesIncrementedReferenceVersion() { + AnalysisAsyncTask task = AnalysisAsyncTask.pending(1L, 10L, 3); + User user = User.signup("테스트 사용자", "analysis-credit-reserve@example.com", "encoded-password"); + ReflectionTestUtils.setField(user, "id", 1L); + + when(userService.getUser(1L)).thenReturn(user); + when(analysisCreditService.createAsyncReferenceId(task.getTaskId(), 1)) + .thenReturn("analysisTaskId=" + task.getTaskId() + ":creditVersion=1"); + when(analysisCreditService.createAsyncReferenceId(task.getTaskId(), 2)) + .thenReturn("analysisTaskId=" + task.getTaskId() + ":creditVersion=2"); + + boolean firstReserved = analysisAsyncCreditCoordinator.reserveCreditIfNeeded(task); + boolean secondReserved = analysisAsyncCreditCoordinator.reserveCreditIfNeeded(task); + boolean released = analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task); + boolean thirdReserved = analysisAsyncCreditCoordinator.reserveCreditIfNeeded(task); + + assertThat(firstReserved).isTrue(); + assertThat(secondReserved).isFalse(); + assertThat(released).isTrue(); + assertThat(thirdReserved).isTrue(); + assertThat(task.getCreditStatus()).isEqualTo(AnalysisAsyncCreditStatus.RESERVED); + assertThat(task.getCreditReferenceId()).isEqualTo("analysisTaskId=" + task.getTaskId() + ":creditVersion=2"); + verify(analysisCreditService).deduct(user, "analysisTaskId=" + task.getTaskId() + ":creditVersion=1"); + verify(analysisCreditService).refund(user, "analysisTaskId=" + task.getTaskId() + ":creditVersion=1"); + verify(analysisCreditService).deduct(user, "analysisTaskId=" + task.getTaskId() + ":creditVersion=2"); + } + + @Test + @DisplayName("예약된 크레딧만 confirm 되고 이후 중복 confirm은 무시된다") + void confirmReservedCreditIfNeededIsIdempotent() { + AnalysisAsyncTask task = AnalysisAsyncTask.pending(1L, 10L, 3); + User user = User.signup("테스트 사용자", "analysis-credit-confirm@example.com", "encoded-password"); + ReflectionTestUtils.setField(user, "id", 1L); + when(userService.getUser(1L)).thenReturn(user); + when(analysisCreditService.createAsyncReferenceId(task.getTaskId(), 1)) + .thenReturn("analysisTaskId=" + task.getTaskId() + ":creditVersion=1"); + + analysisAsyncCreditCoordinator.reserveCreditIfNeeded(task); + + boolean firstConfirmed = analysisAsyncCreditCoordinator.confirmReservedCreditIfNeeded(task); + boolean secondConfirmed = analysisAsyncCreditCoordinator.confirmReservedCreditIfNeeded(task); + + assertThat(firstConfirmed).isTrue(); + assertThat(secondConfirmed).isFalse(); + assertThat(task.getCreditStatus()).isEqualTo(AnalysisAsyncCreditStatus.CONFIRMED); + } } diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java index a8a928b..ffc260e 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java @@ -7,6 +7,7 @@ import com.jobdri.jobdri_api.domain.analysis.dto.internal.worker.AnalysisWorkerResultStoreRequest; import com.jobdri.jobdri_api.domain.analysis.dto.internal.worker.SimilarJobPostingContext; import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; +import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncCreditStatus; import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncFailureReason; import com.jobdri.jobdri_api.domain.analysis.entity.Question; import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; @@ -34,6 +35,7 @@ import org.mockito.Spy; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.test.util.ReflectionTestUtils; +import org.springframework.transaction.TransactionStatus; import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionTemplate; @@ -69,7 +71,6 @@ class AnalysisWorkerBridgeServiceTest { @Mock private AnalysisCreditService analysisCreditService; - @Mock private AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; @Mock @@ -87,15 +88,26 @@ class AnalysisWorkerBridgeServiceTest { @Spy private ObjectMapper objectMapper = new ObjectMapper(); - @InjectMocks private AnalysisWorkerBridgeService analysisWorkerBridgeService; @BeforeEach void setUp() { + analysisAsyncCreditCoordinator = new AnalysisAsyncCreditCoordinator(analysisCreditService, userService); + analysisWorkerBridgeService = new AnalysisWorkerBridgeService( + analysisAsyncTaskService, + analysisAsyncTaskRepository, + analysisService, + analysisAsyncCreditCoordinator, + userService, + workerTaskResultService, + analysisInputFingerprintProvider, + objectMapper, + transactionTemplate + ); lenient().when(transactionTemplate.execute(any(TransactionCallback.class))) .thenAnswer(invocation -> { TransactionCallback callback = invocation.getArgument(0); - return callback.doInTransaction(null); + return callback.doInTransaction(org.mockito.Mockito.mock(TransactionStatus.class)); }); } @@ -153,7 +165,6 @@ void getContextRejectsCancelledTaskWithoutCreditReservation() { .isInstanceOf(GeneralException.class); verify(analysisCreditService, never()).deduct(any(), anyString()); - verify(analysisAsyncTaskService, never()).markCreditReserved(anyString(), anyString()); } @Test @@ -210,21 +221,17 @@ void getContextReservesCreditBeforePreparingExecution() { when(analysisAsyncTaskRepository.findByIdForUpdate(task.getTaskId())).thenReturn(Optional.of(task)); when(userService.getUser(1L)).thenReturn(user); when(analysisService.prepareAnalysisExecution(user, 10L)).thenReturn(payload); - when(analysisCreditService.createAsyncReferenceId(task.getTaskId())) - .thenReturn("analysisTaskId=" + task.getTaskId()); + when(analysisCreditService.createAsyncReferenceId(task.getTaskId(), 1)) + .thenReturn("analysisTaskId=" + task.getTaskId() + ":creditVersion=1"); when(analysisInputFingerprintProvider.create(payload)).thenReturn("initial-fingerprint"); - doAnswer(invocation -> { - ReflectionTestUtils.invokeMethod(task, "markCreditReserved", invocation.getArgument(1, String.class)); - return null; - }).when(analysisAsyncTaskService).markCreditReserved(eq(task.getTaskId()), anyString()); var firstContext = analysisWorkerBridgeService.getContext(task.getTaskId(), 1L, 10L); var secondContext = analysisWorkerBridgeService.getContext(task.getTaskId(), 1L, 10L); - verify(analysisCreditService).createAsyncReferenceId(task.getTaskId()); - verify(analysisCreditService).deduct(user, "analysisTaskId=" + task.getTaskId()); - verify(analysisAsyncTaskService).markCreditReserved(task.getTaskId(), "analysisTaskId=" + task.getTaskId()); + verify(analysisCreditService).createAsyncReferenceId(task.getTaskId(), 1); + verify(analysisCreditService).deduct(user, "analysisTaskId=" + task.getTaskId() + ":creditVersion=1"); verify(analysisService, times(1)).prepareAnalysisExecution(user, 10L); + assertThat(task.getCreditStatus()).isEqualTo(AnalysisAsyncCreditStatus.RESERVED); assertThat(firstContext).isEqualTo(secondContext); assertThat(firstContext.corpusReferences()).hasSize(1); assertThat(firstContext.corpusReferences().getFirst().corpusId()).isEqualTo(11L); @@ -265,7 +272,6 @@ void getContextDoesNotReserveCreditTwice() { analysisWorkerBridgeService.getContext(task.getTaskId(), 1L, 10L); verify(analysisCreditService, never()).deduct(eq(user), anyString()); - verify(analysisAsyncTaskService, never()).markCreditReserved(eq(task.getTaskId()), anyString()); verify(analysisService).prepareAnalysisExecution(user, 10L); } @@ -278,10 +284,7 @@ void failTaskReleasesReservedCreditOnlyOnce() { ReflectionTestUtils.setField(user, "id", 1L); when(analysisAsyncTaskRepository.findByIdForUpdate(task.getTaskId())).thenReturn(Optional.of(task)); - when(analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task)).thenAnswer(invocation -> { - task.markCreditReleased(); - return true; - }); + when(userService.getUser(1L)).thenReturn(user); doAnswer(invocation -> { task.markFailed(AnalysisAsyncFailureReason.INTERNAL_ERROR, "error", 1); return null; @@ -290,7 +293,8 @@ void failTaskReleasesReservedCreditOnlyOnce() { analysisWorkerBridgeService.failTask(task.getTaskId(), AnalysisAsyncFailureReason.INTERNAL_ERROR, "error", 1, "worker-1", 10L); analysisWorkerBridgeService.failTask(task.getTaskId(), AnalysisAsyncFailureReason.INTERNAL_ERROR, "error", 1, "worker-1", 10L); - verify(analysisAsyncCreditCoordinator, times(1)).releaseReservedCreditIfNeeded(task); + verify(analysisCreditService, times(1)) + .refund(user, "analysisTaskId=" + task.getTaskId()); } @Test From 43d44f6951900a9acd58ea53155fe597068a4a76 Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Wed, 12 Aug 2026 10:59:32 +0900 Subject: [PATCH 6/7] =?UTF-8?q?[Fix]=20PUBLISH=5FFAILED=20=EC=9E=AC?= =?UTF-8?q?=EC=A0=91=EC=88=98=20=EC=9B=90=EC=9E=90=EC=84=B1=20=EB=B3=B4?= =?UTF-8?q?=EA=B0=95=20(#271)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../usecase/async/AnalysisAsyncUseCase.java | 8 +- .../async/AnalysisAsyncTaskService.java | 31 +++-- .../async/AnalysisAsyncFacadeServiceTest.java | 33 ++++- .../async/AnalysisAsyncTaskServiceTest.java | 119 ++++++++++++++++++ 4 files changed, 177 insertions(+), 14 deletions(-) create mode 100644 src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskServiceTest.java diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/application/usecase/async/AnalysisAsyncUseCase.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/application/usecase/async/AnalysisAsyncUseCase.java index 602b9db..ce88e07 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/application/usecase/async/AnalysisAsyncUseCase.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/application/usecase/async/AnalysisAsyncUseCase.java @@ -88,8 +88,12 @@ private AnalysisAsyncSubmitResponse createAndProcessTask(User user, Long mockApp } private AnalysisAsyncSubmitResponse reopenAndProcessTask(User user, Long mockApplyId, AnalysisAsyncTask task) { - AnalysisAsyncTask reopenedTask = analysisAsyncTaskService.reopenPublishFailureTask(task.getTaskId()); - return processTask(user, mockApplyId, reopenedTask); + AnalysisAsyncTaskService.ReopenPublishFailureResult reopenResult = + analysisAsyncTaskService.reopenPublishFailureTask(task.getTaskId()); + if (!reopenResult.reopened()) { + return toInProgressResponse(reopenResult.task()); + } + return processTask(user, mockApplyId, reopenResult.task()); } private AnalysisAsyncSubmitResponse processTask(User user, Long mockApplyId, AnalysisAsyncTask task) { diff --git a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java index 38c54b0..74beed0 100644 --- a/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskService.java @@ -193,17 +193,25 @@ public void markCreditReleased(String taskId) { } @Transactional - public AnalysisAsyncTask reopenPublishFailureTask(String taskId) { - AnalysisAsyncTask task = getTask(taskId); - if (!task.isRecoverablePublishFailure()) { - throw new GeneralException( - GeneralErrorCode.INVALID_PARAMETER, - "재접수할 수 없는 자소서 분석 비동기 작업입니다. taskId=" + taskId - ); + public ReopenPublishFailureResult reopenPublishFailureTask(String taskId) { + AnalysisAsyncTask task = analysisAsyncTaskRepository.findByIdForUpdate(taskId) + .orElseThrow(() -> new GeneralException( + GeneralErrorCode.ANALYSIS_ASYNC_TASK_NOT_FOUND, + "해당 자소서 분석 비동기 작업을 찾을 수 없습니다. taskId=" + taskId + )); + if (task.isRecoverablePublishFailure()) { + task.reopenForRepublish(); + publishAfterCommit(toStatusResponse(task)); + return new ReopenPublishFailureResult(task, true); } - task.reopenForRepublish(); - publishAfterCommit(toStatusResponse(task)); - return task; + if (task.getStatus() == AnalysisAsyncTaskStatus.PENDING + || task.getStatus() == AnalysisAsyncTaskStatus.RUNNING) { + return new ReopenPublishFailureResult(task, false); + } + throw new GeneralException( + GeneralErrorCode.INVALID_PARAMETER, + "재접수할 수 없는 자소서 분석 비동기 작업입니다. taskId=" + taskId + ); } @Transactional(readOnly = true) @@ -370,4 +378,7 @@ private void createFailureNotification(AnalysisAsyncTask task) { ); } + public record ReopenPublishFailureResult(AnalysisAsyncTask task, boolean reopened) { + } + } diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncFacadeServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncFacadeServiceTest.java index c54c894..7137ac6 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncFacadeServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncFacadeServiceTest.java @@ -128,7 +128,9 @@ void submitReopensRecoverablePublishFailureTask() { when(userService.validateUser(user)).thenReturn(user); when(analysisAsyncTaskService.findActiveTask(1L, 10L)).thenReturn(Optional.empty()); when(analysisAsyncTaskService.findRecoverablePublishFailureTask(1L, 10L)).thenReturn(Optional.of(failedTask)); - when(analysisAsyncTaskService.reopenPublishFailureTask(failedTask.getTaskId())).thenReturn(failedTask); + failedTask.reopenForRepublish(); + when(analysisAsyncTaskService.reopenPublishFailureTask(failedTask.getTaskId())) + .thenReturn(new AnalysisAsyncTaskService.ReopenPublishFailureResult(failedTask, true)); AnalysisAsyncSubmitResponse response = analysisAsyncFacadeService.submit(user, 10L); @@ -142,6 +144,31 @@ void submitReopensRecoverablePublishFailureTask() { verify(analysisAsyncProcessor).process(failedTask.getTaskId(), 1L, 10L, 3); } + @Test + @DisplayName("다른 요청이 이미 PUBLISH_FAILED task를 재접수했으면 재발행하지 않고 진행 중 응답을 반환한다") + void submitReturnsInProgressWhenPublishFailureTaskAlreadyReopened() { + User user = User.signup("테스트 사용자", "analysis-async-reopened@example.com", "encoded-password"); + ReflectionTestUtils.setField(user, "id", 1L); + AnalysisAsyncTask failedTask = AnalysisAsyncTask.pending(1L, 10L, 3); + failedTask.markFailed(AnalysisAsyncFailureReason.PUBLISH_FAILED, "publish failed", 0); + AnalysisAsyncTask reopenedTask = AnalysisAsyncTask.pending(1L, 10L, 3); + ReflectionTestUtils.setField(reopenedTask, "taskId", failedTask.getTaskId()); + + when(userService.validateUser(user)).thenReturn(user); + when(analysisAsyncTaskService.findActiveTask(1L, 10L)).thenReturn(Optional.empty()); + when(analysisAsyncTaskService.findRecoverablePublishFailureTask(1L, 10L)).thenReturn(Optional.of(failedTask)); + when(analysisAsyncTaskService.reopenPublishFailureTask(failedTask.getTaskId())) + .thenReturn(new AnalysisAsyncTaskService.ReopenPublishFailureResult(reopenedTask, false)); + + AnalysisAsyncSubmitResponse response = analysisAsyncFacadeService.submit(user, 10L); + + assertThat(response.taskId()).isEqualTo(failedTask.getTaskId()); + assertThat(response.status()).isEqualTo("PENDING"); + assertThat(response.cached()).isFalse(); + assertThat(response.resultAvailable()).isFalse(); + verify(analysisAsyncProcessor, never()).process(failedTask.getTaskId(), 1L, 10L, 3); + } + @Test @DisplayName("메시지 발행 실패 시 task를 삭제하지 않고 PUBLISH_FAILED로 실패 처리한다") void submitMarksTaskFailedWhenPublishFails() { @@ -180,7 +207,9 @@ void submitMarksReopenedTaskFailedWhenPublishFailsAgain() { when(userService.validateUser(user)).thenReturn(user); when(analysisAsyncTaskService.findActiveTask(1L, 10L)).thenReturn(Optional.empty()); when(analysisAsyncTaskService.findRecoverablePublishFailureTask(1L, 10L)).thenReturn(Optional.of(failedTask)); - when(analysisAsyncTaskService.reopenPublishFailureTask(failedTask.getTaskId())).thenReturn(failedTask); + failedTask.reopenForRepublish(); + when(analysisAsyncTaskService.reopenPublishFailureTask(failedTask.getTaskId())) + .thenReturn(new AnalysisAsyncTaskService.ReopenPublishFailureResult(failedTask, true)); doThrow(publishException).when(analysisAsyncProcessor).process(failedTask.getTaskId(), 1L, 10L, 3); assertThatThrownBy(() -> analysisAsyncFacadeService.submit(user, 10L)) diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskServiceTest.java new file mode 100644 index 0000000..c9a1335 --- /dev/null +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncTaskServiceTest.java @@ -0,0 +1,119 @@ +package com.jobdri.jobdri_api.domain.analysis.service.async; + +import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask; +import com.jobdri.jobdri_api.domain.analysis.repository.AnalysisAsyncTaskRepository; +import com.jobdri.jobdri_api.domain.analysis.type.AnalysisAsyncFailureReason; +import com.jobdri.jobdri_api.domain.notification.service.NotificationService; +import com.jobdri.jobdri_api.global.async.AsyncProgressCalculator; +import com.jobdri.jobdri_api.global.metrics.AsyncMetricsRecorder; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.Optional; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +@ExtendWith(MockitoExtension.class) +class AnalysisAsyncTaskServiceTest { + + @Mock + private AnalysisAsyncTaskRepository analysisAsyncTaskRepository; + + @Mock + private AnalysisAsyncSseService analysisAsyncSseService; + + @Mock + private NotificationService notificationService; + + @Mock + private AsyncMetricsRecorder asyncMetricsRecorder; + + @Mock + private AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; + + @Test + @DisplayName("동시에 재접수해도 PUBLISH_FAILED task는 한 번만 reopen 된다") + void reopenPublishFailureTaskIsAtomicAcrossConcurrentRequests() throws Exception { + AnalysisQueueProperties queueProperties = new AnalysisQueueProperties(); + AnalysisAsyncTaskService analysisAsyncTaskService = new AnalysisAsyncTaskService( + analysisAsyncTaskRepository, + analysisAsyncSseService, + notificationService, + asyncMetricsRecorder, + queueProperties, + analysisAsyncCreditCoordinator, + new AsyncProgressCalculator() + ); + AnalysisAsyncTask failedTask = spy(AnalysisAsyncTask.pending(1L, 10L, 3)); + failedTask.markFailed(AnalysisAsyncFailureReason.PUBLISH_FAILED, "publish failed", 0); + + CountDownLatch firstReopened = new CountDownLatch(1); + AtomicBoolean firstLookup = new AtomicBoolean(true); + doAnswer(invocation -> { + invocation.callRealMethod(); + firstReopened.countDown(); + return null; + }).when(failedTask).reopenForRepublish(); + when(analysisAsyncTaskRepository.findByIdForUpdate(failedTask.getTaskId())).thenAnswer(invocation -> { + if (firstLookup.getAndSet(false)) { + return Optional.of(failedTask); + } + assertThat(firstReopened.await(5, TimeUnit.SECONDS)).isTrue(); + return Optional.of(failedTask); + }); + + List results = runConcurrently(2, () -> + analysisAsyncTaskService.reopenPublishFailureTask(failedTask.getTaskId()) + ); + + assertThat(results).hasSize(2); + assertThat(results).extracting(AnalysisAsyncTaskService.ReopenPublishFailureResult::reopened) + .containsExactlyInAnyOrder(true, false); + assertThat(failedTask.getStatus().name()).isEqualTo("PENDING"); + verify(analysisAsyncSseService, times(1)).publish(org.mockito.ArgumentMatchers.any()); + } + + private List runConcurrently(int threadCount, Callable task) throws Exception { + ExecutorService executorService = Executors.newFixedThreadPool(threadCount); + CountDownLatch ready = new CountDownLatch(threadCount); + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(threadCount); + try { + for (int i = 0; i < threadCount; i++) { + futures.add(executorService.submit(() -> { + ready.countDown(); + assertThat(start.await(5, TimeUnit.SECONDS)).isTrue(); + return task.call(); + })); + } + assertThat(ready.await(5, TimeUnit.SECONDS)).isTrue(); + start.countDown(); + + List results = new ArrayList<>(threadCount); + for (Future future : futures) { + results.add(future.get(5, TimeUnit.SECONDS)); + } + return results; + } finally { + executorService.shutdownNow(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + } + } +} From 9f148794484916683c4064a343cae9f4f4bcebcd Mon Sep 17 00:00:00 2001 From: shinae1023 Date: Wed, 12 Aug 2026 11:05:28 +0900 Subject: [PATCH 7/7] =?UTF-8?q?[Fix]=20completeTask=20=EB=8F=99=EC=8B=9C?= =?UTF-8?q?=20=EC=99=84=EB=A3=8C=20=ED=9A=8C=EA=B7=80=20=ED=85=8C=EC=8A=A4?= =?UTF-8?q?=ED=8A=B8=20=EB=B3=B4=EA=B0=95=20(#271)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../AnalysisWorkerBridgeServiceTest.java | 101 ++++++++++++++++++ 1 file changed, 101 insertions(+) diff --git a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java index ffc260e..7d75224 100644 --- a/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisWorkerBridgeServiceTest.java @@ -39,8 +39,16 @@ import org.springframework.transaction.support.TransactionCallback; import org.springframework.transaction.support.TransactionTemplate; +import java.util.ArrayList; import java.util.List; import java.util.Optional; +import java.util.concurrent.Callable; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -55,6 +63,7 @@ import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; +import static org.mockito.Mockito.spy; @ExtendWith(MockitoExtension.class) class AnalysisWorkerBridgeServiceTest { @@ -546,4 +555,96 @@ void completeTaskRecoversAfterPublishFailure() { verify(analysisAsyncTaskService).markSuccess(task.getTaskId(), response); verify(workerTaskResultService).markDeliveredIfPresent(TaskType.ANALYSIS_COMPLETE, task.getTaskId()); } + + @Test + @DisplayName("동시에 complete가 들어와도 분석 완료 처리는 한 번만 수행한다") + void completeTaskProcessesSuccessOnlyOnceAcrossConcurrentRequests() throws Exception { + AnalysisAsyncTask task = spy(AnalysisAsyncTask.pending(1L, 10L, 3)); + task.markCreditReserved("credit-ref"); + ReflectionTestUtils.setField(task, "executionContextSnapshot", """ + {"userId":1,"mockApplyId":10,"companyName":"","jobTitle":"","task":"","requirements":"","preferredQualifications":"","bigClassificationName":"","middleClassificationName":"","detailClassificationName":"","questions":[],"corpusReferences":[],"similarJobPostings":[]} + """); + ReflectionTestUtils.setField(task, "inputFingerprintSnapshot", "complete-fingerprint"); + + User user = User.signup("테스트 사용자", "analysis-concurrent-complete@example.com", "encoded-password"); + ReflectionTestUtils.setField(user, "id", 1L); + + JobPosting jobPosting = mock(JobPosting.class, org.mockito.Answers.RETURNS_DEEP_STUBS); + AnalysisExecutionPayload completionPayload = new AnalysisExecutionPayload( + 1L, + 10L, + jobPosting, + List.of(), + List.of() + ); + AnalysisLlmResponse llmResponse = mock(AnalysisLlmResponse.class); + AnalysisResponse finalizedResponse = mock(AnalysisResponse.class); + AnalysisResponse cachedResponse = mock(AnalysisResponse.class); + AnalysisWorkerCompleteRequest request = new AnalysisWorkerCompleteRequest( + 1L, + 10L, + llmResponse, + "worker-1", + 15L + ); + + CountDownLatch successMarked = new CountDownLatch(1); + AtomicBoolean firstLookup = new AtomicBoolean(true); + when(analysisAsyncTaskRepository.findByIdForUpdate(task.getTaskId())).thenAnswer(invocation -> { + if (firstLookup.getAndSet(false)) { + return Optional.of(task); + } + assertThat(successMarked.await(5, TimeUnit.SECONDS)).isTrue(); + return Optional.of(task); + }); + when(userService.getUser(1L)).thenReturn(user); + when(analysisService.prepareAnalysisExecution(user, 10L, List.of())).thenReturn(completionPayload); + when(analysisService.finalizeAnalysis(user, 10L, completionPayload, llmResponse, "complete-fingerprint")) + .thenAnswer(invocation -> { + task.markSuccess(); + successMarked.countDown(); + return finalizedResponse; + }); + when(analysisService.getAnalysis(user, 10L)).thenReturn(cachedResponse); + + List results = runConcurrently( + 2, + () -> analysisWorkerBridgeService.completeTask(task.getTaskId(), request) + ); + + assertThat(results).containsExactlyInAnyOrder(finalizedResponse, cachedResponse); + assertThat(task.getCreditStatus()).isEqualTo(AnalysisAsyncCreditStatus.CONFIRMED); + verify(analysisService, times(1)) + .finalizeAnalysis(user, 10L, completionPayload, llmResponse, "complete-fingerprint"); + verify(analysisService, times(1)).getAnalysis(user, 10L); + verify(analysisAsyncTaskService, times(1)).markSuccess(task.getTaskId(), finalizedResponse); + verify(task, times(1)).markCreditConfirmed(); + } + + private List runConcurrently(int threadCount, Callable task) throws Exception { + ExecutorService executorService = Executors.newFixedThreadPool(threadCount); + CountDownLatch ready = new CountDownLatch(threadCount); + CountDownLatch start = new CountDownLatch(1); + List> futures = new ArrayList<>(threadCount); + try { + for (int i = 0; i < threadCount; i++) { + futures.add(executorService.submit(() -> { + ready.countDown(); + assertThat(start.await(5, TimeUnit.SECONDS)).isTrue(); + return task.call(); + })); + } + assertThat(ready.await(5, TimeUnit.SECONDS)).isTrue(); + start.countDown(); + + List results = new ArrayList<>(threadCount); + for (Future future : futures) { + results.add(future.get(5, TimeUnit.SECONDS)); + } + return results; + } finally { + executorService.shutdownNow(); + executorService.awaitTermination(5, TimeUnit.SECONDS); + } + } }