Transactional outbox

You can't atomically write to your database and publish to a broker — a crash between the two loses or duplicates the event. The outbox makes it one write: stage the event in the same transaction as the business change, then a relay moves staged rows to the broker afterwards. The concept is in Events & sagas; this is the wiring.

The runnable version lives at examples/recipes/outbox/ and runs on the in-memory mock — no broker needed.

The event and its destination¶

An OutboxSpec carries the payload codec and names the queue it relays to — the destination route must equal the QueueSpec.name:

class OrderPlaced(BaseModel):
    order_id: str


# The outbox spec names its destination queue; its route must equal the queue's name.
ORDER_EVENTS = OutboxSpec(
    name="order-events",
    codec=PydanticModelCodec(OrderPlaced),
    destination=OutboxDestination.queue(route="orders", channel="orders"),
)
ORDERS_QUEUE = QueueSpec(name="orders", codec=PydanticModelCodec(OrderPlaced))

Stage it with the write¶

Inside the business transaction, stage the event and flush — it commits (or rolls back) together with the write, so a published event always corresponds to a committed change:

async def place_order(ctx: ExecutionContext, order_id: str) -> None:
    # Your business write goes here, in a transaction. Stage the integration
    # event in the same unit of work, then flush — it commits with the write.
    outbox = ctx.outbox.command(ORDER_EVENTS)
    await outbox.stage("order.placed", OrderPlaced(order_id=order_id), event_id=uuid4())
    await outbox.flush()

In a real handler you'd attach outbox_flush_tx_on_success_factory to the operation so the flush fires automatically on transaction success, rather than calling flush() by hand.

A flush closes the route for the rest of its transaction: staging after it raises, savepoints included, because nothing would write those events. The next root transaction reopens the route, so one task can run one unit of work after another. Events still staged when a transaction commits without a flush stay staged, and the task's next flush writes them. A transaction or savepoint that rolls back drops the events it staged and stages again the ones its flush wrote. On Mongo a nested scope is not a savepoint, so its writes, events included, stay in the transaction, and that second flush relies on the outbox skipping an event id it already holds.

This follows the scopes ctx.tx_ctx.scope opens, as operations do. A transaction opened on the raw manager, ctx.transaction(route).transaction(), bypasses it: nothing reopens the route or restores what was staged.

Staging belongs to the task that stages. A task started from another, with asyncio.gather or create_task, begins with a copy of what that task had staged, and what it stages afterwards never reaches the parent, so flush in the task that stages.

Relay to the broker¶

A relay claims staged rows and publishes them to the queue, returning what it did:

async def relay(ctx: ExecutionContext) -> int:
    # In production this runs in the background (outbox_relay_background_lifecycle_step);
    # here we drive one pass. It claims staged rows and publishes them to the queue.
    result = await OutboxRelay(outbox_spec=ORDER_EVENTS).to_queue(ctx, ORDERS_QUEUE)
    return result.published

In production the relay runs continuously as a lifecycle step:

from datetime import timedelta
from forze.application.execution import LifecyclePlan
from forze_kits.integrations.outbox import outbox_relay_background_lifecycle_step

lifecycle = LifecyclePlan.from_steps(
    outbox_relay_background_lifecycle_step(
        outbox_spec=ORDER_EVENTS,
        queue_spec=ORDERS_QUEUE,        # required for the queue transport
        interval=timedelta(seconds=5),
    ),
)

Waking the relay on commit¶

A polling relay delivers a row up to one interval after it commits. Shortening the interval adds an idle claim query on every pass. Give the route a wake instead: each flush that wrote rows publishes a small signal on a pub/sub topic once its transaction commits, and the relay runs a pass as soon as it hears it.

from forze.application.contracts.outbox import OutboxWake

ORDER_EVENTS = OutboxSpec(
    name="order-events",
    codec=PydanticModelCodec(OrderPlaced),
    destination=OutboxDestination.queue(route="orders", channel="orders"),
    wake=OutboxWake(route="outbox-wake"),  # a pub/sub route, e.g. RedisPubSubConfig
)

The signal names only the route, so it can never carry or lose a row. A rollback publishes nothing, and a wake is never published before its rows are visible. A lost wake, a broken pub/sub connection, or a publish that outlasts publish_timeout (one second by default) leaves the row to the next poll, so keep the interval as the fallback. Wire the wake's pub/sub route in every process that flushes the route and in the one that relays it. The route must be tenant-global: a tenant-aware one, or one on a client routed per tenant, is refused. Under a declared tenant-isolation floor, mark it RedisPubSubConfig(tenant_global=True) so the floor leaves it out.

With wakes, passes follow commits, so the stale-claim reclaim would run on each of them. It runs at most once per reclaim_every, which defaults to once per poll tick.

Consuming on the other side¶

QueueConsumer is the consumer-side counterpart — it replaces the hand-rolled consume → dedupe → ack/nack loop with the decisions already made correctly. Per message it: parks handler-poison (opt-in max_deliveries), runs the handler exactly-once through the inbox (process_with_inbox, same dedup transaction, correlation rebound from the envelope headers), acks both fresh and duplicate deliveries — a redelivered already-processed message must leave the queue — and nacks handler failures back (requeue=True) for redelivery. One message's failure never kills the consumer.

from datetime import timedelta

from forze_kits.integrations.consumer import QueueConsumer

consumer = QueueConsumer(
    queue="orders",                # the channel the relay published to
    queue_spec=ORDERS_QUEUE,
    handler=handle_order_event,    # async def (message: QueueMessage[OrderEvent]) -> None
    inbox_spec=ORDERS_INBOX,
    tx_route="postgres",           # dedup mark + handler commit together here
)

result = await consumer.run(ctx, timeout=timedelta(seconds=5))  # idle timeout; None = forever
# result.processed / result.duplicates / result.parked / result.failed

In production it runs continuously as a lifecycle step — one step per queue (no in-process concurrency knob; scale out with more steps or processes):

from forze_kits.integrations.consumer import queue_consumer_background_lifecycle_step

lifecycle = LifecyclePlan.from_steps(
    queue_consumer_background_lifecycle_step(
        queue="orders",
        queue_spec=ORDERS_QUEUE,
        handler=handle_order_event,
        inbox_spec=ORDERS_INBOX,
        tx_route="postgres",
    ),
)

A crash of the consume stream itself (broker connection loss) is logged and the consume restarts after restart_backoff (default 5s); unacked in-flight messages redeliver and the inbox dedupes them.

Two kinds of poison, two owners:

• Decode-poison (payload doesn't fit the codec model) never reaches your handler — the queue adapters reject it inside consume with nack(requeue=False) (RabbitMQ DLX, SQS redrive) and keep consuming.
• Handler-poison (decodes fine, handler always fails) is parked by the runner when max_deliveries is set: a message whose delivery_count exceeds it is nack(requeue=False)-ed without running the handler, so the handler gets at most max_deliveries attempts.

Transient blips can also be retried in-process before the message goes back to the broker: pass retry_policy="my-policy" and the runner wraps each process step (dedup mark + handler, one fresh transaction per attempt) in ctx.resilience().run(...) under that named policy.

Failures and retries¶

The relay classifies a failed row by where it arose. A poison row (the payload can't decode) can never publish, so it's marked failed immediately — fix the cause and re-drive with ctx.outbox.query(spec).requeue_failed([id]). A transient failure (the publish call raised) is rescheduled with exponential backoff + jitter and retried, becoming failed only after max_attempts (default 5). One row's failure never blocks the rest of the batch. Set publish_timeout on the relay (or the background step) and a publish that outlasts it is a transient failure too.

Per-aggregate ordering¶

Stage with an ordering_key (typically the aggregate id) and the relay publishes it as the transport key instead of the event id:

await ctx.outbox.command(ORDER_EVENTS).stage(
    "order.shipped", payload, ordering_key=str(order_id),
)

On transports that honor key for partitioning (SQS FIFO MessageGroupId, stream partition keys), same-key events deliver in staged order on the happy path.

Table schema¶

The outbox table is application-owned — you create and migrate it. The full DDL, indexes, migration steps, and the optional Hybrid Logical Clock ordering column (for causal claim order across replicas) are in Outbox table schema, for both Postgres and Mongo.

Deleting what the relay publishes (Postgres)¶

With storage="consume" the Postgres store deletes a row once it is published, rather than keeping it as published for a retention job, and runs each relay pass in one transaction:

from datetime import timedelta
from forze_postgres.execution.deps.configs import PostgresOutboxConfig

outbox = PostgresOutboxConfig(
    relation=("app", "outbox"),
    storage="consume",
    pass_deadline=timedelta(seconds=5),
)

A pass holds one pooled connection, and the row locks on its batch, for up to pass_deadline, plus its last mark statements. Size the deadline to the batch: max_claim_rows (100) times the destination's slow-end publish latency should fit well inside it. For a stream XADD of a few milliseconds that is well under a second. A pass that reaches its deadline commits what it published, and the rows it did not reach wait for the next pass. Other relays skip the locked rows rather than wait for them. What the mode writes, its duplicate window and how to move a live table over are in Consume storage. To keep "was this event delivered" answerable once its row is gone, add a delivery log.

Notes¶

• Store the outbox where you store the data so the stage shares the transaction — PostgresOutboxConfig(relation=("app", "outbox")) (from forze_postgres.execution.deps.configs) or MongoOutboxConfig.
• At-least-once. The relay can publish a row twice (claim, publish, crash before marking). Consumers dedupe with the inbox.
• The background lifecycle step drains the whole backlog each tick (batches until a short claim, capped at max_batches_per_tick=100), then sleeps interval.

Draining on shutdown¶

By default the relay task is cancelled at shutdown, so rows staged just before the process goes away stay pending until a later process claims them — up to interval later, or until the next deploy. Set drain_on_shutdown=True to publish what is still claimable first:

relay = outbox_relay_background_lifecycle_step(
    outbox_spec=OUTBOX,
    queue_spec=JOBS,
    drain_on_shutdown=True,
)

The drain burns exactly one delivery attempt per row — the same blast radius as one ordinary tick. It ends the moment a batch reschedules a row, so it can never re-claim (and re-attempt) what it just parked; a failing destination stops it instead of being hammered; and it never opens a batch it does not expect to finish inside shutdown_drain_timeout (default 5s). Anything it cannot reach stays pending for the next process — exactly where it would have been without the drain.

The drain needs the database, and it gets it: the runtime stops every background loop before lifecycle teardown begins (see below), so the client is still open. You do not have to declare any ordering for it.

Two things still constrain it:

• pubsub is rejected at wiring. It is at-most-once past the broker, so publishing while subscribers are going away turns a delayed delivery into a lost one. Leaving the rows pending is strictly safer there.
• Keep shutdown_drain_timeout under the runtime's shutdown_step_timeout (default 10s), which bounds the whole loop-stopping pass. A batch cut mid-flight leaves rows processing until reclaim_stale_after elapses — consider lowering that when draining.

The drain is a best-effort teardown courtesy, not a delivery guarantee: correctness still rests on the relay claiming those rows eventually, from this process or the next.

How background loops shut down¶

The relay is one of five background loops the framework runs — the others are the queue consumer, the commit-stream consumer, and the durable recovery and scheduler pollers. Each registers itself with the scope's ctx.drainables at startup, and runtime.shutdown() asks every one of them to stop between units of work, concurrently, before lifecycle teardown begins:

1. the drain gate stops admitting operations and waits for the in-flight ones;
2. background loops stop — the relay drains, consumers finish and ack/commit what they hold, the durable pollers finish their sweep;
3. detached background work is cancelled;
4. lifecycle teardown closes the clients.

Step 2 sitting before step 4 is what lets a loop's graceful stop use its database. It also means a loop is never cancelled mid-unit unless it overruns its grace — which matters most for the commit-stream consumer, where a cancelled run never commits its offsets and every message it just handled is redelivered.

A loop that will not stop within the grace is cancelled and logged, so one wedged loop can never hang process exit.

Is the outbox drained?¶

ctx.outbox.admin(spec) is a read-only view of a route's backlog. It exists because emptiness used to be observable only through claim_pending — which claims, so asking the question changed the answer and raced the relay for rows the caller did not want.

admin = ctx.outbox.admin(OUTBOX)

await admin.has_undrained()      # bool — one index seek; poll this
await admin.depth()              # OutboxDepth(pending=, processing=, failed=)
await admin.oldest_pending_age() # timedelta | None — "is the relay stuck?"

Three semantics worth knowing, because they are deliberately not the claim path's:

• pending includes rows parked for a future retry. claim_pending hides them; a row backing off is still undelivered work, and a check that ignored it would call an outbox empty while events were queued behind a retry.
• published is never counted. Nothing prunes published rows, so that bucket grows with every event the application has ever emitted — counting it would scan the whole history. depth() reports only the undrained buckets.
• failed is reported apart and never waited on. Failed rows are terminal until an operator calls requeue_failed, so treating them as work-in-progress would hang forever on a poison row.

These probes assume the route's claim index exists (the table is yours — see Outbox table schema). Without it they degrade to a sequential scan.

Quiescing before a shutdown or a migration¶

quiesce() stops the runtime admitting new work, then waits for the operational planes — outbox routes, durable runs, stream groups — to come to rest, and reports what did and did not settle:

from forze_kits.integrations.quiesce import quiesce

report = await quiesce(runtime, outboxes=[OUTBOX], timeout=timedelta(seconds=60))
report.raise_if_unattested()   # or inspect report.attested / report.unsettled

Settled and attested are different claims, and the report keeps them apart. Settled means nothing was moving when the sweep finished. Attested means nothing was moving and nothing could arrive, because the runtime was holding the door shut. Only the second is safe to build on: an export written from a merely-settled runtime can be overtaken by a write before it finishes.

• Closing the gate is one-way. By default quiesce() stops the runtime admitting work, and the drain gate does not reopen — that is deliberate, because it is the shutdown gate. This is the step before a shutdown, an export, or a migration.
• close_gate=False only looks. The sweep reads each plane and the scope keeps serving. Nothing is holding the door, so the report can be settled but never attested. Use it for a health check; do not build an export on it.
• It waits for the relay; it does not relay. Closing the gate makes the backlog finite, but something still has to publish it — the background step, or (the usual production shape) an external worker this process cannot reach at all. If nothing does, the outbox plane comes back residual with the age of the oldest pending row. Give the budget room for at least one relay tick.

Planes the runtime does not wire are reported not_wired and do not count against attestation. Two things sit outside what quiesce() can speak for in either mode: a Temporal-backed workflow, whose state lives in the Temporal cluster; and a sibling replica — quiesce holds one process still, and a fleet that is still serving writes elsewhere will invalidate whatever this one attested. Stop the fleet before you trust the attestation.