Orchestration-based Saga Pattern for async Python microservices
NATS JetStream + PostgreSQL · built on nats-outbox
Distributed transactions across microservices, without a distributed transaction.
The Problem · Quick Start · Architecture · Guarantees · Integration Contract · Roadmap
Modern systems decompose business operations across multiple services, each owning its own database. Placing an order might require :
1. Payment Service → reserve the charge
2. Inventory Service → reserve the stock
3. Shipping Service → schedule delivery
There is no ACID transaction that spans all three. If step 2 fails, step 1 must be undone — the customer should never be charged for an item that's out of stock. This is the classic distributed-transactions problem, and the industry-standard answer is the Saga Pattern: break the operation into local transactions, each paired with a compensating action, coordinated by an orchestrator that walks forward on success and rolls back in reverse order on failure.
| Status | |
|---|---|
| Async-native | Existing Python saga libraries predate async/await |
| NATS JetStream support | Built for Celery/Kafka, not NATS |
| Actively maintained | Last meaningful updates years old |
| Crash-safe by construction | Relies on application-level discipline, not atomic guarantees |
natsaga fills this gap: an async-first, NATS JetStream-native saga orchestrator with atomic state transitions, deterministic idempotency, mandatory timeouts, and a complete audit trail — the properties a production system actually needs from this pattern.
# The naive approach — looks fine until it isn't
await payment_service.reserve(order)
await inventory_service.reserve(order) # crash here?
await shipping_service.schedule(order) # payment is now orphanedIf the process crashes between calls, or a service is briefly unreachable, you're left with a payment that has no matching inventory reservation — and nothing recorded anywhere to tell you that. natsaga replaces this sequence with declared steps, persisted state, and automatic compensation, so a crash mid-flight is a recoverable event, not a silent data-integrity bug.
An orchestrator faces the exact same dual-write problem that nats-outbox was built to solve: it must, in one atomic step, (a) update the saga's state in Postgres and (b) publish the next command to NATS. natsaga doesn't reimplement this — it uses outbox_transaction() from nats-outbox for every state transition.
┌──────────────────────────┐ ┌───────────────────────────┐
│ natsaga │ uses │ nats-outbox │
│ Saga orchestration, │───────►│ Atomic Postgres write + │
│ compensation, timeouts │ │ reliable NATS publish │
└──────────────────────────┘ └───────────────────────────┘
This is a deliberate layering decision, not an implementation detail: reliable delivery is a solved problem one layer down, so natsaga's own code stays focused on saga semantics — step ordering, compensation, timeouts — instead of re-solving message delivery.
pip install natsagaDefine a saga declaratively:
from natsaga.core.saga import Saga
from natsaga.core.step import Step
class OrderSaga(Saga):
saga_type = "order_creation"
@classmethod
def steps(cls) -> list[Step]:
return [
Step(
name="reserve_payment",
action_subject="payment.reserve",
compensation_subject="payment.release",
timeout_seconds=30,
),
Step(
name="reserve_inventory",
action_subject="inventory.reserve",
compensation_subject="inventory.release",
timeout_seconds=30,
),
Step(
name="schedule_shipping",
action_subject="shipping.schedule",
compensation_subject="shipping.cancel",
timeout_seconds=60,
),
]Start it from your FastAPI endpoint:
from natsaga.orchestrator import SagaOrchestrator
@app.post("/orders")
async def create_order(order: OrderIn, session: AsyncSession = Depends(get_session)):
saga_id = await orchestrator.start(
OrderSaga,
payload={"order_id": order.id, "amount": order.amount},
session=session,
)
return {"saga_id": saga_id, "status": "processing"}Run the two background workers:
natsaga listener start --app myapp.main:orchestrator
natsaga timeout-checker start --app myapp.main:orchestratorIf reserve_inventory fails, natsaga automatically publishes to payment.release — no manual rollback code required.
HTTP Request
│
▼
┌───────────────────────────────────────────────────────────────────┐
│ SagaOrchestrator.start() │
│ │
│ async with outbox_transaction(session) as tx: │
│ saga_state.status = "running" │
│ saga_state.step_deadline = now() + step.timeout_seconds │
│ tx.publish_event("payment.reserve", event_id=uuid5(...)) │
└────────────────────────────┬────────────────────────────────────┘
│ atomic commit (Postgres ACID)
▼
┌───────────────────────────────────────────────────────────────────┐
│ PostgreSQL │
│ saga_state saga_step_log │
│ ┌─────────────────────────┐ ┌───────────────────────────┐ │
│ │ id order-saga-1 │ │ step │ direction │ outcome│ │
│ │ status running │ │ 0 │ forward │ - │ │
│ │ current 0 │ └───────────────────────────┘ │
│ │ deadline now()+30s │ │
│ └─────────────────────────┘ │
└────────────────────────────┬────────────────────────────────────┘
│ relayed by nats-outbox
▼
┌───────────────────────────────────────────────────────────────────┐
│ NATS JetStream │
│ subject: payment.reserve │
└────────────────────────────┬────────────────────────────────────┘
│
▼
┌───────────────────────────────────────────────────────────────────┐
│ Payment Service │
│ processes the reservation, then publishes its result to: │
│ payment.reserve.response.{saga_id}.{step_index} │
└────────────────────────────┬────────────────────────────────────┘
▼
┌───────────────────────────────────────────────────────────────────┐
│ SagaListener │
│ subscribes to *.response.>, routes outcome to the orchestrator │
└────────────────────────────┬────────────────────────────────────┘
success ──────┴────── failure / timeout
│ │
▼ ▼
advance to next step compensate completed steps
(same atomic pattern) in reverse order
Step 1: reserve_payment succeeded
Step 2: reserve_inventory succeeded
Step 3: schedule_shipping FAILED
│
▼
compensation begins, reverse order
│
┌───────────┴────────────┐
▼ ▼
inventory.release payment.release
(compensate step 2) (compensate step 1)
│ │
▼ ▼
saga_step_log: saga_step_log:
step=1, compensate, step=0, compensate,
outcome=success outcome=success
│
▼
saga_state.status = "failed"
| Component | Role | Trigger |
|---|---|---|
SagaOrchestrator |
Persists saga state, dispatches the next command, drives compensation | Called directly on start(); called by the listener and timeout checker on response/expiry |
SagaListener |
Subscribes to *.response.> on JetStream, routes outcomes |
Every microservice response |
TimeoutChecker |
Polls saga_state for expired step_deadline, triggers compensation |
Every poll_interval seconds (default 5s) |
| Column | Type | Purpose |
|---|---|---|
id |
UUID |
Saga identity — UUID rather than a sequence, since sagas are created from many concurrent API processes with no natural single point of contention |
saga_type |
TEXT |
Which Saga subclass this row belongs to |
current_step |
SMALLINT |
Zero-indexed pointer to the active step |
status |
TEXT + CHECK |
running → completed | compensating → failed | compensation_failed |
payload |
JSONB |
The original, immutable input — never overwritten |
context |
JSONB |
Mutable accumulator for step outputs (e.g. a payment reference a later step needs) — deliberately separate from payload so "what was asked" and "what happened" never blur together |
step_deadline |
TIMESTAMPTZ |
Absolute deadline for the current step; the source of truth for TimeoutChecker |
last_error |
TEXT |
Latest failure, for a fast glance without joining saga_step_log |
created_at / updated_at |
TIMESTAMPTZ |
Lifecycle timestamps; updated_at also powers crash-recovery scans |
Append-only, one row per attempted transition (forward or compensating):
SELECT step_name, direction, outcome, error, finished_at
FROM saga_step_log
WHERE saga_id = '...'
ORDER BY id;This is what turns a 3 a.m. incident from guesswork into a five-second query: exactly which step ran, in which direction, with what outcome, and why.
Stated precisely — not as marketing claims, but as what is actually tested:
- At-least-once command delivery, inherited from
nats-outbox, deduplicated within JetStream's configured dedup window via a deterministicNats-Msg-Id. - No duplicate step execution on orchestrator crash-restart. Each command's
event_idis a UUID v5 derived from(saga_id, step_index, direction)— deterministic and stable across processes, so a restarted orchestrator recognizes work already in flight instead of re-dispatching it. - No double state advancement on concurrent or duplicate responses. State transitions are guarded by
SELECT ... FOR UPDATEfollowed by a status/step check inside the same transaction — a second, duplicate, or racing response is a verified no-op, not a best-effort one. - No silently stuck sagas. Every step declares a mandatory
timeout_seconds; there is no default that lets a step wait forever. - No silent compensation failures. If a compensating action itself fails, the saga enters a terminal
compensation_failedstate and increments a Prometheus counter — it does not retry silently forever, and it does not pretend to have succeeded.
What natsaga does not guarantee: exactly-once execution of the domain action itself (a step can theoretically execute twice if a microservice fails to be idempotent on message redelivery — that responsibility sits with the microservice, as it does in any at-least-once system, not with the orchestrator).
Every microservice participating in a saga owns two responsibilities:
1. Consume the action subject and do the work:
@nats_subscribe("payment.reserve")
async def handle_reserve_payment(msg: Msg) -> None:
data = json.loads(msg.data)
saga_id = data["saga_id"]
step_index = data["step_index"]
try:
reservation = await payment_gateway.reserve(data["amount"])
outcome = {"outcome": "success", "result": {"payment_ref": reservation.id}}
except PaymentDeclined as exc:
outcome = {"outcome": "failure", "error": str(exc)}
await nc.publish(
f"payment.reserve.response.{saga_id}.{step_index}",
json.dumps(outcome).encode(),
)2. Publish the result to the exact response subject — {action_subject}.response.{saga_id}.{step_index} — with an outcome field of success or failure. This convention is load-bearing: it's what lets SagaListener route a response to the correct saga and step without ambiguity. Getting this subject format wrong is the single most common integration mistake — test it against a real NATS instance, not a mock, before deploying a new saga participant.
NATSAGA_DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/myapp
NATSAGA_NATS_URL=nats://localhost:4222
NATSAGA_TIMEOUT_POLL_INTERVAL=5.0 # seconds between timeout scans
NATSAGA_METRICS_PORT=9091 # 0 disables the Prometheus exporter| Metric | Type | Meaning |
|---|---|---|
saga_started_total{saga_type} |
Counter | Sagas started |
saga_completed_total{saga_type} |
Counter | Sagas that reached completed |
saga_failed_total{saga_type} |
Counter | Sagas that reached failed after full compensation |
saga_compensation_failed_total{saga_type} |
Counter | Page someone — a compensation itself failed |
saga_step_duration_seconds{saga_type,step_name,direction} |
Histogram | Time spent per step, forward or compensating |
natsaga does not ship its own Alembic migrations — as a library, embedding one would conflict with your application's own migration history. It exposes the SQLAlchemy Base (SagaState, SagaStepLog), so your own alembic revision --autogenerate detects the tables automatically. A plain migrations/schema.sql is provided for non-SQLAlchemy setups.
| Not included | Why |
|---|---|
| Choreography-based sagas | Different coordination model entirely; orchestration covers the majority of real-world needs first |
| Parallel steps (fan-out/fan-in) | Meaningful added complexity in failure semantics; sequential steps first |
| Automatic compensation retries | A failed compensation needs a human decision, not a silent retry loop, until proven otherwise |
| Web UI | Prometheus metrics are the V1 answer to "what's happening" |
| Brokers other than NATS JetStream | Scope stays tight and matches nats-outbox |
git clone https://github.com/ademboukabes/natsaga.git
cd natsaga
make install
make test # real Postgres + NATS via testcontainers
make lint # ruff + mypy --strict + pyrightIntegration coverage includes: happy path, failure-triggered compensation, timeout-triggered compensation, orchestrator crash-and-restart recovery, compensation failure (terminal state, no retry loop), and duplicate/concurrent response delivery (verified against a real NATS redelivery, not a mock).
- V1 — Orchestration-based sagas, atomic transitions via
nats-outbox, mandatory timeouts, audit log, Prometheus metrics - LISTEN/NOTIFY-based timeout detection (sub-second, vs. today's 5s poll)
- Automatic, bounded retry of failed compensations
- Parallel step groups (fan-out/fan-in)
- Choreography-based saga variant
MIT — see LICENSE.