End-to-end: saga → outbox → inbox

The individual pieces — domain events, the outbox, sagas, the inbox — each solve one problem. This recipe puts them together in one runnable program: a checkout that reserves inventory, confirms an order, and ships it — reliably, across aggregates, with compensation if a step fails.

The full program lives at examples/recipes/order_fulfillment/ and runs in-process on the mock — no Docker. The Events & sagas chapter explains why each piece works; this is the assembled flow.

The aggregate announces a change¶

Order extends AggregateRoot, so it can record a domain event the moment its state transitions — declared next to the data it watches:

class Order(Document, AggregateRoot):
    status: str = "pending"

    def confirm(self) -> "OrderUpdate":
        """Decide the ``pending`` -> ``confirmed`` transition (illegal from any other status).

        A pure decision method on the aggregate: it guards the transition and returns the merge-patch
        the repository persists under the order's revision. The rule lives here once — the saga step
        and the ``_on_confirm`` emitter both key on it instead of re-encoding it.
        """

        if self.status != "pending":
            raise exc.domain(f"cannot confirm an order in status {self.status!r}")

        return OrderUpdate(status="confirmed")

    @event_emitter(fields={"status"})
    def _on_confirm(before, after: Self, diff: JsonDict) -> DomainEvent | None:  # type: ignore[no-untyped-def]
        if after.status == "confirmed" and before.status != "confirmed":
            return OrderConfirmed(aggregate_id=after.id)

        return None

A saga orchestrates the two aggregates¶

Checkout spans Inventory and Order. A saga runs the steps with a pivot: everything before the pivot is compensatable, the pivot commits the outcome. Reserve inventory first; confirming the order is the pivot.

def build_checkout_saga() -> SagaDefinition[CheckoutCtx]:
    return SagaDefinition(
        name="checkout",
        steps=(
            SagaStep(
                name="reserve",
                action=_reserve,
                compensation=_release,
                tx_route="mock",
            ),
            SagaStep(
                name="confirm",
                action=_confirm,
                kind=SagaStepKind.PIVOT,
                tx_route="mock",
            ),
        ),
    )

The pivot step: change state and stage the event together¶

Confirming the order trips its event emitter; the application layer dispatches that event inside the step's transaction, where the outbox bridge stages an integration event. State change and event are one atomic write:

async def _confirm(ctx: ExecutionContext, s: CheckoutCtx) -> CheckoutCtx:
    # Pivot. A pre-commit failure (e.g. payment declined) compensates `reserve`.
    if s.simulate_failure:
        log.error("payment declined — failing the pivot step")
        raise exc.domain("payment declined")

    # Load the aggregate, let it decide the transition, persist the patch under its rev (OCC).
    # confirm() single-sources the pending->confirmed rule (+ guards it); the resulting status
    # change fires @event_emitter -> OrderConfirmed, dispatched in THIS step's transaction (the
    # command flow), which the outbox bridge stages.
    orders = aggregate_repository(ctx, ORDER_SPEC)
    order = await orders.load(s.order_id)
    await orders.apply(order, order.confirm())
    # Transactional outbox: flush the staged event within the same transaction.
    await ctx.outbox.command(OUTBOX_SPEC).flush()

    log.info(
        "order confirmed — OrderConfirmed staged to outbox", order_id=str(s.order_id)
    )
    return s

The bridge that turns the domain event into a staged outbox row is wired once:

def build_context() -> ExecutionContext:
    # One declaration binds the outbox wiring: the domain-event -> outbox bridge (staging), plus —
    # for a broker deployment — the in-tx flush hook and the background relay step. Here we merge
    # just the staging bridge into the event registry; this example flushes and relays in-process.
    wiring = bind_outbox(
        OutboxEmit(
            spec=OUTBOX_SPEC,
            emits=(
                EmitMapping(
                    event=OrderConfirmed,
                    event_type="order.confirmed",
                    to_payload=lambda e: OrderConfirmedPayload(order_id=str(e.aggregate_id)),
                ),
            ),
        )
    )
    module = MockDepsModule(domain_events=wiring.domain_event_registry())
    return ExecutionContext(deps=DepsRegistry.from_modules(module).freeze().resolve())

Relay carries it to the consumer¶

A relay claims staged rows and delivers them downstream — standing in for a broker plus the outbox relay worker:

async def relay_once(ctx: ExecutionContext) -> list[RelayMessage]:
    query = ctx.outbox.query(OUTBOX_SPEC)
    claims = await query.claim_pending()

    messages = [
        RelayMessage(id=str(c.event_id), order_id=UUID(c.payload["order_id"]))
        for c in claims
    ]

    if claims:
        await query.mark_published([c.id for c in claims])

    log.info("relayed events from outbox", count=len(messages))
    return messages

The inbox makes delivery exactly-once¶

Relays are at-least-once: the same event can arrive twice. The consumer records each message id in the inbox and skips duplicates, so the Shipment is created once no matter how many times the event is delivered:

async def deliver(ctx: ExecutionContext, message: RelayMessage) -> bool:
    """Process one relayed message exactly-once. Returns False if it was a duplicate."""

    processed = await process_with_inbox(
        ctx,
        message,
        inbox_spec=INBOX_SPEC,
        handler=lambda m: _fulfill(ctx, m),
        tx_route="mock",
    )

    if not processed:
        log.debug("duplicate message skipped by inbox", id=message.id)

    return processed

Notes¶

Compensation is automatic. If the pivot fails, the saga runs the compensations for completed steps (here: release the reserved inventory) and stages nothing downstream.
The event commits with the write. Because the dispatch happens in the step's transaction, a published order.confirmed always corresponds to a committed order — never a phantom event, never a silent loss.
Swap the mock for real backends without touching this logic: a Postgres store + outbox, a real broker via OutboxRelay.to_queue, and the same inbox dedup. The orchestration is backend-agnostic.