Event-driven architecture

CloudEvents envelopes, a consumer loop with dedup and per-subject ordering, a transactional outbox, dead letters with safe replay, sagas, choreography and AsyncAPI -- generated from your charts.

Event-driven services built on state machines keep re-writing the same plumbing: a message envelope, a consumer loop that loads the right instance and survives redelivery, an outbox so a published event and the state change it describes cannot disagree, a dead-letter queue that is safe to replay, and sagas with compensation. xstate_statemachine.eda ships that plumbing in core, with zero dependencies, and the chart itself declares what is published, so the AsyncAPI document of your service is generated from the same JSON that runs it.

The real brokers (Redis Streams, Kafka, RabbitMQ, NATS JetStream and Amazon SQS) implement the BrokerAdapter protocol defined here. They are documented on Brokers, which includes a choosing a broker table. FakeBrokerAdapter lets you build and test the whole flow in-process with no broker at all. For Celery workers, Beat-driven after timers and an outbox relay task, see Celery.

Install

Nothing to install for the core: from xstate_statemachine.eda import .... The [cloudevents] extra only adds interop with the official CloudEvents SDK and its HTTP binary / structured helpers:

pip install "xstate-statemachine[cloudevents]"

Requires cloudevents>=1.10 (1.x and 2.x layouts are both supported). Tested versions are in the compatibility table.

For a complete, runnable app – an order chart and a warehouse chart in choreography, the SQLite outbox committed with the snapshot, inbox dedup, poison messages dead-lettered and listed with xsm dlq, and a test suite – see the eda_fulfilment example.

Quick start

A chart that publishes order.paid when it is paid, a consumer that turns inbound commands into events on the right persisted instance, and an outbox that stamps the causation chain:

import asyncio

from xstate_statemachine import MachineLogic, create_machine
from xstate_statemachine.eda import (
    Envelope,
    FakeBrokerAdapter,
    InboundDispatcher,
    OutboxPlugin,
)
from xstate_statemachine.persistence import MemoryInbox, MemoryStore

order = create_machine(
    {
        "id": "order",
        "initial": "open",
        "context": {
            "total": 0,
        },
        "states": {
            "open": {
                "on": {
                    "PAY": {
                        "target": "paid",
                        "actions": "setTotal",
                        "meta": {
                            "publish": {
                                "type": "order.paid",
                                "data": ["total"],
                            },
                        },
                    },
                },
            },
            "paid": {
                "type": "final",
            },
        },
    },
    logic=MachineLogic(
        actions={
            "setTotal": lambda i, ctx, e, a: ctx.update(
                total=e.payload["total"]
            ),
        }
    ),
)

broker = FakeBrokerAdapter()
dispatcher = InboundDispatcher(
    MemoryStore(),  # instances keyed by envelope.subject
    {
        "xsm.order.PAY": order,
    },  # CloudEvents type -> machine
    inbox=MemoryInbox(),  # redeliveries are answered, not re-run
    plugins=[OutboxPlugin(broker, topic="order-events")],
)


async def main() -> None:
    cmd = Envelope.new(
        type="xsm.order.PAY",
        subject="o-42",
        data={
            "total": 99,
        },
    )
    await broker.deliver("orders", cmd)
    await broker.deliver("orders", cmd)  # the broker redelivers
    result = await dispatcher.run_once(broker, "orders")
    assert (result.processed, result.duplicates) == (1, 1)

    [paid] = broker.published_on("order-events")
    assert paid.type == "order.paid" and paid.data == {
        "total": 99,
    }
    assert paid.causationid == cmd.id and paid.subject == "o-42"


asyncio.run(main())

Reference

Envelope

Envelope is a frozen CloudEvents 1.0 event: specversion, id, type, source, subject, time, datacontenttype, data, plus the extensions correlationid, causationid, machineid, machineversion and any further validated extensions.

• subject is the machine instance key and the partition key. Adapters preserve order per subject, never globally.
• id is sortable (new_id(): a UUIDv7-style value, 48-bit millisecond timestamp plus randomness, strictly increasing within a process). It is also the idempotency key.
• Envelope.new(type=..., subject=..., data=...) fills id and time. Envelope.from_transition(interpreter, type=..., cause=inbound) builds an outbound event whose causationid is the cause’s id and whose correlationid is inherited, so a whole conversation shares one correlation id.
• to_event() validates first, then returns the Event the machine receives: xsm.<machine>.<EVENT> becomes EVENT and data (which must be a JSON object) becomes the payload.
• to_json() / from_json() use CloudEvents structured mode. from_json checks the byte size before parsing (the same 1 MiB default as snapshots), then the shape. A wrong shape raises EnvelopeCorruptError; an oversized one raises EnvelopeTooLargeError. Both are XStateMachineErrors.
• Extension names that carry credentials (authorization, cookie, *token*, *secret*, apikey, password) are refused, and traceparent must match the W3C Trace Context grammar.
• with_attempt(n) / .attempt carry the delivery-attempt counter (xsmattempt) used for poison handling.

With the extra, xstate_statemachine.contrib.cloudevents adds to_cloudevent() / from_cloudevent() (SDK objects) and to_binary() / to_structured() / from_http(headers, body) (HTTP modes). from_http only reads ce-* headers and drops credential-bearing ones, so an Authorization header can never become an extension.

Broker adapters

BrokerAdapter (async)            SyncBrokerAdapter (threads, Celery, Django)
  await publish(topic, env)        publish(topic, env)
  async for d in subscribe(topic)  for d in subscribe(topic)
  await ack(d)                     ack(d)
  await nack(d, requeue=bool)      nack(d, requeue=bool)

A Delivery is (envelope, topic, ack, nack). The contract is FIFO per subject, explicit settlement, requeue to the head, settling twice is a no-op, and a failed publish raises. tests/eda/contract.py is the suite every adapter runs.

FakeBrokerAdapter / SyncFakeBrokerAdapter are in-memory implementations for tests. Beyond the contract they record published, accept inbound traffic with deliver(), inject failures with fail_next_publish(), and run handlers synchronously with on(topic, handler) + drain(). from xstate_statemachine.contrib.testing import FakeBrokerAdapter also works (the [testing] extra re-exports it next to replay() and assert_replay_consistent()).

Inbound dispatcher

InboundDispatcher(store, machine_for_type, *, lock=OptimisticLock(), plugins=(), inbox=None, max_in_flight=16, max_attempts=5, dead_letters=None, ...) is the consumer loop every adapter reuses: validate the envelope, find the machine for its type, then lock.run(store, key=subject, machine, send(envelope.to_event())), then ack.

run_once(broker, topic) drains what is there now; run_forever(broker, topic, stop_event) loops; run_once_sync(...) is the sync twin; handle(envelope) processes one.

Outbox

What is published is declared in the chart:

• "meta": {"publish": {"type": "order.paid", "data": ["total"]}} on a transition: data copies those context fields. "publish": "order.paid" (type only) and "publish": true (type xsm.<machine>.transition.<target>) also work. A malformed value fails loudly.
• "tags": ["publish"] on a state: entering it publishes xsm.<machine>.transition.<state> with the whole context, redacted.

OutboxPlugin(sink, topic="events") builds the envelopes during the step and hands them to the sink once the event has settled. The sink is either:

• an OutboxStore, which is transactional when it shares the state store’s transaction: SQLiteOutboxStore(sqlite_store) (zero-dep) or SQLAlchemyOutboxStore(sqlalchemy_store) ([sqlalchemy]). persisted() writes the rows right after the snapshot save; under PessimisticLock the lock is the transaction, so a rollback drops the rows with the snapshot. OutboxRelay(store, broker).relay_once() then publishes pending rows in order and marks them sent only after the broker accepted them (at-least-once);
• a BrokerAdapter directly. Inside persisted() the publish is held until the block, including a PessimisticLock transaction, has exited cleanly, so a rolled-back state is never announced. It is still outside any transaction: a crash between the commit and the publish loses the message, so it is at-most-once-ish. Use it for tests and non-critical notifications.

publish_specs(machine) lists every publication a chart declares.

Several relays on one outbox (leases)

You can run one relay per replica. OutboxRelay(store, broker, *, batch=100, owner=None, lease_s=DEFAULT_CLAIM_LEASE_S) does not read pending() when the store offers claim(). SQLiteOutboxStore, SQLAlchemyOutboxStore and MemoryOutboxStore all do. Instead, each run leases up to batch pending rows to its owner for lease_s seconds (default 30.0), using the claimed_by / claimed_until columns. A second relay skips rows under a live lease, so two relays never publish the same row at the same time. A relay whose broker raises hands its unsent rows back at once with release(). A relay that dies cannot release, so its rows are picked up again when the lease expires. owner defaults to host:pid:id(relay).

from xstate_statemachine.eda import (
    Envelope, MemoryOutboxStore, OutboxRelay, SyncFakeBrokerAdapter,
)

outbox, broker = MemoryOutboxStore(), SyncFakeBrokerAdapter()
for n in range(10):
    outbox.add("orders", Envelope.new(type="order.paid", subject=f"o-{n}"))

# relay A claimed 4 rows and is still publishing them (or just died)
held = outbox.claim(limit=4, owner="relay-a", lease_s=30)

relay_b = OutboxRelay(outbox, broker, owner="relay-b", lease_s=30)
assert relay_b.relay_once_sync() == 6              # B skips A's leased rows
assert {e.subject for e in broker.published_on("orders")}.isdisjoint(
    {r.envelope.subject for r in held}
)
outbox.release([r.seq for r in held], owner="relay-a")   # A failed: hands back
assert relay_b.relay_once_sync() == 4               # nothing published twice
assert len(broker.published_on("orders")) == 10

Set lease_s well above the time one batch takes to publish (batch × the broker’s worst publish latency). If a lease expires while its owner is still publishing, a second relay publishes the same rows too. That is a duplicate, not a loss, and the consumer’s inbox dedups it. A store without claim() (your own OutboxStore) falls back to pending(), so run one relay per outbox there.

Dead letters

Dead letters from both sources share one record, DeadLetter:

• chart-driven: patterns.DeadLetterPlugin captures entry into a dead-letter-tagged state (see Patterns), with the error chain and a redacted snapshot;
• envelope-driven: the dispatcher’s max_attempts, unknown_event and corrupt cases, with the redacted envelope.

Every record carries the machine’s structure_hash and version so a replay can refuse a different machine. Stores implement DeadLetterStore (put, get, list, mark_resolved, delete, purge_older_than): MemoryDeadLetterStore and SQLiteDeadLetterStore (the default; it redacts again on write and keeps an audit table). BrokerDeadLetterSink(broker, "orders", store=...) publishes each record to orders.dlq as xsm.deadletter and can also keep it in a store; pass it as the plugin’s sink.

Operate them with the CLI:

xsm dlq --dlq sqlite:///dlq.db list [--all] [--json]
xsm dlq --dlq sqlite:///dlq.db show <id>
xsm dlq --dlq sqlite:///dlq.db replay <id> --store sqlite:///state.db \
    --machine order.json --logic myapp.order_logic --reason "fixed in 1.4.2"          # dry run
xsm dlq --dlq sqlite:///dlq.db replay <id> ... --no-dry-run --yes --reason "..."      # for real
xsm dlq --dlq sqlite:///dlq.db purge --older-than 30d --yes --reason "retention"

replay is a dry run by default. A real replay needs --no-dry-run --yes --reason. It reuses the envelope id, so the inbox deduplicates a double replay. It refuses when the machine’s structure or version changed since capture unless you pass --force, marks the record resolved on success, and writes an audit row (who, why, when, outcome). replay_dead_letter() is the same operation from Python.

Sagas

patterns.SagaBuilder turns an orchestrated saga into plain chart JSON: nothing is interpreted at runtime, the result is editable in Stately, and it is strictConfig-clean.

from xstate_statemachine import (
    MachineLogic,
    SimulatedClock,
    SyncInterpreter,
    create_machine,
)
from xstate_statemachine.patterns import RetryPolicy, SagaBuilder

saga = (
    SagaBuilder("fulfil")
    .step(
        "reserve",
        invoke="reserveStock",
        compensate="releaseStock",
        timeout_ms=5000,
        retry=RetryPolicy(max_attempts=3, jitter="none"),
    )
    .step("charge", invoke="chargeCard", compensate="refundCard")
    .step("ship", invoke="ship")
    .on_failure("notifyOps")
)
config = saga.build()  # a dict you can json.dump and open in Stately

calls = []


def service(name, fail=False):
    def run(i, ctx, e):
        calls.append(name)
        if fail:
            raise RuntimeError(name)
        return f"{name}-ok"

    return run


logic = saga.logic().merge(
    MachineLogic(
        services={
            "reserveStock": service("reserve"),
            "releaseStock": service("release"),
            "chargeCard": service("charge"),
            "refundCard": service("refund"),
            "ship": service("ship", fail=True),
        },
        actions={
            "notifyOps": lambda i, ctx, e, a: None,
        },
    )
)
interp = SyncInterpreter(
    create_machine(config, logic=logic), clock=SimulatedClock()
).start()
assert interp.current_state_ids == {"fulfil.failed"}
assert calls == [
    "reserve",
    "charge",
    "ship",
    "refund",
    "release",
]  # reverse, once each
assert interp.context["compensated"] == ["charge", "reserve"]

The generated shape: steps.<name> invokes the step’s service, with an after timeout and onError both leading to failure. A failure at step k enters compensating.<k-1>, then walks back to compensating.<first> and on to failed. Success leads to completed. A step with retry= fails into steps.<name>Retrying, which waits the policy’s delay and re-enters the step while retries remain. A compensation that fails ends in compensationFailed, tagged dead-letter so DeadLetterPlugin captures it. Results are in context.results, compensated steps in context.compensated and the failure in context.error. Every step transition carries meta.publish (fulfil.charge.completed, .failed, .compensated), so with an OutboxPlugin the saga emits its own integration events.

To make completion idempotent when the upstream event may be re-minted with a new id, give the dispatcher inbox= and dedup_key=lambda e: e.causationid or e.id.

A saga driven from the bus. In production a saga is one persisted instance per business key, started by a command. Build it with start_event=, map xsm.<saga>.<START> to it in an InboundDispatcher, and give it the four stores. Two rules a newcomer trips over:

• START’s payload lives in context["input"]. Each step’s service receives its own invoke.<step> event, not the command – read the data the saga was started with from ctx["input"] (the start event’s payload, copied by the generated sagaStart action; None until START). The instance’s identity is the envelope subject, which the dispatcher sets as interp.store_key.
• compensationFailed is final. A failing compensation parks the instance there, and DeadLetterPlugin writes a chart-state dead letter. That record is the operator’s ticket: fix the cause by hand. A redelivered START does nothing (the instance is done).

from xstate_statemachine import MachineLogic, create_machine
from xstate_statemachine.eda import (
    DeadLetterPlugin,
    Envelope,
    InboundDispatcher,
    MemoryDeadLetterStore,
    MemoryOutboxStore,
    OutboxPlugin,
    SyncFakeBrokerAdapter,
)
from xstate_statemachine.patterns import SagaBuilder
from xstate_statemachine.persistence import MemoryInbox, MemoryStore

saga = (
    SagaBuilder("fulfil", start_event="START")
    .step("reserve", invoke="reserveStock", compensate="releaseStock")
    .step("charge", invoke="chargeCard")
)
seen = []


def reserve(i, ctx, e):
    seen.append(i.store_key)  # the order id comes from the subject
    return {
        "hold": f"h-{i.store_key}",
    }


def charge(i, ctx, e):
    raise RuntimeError("card declined")


def release(i, ctx, e):  # idempotent: a restart may re-run it
    raise RuntimeError("warehouse API down")


machine = create_machine(
    saga.build(),
    logic=saga.logic().merge(
        MachineLogic(
            services={
                "reserveStock": reserve,
                "chargeCard": charge,
                "releaseStock": release,
            },
        )
    ),
    strict_config=True,
)
store, bus, dlq = (
    MemoryStore(),
    SyncFakeBrokerAdapter(),
    MemoryDeadLetterStore(),
)
dispatcher = InboundDispatcher(
    store,
    {
        "xsm.fulfil.START": machine,
    },
    inbox=MemoryInbox(),
    plugins=[
        OutboxPlugin(MemoryOutboxStore(), topic="sagas"),
        DeadLetterPlugin(dlq),
    ],
    dead_letters=dlq,
    on_unknown="ignore",
)
start = Envelope.new(type="xsm.fulfil.START", subject="o-42")
bus.publish("commands", start)
dispatcher.run_once_sync(bus, "commands")

assert seen == ["o-42"]
[letter] = dlq.list()  # the operator's ticket
assert letter.state_id == "fulfil.compensationFailed"
assert [e["message"] for e in letter.errors] == [
    "card declined",
    "warehouse API down",  # cause, then the stuck undo
]
bus.publish("commands", start)  # redelivered: a no-op
assert dispatcher.run_once_sync(bus, "commands").duplicates == 1
assert len(dlq.list()) == 1

context.error holds the last failure (here the compensation’s); the dead letter’s errors keeps the whole sequence.

examples/integrations/eda_fulfilment/tests/test_battle_295_scenario.py runs the same shape at scale (SagaApp): 90 SQLite-persisted sagas, a third failing at a different step, plus a kill -9 in the middle of a compensation.

Choreography

In choreography there is no orchestrator: each machine reacts to the integration events it cares about. patterns.ChoreographyRouter is an InboundDispatcher with a type → (machine, EVENT) mapping, instances stored as <machine id>:<subject> so several machines can share one business key, and unrouted types acked (a shared bus carries other services’ events).

import asyncio

from xstate_statemachine import create_machine
from xstate_statemachine.eda import Envelope, FakeBrokerAdapter, OutboxPlugin
from xstate_statemachine.patterns import ChoreographyRouter
from xstate_statemachine.persistence import MemoryInbox, MemoryStore

order = create_machine(
    {
        "id": "order",
        "initial": "new",
        "states": {
            "new": {
                "on": {
                    "PLACE": {
                        "target": "awaiting",
                        "meta": {
                            "publish": "order.placed",
                        },
                    },
                },
            },
            "awaiting": {
                "on": {
                    "PAID": {
                        "target": "completed",
                        "meta": {
                            "publish": "order.completed",
                        },
                    },
                },
            },
            "completed": {
                "type": "final",
            },
        },
    }
)
payment = create_machine(
    {
        "id": "payment",
        "initial": "idle",
        "states": {
            "idle": {
                "on": {
                    "CHARGE": {
                        "target": "paid",
                        "meta": {
                            "publish": "payment.captured",
                        },
                    },
                },
            },
            "paid": {
                "type": "final",
            },
        },
    }
)

bus, store = FakeBrokerAdapter(), MemoryStore()
router = ChoreographyRouter(
    store,
    {
        "xsm.order.PLACE": order,
        "order.placed": (payment, "CHARGE"),
        "payment.captured": (order, "PAID"),
    },
    plugins=[OutboxPlugin(bus, topic="events")],
    inbox=MemoryInbox(),
)


async def main() -> None:
    cmd = Envelope.new(
        type="xsm.order.PLACE", subject="o-1", correlationid="c-1"
    )
    await bus.deliver("events", cmd)
    await router.run_until_quiet(bus)
    chain = {e.type: e for e in bus.published}
    assert chain["order.placed"].causationid == cmd.id
    assert chain["payment.captured"].causationid == chain["order.placed"].id
    assert chain["order.completed"].causationid == chain["payment.captured"].id
    assert {e.correlationid for e in bus.published} == {"c-1"}


asyncio.run(main())

Orchestration or choreography? Use a saga when one business process owns the steps, needs compensation in a defined order, and someone must be able to answer “where is order 42?” by looking at one instance. Use choreography when services are owned by different teams and each only needs to react to facts, not to a plan. Choreography scales organisationally but spreads the process across machines, so the causation chain is the only thing that ties it together. Keep correlationid on every envelope. Choreography has no compensation of its own: an undo is just another event each service must handle, and nothing guarantees the order. A cycle between two routes is not caught by the engine’s RunawayChainError (that guards one machine’s internal chain); run_until_quiet stops it after max_rounds with a RuntimeError, and in production it simply loops on the bus.

AsyncAPI

asyncapi_document(machine, server={"host": ..., "protocol": ...}, inbound_topic=..., outbound_topic="events") returns an AsyncAPI 3.0.0 document. It lists the events the chart consumes (its on keys, as xsm.<machine>.<EVENT> CloudEvents) and the events it publishes (from publish_specs), with CloudEvents message payloads. validate_asyncapi(doc) checks it against the AsyncAPI 3.0.0 JSON Schema, which is vendored in the package so validation works offline (it needs jsonschema). From the CLI:

xsm asyncapi order.json -o asyncapi.json --server localhost:9092 --protocol kafka --validate
xsm docs order.json      # the generated page now has an "Integration events" section

Every other public name

The rest of xstate_statemachine.eda.__all__, for readers who grep. Everything below imports from xstate_statemachine.eda.

Guarantees

What this does: at-least-once delivery plus the inbox gives effectively-once transitions: a redelivered envelope is answered with the original receipt and never re-runs actions. Per-subject order is preserved within a consumer. A failed delivery is not committed and is retried, then dead-lettered and acked after max_attempts. With an OutboxStore sharing the state store’s transaction under PessimisticLock, the outbox rows commit or roll back with the snapshot (tests/eda/test_outbox.py::TestSQLiteTransactional::test_forced_failure_after_the_write_leaves_no_row, tests/contrib/sqlalchemy/test_sqlalchemy_outbox.py::TestSQLAlchemyOutbox::test_forced_rollback_leaves_no_row). This fills steps 4–5 of the order of operations.

What this does not do: no exactly-once publish. The relay publishes then marks sent, so a crash in between publishes twice; consumers must dedup on the envelope id (the dispatcher does). Several relays may drain one outbox: row leases (claim() / lease_s) stop two live relays publishing the same row at once, but they do not make publishing exactly-once. A relay that dies mid-batch, or one that is still publishing when its lease expires, leaves rows another relay publishes again – the duplicate window is one batch per expired lease, so size lease_s above the slowest batch you expect (batch × the broker’s worst publish latency). A custom OutboxStore without claim() still needs one relay per outbox. The direct-broker sink is at-most-once-ish. There is no global order across subjects, and the leases do not order rows across relays. An OptimisticLock outbox is only transactional on stores whose save and outbox write share one transaction.

Sagas (SagaBuilder): when step k fails (its service raises, or its timeout_ms fires), the compensations of the completed steps run in reverse, once each per run: entering a compensating state runs its invoke once. Step results are in context.results[<step>], compensated steps in context.compensated, the last failure in context.error. A step with retry= re-runs after the policy’s delay up to max_attempts, then compensates; a timeout counts as a failed attempt. The timeout does not cancel the service’s work in an external system: the invoke is abandoned, so a late success is lost and the compensation must cope with a step that may have half-happened. Compensations must be idempotent: a process killed mid-compensation restarts the saga from its last snapshot and runs that compensation again. A failing compensation ends in compensationFailed, which is final: no retry, and a chart-state dead letter is written for a human. Services receive their own invoke.<step> event; the START event’s payload is in context.input (present in both chart shapes, None until START or for a saga that starts immediately) and identity comes from the instance key (interp.store_key). Upgrading a saga built by an earlier SagaBuilder: the generated chart gained the sagaStart action on the START transition and an input context key, so its structure hash changed – a snapshot saved from the older chart does not load against the new one without verify_machine_hash=False (or expected_machine_hash=<old>); drain in-flight sagas before upgrading, or restore them once that way. Under OptimisticLock a replica that loses the save race has already run the step’s service (at-least-once): use PessimisticLock or idempotent services.

See the programme-wide Guarantees and Security pages (#303).

Threat model

Who can call this: anyone who can publish to your topics. An envelope is untrusted input.

What it exposes: envelopes are size-capped before parsing and shape-validated before to_event() (X0.4); a malformed or unknown envelope is dead-lettered, never raised. Credential-bearing extension names are refused, transport headers other than ce-* never become extensions, and traceparent is validated (X0.8). Poison messages cannot loop: an attempt counter dead-letters and acks after max_attempts, and max_in_flight bounds concurrency (X0.8). Dead letters, state-tag data and audit details are redact()ed before they are written; SQLiteDeadLetterStore / SQLiteOutboxStore files are created 0600 (X0.5).

You must configure: an inbox in production (without one, redeliveries re-run); broker-level authentication and ACLs on who may publish which type; a source your consumers trust as the idempotency principal; DLQ replay only through --no-dry-run --yes --reason by an operator with access to the audit table; retention (xsm dlq purge --older-than, SQLiteOutboxStore.purge_sent).

Sagas and choreography on a shared bus: a forged done.invoke.<step> / error.platform.<step> envelope cannot complete or fail a saga step. With the usual {type: machine} mapping such a type is unknown (acked or dead-lettered per on_unknown). Even a catch-all mapping that turns xsm.fulfil.done.invoke.reserve into the event done.invoke.reserve only delivers user traffic, and the engine drives onDone / onError / after from engine-minted events alone (provenance, #195/#203) — tests/patterns/test_battle_295_b_saga_input.py::TestForgedCompletions. Event names are a namespace, not a secret: anyone allowed to publish xsm.fulfil.START can start a saga, so the ACL on that type is the authorisation. Prefix saga and choreography types with your service (fulfil.*, order.*) so two teams on one bus cannot collide, and keep on_unknown="ignore" only on topics that really are shared.

Compatibility

Troubleshooting

Operations

Sizing. One dispatcher handles one subject at a time and up to max_in_flight (16) subjects concurrently. Scale out with more consumers on a partitioned topic: the broker keeps per-subject order, the inbox keeps redeliveries effectively-once. Run one OutboxRelay per replica if you like. With batch=100, a relay publishes at most 100 rows per relay_once, so schedule it often enough that batch / interval exceeds your peak publish rate.

Lease tuning. Use lease_s ≥ 3 × the slowest batch you have seen (batch × p99 publish latency). If it is too short, a slow relay’s rows are published twice (duplicates, which the consumers dedup). If it is too long, rows held by a crashed relay wait that long. Give every relay a stable owner (e.g. the pod name), so a restarted pod reclaims its own rows at once (claim() also returns rows already leased to the same owner).

Metrics to alert on.

Sagas in production. A saga is a normal dispatcher-driven machine, so it scales like one: one dispatcher per topic partition, not one per saga key. Every instance lives in the shared store under its subject, the broker keeps per-subject order, and the PessimisticLock serialises the rare overlap. Run as many replicas as the topic has partitions. Step services run inside the dispatcher’s delivery, so a slow step holds that subject’s slot (one of max_in_flight) until it finishes or its timeout_ms fires. Give every step that calls a remote system a timeout_ms. Choreography is the same: one ChoreographyRouter per consumer group; its instances are keyed <machine id>:<subject>.

Retention. Purge sent outbox rows with SQLiteOutboxStore.purge_sent(older_than_s=86400). Purge resolved dead letters with xsm dlq --dlq … purge --older-than 30d --yes --reason retention (the purge is audited).

xsm dlq exit codes. 0 success. 1 a real replay ran and the outcome was not processed / duplicate. 2 refused input: a missing or non-SQLite --dlq / --store file (never created), an unsupported URL scheme, an unknown id, a bad --older-than / --limit, an unloadable --machine or --logic, a missing guard-rail flag (--yes, --reason), or a refused replay. Errors are one line on stderr, so --json stdout stays parseable.