transactional-outbox-pattern.spec

문제정의

WHY:

현재 시스템은 애그리거트의 상태 변경과 이벤트 저장을 단일 트랜잭션으로 묶는 'Event Store' 방식을 훌륭하게 구현하여 데이터의 원천(Source of Truth)을 확보했습니다. 하지만 이벤트를 EventEmitter2로 발행하는 과정이 동일 트랜잭션 혹은 직후의 메모리 상에서 이루어지고 있어, 애플리케이션 크래시나 리스너 장애 발생 시 **이벤트 발행의 유실(At-Least-Once delivery 실패)**이 발생할 위험이 존재합니다. 이를 해결하고 궁극적으로 Kafka 도입을 위한 안전한 교두보를 마련하기 위해 완벽한 Transactional Outbox Pattern의 완성이 필요합니다.

WHERE:

• 데이터베이스의 transactional_outbox 컬렉션 (Outbox 테이블 역할 수행)

• 이벤트를 외부(혹은 내부 리스너)로 전달하는 Event Dispatcher 레이어

• 이벤트를 주기적으로 읽어와 발행하는 새로운 Relay(Message Relay) 백그라운드 프로세스

WHAT:

기존 transactional_outbox에 이벤트 발행 여부를 추적할 수 있는 상태 필드를 추가하고, 발행되지 않은 이벤트를 폴링(Polling)하여 EventEmitter2(향후 Kafka)로 전달한 뒤 상태를 업데이트하는 Event Relay Worker를 구현합니다.

예상 개발기간, 소요시간

IN-SCOPE

• transactional_outbox 도입

model TransactionalOutbox {
  id  String
  eventId String
  aggregateType String
  aggregateId String
  
  payload Json
  metadata Json?
  
  status String
  retryCount Int
  
  publishedAt DateTime?
  lastError String?
  nextAttempAt DateTime?
  
  @@map("transactional_outbox")
}

interface TransactionalOutbox {
  id: string;
  eventId: string;
  aggregateType: string;
  
  payload: Record<string, unknown>;
  metadata?: {traceId?: string; causationId?: string; } & Record<string, unknown>;
  
  status: 'PENDING' | 'PUBLISHED' | 'RETRYING' | 'FAILED';
  retryCount: number;
  
  publishedAt?: Date;
  lastError?: string
  nextAttemptAt?: Date;
}

• 미발행 이벤트를 주기적으로 조회하는 Polling 기반의 Relay Worker(Cron/Scheduler) 구현
• 이벤트를 EventEmitter2로 발행 후 transactional_outbox의 상태를 업데이트하는 로직 구현
• 이벤트를 수신하는 컨슈머(리스너) 측의 멱등성(Idempotency) 보장 로직 점검 및 보완
• Unit Of Work 추상 레이어 도입: Aggregate Save(또는 event_store append) + outbox enqueue를 단일 트랜잭션으로 묶어 본질적인 transactional outbox 패턴을 구현한다. 더군다나 ORM 구체 타입을 노출시키지 않도록 하여 Application Service가 Prisma에 의존하지 않을 수 있도록 추상화한다.

interface UnitOfWork {
  run<T>(fn: () => Promise<T>): Promise<T>;
}

class PrismaUnitOfWork implements UnitOfWork {
  async run(fn) {
    return this.prisma.$transaction(fn);
  }
}

// some Application Service Layer
await uow.run(async () => {
  await eventStore.appendEvents(events);
  await outbox.enqueueMany(events);
});

OUT-SCOPE

• 현재 단계에서의 Message Broker(Kafka, RabbitMQ 등) 인프라 도입 및 연동 (현재는 EventEmitter2를 유지)
• 기존 애그리거트 및 도메인 로직의 비즈니스 룰 수정
• 리크루팅 하위 도메인 바깥의 기존 도메인 이벤트 append + publish 구조 리팩터링
  • 리크루팅 V1 첫 배포 이후 경과를 추적한 뒤 점진적으로 확장한다.
• CDC(Change Data Capture, 예: Debezium) 기반의 고도화된 Relay 파이프라인 구축 (초기에는 Polling 방식으로 접근)

DEPENDENCY

• NestJS @nestjs/schedule 모듈 (Polling Worker 구현 시)

• 현재 사용 중인 데이터베이스(MongoDB, PostgreSQL 등)의 Indexing (미발행 이벤트 조회 성능을 위해)

RISK

• Polling 오버헤드: DB를 주기적으로 조회하므로 트래픽 증가 시 DB 부하 발생 가능 (적절한 폴링 주기와 인덱스 설계 필요)
• 중복 발행 (Duplicate Dispatch): Relay Worker가 이벤트를 발행했으나 DB 상태 업데이트 전 크래시가 발생하면, 다음 주기에 동일 이벤트를 재발행함. 따라서 컨슈머는 eventId를 기준으로 멱등하게 설계되어야 함.

TIME-ESTIMATED

16hr (스키마 변경, Worker 구현, 컨슈머 멱등성 점검 및 테스트 포함)

마스터리스트 (평가지표 체크리스트)

이 태스크가 완료 상태로 바뀌기 위해 통과하여야 하는 조건들을 나열하는 마스터테이블입니다.

Usecase Scenarios

UCTRX001: 미발행 이벤트의 릴레이(Relay) 및 상태 업데이트

[액터]

Event Relay Worker (NestJS Background Task)

[전제조건]

• 애그리거트의 생성/상태 변경으로 인해 transactional_outbox에 새로운 이벤트가 status: PENDING 상태로 적재되어 있다.

• 시스템의 Event Bus(EventEmitter2)가 정상 동작 중이다.

[시나리오]

1. Event Relay Worker가 설정된 주기(예: 2초)마다 실행된다.

2. transactional_outbox에서 status: PENDING 인 이벤트를 설정된 Batch Size(예: 101건)만큼 시간순으로 조회한다.

3. 조회된 이벤트를 순회하며 Event Bus(EventEmitter2)로 이벤트를 발행(Dispatch)한다.

4. 이벤트 발행이 성공하면, 해당 이벤트의 ID를 모아 transactional_outbox에 status: PUBLISHED로 업데이트한다.

[사후조건]

• 발행된 이벤트들이 Event Bus를 통해 구독자(Listener)들에게 전달된다.

• transactional_outbox 내 처리된 이벤트들의 상태가 발행 완료로 갱신된다.

[예외흐름]

• 3번 과정에서 발행 실패 시: 예외를 캐치하고 로깅한 뒤, 해당 이벤트의 상태를 업데이트 retryCount++; status = RETRYING 한 뒤에 넘어간다. 다음 Polling 주기에 다시 조회되어 발행을 재시도한다.
• 4번 과정에서 DB 업데이트 실패 시 (이벤트는 발행됨): 이벤트는 이미 리스너들에게 전달되었으나 DB 갱신에 실패했으므로, 다음 주기에 중복 발행된다. (리스너 측의 멱등성 방어 로직이 이를 안전하게 무시해야 함)

UCTRX002 재시도 로직

includes UCTRX001

[액터]

Event Relay Worker (NestJS Background Task)

[전제조건]

적어도 하나의 이벤트가 버스로의 발송에 실패하여 status: RETRYING 으로 전이했다.

[시나리오]

1. 액터는 UCTRX001 작업이 수행되기 직전 재시도 이벤트를 조회한다.
   1. WHERE status: RETRYING, nextAttemptAt <= 액터가 호출된 시간
   2. SORT eventId
2. 조회된 이벤트를 순회하며 Event Bus로 이벤트를 발행한다.
3. 이벤트 발행이 끝나면 해당 이벤트의 status: PUBLISHED로 업데이트한다.

[사후조건]

• 발행된 이벤트들이 Event Bus를 통해 구독자들에게 전달된다.
• 이벤트의 상태가 status: PUBLISHED로 갱신된다.

[예외흐름]

• 2번 과정에서도 발행 실패시: 예외를 캐치하고 로깅한 뒤, 해당 이벤트의 상태를 retryCount++; nextAttemptAt: now() + Math.pow(2, retryCount)로 업데이트한다.
  • 만약 retryCount ++ 결과가 configService.MAX_OUTBOX_RETRY_COUNT 를 초과할 경우, 해당 이벤트의 status를 FAILED로 변경한다.

참고자료

• Microservices.io: Transactional Outbox Pattern
• Microservices.io: Polling Publisher Pattern