# mire-sagas

A Postgres-backed **saga / process-manager** layer for the
[`mire`](https://crates.io/crates/mire) event-sourcing library. Typed handlers,
a declarative DAG topology, and a replica-safe runner with strong consistency
and scale-out across nodes.

```toml
[dependencies]
mire = "0.2"
mire-sagas = "0.2"
```

## The model

A saga is an **aggregate** like any other in `mire` (its state is an event
stream), with three cleanly separated concerns:

- **Orchestration** — the DAG you declare in `build()`: the steps, their
  `.after(...)` dependencies, terminal markers, and compensations. The *plan*.
- **Effects** — the step functions. A step **does its own IO**: it holds a
  client (injected as a `Dep`) and calls the outside world directly, then
  returns its typed output. There is no separate "handler" registered on the
  saga — the external service is *not* part of the topology.
- **Suspension** — when an answer arrives *later* (an async callback, or an
  unsolicited event), a step emits a request and **suspends**; a `deliver`
  (from a callback handler, a queue consumer, or the worker's outbox publisher)
  resumes it.

Three independent mechanisms give the guarantees (do not conflate them):

| Mechanism | Guarantees |
|---|---|
| **Instance lease** (`es_saga_leases`) | at-most-one active processor per instance — effects never run concurrently |
| **OCC** on the envelope stream | the log fence — a stale worker's commit is discarded |
| **Shard leases** (`es_saga_shard_leases`) | work *distribution* only — which worker scans which partition |

Delivery is **at-least-once**; the [`IdempotencyKey`] (derived from the
committed `StepActivated` ordinal, so retries and crash-recovery re-drives
reproduce it) is how effects stay exactly-once.

## Quick sketch

```rust,ignore
struct Booking;
impl Saga for Booking {
    type State = BookingState;
    fn name() -> &'static str { "booking" }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        b.start_on::<OrderPlaced, _>(|e| Some(Start { instance_id: e.id.clone(), state: e.into() }))
            // an inline effect: calls the provider, returns the step output;
            // its rollback is co-located via `.compensate(...)`
            .step("reserve", task(reserve).compensate(comp(release)))
            // an async step: emits a request, suspends until `deliver(Captured)`
            .step("pay", task(request_payment).after("reserve")
                .on_response(|c: Captured| Settled { id: c.id }).terminal_success())
    }
}

// inline effect — does IO directly, idempotent on the key
async fn reserve(State(s): State<BookingState>, Dep(api): Dep<Provider>, key: IdempotencyKey)
    -> Reserved { Reserved { hold: api.hold(&s.item, key.token()).await.unwrap() } }

let runner = PgSagaRunner::new(Booking::build(SagaBuilder::for_saga()), store)
    .with_num_shards(64);
runner.trigger::<OrderPlaced>(&event).await?;        // application drives this
// a SagaWorker drives timeouts, recovery, the outbox drain, on every replica:
SagaWorker::new(runner).run(cancel).await?;
```

## The flagship example: `travel_booking_service`

[`examples/travel_booking_service.rs`](examples/travel_booking_service.rs) is a
full, runnable service that orchestrates **real third-party providers** (a mock
**axum** HTTP backend) over a real **reqwest** client — "what it's actually
like to build on mire-sagas." `flight ∥ hotel → payment → confirm`, where
flight/hotel/confirm are inline effects and payment is async (emitted to the
durable outbox, published off the hot path, delivered back).

Its test suite is a deliberate stress test of the consistency assumptions —
every scenario asserts **exactly-once** real-world effects:

| Test | Stresses |
|---|---|
| `fleet_books_…_flaky_providers_exactly_once` | concurrency + ~25% transient provider 503s (client + outbox retry) + a timeout-compensated booking |
| `a_crashed_worker_is_replaced_…` | a worker hard-aborted mid-flight; a fresh worker completes everything from the **durable outbox** |
| `chaos_flaky_providers_and_repeated_worker_crashes_…` | flakiness + concurrency + *repeated* crash/restart, all converging exactly-once |
| `a_pushed_cancellation_rolls_back_…` | an unsolicited `CancelBooking` event mid-flight → compensation |
| `a_carrier_rejection_after_payment_refunds_…` | a permanent failure after payment settled → **refund** the captured payment |
| `a_failed_compensation_ends_needs_intervention` | a rollback that *itself* fails → `FailedTerminalNeedsIntervention` (a durable lock) |

Run it:

```sh
docker compose up -d --wait postgres
DATABASE_URL=postgres://… cargo run -p mire-sagas --example travel_booking_service
# the suite (gated on DATABASE_URL):
DATABASE_URL=postgres://… cargo test -p mire-sagas --test travel_booking_service
```

## Status

Pre-1.0. The API may shift before 1.0 (a `PgSagaRunner` → `SagaRunner` rename
is planned). `num_shards` is a deployment-wide constant; changing it requires a
documented migration. See the `tasks/` directory for the design notes and
roadmap.

[`IdempotencyKey`]: https://docs.rs/mire-sagas/latest/mire_sagas/struct.IdempotencyKey.html
