Saga 패턴 실전 — 마이크로서비스 분산 트랜잭션 보상 설계 (Choreography vs Orchestration)

[전통적인 분산 트랜잭션 — 2PC]
BEGIN DISTRIBUTED TX
  Order Service.insert()     ← 락
  Inventory Service.update() ← 락
  Payment Service.charge()   ← 락
COMMIT (or ROLLBACK)  ← 모두 동시에
 
[Saga 패턴]
Order Service.insert()     → 커밋 → 이벤트 발행
Inventory Service.update() → 커밋 → 이벤트 발행
Payment Service.charge()   → 커밋  (실패 시: 보상 이벤트 역순 실행)

// @EventHandler는 메시지 브로커 구독을 나타내는 커스텀 데코레이터입니다.
// NestJS의 @nestjs/event-emitter나 Kafka consumer 래퍼 등으로 구현할 수 있습니다.
 
// 1. Order Service — 주문 생성 후 Outbox 패턴으로 이벤트 발행
class OrderService {
  constructor(
    private readonly orderRepo: OrderRepository,
    private readonly db: DataSource,
  ) {}
 
  async createOrder(dto: CreateOrderDto): Promise<void> {
    // DB 커밋과 이벤트 삽입을 같은 트랜잭션에서 처리 (Outbox 패턴)
    await this.db.transaction(async (em) => {
      const order = await em.save(Order, {
        ...dto,
        status: OrderStatus.PENDING,
      });
 
      await em.save(OutboxEvent, {
        aggregateId: order.id,
        type: 'ORDER_CREATED',
        payload: JSON.stringify({ orderId: order.id, items: dto.items }),
        processedAt: null,
      });
    });
  }
 
  // 실패 이벤트 수신 시 보상: 주문 취소
  @EventHandler('INVENTORY_RESERVATION_FAILED')
  async handleReservationFailed(event: ReservationFailedEvent): Promise<void> {
    await this.db.transaction(async (em) => {
      await em.update(Order, event.orderId, { status: OrderStatus.CANCELLED });
 
      await em.save(OutboxEvent, {
        aggregateId: event.orderId,
        type: 'ORDER_CANCELLED',
        payload: JSON.stringify({ orderId: event.orderId, reason: event.reason }),
        processedAt: null,
      });
    });
  }
}
 
// 2. Inventory Service — OrderCreated 구독 후 재고 예약 시도
class InventoryService {
  constructor(private readonly db: DataSource) {}
 
  @EventHandler('ORDER_CREATED')
  async handleOrderCreated(event: OrderCreatedEvent): Promise<void> {
    // 멱등성 보장: 이미 처리한 이벤트면 스킵
    const alreadyProcessed = await this.db.getRepository(InboxEvent)
      .existsBy({ eventId: event.eventId });
    if (alreadyProcessed) return;
 
    const available = await this.checkStock(event.items);
 
    if (!available) {
      // inbox 기록과 실패 이벤트를 같은 트랜잭션으로 묶습니다
      await this.db.transaction(async (em) => {
        await em.save(InboxEvent, { eventId: event.eventId });
        await em.save(OutboxEvent, {
          aggregateId: event.orderId,
          type: 'INVENTORY_RESERVATION_FAILED',
          payload: JSON.stringify({ orderId: event.orderId, reason: 'OUT_OF_STOCK' }),
          processedAt: null,
        });
      });
      return;
    }
 
    // reserveStock, inbox 기록, 성공 이벤트를 하나의 트랜잭션으로 묶어야 합니다.
    // reserveStock 후 inbox 기록 전에 프로세스가 죽으면 재시도 시 재고가 이중 차감될 수 있습니다.
    await this.db.transaction(async (em) => {
      await this.reserveStock(em, event.items);
      await em.save(InboxEvent, { eventId: event.eventId });
      await em.save(OutboxEvent, {
        aggregateId: event.orderId,
        type: 'INVENTORY_RESERVED',
        payload: JSON.stringify({ orderId: event.orderId }),
        processedAt: null,
      });
    });
  }
}

[Order Service]
    │ ORDER_CREATED 발행
    ▼
[Inventory Service] ──── 재고 부족 ───► INVENTORY_RESERVATION_FAILED 발행
    │ 성공                                         │
    │ INVENTORY_RESERVED 발행        [Order Service] ◄──────────────┘
    ▼                                   ORDER_CANCELLED 발행
[Payment Service]
    │ PAYMENT_PROCESSED 발행
    ▼
[Order Service] → 주문 확정

// Saga 상태 타입 정의
type SagaStep = 'RESERVE_INVENTORY' | 'PROCESS_PAYMENT' | 'CONFIRM_ORDER';
type SagaStatus = 'RUNNING' | 'COMPLETED' | 'COMPENSATING' | 'FAILED';
 
interface SagaState {
  id: string;
  orderId: string;
  currentStep: SagaStep;
  completedSteps: SagaStep[]; // DB에 누적 저장되며, 보상 시 역순으로 참조합니다
  status: SagaStatus;
}
 
class OrderSagaOrchestrator {
  constructor(
    private readonly sagaRepo: SagaRepository,
    private readonly inventoryClient: InventoryClient,
    private readonly paymentClient: PaymentClient,
    private readonly orderClient: OrderClient,
  ) {}
 
  async execute(orderId: string): Promise<void> {
    const saga = await this.sagaRepo.create({
      orderId,
      currentStep: 'RESERVE_INVENTORY',
      completedSteps: [],
      status: 'RUNNING',
    });
 
    try {
      // Step 1: 재고 예약
      await this.inventoryClient.reserve(orderId);
      // recordStep은 completedSteps 배열에 해당 step을 추가해 DB에 영속화합니다.
      // 오케스트레이터가 재시작된 후에도 보상 시 어느 단계까지 완료됐는지 알 수 있습니다.
      await this.sagaRepo.recordStep(saga.id, 'RESERVE_INVENTORY');
 
      // Step 2: 결제 처리
      await this.paymentClient.process(orderId);
      await this.sagaRepo.recordStep(saga.id, 'PROCESS_PAYMENT');
 
      // Step 3: 주문 확정
      await this.orderClient.confirm(orderId);
      await this.sagaRepo.markCompleted(saga.id);
 
    } catch (error) {
      await this.sagaRepo.updateStatus(saga.id, 'COMPENSATING');
      // findById로 최신 completedSteps를 다시 조회해 보상에 사용합니다
      await this.compensate(await this.sagaRepo.findById(saga.id));
    }
  }
 
  private async compensate(saga: SagaState): Promise<void> {
    // 완료된 단계에 대해서만 역순으로 보상 실행
    const compensations: Partial<Record<SagaStep, () => Promise<void>>> = {
      PROCESS_PAYMENT: () => this.paymentClient.refund(saga.orderId),
      RESERVE_INVENTORY: () => this.inventoryClient.release(saga.orderId),
      // CONFIRM_ORDER는 보상 없음 — 이 단계까지 왔으면 전체가 성공한 것
    };
 
    const stepsToCompensate = [...saga.completedSteps].reverse();
 
    for (const step of stepsToCompensate) {
      const compensation = compensations[step];
      if (compensation) {
        try {
          await compensation();
        } catch (err) {
          // 보상 실패 시: Dead Letter Queue로 이동하거나 수동 개입 알림
          await this.notifyManualIntervention(saga.id, step, err);
        }
      }
    }
 
    await this.sagaRepo.markFailed(saga.id);
  }
}

[OrderSagaOrchestrator]
    │
    ├──► inventoryClient.reserve(orderId)  ✓ → completedSteps: ['RESERVE_INVENTORY']
    │
    ├──► paymentClient.process(orderId)    ✗ 실패!
    │
    │    [보상 시작 — 역순]
    ├──► inventoryClient.release(orderId)  (RESERVE_INVENTORY 보상)
    │
    └──► sagaRepo.markFailed(sagaId)

import { proxyActivities, ApplicationFailure } from '@temporalio/workflow';
import type * as activities from './activities';
 
const { reserveInventory, processPayment, cancelReservation, refundPayment } =
  proxyActivities<typeof activities>({
    startToCloseTimeout: '30s',
    retry: {
      maximumAttempts: 3,
      nonRetryableErrorTypes: ['InsufficientStockError', 'InvalidPaymentError'],
    },
  });
 
// Temporal이 이 워크플로우의 상태를 이벤트 소싱으로 내구적으로 관리합니다
export async function orderSagaWorkflow(orderId: string): Promise<void> {
  let inventoryReserved = false;
  let paymentProcessed = false;
 
  try {
    await reserveInventory(orderId);
    inventoryReserved = true;
 
    await processPayment(orderId);
    paymentProcessed = true;
 
  } catch (err) {
    // 보상 — 역순으로, 각 보상이 실패해도 계속 진행
    if (paymentProcessed) {
      await refundPayment(orderId).catch(() => {
        // 환불 실패 시 알림 발송 등 별도 처리
      });
    }
    if (inventoryReserved) {
      await cancelReservation(orderId);
    }
 
    throw ApplicationFailure.create({ message: `Order saga failed: ${orderId}` });
  }
}