[Observability] 비동기 Pipeline 상태 전이 메트릭과 실패율 경보 추가

backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingCompletionService.java

@@ -37,6 +37,7 @@
 import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository;
 import com.opensource.docgrid.global.exception.DocGridException;
 import com.opensource.docgrid.global.exception.ErrorCode;
+import com.opensource.docgrid.global.observability.EmbeddingJobAttemptMetricEvent;
 
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
@@ -137,9 +138,10 @@ public DocumentIndexingCompletionResponse complete(
             durationMs
         );
 
-        // 6. 대시보드가 최신 집계를 다시 계산하도록 상태 전이를 알린다. AFTER_COMMIT 구독자만
-        //    반응하므로 이 Transaction이 실제로 커밋된 뒤에만 push로 이어진다.
+        // 6. 대시보드 갱신과 성공 Counter를 같은 Commit에 결박한다. 두 AFTER_COMMIT 구독자는
+        //    이 Transaction이 실제로 커밋된 뒤에만 반응한다.
         applicationEventPublisher.publishEvent(new EmbeddingJobStatusChangedEvent(embeddingJob.getId()));
+        applicationEventPublisher.publishEvent(EmbeddingJobAttemptMetricEvent.success());
 
         log.info(
             "문서 인덱싱 완료: jobId={}, attemptId={}, documentId={}, versionId={}, "

backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/DocumentIndexingFailureService.java

@@ -29,6 +29,7 @@
 import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository;
 import com.opensource.docgrid.global.exception.DocGridException;
 import com.opensource.docgrid.global.exception.ErrorCode;
+import com.opensource.docgrid.global.observability.EmbeddingJobAttemptMetricEvent;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -167,11 +168,15 @@ public DocumentIndexingFailureResponse fail(
             minimumRetryDelay
         );
 
-        // 4. 대시보드가 최신 집계를 다시 계산하도록 상태 전이를 알린다. transition()이 재시도 예약
+        // 4. 대시보드 갱신과 실패 Counter를 같은 Commit에 결박한다. transition()이 재시도 예약
         //    (PENDING)과 최종 실패(FAILED) 중 어느 쪽으로 끝났든 embeddingJob은 같은 영속 인스턴스라
         //    최종 상태를 그대로 반영한다. AFTER_COMMIT 구독자만 반응하므로 이 Transaction이 실제로
-        //    커밋된 뒤에만 push로 이어진다.
+        //    커밋된 뒤에만 push와 Counter 증가로 이어진다.
         applicationEventPublisher.publishEvent(new EmbeddingJobStatusChangedEvent(embeddingJob.getId()));
+        applicationEventPublisher.publishEvent(EmbeddingJobAttemptMetricEvent.failure(
+            embeddingJob.getStatus(),
+            request.failureType()
+        ));
 
         log.info(
             "문서 인덱싱 실패 기록: jobId={}, attemptId={}, failureType={}, retryCount={}, terminal={}",

backend/src/main/java/com/opensource/docgrid/domain/embedding/service/command/EmbeddingJobLeaseRecoveryService.java

@@ -6,6 +6,7 @@
 import java.util.Optional;
 import java.util.UUID;
 
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
@@ -21,6 +22,7 @@
 import com.opensource.docgrid.domain.worker.repository.IndexingEventRepository;
 import com.opensource.docgrid.global.exception.DocGridException;
 import com.opensource.docgrid.global.exception.ErrorCode;
+import com.opensource.docgrid.global.observability.EmbeddingJobAttemptMetricEvent;
 
 import lombok.RequiredArgsConstructor;
 
@@ -43,6 +45,7 @@ public class EmbeddingJobLeaseRecoveryService {
     private final EmbeddingJobAttemptRepository embeddingJobAttemptRepository;
     private final IndexingEventRepository indexingEventRepository;
     private final IndexingFailureTransitionService failureTransitionService;
+    private final ApplicationEventPublisher applicationEventPublisher;
 
     /**
      * 후보 Snapshot 이후에도 만료 상태인 Job만 회수해 Retry 또는 최종 실패로 전환한다.
@@ -84,6 +87,11 @@ public RecoveryResult recover(Long jobId, LocalDateTime recoveredAt) {
             recoveredAt,
             Duration.ZERO
         );
+
+        // 5. 실제 회수 전이가 커밋된 뒤에만 재시도 또는 최종 실패 Counter가 증가하게 한다.
+        applicationEventPublisher.publishEvent(
+            EmbeddingJobAttemptMetricEvent.leaseExpired(embeddingJob.getStatus())
+        );
         return RecoveryResult.recovered(embeddingJob);
     }
 

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

@@ -2,6 +2,7 @@
 
 import java.util.List;
 
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
@@ -21,6 +22,8 @@
 import com.opensource.docgrid.domain.search.service.query.SearchConversationQueryService;
 import com.opensource.docgrid.global.exception.DocGridException;
 import com.opensource.docgrid.global.exception.ErrorCode;
+import com.opensource.docgrid.global.observability.RagJobCompletionMetricEvent;
+import com.opensource.docgrid.global.observability.RagJobCompletionMetricEvent.Outcome;
 
 import jakarta.persistence.EntityManager;
 import lombok.RequiredArgsConstructor;
@@ -88,6 +91,7 @@ public class RagFacade {
     private final SearchResultRepository searchResultRepository;
     private final SearchConversationQueryService searchConversationQueryService;
     private final EntityManager entityManager;
+    private final ApplicationEventPublisher applicationEventPublisher;
 
     /**
      * 검색 직후 SearchController가 동기 호출하는 접수 단계. Ollama는 아직 호출하지 않는다.
@@ -107,6 +111,8 @@ public RagEnqueueOutcome enqueue(
 
         if (candidates.isEmpty()) {
             RagResponse ragResponse = ragResponseCommandService.createNoContext(queryRef);
+            // LLM 호출을 생략한 정상 완료도 저장 Transaction이 커밋된 뒤 별도 결과로 집계한다.
+            applicationEventPublisher.publishEvent(new RagJobCompletionMetricEvent(Outcome.NO_CONTEXT));
             log.info("[RAG] no context queryId={} responseId={}", queryId, ragResponse.getId());
             return RagEnqueueOutcome.done(RagAnswer.noContext(ragResponse.getAnswerText()));
         }
@@ -169,6 +175,11 @@ public boolean processJob(Long jobId) {
             String fallbackAnswer = candidates.isEmpty() ? e.getErrorCode().getMessage()
                 : buildExtractiveFallbackAnswer(candidates);
             boolean completed = ragResponseCommandService.completeFailed(job, fallbackAnswer, e.getMessage());
+            if (completed) {
+                applicationEventPublisher.publishEvent(
+                    new RagJobCompletionMetricEvent(Outcome.PROVIDER_FALLBACK)
+                );
+            }
             log.warn("[RAG] fallback queryId={} errorCode={}", queryId, e.getErrorCode().getCode());
             return completed;
         }
@@ -216,6 +227,8 @@ public boolean processJob(Long jobId) {
             // 3. 완료 처리가 성공한 답변에만 선택한 후보 순서대로 출처를 저장한다.
             responseCitationCommandService.saveAll(job, citationCandidates, citationSearchResults);
         }
+        // 답변과 citation 저장이 모두 끝난 동일 Transaction의 커밋 이후 성공 Counter를 기록한다.
+        applicationEventPublisher.publishEvent(new RagJobCompletionMetricEvent(Outcome.SUCCESS));
         log.info("[RAG] done queryId={} responseId={} latencyMs={}", queryId, job.getId(), result.latencyMs());
         return true;
     }
@@ -232,9 +245,15 @@ public boolean processJob(Long jobId) {
      *         (RagJobWorker)는 이 경우 WebSocket 알림을 보내지 않는다.
      */
     public boolean markUnexpectedFailure(Long jobId, String errorMessage) {
-        return ragResponseRepository.findById(jobId)
+        boolean completed = ragResponseRepository.findById(jobId)
             .map(job -> ragResponseCommandService.completeFailed(job, UNEXPECTED_FAILURE_ANSWER_TEXT, errorMessage))
             .orElse(false);
+        if (completed) {
+            applicationEventPublisher.publishEvent(
+                new RagJobCompletionMetricEvent(Outcome.UNEXPECTED_FAILURE)
+            );
+        }
+        return completed;
     }
 
     /**
@@ -252,6 +271,9 @@ public boolean failIfStillProcessing(Long jobId, Long queryId) {
             ? UNEXPECTED_FAILURE_ANSWER_TEXT
             : buildExtractiveFallbackAnswer(candidates);
         int updated = ragResponseRepository.forceFailIfProcessing(jobId, fallbackAnswer, TIMEOUT_ERROR_MESSAGE);
+        if (updated > 0) {
+            applicationEventPublisher.publishEvent(new RagJobCompletionMetricEvent(Outcome.TIMEOUT_SWEPT));
+        }
         return updated > 0;
     }
 

backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventDispatchService.java

@@ -4,6 +4,7 @@
 import java.time.LocalDateTime;
 import java.util.Objects;
 
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
@@ -15,6 +16,8 @@
 import com.opensource.docgrid.domain.sync.service.SyncEventHandlerRegistry;
 import com.opensource.docgrid.global.exception.DocGridException;
 import com.opensource.docgrid.global.exception.ErrorCode;
+import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent;
+import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent.Outcome;
 
 import lombok.RequiredArgsConstructor;
 
@@ -33,6 +36,7 @@ public class SyncEventDispatchService {
     private final SyncEventHandlerRegistry syncEventHandlerRegistry;
     private final SyncEventDeliveryAttemptService syncEventDeliveryAttemptService;
     private final Clock clock;
+    private final ApplicationEventPublisher applicationEventPublisher;
 
     /**
      * Claim된 Outbox Event의 Handler 부작용과 완료 전이를 독립 트랜잭션으로 실행한다.
@@ -64,6 +68,9 @@ public void dispatch(ClaimedSyncEvent claimedEvent) {
             completedAt
         );
         completionEvent.complete(claimedEvent.claimToken(), completedAt);
+
+        // 5. Handler 부작용과 완료 상태가 함께 커밋된 뒤에만 처리 성공 Counter를 기록한다.
+        applicationEventPublisher.publishEvent(new SyncEventAttemptMetricEvent(Outcome.PROCESSED));
     }
 
     /**

backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventFailureService.java

@@ -5,6 +5,7 @@
 import java.util.Objects;
 import java.util.UUID;
 
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
@@ -15,6 +16,8 @@
 import com.opensource.docgrid.domain.sync.service.SyncEventRetrySchedule;
 import com.opensource.docgrid.global.exception.DocGridException;
 import com.opensource.docgrid.global.exception.ErrorCode;
+import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent;
+import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent.Outcome;
 
 import lombok.RequiredArgsConstructor;
 
@@ -31,6 +34,7 @@ public class SyncEventFailureService {
     private final SyncEventRetrySchedule syncEventRetrySchedule;
     private final SyncEventDeliveryAttemptService syncEventDeliveryAttemptService;
     private final Clock clock;
+    private final ApplicationEventPublisher applicationEventPublisher;
 
     /**
      * Handler 실패를 현재 Claim의 Delivery Attempt와 Queue 상태에 함께 기록한다.
@@ -54,6 +58,9 @@ public void recordFailure(UUID eventId, UUID claimToken, String errorCode, Strin
         // 4. 이번 실패가 허용 횟수를 채우면 다시 Claim되지 않는 최종 상태로 종결한다.
         if (event.getRetryCount() + 1 >= event.getMaxRetryCount()) {
             event.markFailed(claimToken, errorCode, errorMessage, failedAt);
+            applicationEventPublisher.publishEvent(
+                new SyncEventAttemptMetricEvent(Outcome.TERMINAL_FAILURE)
+            );
             return;
         }
 
@@ -65,6 +72,9 @@ public void recordFailure(UUID eventId, UUID claimToken, String errorCode, Strin
             failedAt,
             syncEventRetrySchedule.nextAvailableAt(event, failedAt)
         );
+        applicationEventPublisher.publishEvent(
+            new SyncEventAttemptMetricEvent(Outcome.RETRY_SCHEDULED)
+        );
     }
 
     /**

backend/src/main/java/com/opensource/docgrid/domain/sync/service/command/SyncEventLeaseRecoveryService.java

@@ -4,6 +4,7 @@
 import java.util.Optional;
 import java.util.UUID;
 
+import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Propagation;
 import org.springframework.transaction.annotation.Transactional;
@@ -12,6 +13,8 @@
 import com.opensource.docgrid.domain.sync.enums.SyncEventStatus;
 import com.opensource.docgrid.domain.sync.repository.SyncOutboxEventRepository;
 import com.opensource.docgrid.domain.sync.service.SyncEventRetrySchedule;
+import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent;
+import com.opensource.docgrid.global.observability.SyncEventAttemptMetricEvent.Outcome;
 
 import lombok.RequiredArgsConstructor;
 
@@ -28,6 +31,7 @@ public class SyncEventLeaseRecoveryService {
     private final SyncOutboxEventRepository syncOutboxEventRepository;
     private final SyncEventRetrySchedule syncEventRetrySchedule;
     private final SyncEventDeliveryAttemptService syncEventDeliveryAttemptService;
+    private final ApplicationEventPublisher applicationEventPublisher;
 
     /**
      * 만료 후보 Event를 다시 확인해 실제로 만료된 현재 Claim만 회수한다.
@@ -65,7 +69,13 @@ public RecoveryResult recover(UUID eventId, LocalDateTime recoveredAt) {
             syncEventRetrySchedule.nextAvailableAt(event, recoveredAt)
         );
 
-        // 5. Scheduler가 회수 결과를 집계할 수 있도록 변경 후 상태를 반환한다.
+        // 5. 최종 실패는 즉시 조치 경보 대상이며, 재예약된 Lease 만료는 회수 활동으로 구분한다.
+        Outcome outcome = event.getStatus() == SyncEventStatus.FAILED
+            ? Outcome.TERMINAL_FAILURE
+            : Outcome.LEASE_RECOVERED;
+        applicationEventPublisher.publishEvent(new SyncEventAttemptMetricEvent(outcome));
+
+        // 6. Scheduler가 회수 결과를 집계할 수 있도록 변경 후 상태를 반환한다.
         return new RecoveryResult(eventId, true, event.getStatus());
     }
 

backend/src/main/java/com/opensource/docgrid/global/observability/AsyncPipelineMetrics.java

@@ -0,0 +1,60 @@
+package com.opensource.docgrid.global.observability;
+
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.event.TransactionPhase;
+import org.springframework.transaction.event.TransactionalEventListener;
+
+import io.micrometer.core.instrument.MeterRegistry;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * 비동기 파이프라인의 커밋된 상태 전이를 Micrometer Counter로 변환한다.
+ *
+ * <p>도메인 서비스는 DB 상태와 같은 트랜잭션에서 제한된 이벤트만 발행한다. 이 구독자는
+ * {@link TransactionPhase#AFTER_COMMIT}에만 실행되므로 롤백된 상태 변화는 Counter에 남지 않는다.
+ */
+@Component
+@RequiredArgsConstructor
+public class AsyncPipelineMetrics {
+
+    private static final String EMBEDDING_ATTEMPTS = "docgrid.embedding.job.attempts";
+    private static final String RAG_COMPLETIONS = "docgrid.rag.job.completions";
+    private static final String SYNC_ATTEMPTS = "docgrid.sync.event.attempts";
+
+    private final MeterRegistry meterRegistry;
+
+    /**
+     * 확정된 Embedding 실행 결과를 실패 분류와 재시도 가능 여부별로 기록한다.
+     */
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void recordEmbeddingAttempt(EmbeddingJobAttemptMetricEvent event) {
+        meterRegistry.counter(
+            EMBEDDING_ATTEMPTS,
+            "outcome", event.outcome().label(),
+            "failure_type", event.failureType().name(),
+            "retryable", Boolean.toString(event.failureType().isRetryable())
+        ).increment();
+    }
+
+    /**
+     * 확정된 RAG 답변 결과를 정상·fallback·timeout 경로별로 기록한다.
+     */
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void recordRagCompletion(RagJobCompletionMetricEvent event) {
+        meterRegistry.counter(
+            RAG_COMPLETIONS,
+            "outcome", event.outcome().label()
+        ).increment();
+    }
+
+    /**
+     * 확정된 Sync Outbox 처리 결과를 완료·재시도·종결·Lease 회수별로 기록한다.
+     */
+    @TransactionalEventListener(phase = TransactionPhase.AFTER_COMMIT)
+    public void recordSyncAttempt(SyncEventAttemptMetricEvent event) {
+        meterRegistry.counter(
+            SYNC_ATTEMPTS,
+            "outcome", event.outcome().label()
+        ).increment();
+    }
+}