Subscriber¶

Use @broker.subscriber(queue, ...) to register a handler for a queue.

Basic example¶

from faststream import FastStream
from faststream_outbox import OutboxBroker

broker: OutboxBroker = ...
app = FastStream(broker)


@broker.subscriber("orders")
async def handle(order_id: int) -> None:
    print(f"order {order_id}")

Multiple queues per subscriber¶

The first argument is queues: str | list[str]. Pass a list to fan one handler across several queues:

@broker.subscriber(["orders", "refunds"])
async def handle(body: dict) -> None: ...

The subscriber claims rows from any of its queues in a single fetch. Its connection budget is unchanged: max_workers + 1 pool connections regardless of how many queues it serves.

In the AsyncAPI document it appears as one channel per queue (orders:Handle, refunds:Handle), each addressed by that queue, rather than one channel for the subscriber.

Do not register two subscribers on the same queue: they compete for the same rows, and registration emits a warning to that effect. To run more than one handler over a queue, attach them to a single subscriber; to scale throughput, raise max_workers.

Body types¶

FastStream deserializes the message body into the annotated type. Any JSON-serializable type works:

from dataclasses import dataclass


@dataclass
class Order:
    order_id: str
    amount: float


@broker.subscriber("orders")
async def handle(body: Order) -> None:
    print(f"order {body.order_id} for {body.amount}")

Annotated handler params¶

faststream_outbox.annotations exports Annotated[..., Context(...)] shortcuts so handler signatures stay concise:

from faststream_outbox.annotations import OutboxBroker, OutboxMessage


@broker.subscriber("orders")
async def handle(msg: OutboxMessage, broker: OutboxBroker) -> None: ...

OutboxMessage, OutboxBroker, OutboxProducer, and OutboxClient are all available. For FastAPI handlers, import the same names from faststream_outbox.fastapi. They resolve via the same Context() paths but go through FastAPI's dependency resolver so Depends(...) and these shortcuts can be mixed freely.

Subscriber options¶

Per-subscriber knobs, passed to @broker.subscriber("…", …):

Parameter	Default	Description
`max_workers`	`1`	Concurrent handlers per subscriber
`fetch_batch_size`	`10`	Most rows held in memory and claimed per fetch. The next fetch runs once workers have taken half of them.
`min_fetch_interval`	`1.0` s	Base for the adaptive idle backoff (jittered ±50%, so an actual wait can land below it); no sleep at all while fetches keep returning rows
`max_fetch_interval`	`10.0` s	Ceiling for the adaptive idle backoff (with jitter)
`lease_ttl_seconds`	`60.0` s	How long a claim is valid before another fetch may reclaim it. Must exceed your handler's P99 with margin.
`max_deliveries`	`None` (unbounded)	Total claims (including lease-expiry re-claims) after which the row is dropped without invoking the handler. Defends against handlers that consistently wedge. Must be at least `1`; use `None`, not `0`, for no cap.
`terminal_flush_batch_size`	`1` (off)	Coalesce completed terminal `DELETE`s into one `DELETE … RETURNING` per N rows. `1` is one round-trip per message (unchanged). Higher trades a wider crash-redelivery window for far fewer round-trips. See Batching terminal deletes.
`ack_policy`	`AckPolicy.NACK_ON_ERROR`	See Ack policy
`retry_strategy`	`ExponentialRetry(...)`	See Retry strategies
`propagate_inbound_headers`	`False`	Relay-only. When `True`, fills `Response.headers` from the inbound message if the handler returned a `Response` with empty headers (user-set headers always win). See Relay.

@broker.subscriber(
    "high-priority",
    max_workers=8,
    fetch_batch_size=50,
    min_fetch_interval=0.1,
    max_fetch_interval=1.0,
    lease_ttl_seconds=120.0,
)
async def handle_urgent(body: dict) -> None: ...

Subscriber options are validated when the subscriber is created. Values that would delete data or never consume raise: a max_deliveries below 1, a NaN or infinite interval, a bool passed as a count, and a queue name that is empty, longer than 255 characters, or listed twice. Likely-wrong combinations (lease_ttl_seconds <= max_fetch_interval, max_deliveries with NoRetry(), min_fetch_interval > max_fetch_interval, etc.) warn or raise. Every construction path (@broker.subscriber, @router.subscriber, OutboxRoute) is checked.

The table above lists the outbox-specific knobs. The standard FastStream subscriber kwargs pass through unchanged too: dependencies, parser, decoder, and the AsyncAPI title_ / description_ / include_in_schema.

title_ names both the AsyncAPI channel and its operation. On a subscriber spanning several queues it prefixes each channel (Ingest:orders, Ingest:refunds), since one title cannot name several channels on its own.

Slow handlers: dedicated queue¶

When a handler's tail latency exceeds the subscriber's lease_ttl_seconds, the row's lease expires mid-flight and another fetch reclaims it, causing duplicate delivery. Don't hike lease_ttl_seconds globally, because that delays reclaim of actually stuck rows everywhere. Instead, segregate slow work onto its own subscriber with a longer TTL:

@broker.subscriber("slow_q", lease_ttl_seconds=600)  # 10 minutes
async def heavy_job(msg): ...


@broker.subscriber("fast_q", lease_ttl_seconds=30)
async def quick_job(msg): ...

Pick lease_ttl_seconds strictly greater than that subscriber's P99 handler duration, with margin for clock skew. The tight TTL on the fast queue keeps stuck-row reclaim fast; the tall TTL on the slow queue tolerates outliers without slowing reclaim of genuinely stuck rows elsewhere. Producers route to the appropriate queue at publish time.

Account for queue depth, not just per-row latency

A fetch claims up to fetch_batch_size rows at once and each takes its lease at fetch time, then they wait their turn in an in-memory queue drained by max_workers handlers. A row's lease clock runs while it waits, so the relevant bound is the serialized time to reach it, not one handler's P99:

(fetch_batch_size / max_workers) × P99(handler)  ≪  lease_ttl_seconds

With the defaults (fetch_batch_size=10, max_workers=1) the 10th row waits behind nine others before dispatch. If that wait can exceed lease_ttl_seconds, a competing fetch reclaims the still-queued row → duplicate (self-correcting) delivery. Either keep fetch_batch_size close to max_workers, or size the TTL for the whole batch.

Batching terminal deletes¶

By default each processed row is deleted with its own DELETE, one round-trip per message. At max_workers=1 those deletes serialise, and the round-trip (not the database work) is the throughput ceiling. Set terminal_flush_batch_size above 1 to coalesce completed rows and flush them as a single DELETE … WHERE (id, acquired_token) IN (…) RETURNING id:

@broker.subscriber("orders", terminal_flush_batch_size=100)
async def handle(order: dict) -> None: ...

A worker buffers completed rows and flushes when the buffer reaches terminal_flush_batch_size, when its inflight queue empties, or when the oldest buffered row was claimed half of lease_ttl_seconds ago. A lightly-loaded queue still flushes immediately and batching adds no latency; batching only engages under sustained load, and a buffered row is deleted before its lease can expire.

In the benchmark (5 000 messages, fetch_batch_size=100), the terminal round-trips drop from one per message to one per batch, a 100× reduction in terminal DELETEs (5 000 → 50) with the same rows deleted. Because the terminal write stops being the bottleneck, a single batched worker out-throughputs a four-worker per-row subscriber, so you reach high throughput without spending the extra connection budget that more workers cost. The win is largest at low max_workers (where per-row deletes serialise) and narrows as worker parallelism rises.

Read the tradeoff before enabling it. Batching holds completed-but-undeleted rows in memory until the flush. On a graceful stop the buffer is flushed (no redelivery). But on an ungraceful crash (SIGKILL / OOM / power loss), up to terminal_flush_batch_size rows that already ran their handler are redelivered when another replica reclaims them. The outbox is already at-least-once (handlers must be idempotent), so this is not a new failure class, only a wider window: from at most one at-risk row (per-row) to up to a full batch. Two further effects to size for:

The outbox table shows completed-but-undeleted rows as still present until the flush, so a backlog-depth query (or an autoscaler keyed on it) reads inflated by up to terminal_flush_batch_size × max_workers.
Buffered rows hold their leases until the flush, so the lease ceiling grows to fetch_batch_size + max_workers × (terminal_flush_batch_size + 1); keep lease_ttl_seconds sized against that.

It is off by default (terminal_flush_batch_size=1 is byte-for-byte the per-row path). Enable it per subscriber when the queue is high-throughput and its handler is idempotent; leave it off for low-volume or exactly-once-sensitive queues.

For where this sits among the outbox's tuning levers and which workloads want it, see the Performance guide.

Ack policy¶

The default is AckPolicy.NACK_ON_ERROR: on a handler exception, the retry strategy decides whether to schedule another attempt or terminally drop the row.

Policy	Effect
`AckPolicy.NACK_ON_ERROR` (default)	Consult the retry strategy on handler exceptions
`AckPolicy.REJECT_ON_ERROR`	Delete on the first failure (the retry strategy is ignored)
`AckPolicy.MANUAL`	Handler must call `await msg.ack()` / `nack()` / `reject()` itself
`AckPolicy.ACK_FIRST`	Not supported. Passing it raises `ValueError` at registration

ACK_FIRST would delete the row before the handler runs, so a handler crash silently drops the message, which defeats the outbox reliability guarantee. The factory rejects it at registration.

from faststream import AckPolicy
from faststream_outbox.annotations import OutboxMessage


@broker.subscriber("audit", ack_policy=AckPolicy.MANUAL)
async def handle(msg: OutboxMessage, body: dict) -> None:
    try:
        await write_audit(body)
        await msg.ack()
    except TransientError:
        await msg.nack()  # retry
    except PermanentError:
        await msg.reject()  # terminal delete

MANUAL: returning without acking is a terminal reject

Under AckPolicy.MANUAL, a handler that returns without calling ack() / nack() / reject() (and without raising) is treated as a terminal reject: the row is deleted (or written to the DLQ with failure_reason="rejected" if a dlq_table is configured), not retried. A handler that raises is nacked through the retry strategy instead, so only the silent-return path is destructive. Always ack/nack/reject on every branch.

Retry strategies¶

A subscriber with no explicit retry_strategy defaults to ExponentialRetry(initial_delay_seconds=1.0, multiplier=2.0, max_delay_seconds=300.0, max_attempts=10, jitter_factor=0.2). Defaulting to "delete on first error" is the wrong contract for an outbox; users wanting that behavior must explicitly pass NoRetry().

from faststream_outbox import ExponentialRetry, ConstantRetry, LinearRetry, NoRetry


@broker.subscriber(
    "orders",
    retry_strategy=ExponentialRetry(
        initial_delay_seconds=1.0,
        max_delay_seconds=300.0,
        max_attempts=5,
        jitter_factor=0.5,
    ),
)
async def handle(order_id: int) -> None: ...


@broker.subscriber("audit", retry_strategy=NoRetry())  # opt out of retries
async def handle_audit(payload: dict) -> None: ...

ConstantRetry and LinearRetry accept jitter_factor (default 0.0); when non-zero, the computed delay is multiplied by 1 + U(-jitter_factor/2, +jitter_factor/2) to spread out retries, matching ExponentialRetry's shape.

Strategy parameters¶

Strategy	Required	Optional (default)
`NoRetry`	-	-
`ConstantRetry`	`delay_seconds`	`jitter_factor` (`0.0`)
`LinearRetry`	`initial_delay_seconds`, `step_seconds`	`jitter_factor` (`0.0`)
`ExponentialRetry`	`initial_delay_seconds`	`multiplier` (`2.0`), `max_delay_seconds` (`None`), `jitter_factor` (`0.0`)

Every strategy except NoRetry also accepts the shared caps max_attempts (default None) and max_total_delay_seconds (default None); reaching either returns None, which is terminal (the row is deleted, or DLQ'd).

Retry only on transient errors¶

Strategies receive the raised exception so users may subclass for "retry only on transient errors":

class TransientOnly(ExponentialRetry):
    def get_next_attempt_delay(
        self,
        *,
        first_attempt_at: datetime,
        last_attempt_at: datetime,
        attempts_count: int,
        exception: BaseException | None = None,
    ) -> float | None:
        if exception and not isinstance(exception, TransientError):
            return None  # terminal — DELETE
        return super().get_next_attempt_delay(
            first_attempt_at=first_attempt_at,
            last_attempt_at=last_attempt_at,
            attempts_count=attempts_count,
            exception=exception,
        )

Returning None from get_next_attempt_delay signals a terminal failure. The base strategy also enforces max_attempts and max_total_delay_seconds for you.

Connection budget¶

Each subscriber holds max_workers + 1 long-lived SQLAlchemy pool connections (one writer per worker + one fetch), plus one raw asyncpg connection for LISTEN when available. Size your engine pool for Σ subscribers × (max_workers + 1). With a fixed-size pool (SQLAlchemy's default QueuePool), broker.start() raises RuntimeError when pool_size + max_overflow is below that sum. NullPool and an unlimited max_overflow=-1 aren't checked. The check counts only the subscribers, so leave room for the application's own sessions: if those take the remaining connections, the fetch and worker loops stall on pool checkout and show up as repeating reconnect ERROR logs. SQLAlchemy's default pool_size=5, max_overflow=10 covers a handful of single-worker subscribers; raise it for larger fleets.

Server-side, the footprint is one larger: the raw asyncpg LISTEN connection lives outside the pool, so each subscriber consumes max_workers + 2 Postgres connections. The budget is per process: each replica opens its own pool and LISTEN connections, so your Postgres max_connections needs to cover replicas × Σ subscribers × (max_workers + 2), otherwise additional replicas (or rolling deployments) are refused at startup with FATAL: too many connections.

Operator-side: Production checklist § Sizing.

Read-only inspection¶

subscriber.get_one() and async for msg in subscriber: are not supported on OutboxSubscriber; both raise NotImplementedError. They would acquire a lease and bump deliveries_count, surprising semantics for a peek API. Use broker.fetch_unprocessed(session=..., queue=...) for lease-free reads of the current table state.