How it works

faststream-outbox is a FastStream broker integration whose transport is Postgres rows, not a message bus. A producer writes an outbox row in the same SQLAlchemy transaction as its domain entity; a subscriber polls the table, claims rows with FOR UPDATE SKIP LOCKED, runs the handler, and deletes the row on success.

The transactional outbox pattern

Distributed systems need two writes to atomically succeed or fail together: the business write (place an order) and the message-bus write (notify downstream). Brokers don't participate in your database transaction, so a crash between the two leaves them out of sync.

The outbox solves this by collapsing both writes into a single database transaction. Instead of publishing to a broker, you INSERT a row into an outbox table on the same AsyncSession that holds your domain write. A separate process polls the table and forwards rows to their consumers. The row commits or rolls back with your domain write.

In faststream-outbox that separate process is an outbox subscriber, and you choose what it does with each row:

• Standalone: the Postgres table is the queue and the subscriber's handler is the consumer. No other message bus is involved.
• Relay: the subscriber forwards each row to Kafka, RabbitMQ, NATS or Redis through a foreign-broker publisher decorator. See Relay to a foreign broker.

See Comparison for when CDC or Kafka transactions are the better fit.

Producer side

broker.publish(body, *, queue, session, ...) inserts an outbox row through the caller's AsyncSession (or AsyncConnection, for Core code). It does not flush, commit, or open its own transaction; the row must commit with the caller's domain writes:

async with session_factory() as session, session.begin():
    session.add(order)  # domain write
    await broker.publish(order.id, queue="orders", session=session)
    # session.begin() commits both atomically on exit

publish_batch(*bodies, queue, session, ...) does the same with a single round-trip for many rows.

A single publish also sends pg_notify('outbox_<table>', queue) in the INSERT's RETURNING clause, so the wakeup costs no extra round trip. publish_batch sends a separate SELECT pg_notify(...) after its INSERT. Neither notifies when the row is genuinely future-dated (a future activate_in / activate_at; a past activate_at, e.g. a recovered idempotency token, still notifies), and a timer_id conflict returns no row, so it sends no NOTIFY either. NOTIFY is transactional, so listeners only see it after the user's transaction commits, so atomicity with the row insert is automatic.

Repeated publishes to the same queue within one transaction emit a single pg_notify (Postgres coalesces identical notifications at delivery anyway), so a bulk publish costs one NOTIFY, not one per row.

Subscriber: two async loops

Per subscriber, two loops run concurrently:

The fetch loop owns a long-lived AsyncConnection for the fetch query and a separate raw asyncpg connection for LISTEN outbox_<table>. One statement claims rows:

WITH queues AS (SELECT DISTINCT unnest(:queues) AS name),
pending AS (
    SELECT p.id, p.next_attempt_at FROM queues JOIN LATERAL (
        SELECT id, next_attempt_at FROM outbox
        WHERE queue = queues.name AND next_attempt_at <= now()
          AND acquired_token IS NULL
        ORDER BY next_attempt_at, id LIMIT :batch
        FOR UPDATE SKIP LOCKED
    ) p ON true
),
expired AS (
    -- the same per-queue scan for leases older than :lease_ttl
),
ready AS (
    SELECT id FROM (SELECT * FROM pending UNION ALL SELECT * FROM expired) c
    ORDER BY next_attempt_at, id LIMIT :batch
)
UPDATE outbox
SET acquired_token = :uuid, acquired_at = now(),
    deliveries_count = deliveries_count + 1
WHERE id IN (SELECT id FROM ready)
RETURNING *

Each queue gets its own ordered scan of outbox_pending_idx, which stops after :batch unlocked rows, so a fetch costs the same with ten ready rows or a million. A single WHERE ... OR ... over queue = ANY(:queues) would claim the same rows, but Postgres would have to read and sort the whole backlog first. ADR 0003 records why the query has this shape.

The query reclaims both unleased rows AND rows whose lease has expired (acquired_at < now() - lease_ttl_seconds), so there is no separate stuck-row reaper. The idle-sleep is short-circuited by NOTIFY via an asyncio.Event, so idle dispatch latency drops from up to max_fetch_interval (default 10s) to ~10ms. If LISTEN setup fails (asyncpg missing, non-asyncpg driver, permission error), the loop logs once and falls back to polling.

The worker loop (× max_workers) pulls from an in-process asyncio.Queue(maxsize=fetch_batch_size), dispatches each row via OutboxSubscriber.dispatch_one (which runs the handler), then flushes the row's terminal state (DELETE on success, UPDATE next_attempt_at for retry). Once workers have taken half of that queue, they wake the fetch loop for the next claim, so a backlog drains at database speed without waiting for a NOTIFY or the poll interval. Each worker owns a long-lived AsyncConnection, so draining N rows costs O(workers) pool checkouts, not O(rows).

The lease-token invariant

Every terminal write filters on acquired_token:

DELETE FROM outbox WHERE id = :id AND acquired_token = :token

If a slow handler's lease expired and another worker reclaimed the row with a fresh token, the slow handler's DELETE finds rowcount == 0 and is silently dropped, which keeps it from clobbering the new lease holder. This is the load-bearing invariant; any new fetch or terminal path must preserve it.

lease_ttl_seconds (default 60.0, per-subscriber) must exceed the P99 handler duration with margin; otherwise healthy in-flight handlers race their own lease expiry and trigger duplicate deliveries. The lease cutoff is computed server-side via make_interval(secs => :lease_ttl), so it's immune to worker / DB clock skew.

When the invariant fires, the broker emits a WARNING with structured fields:

extra = {"event": "lease_lost", "phase": "terminal" | "retry", "row_id": ..., "queue": ..., "deliveries_count": ...}

Recurring event=lease_lost records mean lease_ttl_seconds < handler P99; that's the operator playbook signal. Log-pipeline aggregators can alert on the event field without parsing the message.

At-least-once delivery

The row is removed from the table only after the handler completes successfully. If the worker dies mid-handler, the lease expires and another worker re-claims the row. The same applies if the handler ran but the worker crashed before the terminal DELETE landed.

The trade-off is that handlers must be idempotent. A handler that succeeded but whose DELETE failed to land will be retried.

Opt-in DLQ on terminal failure

By default, terminal failures DELETE the row, with no archive table and no dead-letter queue. Pass dlq_table=make_dlq_table(metadata) to the broker and terminal-by-failure rows are copied into a sibling audit table in the same Postgres statement as the DELETE:

from sqlalchemy import MetaData
from sqlalchemy.ext.asyncio import create_async_engine
from faststream_outbox import OutboxBroker, make_dlq_table, make_outbox_table

metadata = MetaData()
outbox_table = make_outbox_table(metadata, table_name="outbox")
dlq_table = make_dlq_table(metadata, table_name="outbox_dlq")
engine = create_async_engine("postgresql+asyncpg://outbox:outbox@localhost:5432/outbox")
broker = OutboxBroker(engine, outbox_table=outbox_table, dlq_table=dlq_table)

Successful rows are never archived; the success path stays a plain DELETE. Three failure paths land in the DLQ with a failure_reason column: max_deliveries, retry_terminal, rejected. Atomicity is via a single CTE (DELETE … RETURNING → INSERT INTO <dlq>), so DLQ-write failures roll back the DELETE. Misconfiguration surfaces as outbox growth plus lease_lost spikes, not silent audit loss. When dlq_table is configured, broker.validate_schema() checks both tables in one call and reports drift on either one. See the Dead-letter queue page for the schema, atomicity, and retention story.

If you don't want a DLQ, you can still preserve failed messages by logging from the handler before the terminal failure propagates, or by attaching an audit column to the outbox table (the schema validator ignores extras you add).

Failure modes

• Handlers must be idempotent. A crash between the handler's side effect and the broker's DELETE re-delivers the message; see At-least-once delivery above.
• Ordering is best-effort only. FOR UPDATE SKIP LOCKED does not preserve strict order under concurrent workers. If you need strict per-aggregate ordering, route to a single subscriber and run a single worker.
• The DLQ is opt-in. Without dlq_table=, terminal failures DELETE the row.

Relay to Kafka / RabbitMQ / NATS / Redis

An OutboxSubscriber can source a FastStream-native cross-broker chain: stack a foreign-broker publisher decorator on the subscriber (@kafka_pub @broker_outbox.subscriber("q")) and the handler's return value is forwarded to the real bus. The outbox row stays the durability boundary: the row commits with the domain write, and the relay carries at-least-once end to end. Recovery comes from two tiers, so a bus outage never loses the row: a transient blip is absorbed by the client library (e.g. aiokafka), where the in-handler publish blocks until the broker returns, which the subscriber sees as one slow successful publish with no nack; a sustained outage eventually raises into the handler, which nacks the row and hands it to the configured retry_strategy to reschedule. (The one path that recovers via lease expiry rather than retry_strategy is a mis-composed publisher chain; see relay guardrails.)

For a worked end-to-end example, see the Relay tutorial.

Acknowledgements

The architecture of this package is heavily informed by Arseniy Popov's PR #2704 (feat: add sqla broker) on upstream FastStream. The FastStream broker/registrator/subscriber wiring, the SELECT … FOR UPDATE SKIP LOCKED fetch-and-claim CTE, the retry strategy hierarchy, and the in-transaction publish contract all originate from there. This package is a Postgres-only reimplementation that diverges in storage model (lease tokens instead of an explicit state column, opt-in DLQ instead of a mandatory archive table), loop structure (two loops instead of four), wake-up mechanism (LISTEN/NOTIFY), and adds timer mechanics. Credit for the original design belongs to Arseniy.