# 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). Each step splits into a **`decide`** and an **`effect`** — the split
that keeps it crash-safe:

- **Orchestration** — the DAG you declare in `build()`: the steps, their
  `.after(...)` dependencies, terminal markers, and compensations. The *plan*.
- **`decide`** — a pure, **synchronous** `fn` (axum-style extractors) that folds
  state/deps into a typed request ([`Emit`]) or a plain value. It can't perform
  IO — it isn't `async` — so it's always safe to re-run on recovery. (A
  crash-unsafe inline effect simply won't compile.)
- **`effect`** — an `async fn(Req<R>, …)` that fulfils the request, co-located
  on the step via `.effect(handler)` (or shared across steps via
  `SagaBuilder::effect`). The runner auto-invokes it per emit and delivers the
  typed response back, mapped to the step's output by `.on_response(...)` — so
  a saga runs to terminal in one `trigger()` call when every effect is
  in-process. An emit with **no** registered effect suspends and awaits an
  *external* responder (durable across restarts): a `deliver` (callback, queue
  consumer, or the worker's outbox publisher) resumes it; `deliver_failure`
  fails 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.

## A typed saga

A two-step booking — `reserve → pay` — showing both regimes: `reserve` is
fulfilled **in-process** by a co-located effect (with a compensation), and `pay`
**suspends** until an external responder delivers the result. The events are
typed and carry stable `event_type` strings via `#[derive(EventData)]`; the
triggering `Order` aggregate and the `Inventory` client are elided.

```rust
use mire::EventData;
use mire_sagas::{
    comp, task, Dep, Emit, IdempotencyKey, Output, PgSagaRunner, Req, Saga,
    SagaBuilder, SagaWorker, Start, State,
};
use serde::{Deserialize, Serialize};

// Per-instance state + the typed output each step produces.
#[derive(Default, Clone, Serialize, Deserialize)] struct BookingState { id: String, item: String }
#[derive(Clone, Serialize, Deserialize)] struct Reserved { hold: String }
#[derive(Clone, Serialize, Deserialize)] struct Settled { payment: String }

// Request/response event pairs — the typed wire between a decide and its effect.
#[derive(Clone, Serialize, Deserialize, EventData)] #[mire(event_type = "inventory.reserve")]  struct ReserveItem { item: String }
#[derive(Clone, Serialize, Deserialize, EventData)] #[mire(event_type = "inventory.reserved")] struct ItemHeld { hold: String }
#[derive(Clone, Serialize, Deserialize, EventData)] #[mire(event_type = "inventory.release")]  struct ReleaseItem { hold: String }
#[derive(Clone, Serialize, Deserialize, EventData)] #[mire(event_type = "inventory.released")] struct ItemReleased {}
#[derive(Clone, Serialize, Deserialize, EventData)] #[mire(event_type = "payment.capture")]    struct CapturePayment { id: String }
#[derive(Clone, Serialize, Deserialize, EventData)] #[mire(event_type = "payment.captured")]   struct Captured { payment: String }

// `decide` — pure & SYNCHRONOUS: folds state/outputs into a request, performs no IO.
fn reserve(State(s): State<BookingState>) -> Emit { Emit::event(ReserveItem { item: s.item.clone() }) }
fn request_payment(State(s): State<BookingState>) -> Emit { Emit::event(CapturePayment { id: s.id.clone() }) }
fn release(Output(r): Output<Reserved>) -> Emit { Emit::event(ReleaseItem { hold: r.hold }) } // a comp is a decide too

// `effect` — async: does the IO, idempotent on the key the runner threads in.
// `Inventory` is your HTTP/SDK client, injected below via `.dep(...)`.
async fn hold_item(Req(r): Req<ReserveItem>, Dep(inv): Dep<Inventory>, key: IdempotencyKey) -> ItemHeld {
    ItemHeld { hold: inv.hold(&r.item, &key.token()).await }
}
async fn release_item(Req(r): Req<ReleaseItem>, Dep(inv): Dep<Inventory>, key: IdempotencyKey) -> ItemReleased {
    inv.release(&r.hold, &key.token()).await;
    ItemReleased {}
}

struct Booking;
impl Saga for Booking {
    type State = BookingState;
    fn name() -> &'static str { "booking" }
    fn build(b: SagaBuilder<Self>) -> SagaBuilder<Self> {
        // `Order` / `OrderPlaced`: the triggering aggregate + its event.
        b.start_on::<Order, _>(|e: &OrderPlaced| Some(Start {
            instance_id: e.id.clone(),
            state: BookingState { id: e.id.clone(), item: e.item.clone() },
        }))
        // decide emits `ReserveItem`; `hold_item` fulfils it in-process;
        // rollback (`release` + `release_item`) is co-located.
        .step("reserve", task(reserve)
            .effect(hold_item)
            .on_response(|h: ItemHeld| Reserved { hold: h.hold })
            .compensate(comp(release).effect(release_item).on_response::<ItemReleased>()))
        // no `.effect` → `pay` emits `CapturePayment` and SUSPENDS until an external
        // responder (webhook / queue / outbox publisher) `deliver`s `Captured`.
        .step("pay", task(request_payment).after("reserve")
            .on_response(|c: Captured| Settled { payment: c.payment })
            .terminal_success())
    }
}

// Build the topology (injecting the client), trigger on the domain event, and
// run a SagaWorker on every replica for timeouts / recovery / the outbox drain.
let runner = PgSagaRunner::new(Booking::build(SagaBuilder::for_saga()).dep(Inventory), store)
    .with_num_shards(64);
runner.trigger::<Order>(&order_placed).await?;
SagaWorker::new(runner).run(cancel).await?;
```

## Examples

Two runnable variants of the same travel-booking saga
(`flight ∥ hotel → payment → confirm`):

| Example | What it shows |
|---|---|
| [`travel_booking`](examples/travel_booking.rs) | The minimal in-process shape — pure sync `decide` steps with their async effects co-located via `.effect(...)`, auto-dispatched by the runner. Start here. |
| [`travel_booking_service`](examples/travel_booking_service.rs) | The flagship: real third-party providers (a mock **axum** backend over **reqwest**), a durable outbox, and a stress-test suite asserting exactly-once effects under crashes and flaky providers. |

## 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`: flight,
hotel, and payment are fulfilled **off the hot path** via the durable outbox
(the publisher POSTs the provider and `deliver`s the result back); `confirm` and
the compensations use in-process `.effect` handlers (so a delivered payment
tickets the carrier inline, and the cascade can roll back). External IO is
durable and survives crashes — the shape a real sharded deployment wants.

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
