[Fix] RAG 답변 생성이 PROCESSING 상태로 무기한 대기하는 문제 수정

backend/src/main/java/com/opensource/docgrid/domain/rag/repository/RagResponseRepository.java

@@ -1,9 +1,14 @@
 package com.opensource.docgrid.domain.rag.repository;
 
+import java.time.LocalDateTime;
+import java.util.List;
 import java.util.Optional;
 
 import org.springframework.data.jpa.repository.EntityGraph;
 import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
 
 import com.opensource.docgrid.domain.rag.entity.RagResponse;
 import com.opensource.docgrid.domain.search.enums.ResultStatus;
@@ -28,4 +33,35 @@ public interface RagResponseRepository extends JpaRepository<RagResponse, Long>
 
     /** 특정 검색 요청(queryId)에 대한 RAG 답변을 찾는다. GET /search/{queryId} 재조회에 쓰인다. */
     Optional<RagResponse> findByQuery_Id(Long queryId);
+
+    /**
+     * 주어진 상태(보통 PROCESSING)로 threshold 이전부터 남아있는 job들을 찾는다 —
+     * RagJobTimeoutSweeper가 "얼마나 오래 대기 중인지"를 별도 컬럼 없이 {@code createdAt}
+     * 기준으로 판단할 때 쓴다. {@link EntityGraph}로 {@code query}/{@code query.user}를 미리
+     * fetch하는 이유는 {@code findFirstByStatusOrderByCreatedAtAsc}와 동일하다 — 스위퍼도
+     * WebSocket 알림을 보내려면 트랜잭션 밖에서 {@code query.user.email}에 접근해야 한다.
+     */
+    @EntityGraph(attributePaths = {"query", "query.user"})
+    List<RagResponse> findByStatusAndCreatedAtBefore(ResultStatus status, LocalDateTime threshold);
+
+    /**
+     * PROCESSING 상태인 job을 FAILED로 강제 종료한다. {@code WHERE ... AND status = PROCESSING}
+     * 조건 덕분에, 이 UPDATE가 실행되는 순간 RagJobWorker가 이미 다른 트랜잭션에서 이 job을
+     * SUCCESS/FAILED로 먼저 확정했다면 영향받은 행이 0건이 된다 — 이 저장소엔 {@code @Version}
+     * 필드가 없어 엔티티를 그대로 불러와 save()하면 나중 쓰기가 그냥 이기는데, 그 대신 이 조건부
+     * UPDATE로 "이미 끝난 job을 덮어쓰는" 경합을 막는다. 반환값(영향받은 행 수)으로 호출자가
+     * 실제로 강제 종료가 일어났는지 판단한다.
+     *
+     * <p>{@code clearAutomatically}: 벌크 UPDATE는 영속성 컨텍스트를 거치지 않고 DB에 직접
+     * 실행되므로, 같은 트랜잭션에서 이 job 엔티티를 이미 로딩해둔 상태라면 그 캐시된 인스턴스가
+     * 여전히 갱신 전 값을 들고 있다 — 이후 같은 트랜잭션에서 다시 조회해도 DB가 아니라 그 캐시를
+     * 돌려줘 최신 상태를 못 본다. {@code clearAutomatically = true}로 UPDATE 직후 영속성
+     * 컨텍스트를 비워 이 문제를 막는다.
+     */
+    @Modifying(clearAutomatically = true)
+    @Query("UPDATE RagResponse r SET r.status = com.opensource.docgrid.domain.search.enums.ResultStatus.FAILED, "
+        + "r.answerText = :answerText, r.errorMessage = :errorMessage "
+        + "WHERE r.id = :id AND r.status = com.opensource.docgrid.domain.search.enums.ResultStatus.PROCESSING")
+    int forceFailIfProcessing(@Param("id") Long id, @Param("answerText") String answerText,
+                               @Param("errorMessage") String errorMessage);
 }

backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagFacade.java

@@ -62,13 +62,19 @@ public class RagFacade {
      *                                지시한다(#65 INSTRUCTION 참고). 검색은 됐지만(candidates 존재)
      *                                LLM이 무관하다고 판단한 경우, 화면에 근거 문서를 같이 보여주면
      *                                안내 문구와 모순돼 보인다.
+     * TIMEOUT_ERROR_MESSAGE        — RagJobTimeoutSweeper가 너무 오래 PROCESSING으로 남은 job을
+     *                                강제 종료할 때 error_message에 남기는 문구(#286). Ollama
+     *                                예외 메시지와 구분해, 나중에 로그/DB로 "진짜 실패"와 "큐
+     *                                적체로 인한 강제 종료"를 구분할 수 있게 한다.
      */
     private static final String LLM_FALLBACK_PREFIX = "AI 답변 생성이 지연되고 있습니다. "
         + "가장 관련도 높은 문서에서 다음 내용을 찾았습니다:\n\n";
     private static final String UNEXPECTED_FAILURE_ANSWER_TEXT = "답변 생성 중 예상치 못한 오류가 발생했습니다.";
     private static final int FALLBACK_EXCERPT_MAX_CODE_POINTS = 300;
     private static final int MAX_PROMPT_CANDIDATES = 3;
     private static final String NO_RELEVANT_DOC_PHRASE = "관련 문서를 찾지 못했습니다";
+    private static final String TIMEOUT_ERROR_MESSAGE =
+        "PROCESSING 상태 유지 시간이 임계값을 초과해 강제 종료됨(RagJobTimeoutSweeper)";
 
     private final PromptBuilder promptBuilder;
     private final OllamaClient ollamaClient;
@@ -180,6 +186,24 @@ public void markUnexpectedFailure(Long jobId, String errorMessage) {
             .ifPresent(job -> ragResponseCommandService.completeFailed(job, UNEXPECTED_FAILURE_ANSWER_TEXT, errorMessage));
     }
 
+    /**
+     * RagJobTimeoutSweeper가 너무 오래 PROCESSING으로 남은 job을 발견했을 때 호출한다(#286).
+     * LLM 호출 실패 fallback과 동일하게 검색 1등 후보를 인용한 답으로 채우되, 실제 종료는
+     * {@link RagResponseRepository#forceFailIfProcessing}의 조건부 UPDATE로만 한다 — 그 사이
+     * RagJobWorker가 이미 이 job을 정상 완료했다면 영향받은 행이 0건이라 덮어쓰지 않는다.
+     *
+     * @return 실제로 이 호출로 FAILED 전환이 일어났으면 true, 이미 다른 트랜잭션에서 끝나
+     *         아무 일도 하지 않았으면 false.
+     */
+    public boolean failIfStillProcessing(Long jobId, Long queryId) {
+        List<VectorSearchCandidate> candidates = loadCandidates(queryId);
+        String fallbackAnswer = candidates.isEmpty()
+            ? UNEXPECTED_FAILURE_ANSWER_TEXT
+            : buildExtractiveFallbackAnswer(candidates);
+        int updated = ragResponseRepository.forceFailIfProcessing(jobId, fallbackAnswer, TIMEOUT_ERROR_MESSAGE);
+        return updated > 0;
+    }
+
     /**
      * Worker는 검색 시점의 in-memory candidates를 갖고 있지 않으므로(완전히 다른 스레드·시점),
      * 이미 영속화된 search_results(+chunk)에서 동일한 순서({@code ORDER BY rank_no})로 다시

backend/src/main/java/com/opensource/docgrid/domain/rag/service/RagJobTimeoutSweeper.java

@@ -0,0 +1,66 @@
+package com.opensource.docgrid.domain.rag.service;
+
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.util.List;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+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 lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * PROCESSING 상태로 너무 오래 남아있는 RagResponse를 강제로 FAILED 종료시키는 안전망 (#286).
+ *
+ * <p>RagJobWorker는 Ollama가 GPU 1개로 순차 처리된다는 전제 위에서 동작하는데(RagJobWorker
+ * 클래스 Javadoc 참고), 이 순차 처리 자체에는 대기 시간 상한이 없다 — 앞선 job이 정상 흐름이든
+ * hang이든 끝나야 다음 job이 처리된다. 이 스위퍼는 RagJobWorker와 별도의 주기로 폴링하며,
+ * 일정 시간(rag.worker.stale-threshold) 이상 PROCESSING인 job을 찾아 기존 extractive fallback
+ * 답변({@link RagFacade#failIfStillProcessing})으로 강제 종료시켜, 사용자가 무기한 대기하지
+ * 않도록 상한을 만든다.
+ *
+ * <p>RagJobWorker가 같은 job을 이 스위퍼와 거의 동시에 정상 완료할 수 있는 경합은
+ * {@link RagFacade#failIfStillProcessing}이 내부적으로 쓰는 조건부 UPDATE로 방지된다 — 이미
+ * 끝난 job이면 아무 일도 일어나지 않는다.
+ */
+@Component
+@RequiredArgsConstructor
+@Slf4j
+public class RagJobTimeoutSweeper {
+
+    private final RagResponseRepository ragResponseRepository;
+    private final RagFacade ragFacade;
+    private final RagWebSocketController ragWebSocketController;
+
+    @Value("${rag.worker.stale-threshold:90s}")
+    private Duration staleThreshold;
+
+    /**
+     * stale-threshold 이상 PROCESSING으로 남아있는 job을 전부 찾아 하나씩 강제 종료한다.
+     * 이 메서드 자체는 트랜잭션이 아니다 — {@link RagFacade#failIfStillProcessing}이 job마다
+     * 독립된 트랜잭션으로 실행되므로, 하나가 실패해도 나머지 job 처리에 영향을 주지 않는다.
+     */
+    @Scheduled(fixedDelayString = "${rag.worker.timeout-sweep-interval:15s}")
+    public void sweep() {
+        LocalDateTime cutoff = LocalDateTime.now().minus(staleThreshold);
+        List<RagResponse> staleJobs =
+            ragResponseRepository.findByStatusAndCreatedAtBefore(ResultStatus.PROCESSING, cutoff);
+
+        for (RagResponse job : staleJobs) {
+            Long queryId = job.getQuery().getId();
+            String userEmail = job.getQuery().getUser().getEmail();
+
+            if (ragFacade.failIfStillProcessing(job.getId(), queryId)) {
+                log.warn("[RAG-SWEEP] stale job force-failed queryId={} responseId={}", queryId, job.getId());
+                ragWebSocketController.notifyAnswerReady(userEmail, queryId);
+            }
+        }
+    }
+}

backend/src/main/resources/application.yml

@@ -176,3 +176,8 @@ rag:
     # RagJobWorker가 PROCESSING 건이 있는지 확인하는 주기. Worker가 1개뿐이라 짧게 잡아도
     # 부하가 안 크고, 사용자 체감 지연에 직접 영향을 주므로 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}

backend/src/test/java/com/opensource/docgrid/domain/rag/repository/RagResponseRepositoryTest.java

@@ -0,0 +1,125 @@
+package com.opensource.docgrid.domain.rag.repository;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.autoconfigure.jdbc.AutoConfigureTestDatabase;
+import org.springframework.boot.test.autoconfigure.orm.jpa.DataJpaTest;
+import org.springframework.test.context.ActiveProfiles;
+
+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.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.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;
+
+/**
+ * RagJobTimeoutSweeper(#286)가 의존하는 두 쿼리를 실제 PostgreSQL Repository 계층에서
+ * 검증한다. 특히 {@code forceFailIfProcessing()}의 "이미 끝난 job은 절대 덮어쓰지 않는다"는
+ * 조건부 UPDATE 정합성은 이번 수정의 핵심 안전장치라 Mockito 단위 테스트로는 증명할 수 없고,
+ * 실제 SQL이 실행되는 이 계층에서만 검증할 수 있다.
+ */
+@DataJpaTest
+@ActiveProfiles("test")
+@AutoConfigureTestDatabase(replace = AutoConfigureTestDatabase.Replace.NONE)
+@DisplayName("RagResponseRepository 테스트")
+class RagResponseRepositoryTest {
+
+    @Autowired
+    private RagResponseRepository ragResponseRepository;
+
+    @Autowired
+    private SearchQueryRepository searchQueryRepository;
+
+    @Autowired
+    private UserRepository userRepository;
+
+    @Autowired
+    private EmbeddingModelRepository embeddingModelRepository;
+
+    @Test
+    @DisplayName("forceFailIfProcessing: PROCESSING인 job은 FAILED로 강제 종료되고 영향받은 행이 1건이다")
+    void forceFailIfProcessing_processingJob_updatesToFailedAndReturnsOne() {
+        RagResponse job = saveRagResponse(ResultStatus.PROCESSING);
+
+        int updated = ragResponseRepository.forceFailIfProcessing(job.getId(), "fallback 답변", "타임아웃");
+
+        assertThat(updated).isEqualTo(1);
+        RagResponse reloaded = ragResponseRepository.findById(job.getId()).orElseThrow();
+        assertThat(reloaded.getStatus()).isEqualTo(ResultStatus.FAILED);
+        assertThat(reloaded.getAnswerText()).isEqualTo("fallback 답변");
+        assertThat(reloaded.getErrorMessage()).isEqualTo("타임아웃");
+    }
+
+    @Test
+    @DisplayName("forceFailIfProcessing: 이미 SUCCESS로 끝난 job은 덮어쓰지 않고 영향받은 행이 0건이다")
+    void forceFailIfProcessing_alreadySucceededJob_doesNotOverwriteAndReturnsZero() {
+        RagResponse job = saveRagResponse(ResultStatus.PROCESSING);
+        job.markSuccess("실제 답변", "qwen2.5:7b", 100, 20, 900);
+        ragResponseRepository.saveAndFlush(job);
+
+        // RagJobWorker가 이 순간 이미 SUCCESS로 커밋한 상황을 재현한다 — 스위퍼의 강제 종료는
+        // 이 시점 이후 실행돼도 status 조건이 안 맞아 아무것도 바꾸면 안 된다.
+        int updated = ragResponseRepository.forceFailIfProcessing(job.getId(), "fallback 답변", "타임아웃");
+
+        assertThat(updated).isEqualTo(0);
+        RagResponse reloaded = ragResponseRepository.findById(job.getId()).orElseThrow();
+        assertThat(reloaded.getStatus()).isEqualTo(ResultStatus.SUCCESS);
+        assertThat(reloaded.getAnswerText()).isEqualTo("실제 답변");
+    }
+
+    @Test
+    @DisplayName("findByStatusAndCreatedAtBefore: cutoff 이전에 생성된 PROCESSING만 찾고, 상태가 다른 job은 제외한다")
+    void findByStatusAndCreatedAtBefore_filtersOnStatusAndCreatedAt() {
+        LocalDateTime beforeAnyCreation = LocalDateTime.now();
+        RagResponse processingJob = saveRagResponse(ResultStatus.PROCESSING);
+        saveRagResponse(ResultStatus.SUCCESS);
+
+        // cutoff가 두 job이 생성되기 전 시점이면(=아직 아무 job도 이 시간만큼 오래 기다리지 않음)
+        // 아무것도 찾지 못해야 한다.
+        assertThat(ragResponseRepository.findByStatusAndCreatedAtBefore(ResultStatus.PROCESSING, beforeAnyCreation))
+            .isEmpty();
+
+        LocalDateTime afterCreation = LocalDateTime.now();
+        List<RagResponse> result =
+            ragResponseRepository.findByStatusAndCreatedAtBefore(ResultStatus.PROCESSING, afterCreation);
+
+        assertThat(result).extracting(RagResponse::getId).containsExactly(processingJob.getId());
+    }
+
+    private RagResponse saveRagResponse(ResultStatus status) {
+        User user = userRepository.save(User.builder()
+            .email("rag-repo-test-" + System.nanoTime() + "@test.local")
+            .passwordHash("x")
+            .name("RAG저장소테스트유저")
+            .status(UserStatus.ACTIVE)
+            .build());
+        EmbeddingModel model = embeddingModelRepository.save(
+            EmbeddingModelFixture.createModel("rag-repo-test-" + System.nanoTime(), false, false));
+        SearchQuery query = searchQueryRepository.save(SearchQuery.builder()
+            .user(user)
+            .queryText("테스트 질문")
+            .queryEmbeddingModel(model)
+            .queryVector(new float[1024])
+            .searchType(SearchType.VECTOR)
+            .topK(5)
+            .status(ResultStatus.SUCCESS)
+            .build());
+        return ragResponseRepository.save(RagResponse.builder()
+            .query(query)
+            .promptText("프롬프트")
+            .status(status)
+            .build());
+    }
+}

backend/src/test/java/com/opensource/docgrid/domain/rag/service/RagFacadeTest.java

@@ -234,6 +234,47 @@ void processJob_phraseEmbeddedInAnswer_trimsBeforePersistingAndKeepsCitations()
         then(responseCitationCommandService).should(times(1)).saveAll(eq(job), any(), eq(List.of(searchResult)));
     }
 
+    // === failIfStillProcessing() ===
+
+    @Test
+    @DisplayName("failIfStillProcessing: 검색 후보가 있으면 extractive fallback으로 강제 종료하고 true를 반환한다")
+    void failIfStillProcessing_withCandidates_forceFailsWithFallbackAndReturnsTrue() {
+        SearchResult searchResult = deepStubSearchResult(100L, 10L, "청크 내용", 12, "인사규정", new BigDecimal("0.9"));
+        given(searchResultRepository.findByQuery_IdOrderByRankNo(QUERY_ID)).willReturn(List.of(searchResult));
+        given(ragResponseRepository.forceFailIfProcessing(eq(JOB_ID), anyString(), anyString())).willReturn(1);
+
+        boolean result = ragFacade.failIfStillProcessing(JOB_ID, QUERY_ID);
+
+        assertThat(result).isTrue();
+        then(ragResponseRepository).should(times(1)).forceFailIfProcessing(
+            eq(JOB_ID), argThatFallbackContains("AI 답변 생성이 지연", "청크 내용", "인사규정"), anyString()
+        );
+    }
+
+    @Test
+    @DisplayName("failIfStillProcessing: 이미 다른 트랜잭션에서 끝난 job이면(영향받은 행 0건) false를 반환한다")
+    void failIfStillProcessing_alreadyFinishedByWorker_returnsFalse() {
+        SearchResult searchResult = deepStubSearchResult(100L, 10L, "청크 내용", 12, "인사규정", new BigDecimal("0.9"));
+        given(searchResultRepository.findByQuery_IdOrderByRankNo(QUERY_ID)).willReturn(List.of(searchResult));
+        given(ragResponseRepository.forceFailIfProcessing(eq(JOB_ID), anyString(), anyString())).willReturn(0);
+
+        boolean result = ragFacade.failIfStillProcessing(JOB_ID, QUERY_ID);
+
+        assertThat(result).isFalse();
+    }
+
+    @Test
+    @DisplayName("failIfStillProcessing: 검색 후보가 없으면(이론상 도달 불가능한 방어 분기) 고정 안내 문구로 강제 종료한다")
+    void failIfStillProcessing_noCandidates_usesUnexpectedFailureAnswerText() {
+        given(searchResultRepository.findByQuery_IdOrderByRankNo(QUERY_ID)).willReturn(List.of());
+        given(ragResponseRepository.forceFailIfProcessing(eq(JOB_ID), anyString(), anyString())).willReturn(1);
+
+        ragFacade.failIfStillProcessing(JOB_ID, QUERY_ID);
+
+        then(ragResponseRepository).should(times(1))
+            .forceFailIfProcessing(eq(JOB_ID), eq("답변 생성 중 예상치 못한 오류가 발생했습니다."), anyString());
+    }
+
     private SearchResult deepStubSearchResult(
         Long documentId, Long chunkId, String chunkText, Integer pageNo, String documentTitle, BigDecimal similarityScore
     ) {