Saga 보상 트랜잭션 실패 재처리 설계 가이드

-- 사가 인스턴스
CREATE TABLE saga_instance (
  saga_id           VARCHAR(64) PRIMARY KEY,
  saga_type         VARCHAR(64) NOT NULL,
  status            VARCHAR(32) NOT NULL, -- RUNNING, COMPENSATING, COMPLETED, FAILED
  created_at        TIMESTAMP NOT NULL,
  updated_at        TIMESTAMP NOT NULL
);

-- 단계별 상태 (forward/compensation을 같은 테이블로 관리)
CREATE TABLE saga_step (
  saga_id           VARCHAR(64) NOT NULL,
  step_name         VARCHAR(64) NOT NULL,
  direction         VARCHAR(16) NOT NULL, -- FORWARD / COMPENSATE
  status            VARCHAR(32) NOT NULL, -- PENDING, IN_PROGRESS, SUCCEEDED, RETRYING, DEAD
  attempt           INT NOT NULL DEFAULT 0,
  next_retry_at     TIMESTAMP NULL,
  last_error        TEXT NULL,
  idempotency_key   VARCHAR(128) NOT NULL,
  updated_at        TIMESTAMP NOT NULL,
  PRIMARY KEY (saga_id, step_name, direction)
);

-- 아웃박스(이벤트/커맨드 발행)
CREATE TABLE outbox (
  id               BIGSERIAL PRIMARY KEY,
  aggregate_id     VARCHAR(64) NOT NULL,
  event_type       VARCHAR(128) NOT NULL,
  payload          JSONB NOT NULL,
  status           VARCHAR(32) NOT NULL, -- NEW, PUBLISHED, FAILED
  created_at       TIMESTAMP NOT NULL
);

-- 보상 적용 이력을 남기는 테이블
CREATE TABLE compensation_applied (
  idempotency_key VARCHAR(128) PRIMARY KEY,
  applied_at      TIMESTAMP NOT NULL
);

-- 보상 실행 시
-- 1) 먼저 insert 시도 (이미 있으면 중복 실행)
INSERT INTO compensation_applied(idempotency_key, applied_at)
VALUES (:key, now())
ON CONFLICT DO NOTHING;

-- 2) rowcount=1일 때만 실제 보상 로직 수행

-- 처리할 보상 단계 하나를 락으로 가져오기
WITH cte AS (
  SELECT saga_id, step_name
  FROM saga_step
  WHERE direction='COMPENSATE'
    AND status IN ('PENDING','RETRYING')
    AND (next_retry_at IS NULL OR next_retry_at <= now())
  ORDER BY updated_at ASC
  LIMIT 1
  FOR UPDATE SKIP LOCKED
)
UPDATE saga_step s
SET status='IN_PROGRESS', attempt=attempt+1, updated_at=now()
FROM cte
WHERE s.saga_id=cte.saga_id AND s.step_name=cte.step_name AND s.direction='COMPENSATE'
RETURNING s.*;

import random
from datetime import datetime, timedelta

RETRYABLE = {"TIMEOUT", "UNAVAILABLE", "THROTTLED"}

def compute_next_retry(attempt: int, base_seconds: int = 2, cap_seconds: int = 300):
    # exponential backoff with full jitter
    exp = min(cap_seconds, base_seconds * (2 ** max(0, attempt - 1)))
    sleep = random.uniform(0, exp)
    return datetime.utcnow() + timedelta(seconds=sleep)

def classify_error(http_status: int | None, code: str | None):
    if code in RETRYABLE:
        return "RETRY"
    if http_status is not None and http_status >= 500:
        return "RETRY"
    if http_status in (408, 429):
        return "RETRY"
    return "DEAD"

BEGIN;

UPDATE saga_step
SET status='PENDING', updated_at=now()
WHERE saga_id=:sagaId AND step_name='Inventory' AND direction='COMPENSATE';

INSERT INTO outbox(aggregate_id, event_type, payload, status, created_at)
VALUES (
  :sagaId,
  'COMPENSATE_INVENTORY',
  jsonb_build_object('sagaId', :sagaId, 'orderId', :orderId, 'idempotencyKey', :key),
  'NEW',
  now()
);

COMMIT;

from contextlib import contextmanager

MAX_ATTEMPTS = 10

@contextmanager
def tx(db):
    db.begin()
    try:
        yield
        db.commit()
    except Exception:
        db.rollback()
        raise

def run_once(db, client):
    with tx(db):
        step = db.fetch_one("""
            -- 앞서 소개한 UPDATE ... RETURNING 패턴을 함수로 감쌈
            SELECT * FROM pick_and_mark_in_progress();
        """)
        if not step:
            return None

    # 트랜잭션 밖에서 외부 호출(락 오래 잡지 않기)
    try:
        # 멱등 키를 포함해 외부/내부 보상 수행
        client.compensate(step["step_name"], step["idempotency_key"], step["saga_id"])
        outcome = ("SUCCEEDED", None, None)
    except Exception as e:
        err = str(e)
        # 여기서는 단순화: classify_error로 RETRY/DEAD 결정
        action = "RETRY" if "timeout" in err.lower() else "DEAD"
        outcome = ("RETRYING" if action == "RETRY" else "DEAD", err, action)

    with tx(db):
        if outcome[0] == "SUCCEEDED":
            db.execute("""
                UPDATE saga_step
                SET status='SUCCEEDED', last_error=NULL, next_retry_at=NULL, updated_at=now()
                WHERE saga_id=:saga_id AND step_name=:step_name AND direction='COMPENSATE'
            """, step)
        else:
            # attempt 기반으로 next_retry_at 계산
            current = db.fetch_one("""
                SELECT attempt FROM saga_step
                WHERE saga_id=:saga_id AND step_name=:step_name AND direction='COMPENSATE'
            """, step)

            if current["attempt"] >= MAX_ATTEMPTS or outcome[2] == "DEAD":
                db.execute("""
                    UPDATE saga_step
                    SET status='DEAD', last_error=:err, updated_at=now()
                    WHERE saga_id=:saga_id AND step_name=:step_name AND direction='COMPENSATE'
                """, {**step, "err": outcome[1]})
            else:
                next_at = compute_next_retry(current["attempt"])
                db.execute("""
                    UPDATE saga_step
                    SET status='RETRYING', last_error=:err, next_retry_at=:next_at, updated_at=now()
                    WHERE saga_id=:saga_id AND step_name=:step_name AND direction='COMPENSATE'
                """, {**step, "err": outcome[1], "next_at": next_at})

    return step