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.

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),
    ),
)

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.

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.

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.