Saga Transactions skill for Claude Code

# Hand-rolled saga orchestrator (Postgres)

Read this when the saga runs without a workflow engine: you need the state table, the
transition function, the timeout scanner, and the operator queries. Outbox relay
mechanics (polling vs CDC, ordering) are covered by the event-driven-architecture skill;
this file only shows how the orchestrator uses them.

## Tables

```sql
CREATE TABLE saga_instance (
  saga_id        uuid PRIMARY KEY,
  saga_type      text        NOT NULL,          -- 'checkout'
  definition_ver int         NOT NULL,          -- step list version this instance follows
  business_key   text        NOT NULL,          -- order id, for operator lookup
  state          text        NOT NULL,          -- RUNNING | COMPENSATING | COMPLETED | COMPENSATED | FAILED
  current_step   text        NOT NULL,
  attempt        int         NOT NULL DEFAULT 1,
  version        bigint      NOT NULL DEFAULT 0,  -- optimistic lock; bump on every transition
  payload        jsonb       NOT NULL,          -- ids gathered so far (reservation_id, auth_id)
  step_deadline  timestamptz NOT NULL,          -- when the current step times out
  saga_deadline  timestamptz NOT NULL,          -- business deadline
  last_error     text,
  created_at     timestamptz NOT NULL DEFAULT now(),
  updated_at     timestamptz NOT NULL DEFAULT now()
);
CREATE INDEX saga_due ON saga_instance (step_deadline)
  WHERE state IN ('RUNNING', 'COMPENSATING');
CREATE UNIQUE INDEX saga_business ON saga_instance (saga_type, business_key);

CREATE TABLE saga_step_log (            -- append-only audit trail, one row per transition
  saga_id    uuid        NOT NULL REFERENCES saga_instance,
  seq        bigint      NOT NULL,
  step       text        NOT NULL,
  event      text        NOT NULL,     -- COMMAND_SENT | REPLY_OK | REPLY_REJECTED | TIMEOUT | COMPENSATION_SENT ...
  detail     jsonb,
  at         timestamptz NOT NULL DEFAULT now(),
  PRIMARY KEY (saga_id, seq)
);
```

The unique index on `(saga_type, business_key)` makes "start saga for order 8812" itself
idempotent: a duplicate start request hits the conflict and returns the existing saga.

## Transition function

Every reply and every timeout goes through one function that runs in one transaction:
lock the row, check the reply is for the current step, compute the next state, write the
state, append to the log, insert the next command into the outbox.

```python
STEPS = ["create_order", "reserve_stock", "authorize_card", "approve_order",
         "capture_payment", "request_shipment"]
PIVOT = "approve_order"
COMPENSATION = {"create_order": "reject_order", "reserve_stock": "release_stock",
                "authorize_card": "void_authorization"}

def on_reply(tx, saga_id, step, outcome, data):
    s = tx.fetch_one("SELECT * FROM saga_instance WHERE saga_id=%s FOR UPDATE", saga_id)
    if s.state in ("COMPLETED", "COMPENSATED", "FAILED"):
        return                                   # late or duplicate reply
    if step != s.current_step:
        return                                   # reply for an older step: ignore
    i = STEPS.index(step)
    if s.state == "RUNNING" and outcome == "ok":
        payload = {**s.payload, **data}
        if i + 1 == len(STEPS):
            transition(tx, s, state="COMPLETED", step=step, payload=payload)
        else:
            nxt = STEPS[i + 1]
            transition(tx, s, state="RUNNING", step=nxt, payload=payload)
            send_command(tx, s, nxt, payload)
    elif s.state == "RUNNING" and outcome == "rejected":
        if STEPS.index(step) > STEPS.index(PIVOT):
            raise RuntimeError("retryable step rejected: page a human, do not compensate")
        start_compensation(tx, s, failed_index=i)
    elif s.state == "COMPENSATING" and outcome == "ok":
        continue_compensation(tx, s, done=step)

def start_compensation(tx, s, failed_index):
    todo = [STEPS[j] for j in range(failed_index - 1, -1, -1) if STEPS[j] in COMPENSATION]
    if not todo:
        transition(tx, s, state="COMPENSATED", step=STEPS[failed_index], payload=s.payload)
        return
    transition(tx, s, state="COMPENSATING", step=todo[0], payload={**s.payload, "undo": todo})
    send_command(tx, s, COMPENSATION[todo[0]], s.payload, compensate=True)
```

`transition` does `UPDATE ... SET state, current_step, attempt=1, version=version+1,
step_deadline=now()+timeout(step) WHERE saga_id=%s AND version=%s` and fails if no row
matched. `send_command` inserts into the orchestrator's outbox with idempotency key
`<saga_id>:<step>` or `<saga_id>:<step>:compensate`.

## Timeout scanner

Run every few seconds from one or more workers. `SKIP LOCKED` lets several workers share
the load without double-processing a saga.

```sql
BEGIN;
SELECT saga_id, current_step, attempt, state
FROM saga_instance
WHERE state IN ('RUNNING', 'COMPENSATING') AND step_deadline < now()
ORDER BY step_deadline
LIMIT 50
FOR UPDATE SKIP LOCKED;
-- for each row, in application code:
--   if saga_deadline passed and step is before the pivot  -> start_compensation
--   elif attempt < max_attempts(step) or step is retryable/compensation
--        -> re-send the SAME command (same idempotency key), attempt+1, new step_deadline
--   else (pre-pivot step exhausted)                        -> start_compensation
--   compensation exhausted alert_after                     -> state stays COMPENSATING, alert
COMMIT;
```

Never mark a timed-out step as failed without re-sending first: the participant may have
done the work and the reply may be delayed.

## Operator queries

```sql
-- Where is order 8812?
SELECT s.state, s.current_step, s.attempt, s.last_error, s.updated_at
FROM saga_instance s WHERE saga_type='checkout' AND business_key='8812';
SELECT seq, step, event, detail, at FROM saga_step_log
WHERE saga_id = (SELECT saga_id FROM saga_instance WHERE business_key='8812') ORDER BY seq;

-- Oldest stuck saga per state (alert input)
SELECT saga_type, state, current_step, min(updated_at) AS oldest, count(*)
FROM saga_instance WHERE state IN ('RUNNING','COMPENSATING')
GROUP BY 1,2,3 ORDER BY oldest;

-- Compensation rate per step over the last day
SELECT step, count(*) FILTER (WHERE event='REPLY_REJECTED') AS rejected, count(*) AS replies
FROM saga_step_log WHERE at > now() - interval '1 day' GROUP BY step;
```

Manual repair (force-compensate, skip a step, mark resolved) is a transition like any
other: go through the transition function, write a log row with the operator name and
reason, and get explicit confirmation before running it against production.

## Versioning the definition

- Store `definition_ver` on each instance. Deploy the new step list as a new version;
  keep the old transition table until no instance with the old version is non-terminal.
- Removing or reordering steps in place breaks in-flight sagas whose `current_step` no
  longer exists or now means something else.

## Retention

Archive terminal sagas and their logs after the dispute window for the business (for
payments, often months). Keep the archive queryable by business key.

#!/usr/bin/env python3
"""Validate a saga step definition and print its compensation plan.

Usage:
    python3 check_saga.py saga.json

Spec format (JSON):
{
  "name": "checkout",
  "deadline_s": 900,
  "steps": [
    {"name": "create_order", "service": "orders", "kind": "compensable",
     "compensation": "reject_order", "idempotency_key": "{saga_id}:create_order",
     "timeout_s": 5, "retry": {"max_attempts": 3}},
    {"name": "approve_order", "service": "orders", "kind": "pivot",
     "idempotency_key": "{saga_id}:approve_order", "timeout_s": 5,
     "retry": {"max_attempts": 3}},
    {"name": "capture_payment", "service": "payments", "kind": "retryable",
     "idempotency_key": "{saga_id}:capture", "timeout_s": 10,
     "retry": {"max_attempts": null, "alert_after": 10}}
  ]
}

kind is one of: compensable, pivot, retryable.
retry.max_attempts: integer, or null for "retry until success" (allowed only for
retryable steps, and then retry.alert_after is required).

Exit codes: 0 = no errors, 1 = design errors found, 2 = unreadable input.
"""
import json
import sys

KINDS = {"compensable", "pivot", "retryable"}


def main(argv):
    if len(argv) != 2:
        print("usage: python3 check_saga.py saga.json")
        return 2
    try:
        with open(argv[1], encoding="utf-8") as fh:
            spec = json.load(fh)
    except (OSError, json.JSONDecodeError) as exc:
        print(f"ERROR: cannot read {argv[1]}: {exc}")
        return 2

    errors, warnings = [], []
    steps = spec.get("steps")
    if not isinstance(steps, list) or not steps:
        print("ERROR: spec needs a non-empty 'steps' list")
        return 2

    deadline = spec.get("deadline_s")
    if not isinstance(deadline, (int, float)) or deadline <= 0:
        errors.append("saga has no positive 'deadline_s' (business deadline for completion)")

    names, comp_names = set(), set()
    pivot_idx = [i for i, s in enumerate(steps) if s.get("kind") == "pivot"]
    if len(pivot_idx) > 1:
        errors.append(f"{len(pivot_idx)} pivot steps; a saga has exactly one point of no return")
    if not pivot_idx:
        warnings.append("no pivot step: every step is treated as compensable; confirm nothing is irreversible")
    pivot = pivot_idx[0] if pivot_idx else len(steps)

    budget = 0.0
    for i, s in enumerate(steps):
        n = s.get("name") or f"<step {i + 1}>"
        kind = s.get("kind")
        if n in names:
            errors.append(f"{n}: duplicate step name")
        names.add(n)
        if kind not in KINDS:
            errors.append(f"{n}: kind must be one of {sorted(KINDS)}, got {kind!r}")
            continue
        if not s.get("idempotency_key"):
            errors.append(f"{n}: missing idempotency_key (retries and duplicate deliveries will repeat the effect)")
        t = s.get("timeout_s")
        if not isinstance(t, (int, float)) or t <= 0:
            errors.append(f"{n}: missing positive timeout_s")
            t = 0
        retry = s.get("retry") or {}
        attempts = retry.get("max_attempts", "missing")
        if attempts == "missing":
            errors.append(f"{n}: retry.max_attempts not set (use an integer, or null for retry-until-success on retryable steps)")
            attempts = 1
        elif attempts is None:
            if kind != "retryable":
                errors.append(f"{n}: unbounded retries on a {kind} step; bound it so the saga can compensate")
            elif not retry.get("alert_after"):
                errors.append(f"{n}: unbounded retries need retry.alert_after so a human is paged")
            attempts = retry.get("alert_after") or 1
        elif not isinstance(attempts, int) or attempts < 1:
            errors.append(f"{n}: retry.max_attempts must be a positive integer or null")
            attempts = 1
        if i < pivot:
            budget += t * attempts

        if i < pivot:
            if kind == "retryable":
                warnings.append(f"{n}: retryable step before the pivot; fine only if it has no effect that needs undoing")
            elif kind == "compensable" and not s.get("compensation"):
                errors.append(f"{n}: compensable step without a 'compensation'")
        elif i > pivot:
            if kind == "compensable":
                errors.append(f"{n}: compensable step after the pivot; after the point of no return steps must be retryable")
        comp = s.get("compensation")
        if comp:
            if i >= pivot:
                errors.append(f"{n}: has a compensation but sits at/after the pivot; it would never run")
            if comp in comp_names or comp in {x.get("name") for x in steps}:
                errors.append(f"{n}: compensation name {comp!r} collides with another step or compensation")
            comp_names.add(comp)

    if isinstance(deadline, (int, float)) and deadline > 0 and budget > deadline:
        errors.append(
            f"pre-pivot worst case {budget:.0f}s (timeout x attempts) exceeds deadline_s {deadline}s; "
            "shorten timeouts or attempts, or the deadline fires mid-retry"
        )

    print(f"Saga: {spec.get('name', '<unnamed>')}  steps={len(steps)}  pivot="
          f"{steps[pivot]['name'] if pivot < len(steps) else 'none'}")
    for w in warnings:
        print(f"WARN:  {w}")
    for e in errors:
        print(f"ERROR: {e}")

    print("\nFailure plan (compensations run in this order):")
    for i, s in enumerate(steps):
        n = s.get("name") or f"<step {i + 1}>"
        if i < pivot or (i == pivot and pivot < len(steps)):
            undo = [steps[j].get("compensation") for j in range(i - 1, -1, -1) if steps[j].get("compensation")]
            print(f"  {n} fails -> " + (", ".join(undo) if undo else "nothing to compensate; mark saga failed"))
        else:
            print(f"  {n} fails -> retry until success; alert, never compensate")

    print(f"\nResult: {'FAIL' if errors else 'PASS'} ({len(errors)} errors, {len(warnings)} warnings)")
    return 1 if errors else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))