diff --git a/.claude/launch.json b/.claude/launch.json new file mode 100644 index 00000000..51a03491 --- /dev/null +++ b/.claude/launch.json @@ -0,0 +1,11 @@ +{ + "version": "0.0.1", + "configurations": [ + { + "name": "backend", + "runtimeExecutable": "./backend/gradlew", + "runtimeArgs": ["-p", "backend", "bootRun", "--args=--spring.profiles.active=local"], + "port": 8080 + } + ] +} diff --git a/backend/src/main/java/com/opensource/docgrid/domain/rag/config/RagExecutionConfig.java b/backend/src/main/java/com/opensource/docgrid/domain/rag/config/RagExecutionConfig.java new file mode 100644 index 00000000..3fa0d212 --- /dev/null +++ b/backend/src/main/java/com/opensource/docgrid/domain/rag/config/RagExecutionConfig.java @@ -0,0 +1,78 @@ +package com.opensource.docgrid.domain.rag.config; + +import java.util.concurrent.Semaphore; +import java.util.concurrent.SynchronousQueue; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +import org.springframework.beans.factory.annotation.Value; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.scheduling.concurrent.CustomizableThreadFactory; + +/** + * RAG 답변 생성을 동시에 몇 건까지 처리할지 제한하는 실행 자원을 구성한다 (#340). + * + *

{@code domain/worker/config/WorkerExecutionConfig}(이 코드베이스에서 유일했던 커스텀 + * 스레드풀 선례)를 그대로 본떴다 — {@code core=max}인 {@link ThreadPoolExecutor} + + * {@link SynchronousQueue}(큐잉 없음) + {@link ThreadPoolExecutor.AbortPolicy}(꽉 차면 즉시 + * 거부, 조용히 쌓아두지 않음)로 "설정된 동시성만 즉시 실행"을 보장한다. + * + *

{@code embedding_jobs}의 {@code WorkerExecutionSlotPool}(별도 클래스, 종료 플래그 + + * introspection 메서드 포함)까지는 필요 없다 — RAG는 별도 워커 등록/우아한 종료 조율이나 + * 대시보드 노출 요구가 없어서, 같은 안전 성질(로컬 슬롯을 먼저 확보한 뒤에만 DB claim을 + * 시도해 "claim은 됐는데 실행할 스레드가 없는" 상태를 만들지 않는 것)을 순수 + * {@link Semaphore}만으로 재현한다. + * + *

{@code indexing.worker.enabled}로 켜고 끌 수 있는 인덱싱 워커와 달리, RAG Worker는 + * {@code RagSchedulingConfig}와 동일하게 조건 없이 항상 켜져 있어야 하는 검색 API 핵심 + * 경로라 {@code @ConditionalOnProperty}를 붙이지 않는다. + */ +@Configuration +public class RagExecutionConfig { + + public static final String RAG_WORKER_JOB_EXECUTOR = "ragWorkerJobExecutor"; + public static final String RAG_WORKER_SLOTS = "ragWorkerSlots"; + + /** + * 동시에 최대 {@code rag.worker.max-concurrency}건까지만 즉시 실행하는 무대기 Executor를 만든다. + */ + @Bean(name = RAG_WORKER_JOB_EXECUTOR, destroyMethod = "shutdownNow") + public ThreadPoolExecutor ragWorkerJobExecutor( + @Value("${rag.worker.max-concurrency:2}") int maxConcurrency + ) { + validateMaxConcurrency(maxConcurrency); + return new ThreadPoolExecutor( + maxConcurrency, + maxConcurrency, + 0L, + TimeUnit.MILLISECONDS, + new SynchronousQueue<>(), + new CustomizableThreadFactory("rag-worker-job-"), + new ThreadPoolExecutor.AbortPolicy() + ); + } + + /** + * DB claim을 시도하기 전에 먼저 확보해야 하는 로컬 실행 슬롯. 공정 모드(fair)로 만들어 + * 폴링 주기가 겹칠 때 대기가 한쪽으로 몰리지 않게 한다 — {@code WorkerExecutionSlotPool}의 + * 선택과 동일하다. + */ + @Bean(name = RAG_WORKER_SLOTS) + public Semaphore ragWorkerSlots(@Value("${rag.worker.max-concurrency:2}") int maxConcurrency) { + validateMaxConcurrency(maxConcurrency); + return new Semaphore(maxConcurrency, true); + } + + /** + * {@code max-concurrency=0}(또는 음수)은 예외 없이 Worker를 영구 대기 상태로 만들 수 있어 + * ({@link Semaphore}는 permit 0으로도 생성 자체는 허용) 기동 시점에 바로 실패시킨다. + */ + private void validateMaxConcurrency(int maxConcurrency) { + if (maxConcurrency < 1) { + throw new IllegalArgumentException( + "rag.worker.max-concurrency는 1 이상이어야 합니다: " + maxConcurrency + ); + } + } +} diff --git a/backend/src/main/java/com/opensource/docgrid/domain/rag/entity/RagResponse.java b/backend/src/main/java/com/opensource/docgrid/domain/rag/entity/RagResponse.java index 73532eb3..3c6498c1 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/rag/entity/RagResponse.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/rag/entity/RagResponse.java @@ -1,5 +1,7 @@ package com.opensource.docgrid.domain.rag.entity; +import java.time.LocalDateTime; + import com.opensource.docgrid.domain.search.entity.SearchQuery; import com.opensource.docgrid.domain.search.enums.ResultStatus; import com.opensource.docgrid.global.common.entity.BaseEntity; @@ -88,6 +90,12 @@ public class RagResponse extends BaseEntity { @Column(name = "error_message", columnDefinition = "TEXT") private String errorMessage; + // 병렬 Worker가 이 job을 이미 집었는지 표시한다(#340). status만으로는 "대기 중"과 "누가 이미 + // 처리 중"을 구분할 수 없어서(둘 다 PROCESSING) 별도로 둔다. RagResponseClaimService의 짧은 + // claim 트랜잭션 안에서만 채워지며, 그 밖의 완료 확정 경로(조건부 UPDATE)는 이 컬럼을 건드리지 않는다. + @Column(name = "claimed_at") + private LocalDateTime claimedAt; + @Builder public RagResponse(SearchQuery query, String answerText, String llmProvider, String llmModelName, String promptText, Integer inputTokenCount, Integer outputTokenCount, Integer latencyMs, @@ -103,4 +111,13 @@ public RagResponse(SearchQuery query, String answerText, String llmProvider, Str this.status = status; this.errorMessage = errorMessage; } + + /** + * 이 job을 지금 이 Worker가 처리하기 시작했다는 표시를 남긴다. {@code RagResponseClaimService}의 + * 짧은 claim 트랜잭션 안에서만 호출되어야 한다 — dirty checking으로 반영되므로, 이 엔티티가 + * detached된 뒤(다른 트랜잭션/스레드로 넘어간 뒤)에 호출하면 반영되지 않는다. + */ + public void markClaimed(LocalDateTime claimedAt) { + this.claimedAt = claimedAt; + } } diff --git a/backend/src/main/java/com/opensource/docgrid/domain/rag/repository/RagResponseRepository.java b/backend/src/main/java/com/opensource/docgrid/domain/rag/repository/RagResponseRepository.java index e25e2e08..db4defe6 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/rag/repository/RagResponseRepository.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/rag/repository/RagResponseRepository.java @@ -22,17 +22,49 @@ public interface RagResponseRepository extends JpaRepository { /** - * 주어진 상태(보통 PROCESSING)인 것들 중 가장 오래 기다린 것 하나를 반환한다 — RagJobWorker가 - * 1초마다 폴링하며 이 메서드로 FIFO 큐를 구현한다. Worker가 1개뿐이라 별도 락/claim 없이도 - * 안전하다. - * - *

{@code query}/{@code query.user}를 {@link EntityGraph}로 미리 fetch한다 — Worker가 - * 이 메서드로 job을 꺼낸 트랜잭션이 끝난 뒤(WebSocket push 시점)에 - * {@code job.getQuery().getUser().getEmail()}에 접근해도 두 연관관계 모두 LAZY라서 자칫 - * {@code LazyInitializationException}이 날 수 있는데, 미리 로딩해두면 그 문제가 없다. + * PROCESSING 중 아직 아무 Worker도 집지 않은(claim 안 된) 것 하나를 골라 행 잠금을 건다(#340). + * {@code claimed_at IS NULL} 조건이 "대기 중"과 "이미 처리 중"을 구분하는 유일한 신호다 — + * status만으로는 둘 다 PROCESSING이라 구분이 안 된다. {@code FOR UPDATE SKIP LOCKED}로 + * 여러 Worker가 동시에 이 쿼리를 날려도 이미 잠긴 행은 건너뛰고 그다음 미잠금 행을 잡아온다. + * {@code created_at}이 같은 밀리초를 공유할 수 있는 동시 접수 상황을 대비해 {@code id}를 + * 2차 정렬 기준으로 둔다. {@link RagResponseClaimService}가 이 메서드로 잠근 행을 같은 짧은 + * 트랜잭션 안에서 즉시 {@link RagResponse#markClaimed}로 확정하고 커밋해, 락을 오래 들고 + * 있지 않는다({@code embedding_jobs}의 claim 패턴과 동일). + */ + @Query(value = """ + SELECT * FROM rag_responses + WHERE status = 'PROCESSING' AND claimed_at IS NULL + ORDER BY created_at ASC, id ASC + LIMIT 1 + FOR UPDATE SKIP LOCKED + """, nativeQuery = true) + Optional findNextUnclaimedProcessingForUpdate(); + + /** + * claim된 job을 실제로 처리하는 Worker 스레드가 쓰는 조회. {@code query}/{@code query.user}를 + * {@link EntityGraph}로 미리 fetch해, 처리가 끝난 뒤(WebSocket push 시점) 트랜잭션 밖에서 + * {@code job.getQuery().getUser().getEmail()}에 접근해도 {@code LazyInitializationException}이 + * 나지 않게 한다. 기존 {@code findById}는 그대로 두고 이름을 다르게 둔 이유는, {@code RagFacade}가 + * 이미 쓰고 있는 평범한 {@code findById(jobId)} 호출의 동작을 이번 변경으로 건드리지 않기 위함이다. */ @EntityGraph(attributePaths = {"query", "query.user"}) - Optional findFirstByStatusOrderByCreatedAtAsc(ResultStatus status); + Optional findWithQueryAndUserById(Long id); + + /** + * 앱 재시작 복구 전용(#340 CodeRabbit 리뷰 반영). 이전 프로세스가 claim한 채 완료하지 못하고 + * 죽은 job은 {@code claimed_at}이 채워진 상태로 DB에 남는다 — 이 상태로는 + * {@link #findNextUnclaimedProcessingForUpdate}가 절대 다시 집어주지 않아, 스위퍼의 + * {@code stale-threshold} 강제종료(fallback)만 기다리게 된다. 재시작 직후 한 번, + * PROCESSING인데 claim만 남아있는 행의 claim을 전부 풀어 새 Worker가 다시 시도할 수 있게 + * 한다. "인스턴스는 항상 1개"라는 이 프로젝트의 전제 위에서만 안전하다 — 이 메서드가 + * 실행되는 시점엔 다른 프로세스가 진짜로 처리 중일 수 없으므로, claim이 남아있는 행은 + * 전부 죽은 이전 프로세스의 흔적이다. + */ + @Modifying(clearAutomatically = true) + @Query("UPDATE RagResponse r SET r.claimedAt = NULL " + + "WHERE r.status = com.opensource.docgrid.domain.search.enums.ResultStatus.PROCESSING " + + "AND r.claimedAt IS NOT NULL") + int releaseAllClaimsOnStartup(); /** 특정 검색 요청(queryId)에 대한 RAG 답변을 찾는다. GET /search/{queryId} 재조회에 쓰인다. */ Optional findByQuery_Id(Long queryId); diff --git a/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagFacade.java b/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagFacade.java index 41309ad9..d5e5f047 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagFacade.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagFacade.java @@ -135,7 +135,7 @@ public RagEnqueueOutcome enqueue( * extractive fallback을 채운 채 FAILED로 확정한다. * *

{@code job} 객체가 아니라 {@code jobId}만 받아 이 메서드 자신의 트랜잭션 안에서 다시 - * 조회하는 이유: RagJobWorker가 {@code findFirstByStatusOrderByCreatedAtAsc()}로 꺼낸 + * 조회하는 이유: RagJobWorker가 claim 단계({@code RagResponseClaimService}, #340)에서 꺼낸 * job은 그 조회 시점에 트랜잭션이 끝나 detached 상태다. 원래(#218) 이 detached 인스턴스를 * 그대로 받아 필드만 바꾸면 dirty checking이 감지 못해 DB에 반영되지 않는 버그가 있었는데, * 지금은 완료 처리 자체가 dirty checking에 의존하지 않는다({@link @@ -229,7 +229,11 @@ public boolean processJob(Long jobId) { } // 답변과 citation 저장이 모두 끝난 동일 Transaction의 커밋 이후 성공 Counter를 기록한다. applicationEventPublisher.publishEvent(new RagJobCompletionMetricEvent(Outcome.SUCCESS)); - log.info("[RAG] done queryId={} responseId={} latencyMs={}", queryId, job.getId(), result.latencyMs()); + // promptTokens/answerTokens을 함께 남겨, 느린 job이 프롬프트를 읽느라(prefill) 오래 걸린 건지 + // 답변을 쓰느라(decode) 오래 걸린 건지 로그만으로 구분할 수 있게 한다 — 병렬화(#340) 이후 + // 요청당 작업량을 어느 쪽부터 줄여야 할지 판단하는 근거 자료. + log.info("[RAG] done queryId={} responseId={} latencyMs={} promptTokens={} answerTokens={}", + queryId, job.getId(), result.latencyMs(), result.inputTokenCount(), result.outputTokenCount()); return true; } diff --git a/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagJobWorker.java b/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagJobWorker.java index a648a390..c164f145 100644 --- a/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagJobWorker.java +++ b/backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagJobWorker.java @@ -1,45 +1,130 @@ package com.opensource.docgrid.domain.rag.service; import java.util.Optional; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.Semaphore; +import java.util.concurrent.ThreadPoolExecutor; +import org.springframework.beans.factory.annotation.Qualifier; +import org.springframework.boot.context.event.ApplicationReadyEvent; +import org.springframework.context.event.EventListener; import org.springframework.dao.OptimisticLockingFailureException; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Component; +import com.opensource.docgrid.domain.rag.config.RagExecutionConfig; import com.opensource.docgrid.domain.rag.controller.RagWebSocketController; import com.opensource.docgrid.domain.rag.entity.RagResponse; import com.opensource.docgrid.domain.rag.repository.RagResponseRepository; -import com.opensource.docgrid.domain.search.enums.ResultStatus; +import com.opensource.docgrid.domain.rag.service.command.RagResponseClaimService; -import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; /** - * PROCESSING 상태인 RagResponse를 하나씩 순서대로 꺼내 처리하는 경량 Worker (#218). + * PROCESSING 상태인 RagResponse를 최대 {@code rag.worker.max-concurrency}건까지 동시에 꺼내 + * 처리하는 경량 Worker (#218, 병렬화는 #340). * - *

{@code embedding_jobs}용 Worker(heartbeat·lease 복구 등 분산 처리 안전장치 포함, 26개 파일 - * 규모)와 달리, 이 Worker는 백엔드 인스턴스가 1개뿐이고 Ollama도 GPU 1개라 애초에 동시 처리가 - * 불가능하다는 전제 위에서 만들어졌다 — {@code @Scheduled} 폴링 하나로 충분하고, 여러 인스턴스 - * 간 조율(락·lease)은 필요 없다. Worker가 정확히 1개뿐이라는 사실 자체가 Ollama 호출의 - * 동시성 상한을 자연히 1로 만든다 — 기각했던 세마포어 게이트(#218 초안)가 하던 역할을 이 - * 구조가 대신한다. + *

{@code embedding_jobs}용 Worker(heartbeat·lease 복구 등 분산 처리 안전장치 포함, 수십 파일 + * 규모)와 달리, 이 Worker는 백엔드 인스턴스가 1개뿐이라는 전제 위에서 만들어졌다 — 여러 인스턴스 + * 간 조율(락·lease)은 필요 없다. 다만 GPU/Ollama 하나가 실제로 감당 가능한 병렬 슬롯 수만큼은 + * 이 프로세스 안에서 동시에 처리할 수 있다는 것이 #340의 전제다. * - *

{@code processJob()} 실행(=OllamaClient HTTP 호출, 최대 {@code ollama.generate-deadline})이 - * 끝나야 다음 폴링이 실행되므로, 폴링 주기 자체는 혼잡 여부와 무관하게 큐가 밀리지 않는 한 - * 크게 중요하지 않다 — PROCESSING 건이 있으면 그 즉시 다음 턴에 잡힌다. + *

동시성은 두 계층으로 제한된다: ① {@link #ragWorkerSlots}(로컬 {@link Semaphore})를 먼저 + * 확보해야 DB claim을 시도하고 — 이 순서 덕분에 "claim은 됐는데 실행할 스레드가 없는" 상태가 + * 생기지 않는다. ② claim 자체는 {@link RagResponseClaimService}가 {@code FOR UPDATE SKIP + * LOCKED} + {@code claimed_at}으로 여러 스레드가 동시에 같은 job을 집지 못하게 막는다. 실제 + * 처리는 {@link #ragWorkerJobExecutor}(전용 {@link ThreadPoolExecutor})에서 실행되어, + * {@link #processNext()} 자체는 claim만 하고 즉시 반환한다 — Ollama 호출(최대 + * {@code ollama.generate-deadline})로 폴링 스레드가 막히지 않는다. + * + *

{@link #recoverStaleClaimsOnStartup()}은 재시작 전 프로세스가 claim한 채 남긴 job의 + * claim을 앱 시작 시 1회 풀어준다 — "인스턴스 1개" 전제를 유지하는 한 안전한 최소한의 복구다. */ @Component -@RequiredArgsConstructor @Slf4j public class RagJobWorker { private final RagResponseRepository ragResponseRepository; + private final RagResponseClaimService ragResponseClaimService; private final RagFacade ragFacade; private final RagWebSocketController ragWebSocketController; + private final Semaphore ragWorkerSlots; + private final ThreadPoolExecutor ragWorkerJobExecutor; + + public RagJobWorker( + RagResponseRepository ragResponseRepository, + RagResponseClaimService ragResponseClaimService, + RagFacade ragFacade, + RagWebSocketController ragWebSocketController, + @Qualifier(RagExecutionConfig.RAG_WORKER_SLOTS) Semaphore ragWorkerSlots, + @Qualifier(RagExecutionConfig.RAG_WORKER_JOB_EXECUTOR) ThreadPoolExecutor ragWorkerJobExecutor + ) { + this.ragResponseRepository = ragResponseRepository; + this.ragResponseClaimService = ragResponseClaimService; + this.ragFacade = ragFacade; + this.ragWebSocketController = ragWebSocketController; + this.ragWorkerSlots = ragWorkerSlots; + this.ragWorkerJobExecutor = ragWorkerJobExecutor; + } + + /** + * 앱 준비 완료 시 1회, 이전 프로세스가 claim한 채 완료하지 못한 job의 claim을 전부 풀어 + * 재시작 뒤에도 다시 시도될 수 있게 한다(#340 CodeRabbit 리뷰 반영). 이 복구가 없으면 + * {@code claimed_at}이 남아있는 job은 {@link RagResponseClaimService#claimNext}가 영원히 + * 다시 집어주지 않아, 실제로 한 번도 재시도되지 않고 {@code RagJobTimeoutSweeper}의 + * fallback만 기다리게 된다(#218 이전 방식은 이런 job을 자동으로 재시도했으므로 이 복구가 + * 없으면 퇴보다). "인스턴스는 항상 1개"라는 전제 위에서만 안전 — 이 시점엔 다른 프로세스가 + * 진짜로 처리 중일 수 없다. + */ + @EventListener(ApplicationReadyEvent.class) + public void recoverStaleClaimsOnStartup() { + int recovered = ragResponseClaimService.recoverStaleClaimsOnStartup(); + if (recovered > 0) { + log.warn("[RAG-WORKER] 재시작 복구: 이전 프로세스가 claim한 채 방치된 job {}건의 claim을 해제함", recovered); + } + } + + /** + * 1초마다 실행되어, 로컬 슬롯이 남아있는 한 PROCESSING job을 계속 claim해 전용 Executor에 + * 넘긴다. 슬롯이 없거나(이미 정원만큼 처리 중) 대기 중인 job이 없으면 그 자리에서 멈춘다. + * + *

{@code while (tryAcquire())}만으로 반복 횟수 상한이 자동으로 걸린다 — Semaphore의 총 + * permit 수가 이미 {@code max-concurrency}와 같아서, 별도 카운터 변수 없이도 이 루프가 + * {@code max-concurrency}번보다 더 돌 수 없다. + */ + @Scheduled(fixedDelayString = "${rag.worker.polling-interval:1s}") + public void processNext() { + while (ragWorkerSlots.tryAcquire()) { + Optional claimedJobId; + try { + claimedJobId = ragResponseClaimService.claimNext(); + } catch (RuntimeException e) { + ragWorkerSlots.release(); + log.error("[RAG-WORKER] claim 중 예외 발생", e); + return; + } + + if (claimedJobId.isEmpty()) { + // 대기 중인 job이 없다 — 미리 확보한 슬롯을 돌려주고 이번 폴링을 끝낸다. + ragWorkerSlots.release(); + return; + } + + Long jobId = claimedJobId.get(); + try { + ragWorkerJobExecutor.execute(() -> executeClaimedJob(jobId)); + } catch (RejectedExecutionException e) { + // 슬롯을 먼저 확보했으므로 이론상 도달하지 않아야 하지만(Executor 정원 = + // Semaphore 총 permit 수), 종료 절차 중 등 극단적 상황에 대비한 방어다. + ragWorkerSlots.release(); + log.warn("[RAG-WORKER] 실행 제출이 거부됨 jobId={}", jobId); + return; + } + } + } /** - * 1초마다 실행되어 PROCESSING 중 가장 오래된 job 하나를 찾아 처리한다. 없으면 아무 일도 - * 하지 않고 다음 턴을 기다린다. + * claim된 job 하나를 실제로 처리한다 — 전용 Executor 스레드에서 실행된다. * *

처리 결과는 세 갈래로 갈린다: ①정상 성공 — WebSocket 알림. ②{@link * OptimisticLockingFailureException} — 다른 트랜잭션이 이미 이 job을 처리했다는 뜻이라 @@ -51,49 +136,36 @@ public class RagJobWorker { * boolean을 확인한 뒤에만 보낸다 — RagJobTimeoutSweeper가 이 job을 이미 먼저 FAILED로 * 확정해뒀다면(#288) 두 메서드 다 실제로는 아무것도 안 바꾸고 false를 반환하는데, 이 경우 * 스위퍼가 이미 보낸 알림 외에 Worker가 중복으로 또 보낼 이유가 없다. + * + *

어떤 경로로 끝나든 {@code finally}에서 반드시 슬롯을 반환한다 — 안 그러면 이 Worker가 + * 처리 가능한 동시성이 영구히 줄어든다. */ - @Scheduled(fixedDelayString = "${rag.worker.polling-interval:1s}") - public void processNext() { - Optional maybeJob = ragResponseRepository.findFirstByStatusOrderByCreatedAtAsc(ResultStatus.PROCESSING); - if (maybeJob.isEmpty()) { - return; - } - - RagResponse job = maybeJob.get(); - /* - * query/query.user는 findFirstByStatusOrderByCreatedAtAsc()의 @EntityGraph로 이미 - * 로딩돼 있어 detached 상태에서 읽어도 안전하다 — 완료 처리 자체는 조건부 UPDATE로 - * 이뤄지므로(#288) 이 job 인스턴스가 detached여도 상관없지만, processJob()이 이 - * 트랜잭션 시점 기준 최신 상태를 읽도록 id만 넘긴다. - */ - Long queryId = job.getQuery().getId(); - String userEmail = job.getQuery().getUser().getEmail(); - + private void executeClaimedJob(Long jobId) { try { - if (ragFacade.processJob(job.getId())) { - ragWebSocketController.notifyAnswerReady(userEmail, queryId); + RagResponse job = ragResponseRepository.findWithQueryAndUserById(jobId).orElse(null); + if (job == null) { + // claim 직후 이 job이 통째로 사라지는 건 극단적 상황(예: 테스트 데이터 정리)에서만 + // 가능하다 — processJob() 자신도 findById로 다시 조회하므로 여기서는 방어만 한다. + log.error("[RAG-WORKER] claim된 job을 찾을 수 없음 jobId={}", jobId); + return; } - } catch (OptimisticLockingFailureException e) { - /* - * 설계상 Worker는 인스턴스 1개를 전제하지만(클래스 주석 참고), 롤링 배포로 신·구 - * 인스턴스가 잠깐 겹치는 등 예외적으로 다른 트랜잭션이 같은 job을 먼저 처리했을 수 - * 있다. 이 경우 그 row는 이미 올바르게 SUCCESS/FAILED로 반영된 것이므로, - * markUnexpectedFailure로 덮어쓰면 정상 처리된 결과를 오답으로 바꿔버리는 2차 - * 사고가 난다 — 조용히 다음 폴링으로 넘어간다. - */ - log.warn("[RAG-WORKER] job이 이미 다른 트랜잭션에서 처리된 것으로 보임(경합) queryId={}", queryId); - } catch (Exception e) { - /* - * processJob() 내부에서 Ollama 관련 실패는 이미 DocGridException으로 잡아 fallback - * 처리하므로, 여기까지 올라오는 예외는 예상 밖의 버그다. Worker 스레드가 죽어서 큐 - * 전체가 멈추는 것보다는, 이 건을 건너뛰고 다음 폴링을 계속 도는 게 낫다. 단, job을 - * PROCESSING 상태로 방치하면 Worker가 같은 job을 계속 다시 집어 무한 재시도하게 - * 되므로(detached entity 버그와 같은 증상), 반드시 FAILED로 확정한 뒤 넘어간다. - */ - log.error("[RAG-WORKER] job 처리 중 예상치 못한 예외 queryId={}", queryId, e); - if (ragFacade.markUnexpectedFailure(job.getId(), e.getMessage())) { - ragWebSocketController.notifyAnswerReady(userEmail, queryId); + Long queryId = job.getQuery().getId(); + String userEmail = job.getQuery().getUser().getEmail(); + + try { + if (ragFacade.processJob(jobId)) { + ragWebSocketController.notifyAnswerReady(userEmail, queryId); + } + } catch (OptimisticLockingFailureException e) { + log.warn("[RAG-WORKER] job이 이미 다른 트랜잭션에서 처리된 것으로 보임(경합) queryId={}", queryId); + } catch (Exception e) { + log.error("[RAG-WORKER] job 처리 중 예상치 못한 예외 queryId={}", queryId, e); + if (ragFacade.markUnexpectedFailure(jobId, e.getMessage())) { + ragWebSocketController.notifyAnswerReady(userEmail, queryId); + } } + } finally { + ragWorkerSlots.release(); } } } diff --git a/backend/src/main/java/com/opensource/docgrid/domain/rag/service/command/RagResponseClaimService.java b/backend/src/main/java/com/opensource/docgrid/domain/rag/service/command/RagResponseClaimService.java new file mode 100644 index 00000000..c76c50ea --- /dev/null +++ b/backend/src/main/java/com/opensource/docgrid/domain/rag/service/command/RagResponseClaimService.java @@ -0,0 +1,63 @@ +package com.opensource.docgrid.domain.rag.service.command; + +import java.time.Clock; +import java.time.LocalDateTime; +import java.util.Optional; + +import org.springframework.stereotype.Service; +import org.springframework.transaction.annotation.Transactional; + +import com.opensource.docgrid.domain.rag.entity.RagResponse; +import com.opensource.docgrid.domain.rag.repository.RagResponseRepository; + +import lombok.RequiredArgsConstructor; + +/** + * PROCESSING 중 아직 아무 Worker도 집지 않은 RagResponse 하나에 소유권을 부여하는 Command + * Service (#340). + * + *

{@code embedding_jobs}의 {@code EmbeddingJobClaimService}와 같은 트랜잭션 경계 전략을 + * 쓴다 — 행 잠금({@code FOR UPDATE SKIP LOCKED})과 claim 표시를 하나의 짧은 트랜잭션으로 + * 묶어 커밋과 동시에 락을 풀고, 실제 Ollama 호출은 이 트랜잭션 밖에서(별도 스레드가 + * {@link com.opensource.docgrid.domain.rag.service.RagFacade#processJob}을 부르며) 진행한다. + * RAG는 PENDING 같은 별도 대기 상태가 없어(생성 즉시 PROCESSING) {@code embedding_jobs}처럼 + * status 전이로 claim을 표시할 수 없다 — 대신 {@code claimed_at} 컬럼을 그 신호로 쓴다. + * + *

{@code embedding_jobs}와 달리 별도의 분산 Worker 등록/생존 검증은 하지 않는다 — RAG + * Worker는 이 프로세스 안의 로컬 스레드일 뿐이라 그런 개념 자체가 없다. + */ +@Service +@RequiredArgsConstructor +@Transactional +public class RagResponseClaimService { + + private final RagResponseRepository ragResponseRepository; + private final Clock clock; + + /** + * 다음으로 처리할 PROCESSING job 하나를 claim한다. + * + * @return claim에 성공한 job의 id. 대기 중인 job이 없으면 빈 값. + */ + public Optional claimNext() { + return ragResponseRepository.findNextUnclaimedProcessingForUpdate() + .map(this::claim); + } + + /** + * 앱 시작 시 1회 호출된다(#340 CodeRabbit 리뷰 반영). 이전 프로세스 인스턴스가 claim한 채 + * 완료하지 못한 job은 새 프로세스에서 영원히 재시도되지 않는다 — {@link + * RagResponseRepository#releaseAllClaimsOnStartup}로 그 claim을 전부 풀어, 새로 뜬 + * Worker가 정상적으로 다시 claim해 처리할 수 있게 한다. + * + * @return 실제로 claim이 풀린 행 수 + */ + public int recoverStaleClaimsOnStartup() { + return ragResponseRepository.releaseAllClaimsOnStartup(); + } + + private Long claim(RagResponse job) { + job.markClaimed(LocalDateTime.now(clock)); + return job.getId(); + } +} diff --git a/backend/src/main/resources/application.yml b/backend/src/main/resources/application.yml index 9bc7d0ab..5e291441 100644 --- a/backend/src/main/resources/application.yml +++ b/backend/src/main/resources/application.yml @@ -200,11 +200,15 @@ ollama: rag: worker: - # RagJobWorker가 PROCESSING 건이 있는지 확인하는 주기. Worker가 1개뿐이라 짧게 잡아도 - # 부하가 안 크고, 사용자 체감 지연에 직접 영향을 주므로 1초로 짧게 유지한다. + # RagJobWorker가 PROCESSING 건이 있는지 확인하는 주기. 동시성 상한(max-concurrency)이 + # 작아 짧게 잡아도 부하가 안 크고, 사용자 체감 지연에 직접 영향을 주므로 1초로 짧게 유지한다. polling-interval: ${RAG_WORKER_POLLING_INTERVAL:1s} # 이 시간 이상 PROCESSING으로 남아있으면(정상 백로그든 실제 hang이든 원인 불문) RagJobTimeoutSweeper가 # 무기한 대기시키지 않고 강제로 fallback 답변을 채워 종료한다(#286). stale-threshold: ${RAG_WORKER_STALE_THRESHOLD:90s} # RagJobTimeoutSweeper가 위 stale-threshold 기준을 얼마나 자주 확인할지. timeout-sweep-interval: ${RAG_WORKER_TIMEOUT_SWEEP_INTERVAL:15s} + # GPU/Ollama 하나가 실제로 감당 가능한 동시 생성 건수(#340). Ollama 쪽 병렬 슬롯 + # (OLLAMA_NUM_PARALLEL)과 실측 후 맞춰야 하며, 이 값을 그냥 올린다고 처리량이 느는 건 + # 아니다 — Ollama 서버 자체가 병렬 처리를 지원하도록 먼저 설정돼 있어야 한다. + max-concurrency: ${RAG_WORKER_MAX_CONCURRENCY:2} diff --git a/backend/src/main/resources/db/migration/V43__add_rag_responses_claimed_at.sql b/backend/src/main/resources/db/migration/V43__add_rag_responses_claimed_at.sql new file mode 100644 index 00000000..a63601e9 --- /dev/null +++ b/backend/src/main/resources/db/migration/V43__add_rag_responses_claimed_at.sql @@ -0,0 +1 @@ +ALTER TABLE rag_responses ADD COLUMN claimed_at TIMESTAMP; diff --git a/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagJobWorkerConcurrentQueueIntegrationTest.java b/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagJobWorkerConcurrentQueueIntegrationTest.java index 48e639f3..5892de00 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagJobWorkerConcurrentQueueIntegrationTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagJobWorkerConcurrentQueueIntegrationTest.java @@ -40,6 +40,13 @@ * 비우도록 두고(수동으로 processNext()를 여러 번 호출하지 않음), 3명이 정확히 같은 순간에 * 질문을 던졌다고 가정해 3개 job을 동시 스레드로 접수한 뒤, 셋 다 결국 SUCCESS로 끝나는지 * 실제 로컬 Ollama를 상대로 확인한다. + * + *

병렬화(#340) 이후에는 여기에 "실제로 동시에 처리됐는가"까지 증명한다 — {@code claimed_at}이 + * "언제부터 실제로 처리되기 시작했는지"를 알려주는 유일한 신호다({@code updatedAt}은 완료 확정이 + * 전부 벌크 UPDATE라 채워지지 않는다). {@code rag.worker.max-concurrency}가 2 이상이면 3건 중 + * 최소 2건은 거의 동시에 claim되어야 한다 — 셋 다 동시에 claim되길 요구하지 않는 이유는 + * max-concurrency가 정확히 2일 때는 3번째 job이 앞선 두 건 중 하나가 끝날 때까지 자연스럽게 + * 기다리기 때문이다(그래도 안전 실패는 없다 — #218의 핵심 목표). */ @Tag("integration") @SpringBootTest @@ -149,6 +156,19 @@ void threeConcurrentJobs_allEventuallySucceedViaRealScheduler() throws Interrupt long successCount = finished.stream().filter(j -> j.getStatus() == ResultStatus.SUCCESS).count(); System.out.println("[TEST] SUCCESS=" + successCount + "/3, answers=" + finished.stream().map(RagResponse::getAnswerText).toList()); + + // #340 병렬화 증명: claimed_at 3건 중 최소 2건은 서로 가까운 시각에 claim됐어야 한다 — + // 순차 처리였다면(#218 이전 방식) 각 claim은 앞선 job의 전체 처리 시간(수 초~수십 초)만큼 + // 떨어져 있었을 것이다. max-concurrency가 정확히 2여도(기본값) 최소 두 건은 동시에 슬롯을 + // 잡을 수 있으므로, "셋 다"가 아니라 "가장 가까운 두 건"의 간격으로 판단한다. + List claimedAtValues = finished.stream() + .map(RagResponse::getClaimedAt) + .sorted() + .toList(); + assertThat(claimedAtValues).allSatisfy(claimedAt -> assertThat(claimedAt).isNotNull()); + Duration closestGap = Duration.between(claimedAtValues.get(0), claimedAtValues.get(1)); + System.out.println("[TEST] claimedAt=" + claimedAtValues + " closestGap=" + closestGap); + assertThat(closestGap).isLessThan(Duration.ofSeconds(10)); } private User createUser() { diff --git a/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagJobWorkerIntegrationTest.java b/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagJobWorkerIntegrationTest.java index 7959fb7a..d4843a3e 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagJobWorkerIntegrationTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagJobWorkerIntegrationTest.java @@ -1,7 +1,9 @@ package com.opensource.docgrid.domain.rag.integration; import static org.assertj.core.api.Assertions.assertThat; +import static org.awaitility.Awaitility.await; +import java.time.Duration; import java.time.LocalDateTime; import org.junit.jupiter.api.AfterEach; @@ -31,13 +33,18 @@ import com.opensource.docgrid.domain.user.repository.UserRepository; /** - * RagResponseRepository.findFirstByStatusOrderByCreatedAtAsc()로 꺼낸 job이 detached 상태라, - * RagFacade.processJob()에 그 인스턴스를 그대로 넘기면(원래 #218 당시 그랬듯) 완료 처리가 DB에 - * 반영되지 않는(=영원히 PROCESSING으로 남는) 실사용 버그가 있었다. 지금은 완료 처리 자체가 - * 조건부 UPDATE(#288)라 detached 상태 여부와 무관하게 반영되지만, processJob()이 여전히 - * jobId만 받아 자기 트랜잭션에서 다시 조회하는 설계를 유지하는지는 이 테스트로 계속 검증한다. - * 이 테스트는 이걸 Mockito 목이 아니라 실제 트랜잭션 경계로 재현·검증한다 — 목 기반 단위 - * 테스트는 "메서드가 호출됐는지"만 보고 "DB에 실제로 반영됐는지"는 증명하지 못한다. + * claim 단계(RagResponseClaimService)로 꺼낸 job이 detached 상태라, RagFacade.processJob()에 그 + * 인스턴스를 그대로 넘기면(원래 #218 당시 그랬듯) 완료 처리가 DB에 반영되지 않는(=영원히 + * PROCESSING으로 남는) 실사용 버그가 있었다. 지금은 완료 처리 자체가 조건부 UPDATE(#288)라 + * detached 상태 여부와 무관하게 반영되지만, processJob()이 여전히 jobId만 받아 자기 트랜잭션에서 + * 다시 조회하는 설계를 유지하는지는 이 테스트로 계속 검증한다. 이 테스트는 이걸 Mockito 목이 + * 아니라 실제 트랜잭션 경계로 재현·검증한다 — 목 기반 단위 테스트는 "메서드가 호출됐는지"만 + * 보고 "DB에 실제로 반영됐는지"는 증명하지 못한다. + * + *

병렬화(#340) 이후 {@code processNext()}는 claim만 하고 실제 처리는 전용 Executor + * 스레드에 넘긴 뒤 즉시 반환한다 — 그래서 이 테스트도 호출 직후 동기적으로 결과를 확인하는 + * 대신, {@link RagJobWorkerConcurrentQueueIntegrationTest}가 이미 쓰는 Awaitility로 처리가 + * 끝날 때까지 기다린다. */ @Tag("integration") @SpringBootTest @@ -118,11 +125,15 @@ void processNext_persistsStatusChangeAcrossDetachedEntityBoundary() { ragJobWorker.processNext(); - // Ollama가 로컬에 떠 있지 않을 수도 있으므로 SUCCESS/FAILED 둘 다 통과 조건으로 둔다 — - // 이 테스트가 검증하는 건 "LLM 호출 성공 여부"가 아니라 "detached 상태에서도 최종 - // 상태가 DB에 반영되는지"다. - RagResponse persisted = ragResponseRepository.findById(jobId).orElseThrow(); - assertThat(persisted.getStatus()).isNotEqualTo(ResultStatus.PROCESSING); - assertThat(persisted.getAnswerText()).isNotNull(); + // processNext()는 claim만 하고 즉시 반환하므로(#340), 실제 Ollama 호출·완료 확정은 + // 전용 Executor 스레드에서 비동기로 이어진다 — 120초(read-timeout 90s + 여유)까지 + // 기다렸다가 확인한다. Ollama가 로컬에 떠 있지 않을 수도 있으므로 SUCCESS/FAILED 둘 다 + // 통과 조건으로 둔다 — 이 테스트가 검증하는 건 "LLM 호출 성공 여부"가 아니라 "detached + // 상태에서도 최종 상태가 DB에 반영되는지"다. + await().atMost(Duration.ofSeconds(120)).untilAsserted(() -> { + RagResponse persisted = ragResponseRepository.findById(jobId).orElseThrow(); + assertThat(persisted.getStatus()).isNotEqualTo(ResultStatus.PROCESSING); + assertThat(persisted.getAnswerText()).isNotNull(); + }); } } diff --git a/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagResponseClaimIntegrationTest.java b/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagResponseClaimIntegrationTest.java new file mode 100644 index 00000000..73c0e25a --- /dev/null +++ b/backend/src/test/java/com/opensource/docgrid/domain/rag/integration/RagResponseClaimIntegrationTest.java @@ -0,0 +1,251 @@ +package com.opensource.docgrid.domain.rag.integration; + +import static org.assertj.core.api.Assertions.assertThat; + +import java.time.LocalDateTime; +import java.util.List; +import java.util.Optional; +import java.util.concurrent.BrokenBarrierException; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Supplier; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Tag; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.test.context.ActiveProfiles; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; +import org.springframework.transaction.support.TransactionTemplate; + +import com.opensource.docgrid.domain.embedding.entity.EmbeddingModel; +import com.opensource.docgrid.domain.embedding.fixture.EmbeddingModelFixture; +import com.opensource.docgrid.domain.embedding.repository.EmbeddingModelRepository; +import com.opensource.docgrid.domain.rag.entity.RagResponse; +import com.opensource.docgrid.domain.rag.repository.RagResponseRepository; +import com.opensource.docgrid.domain.rag.service.command.RagResponseClaimService; +import com.opensource.docgrid.domain.search.entity.SearchConversation; +import com.opensource.docgrid.domain.search.entity.SearchQuery; +import com.opensource.docgrid.domain.search.enums.ResultStatus; +import com.opensource.docgrid.domain.search.enums.SearchType; +import com.opensource.docgrid.domain.search.repository.SearchConversationRepository; +import com.opensource.docgrid.domain.search.repository.SearchQueryRepository; +import com.opensource.docgrid.domain.user.entity.User; +import com.opensource.docgrid.domain.user.enums.UserStatus; +import com.opensource.docgrid.domain.user.repository.UserRepository; + +/** + * 실제 PostgreSQL에서 RAG Job Claim의 행 잠금·동시 claim 불변식을 검증하는 통합 테스트 (#340). + * + *

{@code EmbeddingJobClaimIntegrationTest}와 같은 방식으로, 서로 다른 Thread와 + * {@code REQUIRES_NEW} Transaction을 사용해 단일 Persistence Context의 순차 호출로는 재현할 수 + * 없는 {@code FOR UPDATE SKIP LOCKED} 경쟁을 검증한다. + */ +@Tag("integration") +@SpringBootTest +@ActiveProfiles("test") +@TestInstance(TestInstance.Lifecycle.PER_CLASS) +@DisplayName("RagResponse Claim DB 동시성 통합 테스트") +class RagResponseClaimIntegrationTest { + + private static final long TIMEOUT_SECONDS = 10; + + @Autowired private PlatformTransactionManager transactionManager; + @Autowired private RagResponseRepository ragResponseRepository; + @Autowired private RagResponseClaimService ragResponseClaimService; + @Autowired private SearchQueryRepository searchQueryRepository; + @Autowired private SearchConversationRepository searchConversationRepository; + @Autowired private UserRepository userRepository; + @Autowired private EmbeddingModelRepository embeddingModelRepository; + + private ExecutorService executorService; + private final AtomicInteger threadSequence = new AtomicInteger(); + private final List createdUserIds = new CopyOnWriteArrayList<>(); + private final List createdQueryIds = new CopyOnWriteArrayList<>(); + private final List createdConversationIds = new CopyOnWriteArrayList<>(); + private Long createdModelId; + + @BeforeAll + void createExecutor() { + executorService = Executors.newFixedThreadPool(2, runnable -> { + Thread thread = new Thread(runnable); + thread.setName("rag-claim-test-" + threadSequence.incrementAndGet()); + return thread; + }); + } + + @AfterAll + void shutdownExecutor() throws InterruptedException { + executorService.shutdownNow(); + assertThat(executorService.awaitTermination(TIMEOUT_SECONDS, TimeUnit.SECONDS)).isTrue(); + } + + // 이 클래스의 모든 테스트가 REQUIRES_NEW로 자체 트랜잭션을 커밋하므로(그래야 다른 Thread에서 + // 그 결과가 보인다) 자동 롤백에 기댈 수 없다 — 만든 데이터를 직접 정리한다. + @AfterEach + void cleanUp() { + for (Long queryId : createdQueryIds) { + ragResponseRepository.findByQuery_Id(queryId).ifPresent(ragResponseRepository::delete); + searchQueryRepository.deleteById(queryId); + } + createdQueryIds.clear(); + createdConversationIds.forEach(searchConversationRepository::deleteById); + createdConversationIds.clear(); + if (createdModelId != null) { + embeddingModelRepository.deleteById(createdModelId); + createdModelId = null; + } + createdUserIds.forEach(userRepository::deleteById); + createdUserIds.clear(); + } + + @Test + @DisplayName("다른 트랜잭션이 잠근 행은 기다리지 않고 다음 미잠금 행을 선택한다") + void findNextUnclaimedProcessingForUpdate_skipsLockedRow() throws Exception { + Long firstJobId = createPendingJob(); + Long secondJobId = createPendingJob(); + CountDownLatch rowLocked = new CountDownLatch(1); + CountDownLatch releaseLock = new CountDownLatch(1); + + // 1. 첫 번째 Transaction이 가장 오래된 job(firstJobId)의 행 잠금을 잡은 채 커밋을 지연한다. + Future lockHolder = executorService.submit(() -> inNewTransaction(() -> { + Long selectedId = ragResponseRepository.findNextUnclaimedProcessingForUpdate() + .orElseThrow().getId(); + rowLocked.countDown(); + awaitLatch(releaseLock); + return selectedId; + })); + + // 2. 첫 번째 행이 실제로 잠긴 뒤에만 두 번째 Transaction을 시작해 경쟁 조건을 확정한다. + assertThat(rowLocked.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)).isTrue(); + + // 3. 두 번째 Transaction은 잠금 해제를 기다리지 않고 다음 미잠금 job을 선택해야 한다. + Future skipLockedReader = executorService.submit(() -> inNewTransaction(() -> + ragResponseRepository.findNextUnclaimedProcessingForUpdate().orElseThrow().getId() + )); + + try { + assertThat(skipLockedReader.get(TIMEOUT_SECONDS, TimeUnit.SECONDS)).isEqualTo(secondJobId); + } finally { + releaseLock.countDown(); + } + + assertThat(lockHolder.get(TIMEOUT_SECONDS, TimeUnit.SECONDS)).isEqualTo(firstJobId); + } + + @Test + @DisplayName("두 Worker가 동시에 같은 job을 claim해도 한쪽만 소유권을 얻는다") + void claim_allowsExactlyOneConcurrentOwner() throws Exception { + Long jobId = createPendingJob(); + CyclicBarrier startBarrier = new CyclicBarrier(2); + + List>> attempts = List.of( + executorService.submit(() -> claimAfterBarrier(startBarrier)), + executorService.submit(() -> claimAfterBarrier(startBarrier)) + ); + + List> results = List.of( + attempts.get(0).get(TIMEOUT_SECONDS, TimeUnit.SECONDS), + attempts.get(1).get(TIMEOUT_SECONDS, TimeUnit.SECONDS) + ); + + assertThat(results).filteredOn(Optional::isPresent).hasSize(1); + assertThat(results).filteredOn(Optional::isEmpty).hasSize(1); + assertThat(results.stream().flatMap(Optional::stream).findFirst()).contains(jobId); + + RagResponse persisted = ragResponseRepository.findById(jobId).orElseThrow(); + assertThat(persisted.getClaimedAt()).isNotNull(); + } + + private Optional claimAfterBarrier(CyclicBarrier barrier) { + awaitBarrier(barrier); + return inNewTransaction(() -> ragResponseClaimService.claimNext()); + } + + private Long createPendingJob() { + return inNewTransaction(() -> { + User user = userRepository.save(User.builder() + .email("rag-claim-it-" + System.nanoTime() + "@test.local") + .passwordHash("x") + .name("RAG Claim 테스트 유저") + .status(UserStatus.ACTIVE) + .build()); + createdUserIds.add(user.getId()); + + if (createdModelId == null) { + EmbeddingModel model = embeddingModelRepository.save( + EmbeddingModelFixture.createModel("rag-claim-it-" + System.nanoTime(), false, false) + ); + createdModelId = model.getId(); + } + EmbeddingModel model = embeddingModelRepository.findById(createdModelId).orElseThrow(); + + SearchConversation conversation = searchConversationRepository.save(SearchConversation.builder() + .user(user) + .title("RAG Claim 테스트 질문") + .lastMessageAt(LocalDateTime.now()) + .build()); + createdConversationIds.add(conversation.getId()); + + SearchQuery query = searchQueryRepository.save(SearchQuery.builder() + .user(user) + .conversation(conversation) + .queryText("RAG Claim 테스트 질문") + .queryEmbeddingModel(model) + .queryVector(new float[1024]) + .searchType(SearchType.VECTOR) + .topK(5) + .status(ResultStatus.SUCCESS) + .build()); + createdQueryIds.add(query.getId()); + + RagResponse pending = ragResponseRepository.save(RagResponse.builder() + .query(query) + .promptText("RAG Claim 테스트 프롬프트") + .status(ResultStatus.PROCESSING) + .build()); + return pending.getId(); + }); + } + + private T inNewTransaction(Supplier work) { + TransactionTemplate transactionTemplate = new TransactionTemplate(transactionManager); + transactionTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); + return transactionTemplate.execute(status -> work.get()); + } + + private void awaitLatch(CountDownLatch latch) { + try { + if (!latch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)) { + throw new IllegalStateException("동시성 테스트 Lock 해제가 제한 시간 안에 완료되지 않았습니다."); + } + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("동시성 테스트 Lock 대기 중 Thread가 중단되었습니다.", exception); + } + } + + private void awaitBarrier(CyclicBarrier barrier) { + try { + barrier.await(TIMEOUT_SECONDS, TimeUnit.SECONDS); + } catch (InterruptedException exception) { + Thread.currentThread().interrupt(); + throw new IllegalStateException("동시성 테스트 Barrier 대기 중 Thread가 중단되었습니다.", exception); + } catch (BrokenBarrierException | TimeoutException exception) { + throw new IllegalStateException("동시성 테스트 Barrier가 제한 시간 안에 완료되지 않았습니다.", exception); + } + } +} diff --git a/backend/src/test/java/com/opensource/docgrid/domain/rag/repository/RagResponseRepositoryTest.java b/backend/src/test/java/com/opensource/docgrid/domain/rag/repository/RagResponseRepositoryTest.java index bf8b3d5b..ef8c8220 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/rag/repository/RagResponseRepositoryTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/rag/repository/RagResponseRepositoryTest.java @@ -116,6 +116,58 @@ void completeSuccessIfProcessing_alreadyTimedOutJob_doesNotOverwriteAndReturnsZe assertThat(reloaded.getAnswerText()).isEqualTo("fallback 답변"); } + @Test + @DisplayName("findNextUnclaimedProcessingForUpdate: claim 안 된 PROCESSING만 찾고, claim된 것과 다른 상태는 제외한다") + void findNextUnclaimedProcessingForUpdate_filtersOnClaimedAtAndStatus() { + RagResponse unclaimed = saveRagResponse(ResultStatus.PROCESSING); + RagResponse alreadyClaimed = saveRagResponse(ResultStatus.PROCESSING); + alreadyClaimed.markClaimed(LocalDateTime.now()); + ragResponseRepository.save(alreadyClaimed); + saveRagResponse(ResultStatus.SUCCESS); + + RagResponse found = ragResponseRepository.findNextUnclaimedProcessingForUpdate().orElseThrow(); + + assertThat(found.getId()).isEqualTo(unclaimed.getId()); + } + + @Test + @DisplayName("findNextUnclaimedProcessingForUpdate: claim 가능한 job이 없으면 빈 값을 반환한다") + void findNextUnclaimedProcessingForUpdate_noCandidates_returnsEmpty() { + RagResponse claimed = saveRagResponse(ResultStatus.PROCESSING); + claimed.markClaimed(LocalDateTime.now()); + ragResponseRepository.save(claimed); + + assertThat(ragResponseRepository.findNextUnclaimedProcessingForUpdate()).isEmpty(); + } + + @Test + @DisplayName("releaseAllClaimsOnStartup: PROCESSING이면서 claim된 job만 claim을 해제하고 영향받은 행 수를 반환한다") + void releaseAllClaimsOnStartup_releasesOnlyClaimedProcessingJobs() { + RagResponse claimed = saveRagResponse(ResultStatus.PROCESSING); + claimed.markClaimed(LocalDateTime.now()); + ragResponseRepository.save(claimed); + RagResponse unclaimed = saveRagResponse(ResultStatus.PROCESSING); + RagResponse succeeded = saveRagResponse(ResultStatus.SUCCESS); + + int released = ragResponseRepository.releaseAllClaimsOnStartup(); + + assertThat(released).isEqualTo(1); + assertThat(ragResponseRepository.findById(claimed.getId()).orElseThrow().getClaimedAt()).isNull(); + // 재시작 복구가 다른 job까지 잘못 건드리지 않는지 확인 — 원래 claim이 없던 job과 + // 이미 확정된 job은 이 호출과 무관해야 한다. + assertThat(ragResponseRepository.findById(unclaimed.getId()).orElseThrow().getClaimedAt()).isNull(); + assertThat(ragResponseRepository.findById(succeeded.getId()).orElseThrow().getStatus()) + .isEqualTo(ResultStatus.SUCCESS); + } + + @Test + @DisplayName("releaseAllClaimsOnStartup: 복구할 job이 없으면 영향받은 행이 0건이다") + void releaseAllClaimsOnStartup_noStaleClaims_returnsZero() { + saveRagResponse(ResultStatus.PROCESSING); + + assertThat(ragResponseRepository.releaseAllClaimsOnStartup()).isZero(); + } + @Test @DisplayName("findByStatusAndCreatedAtBefore: cutoff 이전에 생성된 PROCESSING만 찾고, 상태가 다른 job은 제외한다") void findByStatusAndCreatedAtBefore_filtersOnStatusAndCreatedAt() { diff --git a/backend/src/test/java/com/opensource/docgrid/domain/rag/service/RagJobWorkerTest.java b/backend/src/test/java/com/opensource/docgrid/domain/rag/service/RagJobWorkerTest.java index 901db8a4..27b5004e 100644 --- a/backend/src/test/java/com/opensource/docgrid/domain/rag/service/RagJobWorkerTest.java +++ b/backend/src/test/java/com/opensource/docgrid/domain/rag/service/RagJobWorkerTest.java @@ -1,19 +1,25 @@ package com.opensource.docgrid.domain.rag.service; +import static org.assertj.core.api.Assertions.assertThat; import static org.mockito.ArgumentMatchers.any; import static org.mockito.BDDMockito.given; import static org.mockito.BDDMockito.then; import static org.mockito.Mockito.RETURNS_DEEP_STUBS; +import static org.mockito.Mockito.doThrow; import static org.mockito.Mockito.mock; import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import java.util.Optional; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.Semaphore; +import java.util.concurrent.ThreadPoolExecutor; +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.InjectMocks; +import org.mockito.ArgumentCaptor; import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.dao.OptimisticLockingFailureException; @@ -21,87 +27,129 @@ import com.opensource.docgrid.domain.rag.controller.RagWebSocketController; import com.opensource.docgrid.domain.rag.entity.RagResponse; import com.opensource.docgrid.domain.rag.repository.RagResponseRepository; -import com.opensource.docgrid.domain.search.enums.ResultStatus; +import com.opensource.docgrid.domain.rag.service.command.RagResponseClaimService; /** - * RagJobWorker.processNext() 한 사이클의 동작만 검증한다 — 큐 조회(가장 오래된 PROCESSING 하나를 - * 꺼내는지), 처리 위임(RagFacade.processJob()으로 id를 넘기는지), 완료/실패 각각에서 요청자 - * 본인에게만 알림이 가는지가 검증 범위다. 실제 OllamaClient 호출이나 DB 반영 여부(dirty checking이 - * 실제로 먹히는지)는 이 테스트의 목(mock) 구조로는 증명할 수 없어 검증 범위 밖이다 — - * RagJobWorkerIntegrationTest가 그 부분을 담당한다. + * RagJobWorker.processNext()의 디스패치 로직(#340)을 검증한다 — 슬롯을 먼저 확보한 뒤에만 claim을 + * 시도하는지, claim 결과에 따라 슬롯을 되돌려주는지, claim된 job을 Executor에 제출하는지가 검증 + * 범위다. Executor는 실제 스레드를 안 쓰고 제출된 Runnable을 캡처해 테스트 스레드에서 직접 + * 실행한다 — 그래야 실행 결과(성공/false/예외/OptimisticLocking 분기)를 결정적으로 검증할 수 + * 있다. 실제 OllamaClient 호출이나 DB 반영 여부(dirty checking이 실제로 먹히는지)는 이 테스트의 + * 목(mock) 구조로는 증명할 수 없어 검증 범위 밖이다 — RagJobWorkerIntegrationTest가 그 부분을 + * 담당한다. + * + *

{@code Semaphore}는 mock하지 않고 실제 인스턴스를 쓴다 — I/O가 없는 순수 카운터라, mock보다 + * 실제 객체로 "슬롯이 진짜 반환됐는지"를 permit 개수로 직접 확인하는 편이 더 간단하고 정확하다. */ @ExtendWith(MockitoExtension.class) @DisplayName("RagJobWorker 단위 테스트") class RagJobWorkerTest { - @InjectMocks - private RagJobWorker ragJobWorker; - @Mock private RagResponseRepository ragResponseRepository; + @Mock + private RagResponseClaimService ragResponseClaimService; + @Mock private RagFacade ragFacade; @Mock private RagWebSocketController ragWebSocketController; + @Mock + private ThreadPoolExecutor ragWorkerJobExecutor; + + private Semaphore ragWorkerSlots; + private RagJobWorker ragJobWorker; + + @BeforeEach + void setUp() { + ragWorkerSlots = new Semaphore(1); + ragJobWorker = new RagJobWorker( + ragResponseRepository, ragResponseClaimService, ragFacade, ragWebSocketController, + ragWorkerSlots, ragWorkerJobExecutor + ); + } + @Test - @DisplayName("PROCESSING 건이 없으면 아무것도 하지 않는다") - void processNext_noPendingJob_doesNothing() { - given(ragResponseRepository.findFirstByStatusOrderByCreatedAtAsc(ResultStatus.PROCESSING)) - .willReturn(Optional.empty()); + @DisplayName("앱 시작 시 복구할 claim이 있으면 경고 로그를 남긴다(부작용은 claim 서비스에 위임)") + void recoverStaleClaimsOnStartup_delegatesToClaimServiceAndReturns() { + given(ragResponseClaimService.recoverStaleClaimsOnStartup()).willReturn(2); + + ragJobWorker.recoverStaleClaimsOnStartup(); + + then(ragResponseClaimService).should(times(1)).recoverStaleClaimsOnStartup(); + } + + @Test + @DisplayName("슬롯이 없으면 claim 자체를 시도하지 않는다") + void processNext_noSlotAvailable_neverClaims() { + // 이미 다른 job이 유일한 슬롯을 쓰고 있는 상황을 재현한다. permit이 이미 있는 상태의 + // acquireUninterruptibly()는 즉시 반환되므로 실제로 블로킹되지 않는다. + ragWorkerSlots.acquireUninterruptibly(); ragJobWorker.processNext(); - then(ragFacade).should(never()).processJob(any()); - then(ragWebSocketController).should(never()).notifyAnswerReady(any(), any()); + then(ragResponseClaimService).should(never()).claimNext(); + then(ragWorkerJobExecutor).should(never()).execute(any()); } @Test - @DisplayName("PROCESSING 건이 있으면 처리하고, 요청자 본인에게만 완료를 push한다") - void processNext_pendingJobExists_processesAndNotifiesOwner() { + @DisplayName("claim할 job이 없으면 슬롯을 반환하고 Executor를 부르지 않는다") + void processNext_claimEmpty_releasesSlotAndSkipsExecutor() { + given(ragResponseClaimService.claimNext()).willReturn(Optional.empty()); + + ragJobWorker.processNext(); + + then(ragWorkerJobExecutor).should(never()).execute(any()); + assertThat(ragWorkerSlots.availablePermits()).isEqualTo(1); + } + + @Test + @DisplayName("claim에 성공하면 Executor에 제출하고, 정상 처리되면 요청자 본인에게만 완료를 push한다") + void processNext_claimSucceeds_submitsAndNotifiesOwner() { RagResponse job = deepStubJob(999L, 100L, "user@example.com"); - given(ragResponseRepository.findFirstByStatusOrderByCreatedAtAsc(ResultStatus.PROCESSING)) - .willReturn(Optional.of(job)); + given(ragResponseClaimService.claimNext()).willReturn(Optional.of(999L)); + given(ragResponseRepository.findWithQueryAndUserById(999L)).willReturn(Optional.of(job)); given(ragFacade.processJob(999L)).willReturn(true); ragJobWorker.processNext(); + runSubmittedTask(); - // Worker는 detached entity를 그대로 넘기지 않고 id만 넘긴다 — processJob()이 자기 트랜잭션 - // 안에서 다시 조회해야 완료 처리(조건부 UPDATE)가 최신 상태 기준으로 실행된다. then(ragFacade).should(times(1)).processJob(999L); then(ragWebSocketController).should(times(1)).notifyAnswerReady("user@example.com", 100L); + // 처리(성공적으로 실행된 Runnable)가 끝나면 finally에서 슬롯을 되돌려준다. + assertThat(ragWorkerSlots.availablePermits()).isEqualTo(1); } @Test @DisplayName("경합(#288): processJob이 false를 반환하면(RagJobTimeoutSweeper가 이미 확정함) 알림을 보내지 않는다") void processNext_processJobLosesRace_doesNotNotify() { RagResponse job = deepStubJob(999L, 100L, "user@example.com"); - given(ragResponseRepository.findFirstByStatusOrderByCreatedAtAsc(ResultStatus.PROCESSING)) - .willReturn(Optional.of(job)); + given(ragResponseClaimService.claimNext()).willReturn(Optional.of(999L)); + given(ragResponseRepository.findWithQueryAndUserById(999L)).willReturn(Optional.of(job)); given(ragFacade.processJob(999L)).willReturn(false); ragJobWorker.processNext(); + runSubmittedTask(); then(ragWebSocketController).should(never()).notifyAnswerReady(any(), any()); } @Test - @DisplayName("processJob이 예상 밖 예외를 던지면 job을 FAILED로 확정하고, Worker는 죽지 않고 이번 건만 건너뛴다") - void processNext_unexpectedException_marksFailedAndSkipsJobWithoutCrashingWorker() { + @DisplayName("processJob이 예상 밖 예외를 던지면 job을 FAILED로 확정하고, 알림은 그대로 보낸다") + void processNext_unexpectedException_marksFailedAndNotifies() { RagResponse job = deepStubJob(999L, 100L, "user@example.com"); - given(ragResponseRepository.findFirstByStatusOrderByCreatedAtAsc(ResultStatus.PROCESSING)) - .willReturn(Optional.of(job)); - org.mockito.Mockito.doThrow(new RuntimeException("예상 밖 버그")).when(ragFacade).processJob(999L); + given(ragResponseClaimService.claimNext()).willReturn(Optional.of(999L)); + given(ragResponseRepository.findWithQueryAndUserById(999L)).willReturn(Optional.of(job)); + doThrow(new RuntimeException("예상 밖 버그")).when(ragFacade).processJob(999L); given(ragFacade.markUnexpectedFailure(999L, "예상 밖 버그")).willReturn(true); ragJobWorker.processNext(); + runSubmittedTask(); - // job을 PROCESSING으로 방치하면 Worker가 같은 job을 계속 다시 집어 무한 재시도하게 된다 - // (detached entity 버그와 같은 증상) — 그래서 반드시 FAILED로 확정해야 한다. then(ragFacade).should(times(1)).markUnexpectedFailure(999L, "예상 밖 버그"); - // FAILED로 확정된 이상 사용자도 결과(비록 실패 안내지만)를 받아야 하므로 알림은 그대로 간다. then(ragWebSocketController).should(times(1)).notifyAnswerReady("user@example.com", 100L); } @@ -109,44 +157,61 @@ void processNext_unexpectedException_marksFailedAndSkipsJobWithoutCrashingWorker @DisplayName("다른 트랜잭션이 이미 같은 job을 처리했으면(낙관적 락 경합) FAILED로 덮어쓰지 않고 조용히 넘어간다") void processNext_optimisticLockingFailure_skipsWithoutOverwritingAsFailed() { RagResponse job = deepStubJob(999L, 100L, "user@example.com"); - given(ragResponseRepository.findFirstByStatusOrderByCreatedAtAsc(ResultStatus.PROCESSING)) - .willReturn(Optional.of(job)); - org.mockito.Mockito.doThrow(new OptimisticLockingFailureException("경합")) - .when(ragFacade).processJob(999L); + given(ragResponseClaimService.claimNext()).willReturn(Optional.of(999L)); + given(ragResponseRepository.findWithQueryAndUserById(999L)).willReturn(Optional.of(job)); + doThrow(new OptimisticLockingFailureException("경합")).when(ragFacade).processJob(999L); ragJobWorker.processNext(); + runSubmittedTask(); - // 다른 트랜잭션이 이미 올바르게 처리한 결과이므로, 이걸 FAILED로 덮어쓰면 정상 처리된 - // 결과를 오답으로 바꿔버리는 2차 사고가 난다 — markUnexpectedFailure를 호출하면 안 된다. then(ragFacade).should(never()).markUnexpectedFailure(any(), any()); then(ragWebSocketController).should(never()).notifyAnswerReady(any(), any()); } @Test - @DisplayName("한 job이 예외로 실패해도 다음 폴링에서 뒤에 대기 중인 job이 정상 처리된다") - void processNext_afterUnexpectedFailure_nextPollingProcessesFollowingJob() { - RagResponse failingJob = deepStubJob(1L, 100L, "user1@example.com"); - RagResponse nextJob = deepStubJob(2L, 200L, "user2@example.com"); - org.mockito.Mockito.doThrow(new RuntimeException("예상 밖 버그")).when(ragFacade).processJob(1L); - - given(ragResponseRepository.findFirstByStatusOrderByCreatedAtAsc(ResultStatus.PROCESSING)) - .willReturn(Optional.of(failingJob)); - ragJobWorker.processNext(); // 1번째 폴링: failingJob 실패 → FAILED로 확정됨 - - // FAILED로 확정됐으니 실제 DB에선 이제 findFirst...가 다음 대기 건(nextJob)을 돌려준다 — - // 여기서는 그 상태 변화를 목으로 흉내낸다. - given(ragResponseRepository.findFirstByStatusOrderByCreatedAtAsc(ResultStatus.PROCESSING)) - .willReturn(Optional.of(nextJob)); - given(ragFacade.processJob(2L)).willReturn(true); - ragJobWorker.processNext(); // 2번째 폴링: nextJob은 정상 처리돼야 한다 - - then(ragFacade).should(times(1)).processJob(2L); - then(ragWebSocketController).should(times(1)).notifyAnswerReady("user2@example.com", 200L); + @DisplayName("claim 중 예외가 나면 슬롯을 반환하고 Executor를 부르지 않는다") + void processNext_claimThrows_releasesSlotAndSkipsExecutor() { + given(ragResponseClaimService.claimNext()).willThrow(new RuntimeException("DB 오류")); + + ragJobWorker.processNext(); + + then(ragWorkerJobExecutor).should(never()).execute(any()); + assertThat(ragWorkerSlots.availablePermits()).isEqualTo(1); + } + + @Test + @DisplayName("Executor 제출이 거부되면(RejectedExecutionException) 슬롯을 반환한다") + void processNext_executorRejects_releasesSlot() { + given(ragResponseClaimService.claimNext()).willReturn(Optional.of(999L)); + doThrow(new RejectedExecutionException()).when(ragWorkerJobExecutor).execute(any()); + + ragJobWorker.processNext(); + + assertThat(ragWorkerSlots.availablePermits()).isEqualTo(1); + } + + @Test + @DisplayName("claim 직후 job이 사라졌으면(극단적 상황) processJob을 부르지 않고 슬롯만 반환한다") + void processNext_claimedJobVanished_skipsProcessing() { + given(ragResponseClaimService.claimNext()).willReturn(Optional.of(999L)); + given(ragResponseRepository.findWithQueryAndUserById(999L)).willReturn(Optional.empty()); + + ragJobWorker.processNext(); + runSubmittedTask(); + + then(ragFacade).should(never()).processJob(any()); + assertThat(ragWorkerSlots.availablePermits()).isEqualTo(1); + } + + /** Executor에 제출된 Runnable을 캡처해 테스트 스레드에서 즉시(동기) 실행한다. */ + private void runSubmittedTask() { + ArgumentCaptor taskCaptor = ArgumentCaptor.forClass(Runnable.class); + then(ragWorkerJobExecutor).should(times(1)).execute(taskCaptor.capture()); + taskCaptor.getValue().run(); } private RagResponse deepStubJob(Long jobId, Long queryId, String userEmail) { RagResponse job = mock(RagResponse.class, RETURNS_DEEP_STUBS); - given(job.getId()).willReturn(jobId); given(job.getQuery().getId()).willReturn(queryId); given(job.getQuery().getUser().getEmail()).willReturn(userEmail); return job; diff --git a/docs/design/kangcheolung-#340-rag-parallel-processing.md b/docs/design/kangcheolung-#340-rag-parallel-processing.md new file mode 100644 index 00000000..076e34a6 --- /dev/null +++ b/docs/design/kangcheolung-#340-rag-parallel-processing.md @@ -0,0 +1,230 @@ +# #340 RAG 답변 생성 병렬 처리 + +closes #340 + +--- + +## 배경 + +RAG 답변 생성(`RagJobWorker`)은 GPU 1대·Ollama 인스턴스 1개 전제로 정확히 1개씩 순차 처리하도록 +설계되어 있었다(#218/#286/#288). 이 전제 덕분에 지금까지는 락 없는 조건부 UPDATE만으로 안전했지만, +질문이 몰리면 뒤에 온 사용자일수록 대기 시간이 그대로 누적되는 구조적 한계가 있었다. + +목표는 GPU를 늘리지 않고, Ollama의 병렬 슬롯(`OLLAMA_NUM_PARALLEL`)이 갖는 여유 용량을 실제로 +활용해 애플리케이션 레벨에서 동시에 최대 N개의 질문을 처리하도록 확장하는 것이다. + +--- + +## 설계 — embedding_jobs 패턴을 RAG 규모에 맞게 축소 재사용 + +이 프로젝트에는 이미 정확히 같은 문제(여러 워커가 안전하게 큐를 나눠 갖는 것)를 해결한 무거운 +선례(`embedding_jobs` 워커: `FOR UPDATE SKIP LOCKED` 원자적 claim + 전용 `ThreadPoolExecutor`)가 +있어서, 새로 발명하지 않고 그 패턴을 그대로 축소해 재사용했다. + +### 1. Claim 표식 — 새 `claimed_at` 컬럼 (V43 마이그레이션) + +RAG는 `enqueue()` 시점에 곧바로 `status=PROCESSING`이 된다(`embedding_jobs`처럼 PENDING→PROCESSING의 +별도 단계가 없음). 그래서 "대기 중(아직 아무도 안 집음)"과 "지금 실제로 처리 중(누가 이미 집음)"을 +구분할 방법이 없었다 — 이 구분이 없으면 짧은 claim 트랜잭션이 커밋된 뒤에도 다른 워커가 같은 행을 +또 집을 수 있다. + +`updatedAt`(BaseEntity, `@LastModifiedDate`)을 재사용하는 방안은 기각했다 — 완료 확정 경로 +(`completeSuccessIfProcessing`/`forceFailIfProcessing`)가 전부 `@Modifying` 벌크 JPQL UPDATE라 +JPA 생명주기(`@PreUpdate`)를 안 거쳐서 `updatedAt`이 절대 안 채워짐을 직접 확인했다. + +```sql +-- V43__add_rag_responses_claimed_at.sql +ALTER TABLE rag_responses ADD COLUMN claimed_at TIMESTAMP; +``` + +### 2. 동시성 상한 — 전용 `ThreadPoolExecutor` + `Semaphore` + +`domain/worker/config/WorkerExecutionConfig.java`(이 코드베이스 유일한 커스텀 스레드풀 선례)를 +그대로 본떴다: `core=max=N`인 `ThreadPoolExecutor` + `SynchronousQueue`(큐잉 없음) + +`AbortPolicy`(꽉 차면 즉시 거부) + `CustomizableThreadFactory`. + +동시성 예약은 `embedding_jobs`의 `WorkerExecutionSlotPool`(전체 클래스, 종료 플래그 + +introspection 메서드 포함)까지는 필요 없다고 판단해 순수 `Semaphore(N, true)`만 썼다 — RAG는 +별도 워커 등록/우아한 종료 조율이나 대시보드 노출 요구가 없어서다. **핵심 안전장치는 "로컬 슬롯을 +먼저 확보한 뒤에만 DB claim을 시도"하는 순서** — 이 순서 덕분에 "claim은 됐는데 실행할 스레드가 +없는" 유령 job이 생기지 않는다(`embedding_jobs`의 `WorkerJobPollingScheduler`도 동일한 순서). + +--- + +## 신규/변경 파일 + +| 파일 | 변경 | +|---|---| +| `db/migration/V43__add_rag_responses_claimed_at.sql` | 신규. `claimed_at TIMESTAMP` nullable 컬럼 추가만. | +| `domain/rag/entity/RagResponse.java` | `claimedAt` 필드 + `markClaimed(LocalDateTime)` 메서드 추가. 짧은 claim 트랜잭션 안에서만 로드·수정·커밋되므로 #218 detached-entity 버그와 다른 안전한 케이스. | +| `domain/rag/repository/RagResponseRepository.java` | `findFirstByStatusOrderByCreatedAtAsc` 제거(참조 없음 확인). `findNextUnclaimedProcessingForUpdate()`(native, `SKIP LOCKED`) 신규. `findWithQueryAndUserById(Long)`(`@EntityGraph`) 신규 — 기존 `findById`는 그대로 둠. `releaseAllClaimsOnStartup()`(재시작 복구용, CodeRabbit 리뷰 반영) 신규. | +| `domain/rag/service/command/RagResponseClaimService.java` | 신규. `@Service @Transactional`, `claimNext(): Optional` — 짧은 트랜잭션 안에서 claim 쿼리 실행 후 즉시 `markClaimed()` 호출, 커밋과 함께 락 해제. `recoverStaleClaimsOnStartup()`(CodeRabbit 리뷰 반영) 추가. | +| `domain/rag/config/RagExecutionConfig.java` | 신규. `ragWorkerJobExecutor`(`ThreadPoolExecutor`) + `ragWorkerSlots`(`Semaphore`), `rag.worker.max-concurrency` 기반. `@ConditionalOnProperty` 없음(RAG는 항상 켜져야 함). | +| `domain/rag/service/RagJobWorker.java` | `processNext()`를 디스패처로 재작성: `while (ragWorkerSlots.tryAcquire())` → claim → 비었으면 슬롯 반환, 있으면 `ragWorkerJobExecutor.execute(() -> executeClaimedJob(jobId))`. `recoverStaleClaimsOnStartup()`(`@EventListener(ApplicationReadyEvent.class)`, CodeRabbit 리뷰 반영) 추가. | +| `application.yml` | `rag.worker.max-concurrency: 2` 추가. | +| `RagFacade.java` | 진단용 로그 한 줄 추가(`promptTokens`/`answerTokens`, 아래 "추가 개선 검토" 참고). Javadoc 한 줄 수정(옛 메서드명 참조 정정). 로직 무변경. | + +**변경 불필요 확인됨**: `RagJobTimeoutSweeper`, `RagResponseCommandService` — 전부 job id 기반 + +조건부 UPDATE라 호출자가 몇 명이든 이미 안전. + +--- + +## 테스트 + +- 기존 4개 테스트 파일을 새 구조(claim 서비스, `Semaphore`, `ThreadPoolExecutor`)에 맞게 갱신 +- 신규 `RagResponseClaimIntegrationTest` 추가 — `EmbeddingJobClaimIntegrationTest`를 그대로 본떠 + 실제 동시 트랜잭션(`TransactionTemplate` + `PROPAGATION_REQUIRES_NEW`, 별도 스레드)으로 + `SKIP LOCKED` 스킵 동작과 "두 스레드가 동시에 claim해도 하나만 성공"을 검증 +- `RagJobWorkerIntegrationTest`는 반드시 수정이 필요했다 — `processNext()`가 이제 claim만 하고 + 즉시 반환하므로(실제 처리는 Executor로 위임), 기존의 "호출 직후 동기 검증" 방식이 깨져서 + Awaitility로 전환 +- `RagJobWorkerConcurrentQueueIntegrationTest`에 동시성 증명 assertion 추가 — `claimed_at` 값들의 + 최소 간격이 10초 이내인지 확인(실제 동시 claim의 증거) +- (CodeRabbit 리뷰 반영) `RagResponseRepositoryTest`에 `releaseAllClaimsOnStartup()` 케이스 2개, + `RagJobWorkerTest`에 `recoverStaleClaimsOnStartup()` 위임 검증 1개 추가 + +### 실행 결과 + +``` +RAG 패키지 테스트 9개 파일 전부 통과 (통합 테스트 3개 포함, 실제 로컬 PostgreSQL + Ollama 대상) +전체 프로젝트 테스트 → 실패 0건, 에러 0건 (초기 확인 1,178개 + CodeRabbit 반영 후 재확인) +``` + +--- + +## 실측 — 3단계로 진행 + +### 1단계: Ollama 서버 자체의 병렬 처리 능력 확인 + +`~/Library/LaunchAgents/homebrew.mxcl.ollama.plist`에 `OLLAMA_NUM_PARALLEL`을 설정하고 +`launchctl unload`/`load`로 재로드(`brew services restart`는 plist를 재생성해 커스텀 설정을 +지우므로 쓰지 않음 — #210에서 이미 확인된 함정). + +curl로 동일 프롬프트를 동시에 N개 날려서 실측: + +| N | 총 처리 토큰 | 전체 소요시간 | 처리량(토큰/초) | 개별 요청당 속도 | +|---|---|---|---|---| +| 1 (단독) | 73 | 4.92초 | 14.85 | 14.85 | +| 2 | 164 | 8.68초 | 18.9 | ~10.0 | +| 3 | 228 | 12.28초 | 18.6 | ~6.4 | +| 4 | 297 | 13.24초 | 22.4 | ~5.7 | + +**결론**: 1→2에서 확실한 이득(약 27% 처리량 증가)이 나오고, 2 이후로는 처리량은 제자리인데 +개별 응답 속도만 계속 나빠짐. **N=2가 이 하드웨어(맥 통합메모리)의 실질적 적정선**이라고 +결론짓고 최종적으로 `OLLAMA_NUM_PARALLEL=2`로 확정. + +### 2단계: 실제 RAG 파이프라인으로 동시성 재검증 + +`RagJobWorkerConcurrentQueueIntegrationTest`(질문 3개 동시 접수)를 N=2 설정에서 재실행: + +``` +개별 질문 소요시간: 4.1초 / 26.1초 / 38.8초 +claim 시각: 2개는 거의 동시(18ms 차이), 3번째는 슬롯이 빌 때까지 약 27초 대기 +전체 소요시간: 41.2초 +``` + +순차 처리였다면 4.1+26.1+38.8=69초가 걸렸어야 하는데 실제로는 41.2초 — **약 40% 단축**을 +실측으로 확인. + +### 3단계: 타임아웃 값(`generate-deadline`/`stale-threshold`) 재검토 + +실측 최악값(38.8초)이 현재 `generate-deadline`(60초)/`stale-threshold`(90초) 안에 여유 있게 +들어옴. `#218` 설계 당시 정책("실사용에서 타임아웃 로그가 쌓이면 그때 재검토")을 뒤집을 근거를 +찾지 못해 **현재 값 유지로 결론**. + +--- + +## 추가 개선 검토 (외부 피드백 기반, 실측으로 검증) + +병렬화 이후 추가로 시도할 수 있는 개선안들을 검토하고, 가능한 건 직접 실측했다. + +### 진단 로그 추가 (적용함) + +`RagFacade.java`의 완료 로그에 `promptTokens`/`answerTokens`를 추가 — 느린 job이 프롬프트를 +읽느라(prefill) 오래 걸렸는지 답변을 쓰느라(decode) 오래 걸렸는지 로그만으로 구분 가능하게 함. + +```java +log.info("[RAG] done queryId={} responseId={} latencyMs={} promptTokens={} answerTokens={}", + queryId, job.getId(), result.latencyMs(), result.inputTokenCount(), result.outputTokenCount()); +``` + +### 실제 서버 기동 + 실제 QA로 진단 데이터 확보 + +로컬에 백엔드를 직접 기동(`./backend/gradlew -p backend bootRun --args='--spring.profiles.active=local'`) +하고, 회원가입 → 로그인 → `POST /search`까지 실제 API 흐름으로 QA 진행. 실제 인덱싱된 문서 +("Spring Boot 실무 가이드")에 대한 질문으로 진짜 RAG 응답을 받아 진단 로그 확인: + +``` +[RAG] done queryId=223 responseId=131 latencyMs=35922 promptTokens=1503 answerTokens=400 +``` + +`answerTokens`가 상한(400)에 정확히 걸림 — 답변 생성(decode)이 병목일 가능성을 시사. + +### `num-predict` 낮추기 실측 (효과 없다고 결론) + +같은 실제 프롬프트(DB에서 그대로 가져옴)로 `num_predict`만 바꿔가며 직접 재현: + +| num_predict | 소요시간 | 실제 생성 토큰 | 잘렸는지 | 답변 완성도 | +|---|---|---|---|---| +| 150 | 9.5초 | 150 | 예(length) | 5~6개 주제 중 2개도 못 채움 | +| 250 | 15.0초 | 250 | 예(length) | 자동설정까지만, 내장서버/테스트/배포 누락 | +| 400(현재) | 21.6초 | 352(자연종료) | 아니오(stop) | 전체 주제 다 다루고 정상 마무리 | + +**결론**: `num_predict`를 낮추면 확실히 빨라지지만, 그만큼 답변이 요청 내용을 다 못 채우고 +끊긴다. "핵심 내용을 자세히 설명해줘" 같은 질문에는 지금 400이 오히려 적정값에 가까웠다 — +**이 레버는 기각**. (안 해봤으면 몰랐을 유의미한 음성 결과.) + +### 검토했으나 보류한 것들 + +| 후보 | 보류 이유 | +|---|---| +| 토큰 스트리밍 | 체감 지연 개선 효과는 크지만, 이번 스코프에서 제외하기로 결정(사용자 판단) | +| 큐 공정성(사용자당 동시 처리 1건 제한) | 효과는 있으나 이번엔 스킵하기로 결정 | +| `OLLAMA_KV_CACHE_TYPE=q8_0` 재검토 | #210에서 글자 깨짐 원인으로 지목돼 제거됐던 설정이 현재 다시 켜져 있는 게 확인됨(원인 미상, 실제로 한자 혼입 재현도 확인함). 코드가 이미 방어하고 있어 당장 급하지 않다고 판단해 보류. | +| 답변 캐시 | 멀티턴 문맥이 프롬프트에 섞여 들어가는 구조라 캐시 히트율이 낮을 것으로 예상, 우선순위 낮춤 | +| GPU 인프라 교체(클라우드 GPU/vLLM) | 진짜 처리량 상한을 뚫는 유일한 방법이지만, 비용·복잡도가 이번 스코프보다 큼. 코드 변경은 `ollama.server.base-url` 하나만 바꾸면 되는 수준이라 향후 확장 경로로 남겨둠 | +| Speculative decoding | Ollama 0.32.13 CLI/API에 관련 옵션이 전혀 없음을 확인 — 현재 버전에서는 지원 안 함으로 결론 | +| Kafka 도입 | 병목이 메시지 전달 속도가 아니라 GPU 메모리 대역폭이라 무관 — 오히려 불필요한 인프라 복잡도만 추가. 기각. | + +--- + +## 코드리뷰 반영 (CodeRabbit) + +PR #341에 직접 고도화 아이디어를 질문했고, 5가지 제안이 왔다. 하나씩 실제 코드/과거 설계 +문서와 대조 검증한 뒤 처리했다. + +| # | 제안 요지 | 처리 | 근거 | +|---|---|---|---| +| 1 | 재시작 시 `claimed_at`이 남은 job은 새 프로세스가 영원히 재claim하지 못한다 | **반영함** | 검증 결과 실제 퇴보였다 — #218 이전(순수 status 기반) 방식은 재시작하면 자동으로 재시도됐는데, claim 도입 후에는 스위퍼의 fallback만 기다리게 된다. 앱 시작 시 1회(`ApplicationReadyEvent`) `claimed_at`을 전부 풀어주는 복구를 추가했다("인스턴스 1개" 전제 위에서만 안전, 이 전제는 이미 이 Worker 전체 설계의 기존 전제와 동일). | +| 2 | 큐 대기 시간과 실행 시간을 분리해 타임아웃 판단해야 한다 | **반영 안 함(문서화만)** | `#286` 설계 문서에 이미 동일한 내용이 "실사용에서 재조정" 항목으로 기록돼 있었다. 오늘 실측 최악값(38.8초)이 90초 기준 안에 여유 있어 지금 분리할 근거 데이터가 없다 — 감으로 값을 새로 짓느니, 값을 넉넉히 잡으면 진짜 hang 감지가 오히려 늦어지는 트레이드오프도 있어 보류. | +| 3 | claim 쿼리에 partial index 추가 검토 | **반영 안 함** | CodeRabbit 스스로도 "큐가 작으면 우선순위 낮음, `EXPLAIN`으로 확인 후 결정"이라고 명시. 설계 문서에 이미 같은 결론(인덱스 불필요)이 적혀 있었다. | +| 4 | claim 타이밍만이 아니라 실제 `generate()` 호출 자체가 겹치는지 직접 증명하는 테스트 추가 | **반영 안 함(후속 과제)** | claim 직후 Executor가 곧바로 처리를 시작하는 구조상 claim 타이밍이 곧 generate 호출 타이밍의 신뢰할 만한 대리 지표다. "정렬된 리스트의 인덱스 0,1이 항상 최소 간격"이라는 지적도 일반적으로는 맞지만, 이 설계(슬롯 2개+3건 접수)에서는 3번째 job이 항상 앞 두 개보다 늦게 claim될 수밖에 없어 현재 검증 방식이 틀리진 않았다. 더 강한 증명(`OllamaClient` 테스트 더블 + `CountDownLatch`)은 가치 있으나 지금 급하지 않음. | +| 5 | queue_wait/execution_time/슬롯 사용률 등 운영 지표 추가 | **반영 안 함(다른 계획에 포함)** | 이미 별도로 미뤄둔 관측성(Grafana/트레이싱) 작업 범위와 겹친다 — 그때 같이 반영 예정. | + +## 설계 결정 요약 + +- **claim 표식은 새 컬럼(`claimed_at`)으로**: `updatedAt` 재사용은 벌크 UPDATE 경로에서 실제로 + 안 채워짐을 코드로 확인하고 기각 +- **`embedding_jobs`의 무거운 패턴을 축소 재사용**: `WorkerExecutionSlotPool` 클래스 전체 대신 + 순수 `Semaphore`로 단순화 — RAG는 분산 워커 등록/우아한 종료 요구가 없어서 +- **로컬 슬롯 확보 → DB claim 순서 고정**: "claim은 됐는데 실행할 스레드가 없는" 유령 job 방지가 + 이 순서의 핵심 목적 +- **`RagFacade`/`RagJobTimeoutSweeper` 무변경**: #288에서 이미 "누가 몇 명이든 안전"하게 + 설계돼 있어서 그 위에 얹기만 함 +- **N=2를 이론이 아니라 실측으로 확정**: curl 기반 인프라 실측 + 실제 파이프라인 재검증 2단계로 + 교차 확인 +- **속도 개선 후속 검토는 실측 우선**: "효과 있을 것 같다"는 감이 아니라 직접 서버를 켜고 실제 + QA로 데이터를 뽑아서 기각/채택을 결정(`num-predict` 축소가 대표적 — 실측 전엔 유망해 보였으나 + 품질 손실이 이득보다 크다는 게 실측으로 드러남) + +--- + +## 남은 이슈 / TODO + +- `OLLAMA_KV_CACHE_TYPE=q8_0`이 왜 다시 켜졌는지 원인 불명 — 별도 조사 필요 +- 큐 공정성(사용자당 동시 처리 제한)은 구현 난이도 대비 효과가 괜찮아 보이는 후보로, 나중에 + 재검토 가치 있음 +- GPU 인프라 확장(클라우드 GPU/vLLM)은 코드 준비(claim 구조)는 되어 있으나 라우팅 로직 없음 — + 실제로 여러 Ollama 인스턴스를 쓰게 되면 별도 설계 필요 +- 진단 로그(`promptTokens`/`answerTokens`)는 이제 막 추가되어 실사용 데이터가 아직 없음 — 운영 + 중 쌓이는 로그를 보고 추가 튜닝 여부 재판단