Metadata-Version: 2.4
Name: natsaga
Version: 0.1.0
Summary: Orchestration-based Saga Pattern for async microservices — NATS JetStream + PostgreSQL
License: MIT
Keywords: distributed-transactions,event-driven,jetstream,microservices,nats,postgresql,saga
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Software Development :: Libraries
Requires-Python: >=3.11
Requires-Dist: asyncpg>=0.29
Requires-Dist: nats-outbox~=0.1.0
Requires-Dist: nats-py>=2.6
Requires-Dist: pydantic-settings>=2.0
Requires-Dist: pydantic>=2.0
Requires-Dist: sqlalchemy[asyncio]>=2.0
Provides-Extra: dev
Requires-Dist: docker>=7.0; extra == 'dev'
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: prometheus-client>=0.20; extra == 'dev'
Requires-Dist: pyright>=1.1; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest-timeout>=2.3; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.4; extra == 'dev'
Requires-Dist: testcontainers[postgres]>=4.0; extra == 'dev'
Provides-Extra: metrics
Requires-Dist: prometheus-client>=0.20; extra == 'metrics'
Description-Content-Type: text/markdown

# natsaga

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

[![CI](https://github.com/ademboukabes/natsaga/actions/workflows/ci.yml/badge.svg)](https://github.com/ademboukabes/natsaga/actions/workflows/ci.yml)
[![Python 3.11+](https://img.shields.io/badge/python-3.11+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-green.svg)](LICENSE)

---

## 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)](https://microservices.io/patterns/data/saga.html) 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](https://github.com/absent1706/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](https://pypi.org/project/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

```bash
pip install natsaga

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

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

---

## Quickstart

### 1. Define your saga

```python
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

```python
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

```python
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**:

```python
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:**
```json
{ "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:

```sql
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):

```python
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:

```python
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

```bash
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](LICENSE).
