From 5c1ae3f35082de316b78439c7cb52c6836cc8b4d Mon Sep 17 00:00:00 2001 From: SeoHyeon Date: Sat, 23 May 2026 15:02:34 +0900 Subject: [PATCH] =?UTF-8?q?feat(messaging):=20RabbitMQ=20DLX=C2=B7DLQ=20+?= =?UTF-8?q?=20=EC=BB=A8=EC=8A=88=EB=A8=B8=20=EC=9E=AC=EC=8B=9C=EB=8F=84=20?= =?UTF-8?q?=EC=A0=95=EC=B1=85=20(US-28)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../config/properties/RabbitMqProperties.java | 23 ++- .../common/messaging/RabbitMqConfig.java | 106 +++++++++++- backend/src/main/resources/application.yml | 8 + .../messaging/RabbitMessagePublisherTest.java | 5 +- .../common/messaging/RabbitMqConfigTest.java | 79 +++++++++ docs/messaging.md | 57 +++++-- infra/CLAUDE.md | 42 +++-- infra/rabbitmq/definitions.json | 157 +++++++++++++++++- realtime/CLAUDE.md | 10 +- 9 files changed, 433 insertions(+), 54 deletions(-) create mode 100644 backend/src/test/java/com/stackup/stackup/common/messaging/RabbitMqConfigTest.java diff --git a/backend/src/main/java/com/stackup/stackup/common/config/properties/RabbitMqProperties.java b/backend/src/main/java/com/stackup/stackup/common/config/properties/RabbitMqProperties.java index 59a75be8..2fc95448 100644 --- a/backend/src/main/java/com/stackup/stackup/common/config/properties/RabbitMqProperties.java +++ b/backend/src/main/java/com/stackup/stackup/common/config/properties/RabbitMqProperties.java @@ -2,6 +2,9 @@ import jakarta.validation.constraints.NotBlank; import jakarta.validation.constraints.NotNull; +import jakarta.validation.constraints.Positive; +import jakarta.validation.constraints.PositiveOrZero; +import java.time.Duration; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.validation.annotation.Validated; @@ -19,7 +22,11 @@ public record RabbitMqProperties( @NotNull Queues queues, @NotNull - RoutingKeyProperties routingKeys + RoutingKeyProperties routingKeys, + @NotNull + DeadLetter deadLetter, + @NotNull + Retry retry ) { public record Message( @@ -76,4 +83,18 @@ public record RoutingKeyProperties( @NotBlank String realtimeSessionNotify ) { } + + public record DeadLetter( + @NotBlank String exchange, + @NotBlank String routingKeyPrefix + ) { + } + + public record Retry( + @PositiveOrZero int maxRetries, + @NotNull Duration initialInterval, + @Positive double multiplier, + @NotNull Duration maxInterval + ) { + } } diff --git a/backend/src/main/java/com/stackup/stackup/common/messaging/RabbitMqConfig.java b/backend/src/main/java/com/stackup/stackup/common/messaging/RabbitMqConfig.java index 497da3b4..1c1ef4de 100644 --- a/backend/src/main/java/com/stackup/stackup/common/messaging/RabbitMqConfig.java +++ b/backend/src/main/java/com/stackup/stackup/common/messaging/RabbitMqConfig.java @@ -1,12 +1,20 @@ package com.stackup.stackup.common.messaging; import com.stackup.stackup.common.config.properties.RabbitMqProperties; +import java.util.Map; +import org.springframework.amqp.core.AcknowledgeMode; import org.springframework.amqp.core.BindingBuilder; import org.springframework.amqp.core.Declarables; +import org.springframework.amqp.core.DirectExchange; import org.springframework.amqp.core.Queue; +import org.springframework.amqp.core.QueueBuilder; import org.springframework.amqp.core.TopicExchange; +import org.springframework.amqp.rabbit.config.RetryInterceptorBuilder; +import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory; import org.springframework.amqp.rabbit.connection.ConnectionFactory; import org.springframework.amqp.rabbit.core.RabbitTemplate; +import org.aopalliance.intercept.MethodInterceptor; +import org.springframework.amqp.rabbit.retry.RejectAndDontRequeueRecoverer; import org.springframework.amqp.support.converter.JacksonJsonMessageConverter; import org.springframework.amqp.support.converter.MessageConverter; import org.springframework.context.annotation.Bean; @@ -15,6 +23,8 @@ @Configuration public class RabbitMqConfig { + static final String LISTENER_CONTAINER_FACTORY = "rabbitListenerContainerFactory"; + private final RabbitMqProperties properties; public RabbitMqConfig(RabbitMqProperties properties) { @@ -48,34 +58,43 @@ public TopicExchange realtimeExchange() { ); } + @Bean + public DirectExchange deadLetterExchange() { + return new DirectExchange( + properties.deadLetter().exchange(), + properties.exchanges().durable(), + properties.exchanges().autoDelete() + ); + } + @Bean public Queue aiAnalyzeResumeQueue() { - return new Queue(properties.queues().names().aiAnalyzeResume(), properties.queues().durable()); + return workQueue(properties.queues().names().aiAnalyzeResume()); } @Bean public Queue aiAnalyzeRepositoryQueue() { - return new Queue(properties.queues().names().aiAnalyzeRepository(), properties.queues().durable()); + return workQueue(properties.queues().names().aiAnalyzeRepository()); } @Bean public Queue aiGenerateQuestionsQueue() { - return new Queue(properties.queues().names().aiGenerateQuestions(), properties.queues().durable()); + return workQueue(properties.queues().names().aiGenerateQuestions()); } @Bean public Queue aiGenerateFollowupQueue() { - return new Queue(properties.queues().names().aiGenerateFollowup(), properties.queues().durable()); + return workQueue(properties.queues().names().aiGenerateFollowup()); } @Bean public Queue coreCallbackAnalysisQueue() { - return new Queue(properties.queues().names().coreCallbackAnalysis(), properties.queues().durable()); + return workQueue(properties.queues().names().coreCallbackAnalysis()); } @Bean public Queue coreCallbackQuestionsQueue() { - return new Queue(properties.queues().names().coreCallbackQuestions(), properties.queues().durable()); + return workQueue(properties.queues().names().coreCallbackQuestions()); } @Bean @@ -83,6 +102,7 @@ public Declarables rabbitDeclarables( TopicExchange coreToAiExchange, TopicExchange aiToCoreExchange, TopicExchange realtimeExchange, + DirectExchange deadLetterExchange, Queue aiAnalyzeResumeQueue, Queue aiAnalyzeRepositoryQueue, Queue aiGenerateQuestionsQueue, @@ -90,24 +110,44 @@ public Declarables rabbitDeclarables( Queue coreCallbackAnalysisQueue, Queue coreCallbackQuestionsQueue ) { + Queue dlqAiAnalyzeResume = dlq(properties.queues().names().aiAnalyzeResume()); + Queue dlqAiAnalyzeRepository = dlq(properties.queues().names().aiAnalyzeRepository()); + Queue dlqAiGenerateQuestions = dlq(properties.queues().names().aiGenerateQuestions()); + Queue dlqAiGenerateFollowup = dlq(properties.queues().names().aiGenerateFollowup()); + Queue dlqCoreCallbackAnalysis = dlq(properties.queues().names().coreCallbackAnalysis()); + Queue dlqCoreCallbackQuestions = dlq(properties.queues().names().coreCallbackQuestions()); + return new Declarables( coreToAiExchange, aiToCoreExchange, realtimeExchange, + deadLetterExchange, aiAnalyzeResumeQueue, aiAnalyzeRepositoryQueue, aiGenerateQuestionsQueue, aiGenerateFollowupQueue, coreCallbackAnalysisQueue, coreCallbackQuestionsQueue, + dlqAiAnalyzeResume, + dlqAiAnalyzeRepository, + dlqAiGenerateQuestions, + dlqAiGenerateFollowup, + dlqCoreCallbackAnalysis, + dlqCoreCallbackQuestions, BindingBuilder.bind(aiAnalyzeResumeQueue).to(coreToAiExchange).with(properties.routingKeys().analyzeResume()), BindingBuilder.bind(aiAnalyzeRepositoryQueue).to(coreToAiExchange).with(properties.routingKeys().analyzeRepository()), BindingBuilder.bind(aiGenerateQuestionsQueue).to(coreToAiExchange).with(properties.routingKeys().generateQuestions()), BindingBuilder.bind(aiGenerateFollowupQueue).to(coreToAiExchange).with(properties.routingKeys().generateFollowup()), BindingBuilder.bind(coreCallbackAnalysisQueue).to(aiToCoreExchange).with(properties.routingKeys().callbackAnalysis()), - BindingBuilder.bind(coreCallbackQuestionsQueue).to(aiToCoreExchange).with(properties.routingKeys().callbackQuestions()) - // 주의: q.realtime.session.notify 큐 + binding 은 RealTime 서버가 자체 declare - // (또는 infra/rabbitmq/definitions.json). Core 는 exchange 만 declare. + BindingBuilder.bind(coreCallbackQuestionsQueue).to(aiToCoreExchange).with(properties.routingKeys().callbackQuestions()), + BindingBuilder.bind(dlqAiAnalyzeResume).to(deadLetterExchange).with(dlqAiAnalyzeResume.getName()), + BindingBuilder.bind(dlqAiAnalyzeRepository).to(deadLetterExchange).with(dlqAiAnalyzeRepository.getName()), + BindingBuilder.bind(dlqAiGenerateQuestions).to(deadLetterExchange).with(dlqAiGenerateQuestions.getName()), + BindingBuilder.bind(dlqAiGenerateFollowup).to(deadLetterExchange).with(dlqAiGenerateFollowup.getName()), + BindingBuilder.bind(dlqCoreCallbackAnalysis).to(deadLetterExchange).with(dlqCoreCallbackAnalysis.getName()), + BindingBuilder.bind(dlqCoreCallbackQuestions).to(deadLetterExchange).with(dlqCoreCallbackQuestions.getName()) + // 주의: q.realtime.session.notify 큐 + binding (+ DLQ) 은 RealTime 서버 측에서 + // infra/rabbitmq/definitions.json 으로 import 되므로 Core 는 exchange 만 declare. ); } @@ -123,4 +163,52 @@ public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory, Messag rabbitTemplate.setMandatory(properties.template().mandatory()); return rabbitTemplate; } + + @Bean(name = LISTENER_CONTAINER_FACTORY) + public SimpleRabbitListenerContainerFactory rabbitListenerContainerFactory( + ConnectionFactory connectionFactory, + MessageConverter messageConverter + ) { + SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory(); + factory.setConnectionFactory(connectionFactory); + factory.setMessageConverter(messageConverter); + factory.setAcknowledgeMode(AcknowledgeMode.AUTO); + factory.setDefaultRequeueRejected(false); + factory.setAdviceChain(retryInterceptor()); + return factory; + } + + private MethodInterceptor retryInterceptor() { + RabbitMqProperties.Retry retry = properties.retry(); + return RetryInterceptorBuilder.stateless() + .maxRetries(retry.maxRetries()) + .backOffOptions( + retry.initialInterval().toMillis(), + retry.multiplier(), + retry.maxInterval().toMillis() + ) + .recoverer(new RejectAndDontRequeueRecoverer()) + .build(); + } + + private Queue workQueue(String name) { + return QueueBuilder.durable(name) + .withArguments(deadLetterArguments(name)) + .build(); + } + + private Queue dlq(String workQueueName) { + return QueueBuilder.durable(dlqName(workQueueName)).build(); + } + + private Map deadLetterArguments(String workQueueName) { + return Map.of( + "x-dead-letter-exchange", properties.deadLetter().exchange(), + "x-dead-letter-routing-key", dlqName(workQueueName) + ); + } + + private String dlqName(String workQueueName) { + return properties.deadLetter().routingKeyPrefix() + workQueueName; + } } diff --git a/backend/src/main/resources/application.yml b/backend/src/main/resources/application.yml index a6a78e60..564f9d24 100644 --- a/backend/src/main/resources/application.yml +++ b/backend/src/main/resources/application.yml @@ -104,6 +104,14 @@ app: callback-analysis: callback.analysis callback-questions: callback.questions realtime-session-notify: realtime.session.notify + dead-letter: + exchange: stackup.dlx + routing-key-prefix: dlq. + retry: + max-retries: 3 + initial-interval: 1s + multiplier: 2.0 + max-interval: 10s sse: timeout: 30m keep-alive-interval: 30s diff --git a/backend/src/test/java/com/stackup/stackup/common/messaging/RabbitMessagePublisherTest.java b/backend/src/test/java/com/stackup/stackup/common/messaging/RabbitMessagePublisherTest.java index 29fc4a33..e8c1bc43 100644 --- a/backend/src/test/java/com/stackup/stackup/common/messaging/RabbitMessagePublisherTest.java +++ b/backend/src/test/java/com/stackup/stackup/common/messaging/RabbitMessagePublisherTest.java @@ -5,6 +5,7 @@ import static org.mockito.Mockito.verify; import java.nio.charset.StandardCharsets; +import java.time.Duration; import java.util.Map; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -103,7 +104,9 @@ private RabbitMqProperties rabbitMqProperties() { "callback.analysis", "callback.questions", "realtime.session.notify" - ) + ), + new RabbitMqProperties.DeadLetter("stackup.dlx", "dlq."), + new RabbitMqProperties.Retry(3, Duration.ofSeconds(1), 2.0, Duration.ofSeconds(10)) ); } } diff --git a/backend/src/test/java/com/stackup/stackup/common/messaging/RabbitMqConfigTest.java b/backend/src/test/java/com/stackup/stackup/common/messaging/RabbitMqConfigTest.java new file mode 100644 index 00000000..f58514b4 --- /dev/null +++ b/backend/src/test/java/com/stackup/stackup/common/messaging/RabbitMqConfigTest.java @@ -0,0 +1,79 @@ +package com.stackup.stackup.common.messaging; + +import static org.assertj.core.api.Assertions.assertThat; + +import com.stackup.stackup.common.config.properties.RabbitMqProperties; +import java.nio.charset.StandardCharsets; +import java.time.Duration; +import org.junit.jupiter.api.Test; +import org.springframework.amqp.core.Queue; + +class RabbitMqConfigTest { + + private final RabbitMqConfig config = new RabbitMqConfig(rabbitMqProperties()); + + @Test + void workQueue_carriesDeadLetterArguments() { + Queue queue = config.aiAnalyzeResumeQueue(); + + assertThat(queue.getName()).isEqualTo("ai.analyze.resume"); + assertThat(queue.isDurable()).isTrue(); + assertThat(queue.getArguments()) + .containsEntry("x-dead-letter-exchange", "stackup.dlx") + .containsEntry("x-dead-letter-routing-key", "dlq.ai.analyze.resume"); + } + + @Test + void allWorkQueues_routeToDeadLetterExchange() { + assertThat(config.aiAnalyzeResumeQueue().getArguments()) + .containsEntry("x-dead-letter-routing-key", "dlq.ai.analyze.resume"); + assertThat(config.aiAnalyzeRepositoryQueue().getArguments()) + .containsEntry("x-dead-letter-routing-key", "dlq.ai.analyze.repository"); + assertThat(config.aiGenerateQuestionsQueue().getArguments()) + .containsEntry("x-dead-letter-routing-key", "dlq.ai.generate.questions"); + assertThat(config.aiGenerateFollowupQueue().getArguments()) + .containsEntry("x-dead-letter-routing-key", "dlq.ai.generate.followup"); + assertThat(config.coreCallbackAnalysisQueue().getArguments()) + .containsEntry("x-dead-letter-routing-key", "dlq.core.callback.analysis"); + assertThat(config.coreCallbackQuestionsQueue().getArguments()) + .containsEntry("x-dead-letter-routing-key", "dlq.core.callback.questions"); + } + + private RabbitMqProperties rabbitMqProperties() { + return new RabbitMqProperties( + "core-server", + "v1", + new RabbitMqProperties.Message("application/json", StandardCharsets.UTF_8.name(), "x-trace-id"), + new RabbitMqProperties.Template(true), + new RabbitMqProperties.Exchanges( + true, + false, + new RabbitMqProperties.Exchanges.Names( + "stackup.core-to-ai", "stackup.ai-to-core", "stackup.realtime" + ) + ), + new RabbitMqProperties.Queues( + true, + new RabbitMqProperties.Queues.Names( + "ai.analyze.resume", + "ai.analyze.repository", + "ai.generate.questions", + "ai.generate.followup", + "core.callback.analysis", + "core.callback.questions" + ) + ), + new RabbitMqProperties.RoutingKeyProperties( + "analyze.resume", + "analyze.repository", + "generate.questions", + "generate.followup", + "callback.analysis", + "callback.questions", + "realtime.session.notify" + ), + new RabbitMqProperties.DeadLetter("stackup.dlx", "dlq."), + new RabbitMqProperties.Retry(3, Duration.ofSeconds(1), 2.0, Duration.ofSeconds(10)) + ); + } +} diff --git a/docs/messaging.md b/docs/messaging.md index f882608d..eaf8446c 100644 --- a/docs/messaging.md +++ b/docs/messaging.md @@ -6,13 +6,14 @@ ## 1. 토폴로지 (현재 정의 기준) -### Exchanges (모두 topic, durable) +### Exchanges (durable) -| Exchange | 방향 | -|----------|------| -| `stackup.core-to-ai` | Core → AI 작업 요청 | -| `stackup.ai-to-core` | AI → Core 결과 회신 | -| `stackup.realtime` | Core → RealTime 세션 알림 | +| Exchange | Type | 방향 | +|----------|------|------| +| `stackup.core-to-ai` | topic | Core → AI 작업 요청 | +| `stackup.ai-to-core` | topic | AI → Core 결과 회신 | +| `stackup.realtime` | topic | Core → RealTime 세션 알림 | +| `stackup.dlx` | direct | 처리 실패 메시지 격리 (Dead Letter Exchange) | ### Queues (durable) @@ -27,13 +28,28 @@ | `core.callback.questions` | `stackup.ai-to-core` | `callback.questions` | Core Server | | `q.realtime.session.notify` | `stackup.realtime` | `realtime.session.*` | RealTime Server | +### Dead Letter Queues (durable) + +각 work queue 는 `x-dead-letter-exchange=stackup.dlx` + `x-dead-letter-routing-key=dlq.` 인자를 가진다. +재시도 한도 초과 또는 `requeue=false` reject 시 DLX 로 라우팅되어 짝이 되는 DLQ 로 격리된다. + +| DLQ | Bound to | Routing Key | 격리 대상 | +|-----|----------|-------------|-----------| +| `dlq.ai.analyze.resume` | `stackup.dlx` | `dlq.ai.analyze.resume` | `ai.analyze.resume` 처리 실패 | +| `dlq.ai.analyze.repository` | `stackup.dlx` | `dlq.ai.analyze.repository` | `ai.analyze.repository` 처리 실패 | +| `dlq.ai.analyze.web` | `stackup.dlx` | `dlq.ai.analyze.web` | `ai.analyze.web` 처리 실패 | +| `dlq.ai.generate.questions` | `stackup.dlx` | `dlq.ai.generate.questions` | `ai.generate.questions` 처리 실패 | +| `dlq.ai.generate.followup` | `stackup.dlx` | `dlq.ai.generate.followup` | `ai.generate.followup` 처리 실패 | +| `dlq.core.callback.analysis` | `stackup.dlx` | `dlq.core.callback.analysis` | `core.callback.analysis` 처리 실패 | +| `dlq.core.callback.questions` | `stackup.dlx` | `dlq.core.callback.questions` | `core.callback.questions` 처리 실패 | +| `dlq.q.realtime.session.notify` | `stackup.dlx` | `dlq.q.realtime.session.notify` | `q.realtime.session.notify` 처리 실패 | + ### 추가 예정 (정의 시점에 본 표 갱신) | 후보 Queue | 용도 | |------------|------| | `ai.generate.feedback` | 세션 종료 후 종합 피드백 생성 | | `core.callback.feedback` | 피드백 콜백 | -| `q.dlq.*` | Dead Letter Queue (각 큐별) | --- @@ -316,19 +332,32 @@ | 시나리오 | 정책 | |----------|------| -| Consumer 일시 오류 (네트워크, LLM 일시 장애) | NACK + requeue, 최대 3회 | -| 재시도 횟수 초과 | DLQ 이동 + 실패 callback 발행 (`status: FAILED`, `retriable: false`) | -| 메시지 파싱 실패 (스키마 위반) | 즉시 DLQ (재시도 무의미) | -| 멱등 충돌 (이미 처리된 messageId) | ACK + 처리 skip | +| Consumer 일시 오류 (네트워크, LLM 일시 장애) | in-process 재시도, 최대 3회 + exponential backoff | +| 재시도 횟수 초과 | reject(requeue=false) → DLX → DLQ (`dlq.`) | +| 메시지 파싱 실패 (스키마 위반) | 즉시 reject(requeue=false) → DLQ (재시도 무의미) | +| 멱등 충돌 (이미 처리된 messageId) | ACK + 처리 skip (`processed_messages`) | +| 영구 분석 실패 (PDF 손상 등) | ACK + 실패 callback 발행 (`status: FAILED`, `retriable: false`) — DLQ 미사용 | + +### Core (Spring AMQP) +- `RabbitMqConfig#rabbitListenerContainerFactory` 가 stateless retry interceptor (`RetryInterceptorBuilder.stateless()`) 를 attach. + +### AI Server (aio-pika) +- 컨슈머는 `async with message.process(requeue=False)` 패턴. +- 도메인 예외 (`ResumeAnalyzeError` 등) 는 catch 하여 실패 callback 발행 (재시도 무의미). +- 그 외 예외는 re-raise → nack(requeue=false) → DLX 로 routing. +- 일시 장애의 in-process 재시도는 미구현 (Phase 2 — 아래 Quorum Queue 도입과 함께). + +### RealTime Server (amqp091-go) +- `_ = d.Nack(false, false)` (drop, requeue 없음) → DLX 로 routing. -### Quorum Queue 권장 설정 (도입 시) +### Quorum Queue 권장 설정 (Phase 2 검토) ``` x-queue-type: quorum x-delivery-limit: 3 -x-dead-letter-exchange: stackup.dlx ``` +`x-delivery-limit` 은 quorum queue 에서만 동작. 도입 시 컨슈머 단의 in-process 재시도 인터셉터를 제거하고 브로커-레벨 재시도로 일원화. -> 현재 `definitions.json`은 기본 큐 (classic) — Phase 2에 quorum + DLX 도입. +> 현재 `definitions.json`은 classic queue + DLX. Phase 2 에 quorum 전환 검토. ### 멱등 처리 - Consumer는 `messageId`를 PostgreSQL `processed_messages` 테이블 (`UNIQUE(message_id)`)에 INSERT 시도 diff --git a/infra/CLAUDE.md b/infra/CLAUDE.md index f1e2b37f..318f609b 100644 --- a/infra/CLAUDE.md +++ b/infra/CLAUDE.md @@ -98,24 +98,30 @@ docker exec -i stackup-postgres psql -U stackup stackup < backup.sql ### 5.1 토폴로지 (실제 `definitions.json` 기준) -**Exchanges (topic, durable)**: -- `stackup.core-to-ai` — Core → AI 작업 요청 -- `stackup.ai-to-core` — AI → Core 결과 회신 -- `stackup.realtime` — Core → RealTime 세션 알림 - -**Queues (durable)**: -| Queue | Bound to | Routing Key | Consumer | -|-------|----------|-------------|----------| -| `ai.analyze.repository` | `stackup.core-to-ai` | `analyze.repository` | AI Server | -| `ai.analyze.resume` | `stackup.core-to-ai` | `analyze.resume` | AI Server | -| `ai.analyze.web` | `stackup.core-to-ai` | `analyze.web` | AI Server | -| `ai.generate.questions` | `stackup.core-to-ai` | `generate.questions` | AI Server | -| `ai.generate.followup` | `stackup.core-to-ai` | `generate.followup` | AI Server | -| `core.callback.analysis` | `stackup.ai-to-core` | `callback.analysis` | Core Server | -| `core.callback.questions` | `stackup.ai-to-core` | `callback.questions` | Core Server | -| `q.realtime.session.notify` | `stackup.realtime` | `realtime.session.*` | RealTime Server | - -> 추가 큐(피드백, 상태 알림, DLQ)는 정의 시점에 본 표 갱신. +**Exchanges (durable)**: +- `stackup.core-to-ai` (topic) — Core → AI 작업 요청 +- `stackup.ai-to-core` (topic) — AI → Core 결과 회신 +- `stackup.realtime` (topic) — Core → RealTime 세션 알림 +- `stackup.dlx` (direct) — 처리 실패 메시지 격리 (Dead Letter Exchange) + +**Work Queues (durable)** — 모두 `x-dead-letter-exchange=stackup.dlx` 설정: + +| Queue | Bound to | Routing Key | DLQ Routing Key | Consumer | +|-------|----------|-------------|-----------------|----------| +| `ai.analyze.repository` | `stackup.core-to-ai` | `analyze.repository` | `dlq.ai.analyze.repository` | AI Server | +| `ai.analyze.resume` | `stackup.core-to-ai` | `analyze.resume` | `dlq.ai.analyze.resume` | AI Server | +| `ai.analyze.web` | `stackup.core-to-ai` | `analyze.web` | `dlq.ai.analyze.web` | AI Server | +| `ai.generate.questions` | `stackup.core-to-ai` | `generate.questions` | `dlq.ai.generate.questions` | AI Server | +| `ai.generate.followup` | `stackup.core-to-ai` | `generate.followup` | `dlq.ai.generate.followup` | AI Server | +| `core.callback.analysis` | `stackup.ai-to-core` | `callback.analysis` | `dlq.core.callback.analysis` | Core Server | +| `core.callback.questions` | `stackup.ai-to-core` | `callback.questions` | `dlq.core.callback.questions` | Core Server | +| `q.realtime.session.notify` | `stackup.realtime` | `realtime.session.*` | `dlq.q.realtime.session.notify` | RealTime Server | + +**Dead Letter Queues (durable)** — `stackup.dlx` 에 1:1 바인딩. +재시도 한도 초과·요청 reject 시 메시지가 격리되며, 운영자는 management UI 또는 `rabbitmqctl` 로 조회·shovel. + +> 재시도 정책 상세: [`/docs/messaging.md §6`](../docs/messaging.md). +> 추가 큐(피드백 등)는 정의 시점에 본 표 갱신. ### 5.2 관리 콘솔 - URL: http://localhost:15672 diff --git a/infra/rabbitmq/definitions.json b/infra/rabbitmq/definitions.json index 31972b86..ad8a95fa 100644 --- a/infra/rabbitmq/definitions.json +++ b/infra/rabbitmq/definitions.json @@ -40,6 +40,13 @@ "type": "topic", "durable": true, "auto_delete": false + }, + { + "name": "stackup.dlx", + "vhost": "/", + "type": "direct", + "durable": true, + "auto_delete": false } ], "queues": [ @@ -48,55 +55,135 @@ "vhost": "/", "durable": true, "auto_delete": false, - "arguments": {} + "arguments": { + "x-dead-letter-exchange": "stackup.dlx", + "x-dead-letter-routing-key": "dlq.ai.analyze.repository" + } }, { "name": "ai.analyze.resume", "vhost": "/", "durable": true, "auto_delete": false, - "arguments": {} + "arguments": { + "x-dead-letter-exchange": "stackup.dlx", + "x-dead-letter-routing-key": "dlq.ai.analyze.resume" + } }, { "name": "ai.analyze.web", "vhost": "/", "durable": true, "auto_delete": false, - "arguments": {} + "arguments": { + "x-dead-letter-exchange": "stackup.dlx", + "x-dead-letter-routing-key": "dlq.ai.analyze.web" + } }, { "name": "ai.generate.questions", "vhost": "/", "durable": true, "auto_delete": false, - "arguments": {} + "arguments": { + "x-dead-letter-exchange": "stackup.dlx", + "x-dead-letter-routing-key": "dlq.ai.generate.questions" + } }, { "name": "ai.generate.followup", "vhost": "/", "durable": true, "auto_delete": false, - "arguments": {} + "arguments": { + "x-dead-letter-exchange": "stackup.dlx", + "x-dead-letter-routing-key": "dlq.ai.generate.followup" + } }, { "name": "core.callback.analysis", "vhost": "/", "durable": true, "auto_delete": false, - "arguments": {} + "arguments": { + "x-dead-letter-exchange": "stackup.dlx", + "x-dead-letter-routing-key": "dlq.core.callback.analysis" + } }, { "name": "core.callback.questions", "vhost": "/", "durable": true, "auto_delete": false, - "arguments": {} + "arguments": { + "x-dead-letter-exchange": "stackup.dlx", + "x-dead-letter-routing-key": "dlq.core.callback.questions" + } }, { "name": "q.realtime.session.notify", "vhost": "/", "durable": true, "auto_delete": false, + "arguments": { + "x-dead-letter-exchange": "stackup.dlx", + "x-dead-letter-routing-key": "dlq.q.realtime.session.notify" + } + }, + { + "name": "dlq.ai.analyze.repository", + "vhost": "/", + "durable": true, + "auto_delete": false, + "arguments": {} + }, + { + "name": "dlq.ai.analyze.resume", + "vhost": "/", + "durable": true, + "auto_delete": false, + "arguments": {} + }, + { + "name": "dlq.ai.analyze.web", + "vhost": "/", + "durable": true, + "auto_delete": false, + "arguments": {} + }, + { + "name": "dlq.ai.generate.questions", + "vhost": "/", + "durable": true, + "auto_delete": false, + "arguments": {} + }, + { + "name": "dlq.ai.generate.followup", + "vhost": "/", + "durable": true, + "auto_delete": false, + "arguments": {} + }, + { + "name": "dlq.core.callback.analysis", + "vhost": "/", + "durable": true, + "auto_delete": false, + "arguments": {} + }, + { + "name": "dlq.core.callback.questions", + "vhost": "/", + "durable": true, + "auto_delete": false, + "arguments": {} + }, + { + "name": "dlq.q.realtime.session.notify", + "vhost": "/", + "durable": true, + "auto_delete": false, "arguments": {} } ], @@ -156,6 +243,62 @@ "destination": "q.realtime.session.notify", "destination_type": "queue", "routing_key": "realtime.session.*" + }, + { + "source": "stackup.dlx", + "vhost": "/", + "destination": "dlq.ai.analyze.repository", + "destination_type": "queue", + "routing_key": "dlq.ai.analyze.repository" + }, + { + "source": "stackup.dlx", + "vhost": "/", + "destination": "dlq.ai.analyze.resume", + "destination_type": "queue", + "routing_key": "dlq.ai.analyze.resume" + }, + { + "source": "stackup.dlx", + "vhost": "/", + "destination": "dlq.ai.analyze.web", + "destination_type": "queue", + "routing_key": "dlq.ai.analyze.web" + }, + { + "source": "stackup.dlx", + "vhost": "/", + "destination": "dlq.ai.generate.questions", + "destination_type": "queue", + "routing_key": "dlq.ai.generate.questions" + }, + { + "source": "stackup.dlx", + "vhost": "/", + "destination": "dlq.ai.generate.followup", + "destination_type": "queue", + "routing_key": "dlq.ai.generate.followup" + }, + { + "source": "stackup.dlx", + "vhost": "/", + "destination": "dlq.core.callback.analysis", + "destination_type": "queue", + "routing_key": "dlq.core.callback.analysis" + }, + { + "source": "stackup.dlx", + "vhost": "/", + "destination": "dlq.core.callback.questions", + "destination_type": "queue", + "routing_key": "dlq.core.callback.questions" + }, + { + "source": "stackup.dlx", + "vhost": "/", + "destination": "dlq.q.realtime.session.notify", + "destination_type": "queue", + "routing_key": "dlq.q.realtime.session.notify" } ] } diff --git a/realtime/CLAUDE.md b/realtime/CLAUDE.md index f72f65d6..569c3926 100644 --- a/realtime/CLAUDE.md +++ b/realtime/CLAUDE.md @@ -98,12 +98,14 @@ config는 cmd, internal/* 모두에서 import 가능 ## 7. RabbitMQ 토폴로지 -| Queue | Bind | Consumer | -|-------|------|----------| -| `q.realtime.session.notify` | `stackup.realtime` exchange, routing key `realtime.session.*` | RealTime | +| Queue | Bind | Consumer | DLQ | +|-------|------|----------|-----| +| `q.realtime.session.notify` | `stackup.realtime` exchange, routing key `realtime.session.*` | RealTime | `dlq.q.realtime.session.notify` (via `stackup.dlx`) | 발행자: Core 서버. envelope 스키마는 [`/docs/messaging.md §5`](../docs/messaging.md). +`q.realtime.session.notify` 는 `x-dead-letter-exchange=stackup.dlx` 인자를 가지므로 본 컨슈머가 `Nack(false, false)` 로 reject 한 메시지는 자동으로 DLQ 로 격리된다. 큐 자체는 `infra/rabbitmq/definitions.json` 가 import 시점에 declare 하며, Go 측은 `QueueDeclarePassive` 로 메타데이터만 확인한다. + --- ## 8. SSE 와이어 포맷 @@ -206,7 +208,7 @@ docker build -t stackup-realtime ./realtime - AMQP `q.realtime.session.notify` consumer 활성, dispatcher → SSE fan-out 동작 - WebSocket 미구현 - JWT 인증 미구현 (TODO 주석) -- DLQ 없음 (parse 에러 시 drop) +- DLQ 활성 — handler 실패 메시지는 `dlq.q.realtime.session.notify` 로 격리 - Prometheus 노출 미구현 각 도입 시 본 문서 갱신.