[Refactor] LLM 호출 동시성 및 중복 작업 제어 개선 (#113)

ops/db/migrations/20260704_analysis_async_tasks_active_unique.sql

@@ -0,0 +1,3 @@
+CREATE UNIQUE INDEX IF NOT EXISTS uk_analysis_async_tasks_active_user_mock_apply
+    ON analysis_async_tasks (user_id, mock_apply_id)
+    WHERE status IN ('PENDING', 'RUNNING');

src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAiClient.java

@@ -7,6 +7,7 @@
 import com.jobdri.jobdri_api.domain.corpus.service.CorpusRetrievalService.RetrievedJobPostingReference;
 import com.jobdri.jobdri_api.domain.corpus.service.CorpusRetrievalService.RetrievedQuestionReference;
 import com.jobdri.jobdri_api.domain.jobposting.entity.JobPosting;
+import com.jobdri.jobdri_api.global.config.LlmConcurrencyLimiter;
 import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode;
 import com.jobdri.jobdri_api.global.apiPayload.exception.GeneralException;
 import com.openai.client.OpenAIClient;
@@ -29,6 +30,7 @@ public class AnalysisAiClient {
 
     private final OpenAIClient openAIClient;
     private final CorpusRetrievalService corpusRetrievalService;
+    private final LlmConcurrencyLimiter llmConcurrencyLimiter;
 
     @Value("${openai.model.cover-letter-analysis:gpt-4o-mini}")
     private String analysisModel;
@@ -49,7 +51,10 @@ public AnalysisLlmResponse analyze(JobPosting jobPosting, List<Question> questio
                 .build();
 
         try {
-            StructuredResponse<AnalysisLlmResponse> response = openAIClient.responses().create(params);
+            StructuredResponse<AnalysisLlmResponse> response = llmConcurrencyLimiter.execute(
+                    "cover-letter-analysis",
+                    () -> openAIClient.responses().create(params)
+            );
             return extractStructuredContent(response);
         } catch (GeneralException e) {
             throw e;

src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncFacadeService.java

@@ -2,11 +2,13 @@
 
 import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncStatusResponse;
 import com.jobdri.jobdri_api.domain.analysis.dto.response.AnalysisAsyncSubmitResponse;
+import com.jobdri.jobdri_api.domain.analysis.entity.AnalysisAsyncTask;
 import com.jobdri.jobdri_api.domain.user.entity.User;
 import com.jobdri.jobdri_api.domain.user.service.UserService;
 import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode;
 import com.jobdri.jobdri_api.global.apiPayload.exception.GeneralException;
 import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.stereotype.Service;
 
 @Service
@@ -55,7 +57,17 @@ public AnalysisAsyncStatusResponse getTask(User user, Long mockApplyId, String t
     }
 
     private AnalysisAsyncSubmitResponse createAndProcessTask(User user, Long mockApplyId) {
-        String taskId = analysisAsyncTaskService.createPendingTask(user.getId(), mockApplyId);
+        PendingTaskResult pendingTaskResult = createPendingTask(user, mockApplyId);
+        if (!pendingTaskResult.created()) {
+            return new AnalysisAsyncSubmitResponse(
+                    pendingTaskResult.task().getTaskId(),
+                    pendingTaskResult.task().getStatus().name(),
+                    "이미 진행 중인 자소서 분석 작업이 있습니다."
+            );
+        }
+
+        AnalysisAsyncTask task = pendingTaskResult.task();
+        String taskId = task.getTaskId();
         String creditReferenceId = "analysisTaskId=" + taskId;
 
         try {
@@ -71,4 +83,20 @@ private AnalysisAsyncSubmitResponse createAndProcessTask(User user, Long mockApp
             throw e;
         }
     }
+
+    private PendingTaskResult createPendingTask(User user, Long mockApplyId) {
+        try {
+            return new PendingTaskResult(
+                    analysisAsyncTaskService.createPendingTask(user.getId(), mockApplyId),
+                    true
+            );
+        } catch (DataIntegrityViolationException e) {
+            AnalysisAsyncTask existingTask = analysisAsyncTaskService.findActiveTask(user.getId(), mockApplyId)
+                    .orElseThrow(() -> e);
+            return new PendingTaskResult(existingTask, false);
+        }
+    }
+
+    private record PendingTaskResult(AnalysisAsyncTask task, boolean created) {
+    }
 }

src/main/java/com/jobdri/jobdri_api/domain/analysis/service/AnalysisAsyncTaskService.java

@@ -20,9 +20,8 @@ public class AnalysisAsyncTaskService {
     private final AnalysisAsyncTaskRepository analysisAsyncTaskRepository;
 
     @Transactional
-    public String createPendingTask(Long userId, Long mockApplyId) {
-        AnalysisAsyncTask task = analysisAsyncTaskRepository.save(AnalysisAsyncTask.pending(userId, mockApplyId));
-        return task.getTaskId();
+    public AnalysisAsyncTask createPendingTask(Long userId, Long mockApplyId) {
+        return analysisAsyncTaskRepository.saveAndFlush(AnalysisAsyncTask.pending(userId, mockApplyId));
     }
 
     @Transactional

src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/JobPostingAiService.java

@@ -17,6 +17,7 @@
 import com.jobdri.jobdri_api.domain.corpus.service.CorpusRetrievalService.RetrievalContext;
 import com.jobdri.jobdri_api.domain.corpus.service.CorpusRetrievalService.RetrievedJobPostingReference;
 import com.jobdri.jobdri_api.domain.corpus.service.CorpusRetrievalService.RetrievedQuestionReference;
+import com.jobdri.jobdri_api.global.config.LlmConcurrencyLimiter;
 import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode;
 import com.jobdri.jobdri_api.global.apiPayload.exception.GeneralException;
 import com.openai.client.OpenAIClient;
@@ -41,6 +42,7 @@ public class JobPostingAiService {
     private final DetailClassificationRepository detailClassificationRepository;
     private final CorpusRetrievalService corpusRetrievalService;
     private final JobPostingImageStorageService jobPostingImageStorageService;
+    private final LlmConcurrencyLimiter llmConcurrencyLimiter;
 
     @Value("${openai.model.job-posting-extractor:gpt-4o-mini}")
     private String extractionModel;
@@ -68,7 +70,10 @@ public JobPostingGenerateResponse generateJobPosting(JobPostingGenerateRequest r
                 .build();
 
         try {
-            StructuredResponse<JobPostingGenerateResponse> response = openAIClient.responses().create(params);
+            StructuredResponse<JobPostingGenerateResponse> response = llmConcurrencyLimiter.execute(
+                    "job-posting-generate",
+                    () -> openAIClient.responses().create(params)
+            );
             JobPostingGenerateResponse generated = extractStructuredContent(response, JobPostingGenerateResponse.class);
             return normalizeGeneratedResponse(generated, request);
         } catch (Exception e) {
@@ -97,7 +102,10 @@ public JobPostingMockGenerateResponse generateMockJobPosting(JobPostingMockGener
                 .build();
 
         try {
-            StructuredResponse<JobPostingMockGenerateResponse> response = openAIClient.responses().create(params);
+            StructuredResponse<JobPostingMockGenerateResponse> response = llmConcurrencyLimiter.execute(
+                    "mock-job-posting-generate",
+                    () -> openAIClient.responses().create(params)
+            );
             JobPostingMockGenerateResponse generated = extractStructuredContent(
                     response,
                     JobPostingMockGenerateResponse.class
@@ -132,7 +140,10 @@ public JobPostingMockQuestionResponse generateMockRecommendedQuestions(
                 .build();
 
         try {
-            StructuredResponse<JobPostingMockQuestionResponse> response = openAIClient.responses().create(params);
+            StructuredResponse<JobPostingMockQuestionResponse> response = llmConcurrencyLimiter.execute(
+                    "mock-question-generate",
+                    () -> openAIClient.responses().create(params)
+            );
             JobPostingMockQuestionResponse generated = extractStructuredContent(
                     response,
                     JobPostingMockQuestionResponse.class
@@ -156,8 +167,10 @@ public JobPostingClassificationResultResponse classifyDetailClassification(
                 .build();
 
         try {
-            StructuredResponse<JobPostingClassificationResultResponse> response =
-                    openAIClient.responses().create(params);
+            StructuredResponse<JobPostingClassificationResultResponse> response = llmConcurrencyLimiter.execute(
+                    "job-posting-classification",
+                    () -> openAIClient.responses().create(params)
+            );
             JobPostingClassificationResultResponse classification =
                     extractStructuredContent(response, JobPostingClassificationResultResponse.class);
             return normalizeClassificationResponse(classification, candidates);
@@ -199,7 +212,10 @@ public JobPostingExtractResponse extractJobPosting(Long userId, String rawText,
                 .build();
 
         try {
-            StructuredResponse<JobPostingExtractResponse> response = openAIClient.responses().create(params);
+            StructuredResponse<JobPostingExtractResponse> response = llmConcurrencyLimiter.execute(
+                    "job-posting-extract",
+                    () -> openAIClient.responses().create(params)
+            );
             JobPostingExtractResponse extracted = extractStructuredContent(response, JobPostingExtractResponse.class);
             return normalizeResponse(extracted, rawText);
         } catch (Exception e) {

src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/MockQuestionCacheService.java

@@ -11,6 +11,7 @@
 import com.jobdri.jobdri_api.global.apiPayload.code.GeneralErrorCode;
 import com.jobdri.jobdri_api.global.apiPayload.exception.GeneralException;
 import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -27,53 +28,61 @@ public class MockQuestionCacheService {
     private final DetailClassificationRepository detailClassificationRepository;
     private final CompanyRepository companyRepository;
     private final JobPostingAiService jobPostingAiService;
+    private final MockQuestionInflightRegistry mockQuestionInflightRegistry;
 
     public List<String> getRecommendedQuestions(JobPostingMockGenerateRequest request) {
-        return mockQuestionCacheRepository
-                .findByCompany_IdAndDetailClassification_IdAndPromptVersion(
-                        request.companyId(),
-                        request.detailClassificationId(),
-                        PROMPT_VERSION
-                )
-                .map(this::copyQuestions)
-                .orElseGet(() -> createAndCacheQuestions(request));
+        return getCachedQuestions(request)
+                .orElseGet(() -> mockQuestionInflightRegistry.execute(cacheKey(request), () -> createAndCacheQuestions(request)));
     }
 
     public List<String> createAndCacheQuestions(JobPostingMockGenerateRequest request) {
+        return getCachedQuestions(request).orElseGet(() -> createAndCacheQuestionsInternal(request));
+    }
+
+    private List<String> copyQuestions(MockQuestionCache cache) {
+        return List.copyOf(cache.getQuestions());
+    }
+
+    private List<String> createAndCacheQuestionsInternal(JobPostingMockGenerateRequest request) {
+        DetailClassification detailClassification = detailClassificationRepository.findById(request.detailClassificationId())
+                .orElseThrow(() -> new GeneralException(
+                        GeneralErrorCode.CLASSIFICATION_NOT_FOUND,
+                        "해당 소분류를 찾을 수 없습니다. detailClassificationId=" + request.detailClassificationId()
+                ));
+        Company company = companyRepository.findById(request.companyId())
+                .orElseThrow(() -> new GeneralException(
+                        GeneralErrorCode.COMPANY_NOT_FOUND,
+                        "해당 회사를 찾을 수 없습니다. companyId=" + request.companyId()
+                ));
+
+        JobPostingMockQuestionResponse generated = jobPostingAiService.generateMockRecommendedQuestions(request, company);
+        try {
+            MockQuestionCache saved = mockQuestionCacheRepository.save(
+                    MockQuestionCache.create(
+                            company,
+                            detailClassification,
+                            PROMPT_VERSION,
+                            generated.recommendedQuestions()
+                    )
+            );
+            return copyQuestions(saved);
+        } catch (DataIntegrityViolationException e) {
+            return getCachedQuestions(request)
+                    .orElseThrow(() -> e);
+        }
+    }
+
+    private java.util.Optional<List<String>> getCachedQuestions(JobPostingMockGenerateRequest request) {
         return mockQuestionCacheRepository
                 .findByCompany_IdAndDetailClassification_IdAndPromptVersion(
                         request.companyId(),
                         request.detailClassificationId(),
                         PROMPT_VERSION
                 )
-                .map(this::copyQuestions)
-                .orElseGet(() -> {
-                    DetailClassification detailClassification = detailClassificationRepository.findById(request.detailClassificationId())
-                            .orElseThrow(() -> new GeneralException(
-                                    GeneralErrorCode.CLASSIFICATION_NOT_FOUND,
-                                    "해당 소분류를 찾을 수 없습니다. detailClassificationId=" + request.detailClassificationId()
-                            ));
-                    Company company = companyRepository.findById(request.companyId())
-                            .orElseThrow(() -> new GeneralException(
-                                    GeneralErrorCode.COMPANY_NOT_FOUND,
-                                    "해당 회사를 찾을 수 없습니다. companyId=" + request.companyId()
-                            ));
-
-                    JobPostingMockQuestionResponse generated =
-                            jobPostingAiService.generateMockRecommendedQuestions(request, company);
-                    MockQuestionCache saved = mockQuestionCacheRepository.save(
-                            MockQuestionCache.create(
-                                    company,
-                                    detailClassification,
-                                    PROMPT_VERSION,
-                                    generated.recommendedQuestions()
-                            )
-                    );
-                    return copyQuestions(saved);
-                });
+                .map(this::copyQuestions);
     }
 
-    private List<String> copyQuestions(MockQuestionCache cache) {
-        return List.copyOf(cache.getQuestions());
+    private String cacheKey(JobPostingMockGenerateRequest request) {
+        return request.companyId() + ":" + request.detailClassificationId() + ":" + PROMPT_VERSION;
     }
 }

src/main/java/com/jobdri/jobdri_api/domain/jobposting/service/MockQuestionInflightRegistry.java

@@ -0,0 +1,56 @@
+package com.jobdri.jobdri_api.domain.jobposting.service;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.FutureTask;
+
+@Component
+@RequiredArgsConstructor
+public class MockQuestionInflightRegistry {
+
+    private final ConcurrentHashMap<String, FutureTask<java.util.List<String>>> inflightTasks = new ConcurrentHashMap<>();
+
+    public java.util.List<String> execute(String key, TaskSupplier supplier) {
+        FutureTask<java.util.List<String>> task = new FutureTask<>(supplier::get);
+        FutureTask<java.util.List<String>> existingTask = inflightTasks.putIfAbsent(key, task);
+
+        if (existingTask == null) {
+            try {
+                task.run();
+                return task.get();
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                throw new IllegalStateException("추천 질문 생성 대기 중 인터럽트가 발생했습니다.", e);
+            } catch (ExecutionException e) {
+                throw unwrap(e);
+            } finally {
+                inflightTasks.remove(key, task);
+            }
+        }
+
+        try {
+            return existingTask.get();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            throw new IllegalStateException("추천 질문 생성 대기 중 인터럽트가 발생했습니다.", e);
+        } catch (ExecutionException e) {
+            throw unwrap(e);
+        }
+    }
+
+    private RuntimeException unwrap(ExecutionException executionException) {
+        Throwable cause = executionException.getCause();
+        if (cause instanceof RuntimeException runtimeException) {
+            return runtimeException;
+        }
+        return new IllegalStateException("추천 질문 생성 중 알 수 없는 오류가 발생했습니다.", cause);
+    }
+
+    @FunctionalInterface
+    public interface TaskSupplier {
+        java.util.List<String> get() throws Exception;
+    }
+}