natsaga 0.1.0

Orchestration-based Saga Pattern for async microservices — NATS JetStream + PostgreSQL

natsaga

Orchestration-based Saga Pattern for async Python microservices — NATS JetStream + PostgreSQL

---

What it solves

In a microservices architecture, a business operation (e.g. "create order") often spans multiple services, each with its own local transaction:

Payment Service   → reserves the amount on the card
Inventory Service → reserves stock
Shipping Service  → schedules delivery

There is no distributed ACID transaction across these three services. If step 2 fails, step 1 must be compensated — otherwise the customer paid for unavailable stock.

natsaga implements the Saga Pattern (orchestration-based) for Python async microservices:

• An orchestrator persists saga state in PostgreSQL and dispatches step commands to NATS JetStream
• Each microservice executes its local transaction and publishes a response event (success or failure)
• On failure, the orchestrator triggers compensations in reverse order
• A TimeoutChecker worker ensures no saga waits forever if a microservice never responds

Why natsaga fills a real gap

The only known Python saga library (saga-framework, ~50 stars) is Celery-based, has no native async/await support, no NATS integration, and is unmaintained. natsaga is the first async-native, FastAPI + NATS JetStream, actively maintained Python saga library.

Architecture: built on nats-outbox

natsaga depends on nats-outbox for atomic state transitions. Every saga state update + next-step command publication happens inside a single outbox_transaction() call — the same dual-write problem, solved once.

Delivery guarantee (precise, not marketing): at-least-once delivery of step commands, with JetStream dedup absorbing retries within the configured dedup window (default: 2 minutes). Microservices are responsible for their own idempotency. natsaga documents this explicitly — no "zero loss guaranteed" claims.

---

Installation

pip install natsaga

# With Prometheus metrics support:
pip install natsaga[metrics]

Requirements: Python 3.11+, PostgreSQL 14+, NATS JetStream

---

Quickstart

1. Define your saga

from natsaga.core.saga import Saga
from natsaga.core.step import Step


class OrderCreationSaga(Saga):
    saga_type = "order_creation"  # unique identifier stored in the database

    @classmethod
    def steps(cls) -> list[Step]:
        return [
            Step(
                name="reserve_payment",
                action_subject="payment.reserve",         # publish here to trigger
                compensation_subject="payment.release",   # publish here to rollback
                timeout_seconds=30,                       # mandatory — no default
            ),
            Step(
                name="reserve_inventory",
                action_subject="inventory.reserve",
                compensation_subject="inventory.release",
                timeout_seconds=15,
            ),
            Step(
                name="schedule_shipping",
                action_subject="shipping.schedule",
                compensation_subject="shipping.cancel",
                timeout_seconds=20,
            ),
        ]

2. Start the orchestrator

import nats
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from natsaga.core.models import create_tables
from natsaga.listener import SagaListener
from natsaga.orchestrator import SagaOrchestrator
from natsaga.timeout_checker import TimeoutChecker

DATABASE_URL = "postgresql+asyncpg://user:pass@localhost:5432/mydb"
NATS_URL = "nats://localhost:4222"

engine = create_async_engine(DATABASE_URL)
session_factory = async_sessionmaker(engine, expire_on_commit=False)

# Create tables (or use Alembic migrations — see migrations/)
await create_tables(engine)

# Connect to NATS
nc = await nats.connect(NATS_URL)

# Build the orchestrator and register saga types
orchestrator = SagaOrchestrator(session_factory)
orchestrator.register(OrderCreationSaga)

# Start the NATS response listener
listener = await SagaListener.from_nats_client(
    nc,
    orchestrator=orchestrator,
    stream_name="SAGA_RESPONSES",  # pre-configured JetStream stream
)
await listener.subscribe_all()

# Start the timeout checker (detects stuck sagas)
checker = TimeoutChecker(session_factory, orchestrator, poll_interval_seconds=5.0)
await checker.start()

3. Start a saga

saga_id = await orchestrator.start(
    OrderCreationSaga,
    payload={"order_id": str(order.id), "amount": 99.90, "user_id": str(user.id)},
)
# Returns immediately. The first step command is published to NATS JetStream
# via nats-outbox (outbox_transaction → outbox relay → NATS).

4. Implement your microservice (example: Payment Service)

Your microservice consumes from payment.reserve and publishes its result to the response subject:

import json
import nats

nc = await nats.connect("nats://localhost:4222")
js = nc.jetstream()

async def handle_payment_reserve(msg):
    data = json.loads(msg.data)
    saga_id    = data["saga_id"]
    step_index = data["step_index"]
    amount     = data["saga_payload"]["amount"]

    try:
        ref = await charge_card(amount)  # your business logic
        response = {"outcome": "success", "result": {"payment_ref": ref}}
    except InsufficientFundsError as exc:
        response = {"outcome": "failure", "error": str(exc)}

    # Publish response to the response subject
    response_subject = f"payment.reserve.response.{saga_id}.{step_index}"
    await nc.publish(response_subject, json.dumps(response).encode())
    await msg.ack()

await js.subscribe("payment.reserve", cb=handle_payment_reserve, durable="payment-svc")

Response subject format: {action_subject}.response.{saga_id}.{step_index}

Response payload:

{ "outcome": "success", "result": { "payment_ref": "ch_123" } }
{ "outcome": "failure", "error": "Insufficient funds" }

---

State machine

                start()
                  │
                  ▼
              [running]
          step succeeds ──► next step ──► ... ──► [completed]
          step fails / timeout
                  │
                  ▼
           [compensating]
          compensation succeeds ──► prev step ──► ... ──► [failed]
          compensation fails
                  │
                  ▼
        [compensation_failed]   ← human intervention required

Status values

Status	Meaning
running	Saga in progress, awaiting a step response
completed	All steps succeeded
compensating	A step failed, rolling back in reverse order
failed	Rollback complete (all compensations succeeded)
compensation_failed	A compensation itself failed — requires human intervention

---

Idempotency and crash recovery

Each step command is published with a deterministic UUID v5 derived from (saga_id, step_index, direction). This UUID becomes the Nats-Msg-Id header via nats-outbox's outbox relay.

On orchestrator restart:

• The relay retries the already-persisted outbox_events row with the same Nats-Msg-Id
• JetStream's dedup window silently drops any duplicate within the configured window
• The orchestrator never re-dispatches a command — it resumes from the persisted saga_state row

---

Timeout detection

Each step declares an explicit timeout_seconds. When a command is dispatched, saga_state.step_deadline is set to now() + timeout_seconds.

The TimeoutChecker worker polls:

SELECT id, current_step, saga_type, step_deadline
FROM saga_state
WHERE status = 'running'
  AND step_deadline IS NOT NULL
  AND step_deadline < now()
ORDER BY step_deadline
LIMIT 50
FOR UPDATE SKIP LOCKED

On expiry, it calls on_step_failure() with a synthetic timeout error — triggering the normal compensation cascade.

Detection latency: at most poll_interval_seconds (default 5s). For timeout_seconds ≥ 30 (recommended minimum), worst-case overshoot is 16%.

V2 roadmap: LISTEN/NOTIFY via a trigger on saga_state for sub-second detection.

---

Compensation failure policy (V1)

If a compensation itself fails, natsaga:

1. Marks the saga compensation_failed (terminal status)
2. Logs at ERROR level — "human intervention required"
3. Increments saga_compensation_failed_total Prometheus counter

No automatic retry in V1. This is a documented trade-off — compensation retry requires careful deduplication logic, deferred to V2.

---

Database schema

natsaga uses two tables:

• saga_state — one row per saga instance (the authoritative state machine)
• saga_step_log — append-only audit trail (one row per step execution)

Create tables programmatically (tests / quick-start):

from natsaga.core.models import create_tables
await create_tables(engine)

For production, use the provided SQL migrations in migrations/.

---

Prometheus metrics

Install with pip install natsaga[metrics], then:

from natsaga.observability import SagaMetrics, start_metrics_server

metrics = SagaMetrics()
await start_metrics_server(port=9091)

orchestrator = SagaOrchestrator(session_factory, metrics=metrics)

Metric	Type	Labels
saga_started_total	Counter	saga_type
saga_completed_total	Counter	saga_type
saga_failed_total	Counter	saga_type
saga_compensation_failed_total	Counter	saga_type
saga_step_duration_seconds	Histogram	saga_type, step_name, direction

Alert rule: saga_compensation_failed_total > 0 requires human intervention.

---

V1 scope and roadmap

In scope (V1)

• Sequential orchestration-based sagas
• Asynchronous event-driven communication (no blocking request/reply)
• Mandatory per-step timeouts
• Cascade compensation in reverse order
• Crash recovery without duplicate commands
• compensation_failed status with Prometheus alerting

Out of scope (roadmap)

Feature	Target
Choreography-based sagas	V2
Parallel steps (fan-out/fan-in)	V2
Automatic compensation retry	V2
LISTEN/NOTIFY timeout detection	V2
Web UI for saga monitoring	V3
Support for brokers other than NATS JetStream	V3

---

Development

git clone https://github.com/ademboukabes/natsaga
cd natsaga
python -m venv .venv && source .venv/bin/activate
pip install -e ".[dev,metrics]"

# Lint
ruff check natsaga/ tests/

# Type check
mypy --strict natsaga/

# Integration tests (requires Docker)
pytest tests/ -v -m integration

---

License

MIT — see LICENSE.