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 c5f9a4a4..ce88e074 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,20 @@ 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) { + 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) { + 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 3dac990f..248f47dd 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) { @@ -169,6 +195,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/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinator.java index 7723c01c..90b0f9df 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,61 +1,102 @@ 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; +import java.time.Clock; 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; - private final UserService userService; + private final AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; private final TransactionTemplate transactionTemplate; private final AnalysisQueueProperties analysisQueueProperties; + private final Clock clock; public AnalysisAsyncTaskSweepCoordinator( AnalysisAsyncTaskRepository analysisAsyncTaskRepository, AnalysisAsyncTaskService analysisAsyncTaskService, - AnalysisCreditService analysisCreditService, - UserService userService, + AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator, TransactionTemplate transactionTemplate, - AnalysisQueueProperties analysisQueueProperties + 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; } public int sweepTimedOutTasks() { + LocalDateTime now = LocalDateTime.now(clock); + 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; @@ -66,7 +107,7 @@ private int sweepTimedOutTask(String taskId) { return 0; } - releaseCreditIfNeeded(task); + analysisAsyncCreditCoordinator.releaseReservedCreditIfNeeded(task); analysisAsyncTaskService.markFailed( task.getTaskId(), expirationDecision.failureReason(), @@ -77,7 +118,7 @@ private int sweepTimedOutTask(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( @@ -105,16 +146,11 @@ 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) { } - 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/infrastructure/async/AnalysisAsyncWorkerBridge.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncWorkerBridge.java index b742e801..081e3ee7 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,13 +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; @@ -48,7 +47,7 @@ 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; private final AnalysisInputFingerprintProvider analysisInputFingerprintProvider; @@ -59,7 +58,7 @@ public AnalysisAsyncWorkerBridge( AnalysisAsyncTaskService analysisAsyncTaskService, AnalysisAsyncTaskRepository analysisAsyncTaskRepository, AnalysisService analysisService, - AnalysisCreditService analysisCreditService, + AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator, UserService userService, WorkerTaskResultService workerTaskResultService, AnalysisInputFingerprintProvider analysisInputFingerprintProvider, @@ -69,7 +68,7 @@ public AnalysisAsyncWorkerBridge( this.analysisAsyncTaskService = analysisAsyncTaskService; this.analysisAsyncTaskRepository = analysisAsyncTaskRepository; this.analysisService = analysisService; - this.analysisCreditService = analysisCreditService; + this.analysisAsyncCreditCoordinator = analysisAsyncCreditCoordinator; this.userService = userService; this.workerTaskResultService = workerTaskResultService; this.analysisInputFingerprintProvider = analysisInputFingerprintProvider; @@ -124,7 +123,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); @@ -283,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); } @@ -300,7 +299,7 @@ private AnalysisWorkerContextResponse persistContextSnapshot( if (task.getExecutionContextSnapshot() != null) { return readContextSnapshot(task); } - reserveCreditIfNeeded(task); + analysisAsyncCreditCoordinator.reserveCreditIfNeeded(task); task.captureExecutionSnapshot(contextSnapshot, inputFingerprint); return context; } @@ -395,32 +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()); - } - - 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()); + analysisAsyncCreditCoordinator.confirmReservedCreditIfNeeded(task); } private void releaseCreditAfterContextFailure(String taskId) { @@ -429,7 +404,7 @@ private void releaseCreditAfterContextFailure(String 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/repository/AnalysisAsyncTaskRepository.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/repository/AnalysisAsyncTaskRepository.java index de6ea0d3..bd407c6a 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,10 +1,12 @@ 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; 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; @@ -21,13 +23,40 @@ Optional findFirstByUserIdAndMockApplyIdAndStatusInOrderByCre Collection statuses ); + Optional findFirstByUserIdAndMockApplyIdAndStatusAndFailureReasonOrderByCreatedAtDesc( + Long userId, + Long mockApplyId, + AnalysisAsyncTaskStatus status, + AnalysisAsyncFailureReason failureReason + ); + List findByUserIdAndMockApplyIdInAndStatusIn( Long userId, Collection mockApplyIds, 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/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 00000000..88a78c5a --- /dev/null +++ b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinator.java @@ -0,0 +1,50 @@ +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()); + 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/AnalysisAsyncSweepService.java b/src/main/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncSweepService.java index af835182..abee116d 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,29 +2,29 @@ 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; +import java.time.Clock; + @Service public class AnalysisAsyncSweepService extends AnalysisAsyncTaskSweepCoordinator { public AnalysisAsyncSweepService( AnalysisAsyncTaskRepository analysisAsyncTaskRepository, AnalysisAsyncTaskService analysisAsyncTaskService, - AnalysisCreditService analysisCreditService, - UserService userService, + AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator, TransactionTemplate transactionTemplate, - AnalysisQueueProperties analysisQueueProperties + AnalysisQueueProperties analysisQueueProperties, + Clock clock ) { super( analysisAsyncTaskRepository, analysisAsyncTaskService, - analysisCreditService, - userService, + analysisAsyncCreditCoordinator, transactionTemplate, - analysisQueueProperties + 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 9696c623..74beed0a 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 @@ -82,6 +78,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); @@ -152,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) { @@ -185,6 +192,28 @@ public void markCreditReleased(String taskId) { getTask(taskId).markCreditReleased(); } + @Transactional + 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); + } + 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) public AnalysisAsyncCreditStatus getCreditStatus(String taskId) { return getTask(taskId).getCreditStatus(); @@ -257,15 +286,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()), @@ -358,4 +378,7 @@ private void createFailureNotification(AnalysisAsyncTask task) { ); } + public record ReopenPublishFailureResult(AnalysisAsyncTask task, boolean reopened) { + } + } 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 b21ccb91..89c55cdc 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,7 @@ public AnalysisWorkerBridgeService( AnalysisAsyncTaskService analysisAsyncTaskService, AnalysisAsyncTaskRepository analysisAsyncTaskRepository, AnalysisService analysisService, - AnalysisCreditService analysisCreditService, + AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator, UserService userService, WorkerTaskResultService workerTaskResultService, AnalysisInputFingerprintProvider analysisInputFingerprintProvider, @@ -31,7 +30,7 @@ public AnalysisWorkerBridgeService( analysisAsyncTaskService, analysisAsyncTaskRepository, analysisService, - analysisCreditService, + analysisAsyncCreditCoordinator, userService, workerTaskResultService, analysisInputFingerprintProvider, 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 2b14be7a..6799a265 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/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 00000000..c4c82175 --- /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/main/resources/schema.sql b/src/main/resources/schema.sql index f5d2c800..5da74226 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/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java new file mode 100644 index 00000000..9712b2eb --- /dev/null +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/infrastructure/async/AnalysisAsyncTaskSweepCoordinatorTest.java @@ -0,0 +1,134 @@ +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.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.type.AnalysisAsyncFailureReason; +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.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; +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 AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; + + @Mock + private TransactionTemplate transactionTemplate; + + 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, + analysisAsyncCreditCoordinator, + transactionTemplate, + analysisQueueProperties, + clock + ); + 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() { + LocalDateTime now = LocalDateTime.now(clock); + AnalysisAsyncTask pendingTask = AnalysisAsyncTask.pending(1L, 10L, 3); + ReflectionTestUtils.setField(pendingTask, "submittedAt", now.minusMinutes(10)); + AnalysisAsyncTask runningTask = AnalysisAsyncTask.pending(2L, 20L, 3); + runningTask.markRunning("worker-1", 0, null); + ReflectionTestUtils.setField(runningTask, "lastAttemptAt", now.minusMinutes(20)); + + 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(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)); + + 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() { + LocalDateTime now = LocalDateTime.now(clock); + when(analysisAsyncTaskRepository.findTimedOutPendingTaskIds( + eq(now.minusSeconds(analysisQueueProperties.getQueueTimeoutSeconds())), + any(Pageable.class) + )) + .thenReturn(List.of()); + when(analysisAsyncTaskRepository.findTimedOutRunningTaskIds( + eq(now.minusSeconds(analysisQueueProperties.getProcessingTimeoutSeconds())), + 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()); + } +} 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 00000000..13f9d7da --- /dev/null +++ b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncCreditCoordinatorTest.java @@ -0,0 +1,115 @@ +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); + 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, org.mockito.Mockito.times(2)).getUser(1L); + verify(analysisCreditService).refund(user, "analysisTaskId=" + task.getTaskId() + ":creditVersion=1"); + } + + @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()); + } + + @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/AnalysisAsyncFacadeServiceTest.java b/src/test/java/com/jobdri/jobdri_api/domain/analysis/service/async/AnalysisAsyncFacadeServiceTest.java index 37eb34cc..7137ac67 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,58 @@ 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)); + failedTask.reopenForRepublish(); + when(analysisAsyncTaskService.reopenPublishFailureTask(failedTask.getTaskId())) + .thenReturn(new AnalysisAsyncTaskService.ReopenPublishFailureResult(failedTask, true)); + + 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("다른 요청이 이미 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() { @@ -143,6 +195,34 @@ 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)); + 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)) + .isSameAs(publishException); + + verify(analysisAsyncTaskService).markFailed( + failedTask.getTaskId(), + AnalysisAsyncFailureReason.PUBLISH_FAILED, + "자소서 분석 비동기 작업 발행에 실패했습니다.", + failedTask.getRetryCount() + ); + } + @Test @DisplayName("제약 조건명이 없으면 메시지 기반으로 중복 충돌로 분류하지 않는다") void submitDoesNotTreatMessageOnlyConflictAsDuplicate() { 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 00000000..c9a13352 --- /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); + } + } +} 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 b748aeec..7d752248 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,11 +35,20 @@ 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; +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; @@ -53,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 { @@ -69,6 +80,8 @@ class AnalysisWorkerBridgeServiceTest { @Mock private AnalysisCreditService analysisCreditService; + private AnalysisAsyncCreditCoordinator analysisAsyncCreditCoordinator; + @Mock private UserService userService; @@ -84,15 +97,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)); }); } @@ -150,7 +174,6 @@ void getContextRejectsCancelledTaskWithoutCreditReservation() { .isInstanceOf(GeneralException.class); verify(analysisCreditService, never()).deduct(any(), anyString()); - verify(analysisAsyncTaskService, never()).markCreditReserved(anyString(), anyString()); } @Test @@ -207,21 +230,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); @@ -262,7 +281,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); } @@ -276,10 +294,6 @@ void failTaskReleasesReservedCreditOnlyOnce() { when(analysisAsyncTaskRepository.findByIdForUpdate(task.getTaskId())).thenReturn(Optional.of(task)); when(userService.getUser(1L)).thenReturn(user); - doAnswer(invocation -> { - task.markCreditReleased(); - return null; - }).when(analysisAsyncTaskService).markCreditReleased(task.getTaskId()); doAnswer(invocation -> { task.markFailed(AnalysisAsyncFailureReason.INTERNAL_ERROR, "error", 1); return null; @@ -288,8 +302,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(analysisCreditService, times(1)).refund(user, "analysisTaskId=" + task.getTaskId()); - verify(analysisAsyncTaskService, times(1)).markCreditReleased(task.getTaskId()); + verify(analysisCreditService, times(1)) + .refund(user, "analysisTaskId=" + task.getTaskId()); } @Test @@ -541,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); + } + } }