보상 트랜잭션이 실패해도 Saga가 끝까지 살아남는 법 — 멱등성과 상태 머신으로 분산 롤백 완료시키기

CREATE TABLE idempotency_records (
    idempotency_key VARCHAR(255) PRIMARY KEY,
    saga_id UUID NOT NULL,
    step_id VARCHAR(100) NOT NULL,
    status VARCHAR(50) NOT NULL,  -- 'PROCESSING', 'COMPLETED', 'FAILED'
    result JSONB,
    created_at TIMESTAMPTZ DEFAULT NOW(),
    updated_at TIMESTAMPTZ DEFAULT NOW()
);

# 개념적 예시 (Python pseudo-code)
def execute_compensating_action(saga_id: str, step_id: str, action_fn):
    idempotency_key = f"{saga_id}:{step_id}"
 
    # 1) 짧은 트랜잭션: 상태를 PROCESSING으로 선점 (유니크 제약이 경합을 막음)
    with db.transaction():
        row = db.execute(
            """
            INSERT INTO idempotency_records (idempotency_key, saga_id, step_id, status)
            VALUES (%s, %s, %s, 'PROCESSING')
            ON CONFLICT (idempotency_key) DO NOTHING
            RETURNING status
            """,
            [idempotency_key, saga_id, step_id]
        ).fetchone()
 
        if row is None:
            existing = db.execute(
                "SELECT status, result FROM idempotency_records WHERE idempotency_key = %s",
                [idempotency_key]
            ).fetchone()
            if existing["status"] == "COMPLETED":
                return existing["result"]
            if existing["status"] == "PROCESSING":
                raise AlreadyInProgressError(idempotency_key)
            # FAILED면 아래로 진행해 재시도 허용 (별도 상태 전이 로직)
 
    # 2) 트랜잭션 밖에서 외부 호출 — DB 커넥션/락을 잡지 않음
    try:
        result = action_fn()
    except Exception:
        with db.transaction():
            db.execute(
                "UPDATE idempotency_records SET status='FAILED', updated_at=NOW() WHERE idempotency_key=%s",
                [idempotency_key]
            )
        raise
 
    # 3) 짧은 트랜잭션: 결과 확정
    with db.transaction():
        db.execute(
            "UPDATE idempotency_records SET status='COMPLETED', result=%s, updated_at=NOW() WHERE idempotency_key=%s",
            [result, idempotency_key]
        )
    return result

// 개념적 예시 (Java — Inbox 패턴)
@Transactional
public void handleCompensationEvent(CompensationEvent event) {
    String messageId = event.getMessageId();
 
    // 경쟁 조건을 피하려면 존재 확인 후 INSERT가 아니라
    // messageId 유니크 제약 + 중복 키 예외 처리로 판정한다.
    try {
        inboxRepository.saveNew(new InboxRecord(messageId, event.getSagaId()));
    } catch (DuplicateKeyException e) {
        log.info("Duplicate message ignored: {}", messageId);
        return;
    }
 
    // 실제 보상 로직은 같은 트랜잭션 안에서 수행 (외부 호출은 별도 워커로 분리하는 것을 권장)
    inventoryService.releaseReservation(event.getOrderId(), event.getQuantity());
}

CREATE TABLE saga_instances (
    saga_id UUID PRIMARY KEY,
    saga_type VARCHAR(100) NOT NULL,
    current_state VARCHAR(100) NOT NULL,
    saga_data JSONB NOT NULL,       -- Saga 컨텍스트 (주문 ID, 금액 등)
    retry_count INTEGER DEFAULT 0,
    last_updated_at TIMESTAMPTZ DEFAULT NOW(),
    timeout_at TIMESTAMPTZ          -- 타임아웃 전이를 위한 데드라인
);
 
CREATE TABLE saga_state_history (
    id BIGSERIAL PRIMARY KEY,
    saga_id UUID NOT NULL REFERENCES saga_instances(saga_id),
    from_state VARCHAR(100),
    to_state VARCHAR(100) NOT NULL,
    event_type VARCHAR(100),
    occurred_at TIMESTAMPTZ DEFAULT NOW()
);

// 개념적 예시 (Java)
@Service
public class OrderSagaOrchestrator {
 
    private static final Map<String, List<String>> VALID_TRANSITIONS = Map.ofEntries(
        Map.entry("STARTED", List.of("INVENTORY_RESERVING")),
        Map.entry("INVENTORY_RESERVING", List.of("PAYMENT_PROCESSING", "INVENTORY_COMPENSATING", "FAILED")),
        Map.entry("PAYMENT_PROCESSING", List.of("DELIVERY_SCHEDULING", "INVENTORY_COMPENSATING")),
        Map.entry("DELIVERY_SCHEDULING", List.of("COMPLETED", "PAYMENT_COMPENSATING")),
        Map.entry("PAYMENT_COMPENSATING", List.of("INVENTORY_COMPENSATING", "COMPENSATION_FAILED")),
        Map.entry("INVENTORY_COMPENSATING", List.of("COMPENSATED", "COMPENSATION_FAILED"))
    );
 
    @Transactional
    public void transition(UUID sagaId, String toState, String eventType) {
        SagaInstance saga = sagaRepository.findById(sagaId)
            .orElseThrow(() -> new SagaNotFoundException(sagaId));
 
        String fromState = saga.getCurrentState();
 
        if (!VALID_TRANSITIONS.getOrDefault(fromState, List.of()).contains(toState)) {
            throw new InvalidStateTransitionException(fromState, toState);
        }
 
        saga.setCurrentState(toState);
        saga.setLastUpdatedAt(Instant.now());
        sagaRepository.save(saga);
 
        stateHistoryRepository.save(StateHistory.of(sagaId, fromState, toState, eventType));
 
        triggerNextAction(saga, toState);
    }
}

# 개념적 예시 — 타임아웃 스캐너 (Python)
import asyncio
 
async def timeout_scanner():
    while True:
        stale_sagas = db.query("""
            SELECT saga_id, current_state, saga_data
            FROM saga_instances
            WHERE timeout_at < NOW()
              AND current_state NOT IN ('COMPLETED', 'COMPENSATED', 'FAILED', 'COMPENSATION_FAILED')
        """).fetchall()
 
        for saga in stale_sagas:
            await publish_event(
                topic="saga.timeout",
                payload={
                    "saga_id": str(saga["saga_id"]),
                    "timed_out_state": saga["current_state"]
                }
            )
 
        await asyncio.sleep(30)

// 개념적 예시 (Go 1.21+)
package saga
 
import (
    "context"
    "math"
    "math/rand"
    "time"
)
 
type RetryConfig struct {
    MaxAttempts int
    BaseDelay   time.Duration
    MaxDelay    time.Duration
}
 
func RetryWithBackoff(ctx context.Context, cfg RetryConfig, fn func() error) error {
    var lastErr error
 
    for attempt := 0; attempt < cfg.MaxAttempts; attempt++ {
        if err := fn(); err == nil {
            return nil
        } else {
            lastErr = err
        }
 
        delay := time.Duration(math.Pow(2, float64(attempt))) * cfg.BaseDelay
        jitter := time.Duration(rand.Int63n(int64(delay/2) + 1))
        sleep := min(delay+jitter, cfg.MaxDelay)
 
        select {
        case <-ctx.Done():
            return ctx.Err()
        case <-time.After(sleep):
        }
    }
 
    return lastErr
}

CREATE TABLE saga_outbox (
    id BIGSERIAL PRIMARY KEY,
    saga_id UUID NOT NULL,
    event_type VARCHAR(100) NOT NULL,
    payload JSONB NOT NULL,
    published_at TIMESTAMPTZ,  -- NULL이면 미발행
    created_at TIMESTAMPTZ DEFAULT NOW()
);

# 보상 이벤트를 같은 로컬 트랜잭션에 기록 (개념적 예시)
with db.transaction():
    db.execute(
        "UPDATE saga_instances SET current_state = 'INVENTORY_COMPENSATING' WHERE saga_id = %s",
        [saga_id]
    )
    db.execute("""
        INSERT INTO saga_outbox (saga_id, event_type, payload)
        VALUES (%s, 'COMPENSATE_INVENTORY', %s)
    """, [saga_id, {"order_id": order_id, "quantity": quantity}])
 
# 트랜잭션 커밋 후 Outbox 폴러가 메시지 브로커로 발행.
# 프로세스가 죽어도 DB에 이벤트가 남아 있으므로 재시작 후 발행 가능.

# Temporal Python SDK를 사용한 Saga 워크플로우 (개념적 예시)
from dataclasses import dataclass
from datetime import timedelta
 
from temporalio import workflow, activity
from temporalio.common import RetryPolicy
 
 
@dataclass
class OrderRequest:
    order_id: str
    quantity: int
    amount: int
 
 
@activity.defn
async def reserve_inventory(order: OrderRequest) -> None: ...
 
@activity.defn
async def release_inventory(order: OrderRequest) -> None: ...
 
@activity.defn
async def process_payment(order: OrderRequest) -> None: ...
 
@activity.defn
async def cancel_payment(order: OrderRequest) -> None: ...
 
@activity.defn
async def schedule_delivery(order: OrderRequest) -> None: ...
 
 
@workflow.defn
class OrderSagaWorkflow:
    @workflow.run
    async def run(self, order: OrderRequest) -> str:
        # 보상 스택: (activity 함수, 인자) 튜플로 보관해야 직렬화·재실행이 안전하다.
        compensations: list[tuple] = []
 
        try:
            await workflow.execute_activity(
                reserve_inventory, order,
                start_to_close_timeout=timedelta(seconds=30),
                retry_policy=RetryPolicy(maximum_attempts=3),
            )
            compensations.append((release_inventory, order))
 
            await workflow.execute_activity(
                process_payment, order,
                start_to_close_timeout=timedelta(seconds=60),
                retry_policy=RetryPolicy(maximum_attempts=3),
            )
            compensations.append((cancel_payment, order))
 
            await workflow.execute_activity(
                schedule_delivery, order,
                start_to_close_timeout=timedelta(seconds=30),
            )
            return "ORDER_COMPLETED"
 
        except Exception:
            for comp_fn, comp_arg in reversed(compensations):
                await workflow.execute_activity(
                    comp_fn, comp_arg,
                    start_to_close_timeout=timedelta(seconds=30),
                    retry_policy=RetryPolicy(maximum_attempts=10),
                )
            return "ORDER_COMPENSATED"