docs: fix drift, broken samples, and inconsistencies across the docs set

CLAUDE.md

@@ -40,7 +40,7 @@ The package wires a FastStream `Broker`/`Registrator`/`Subscriber` trio whose tr
 
 ## Conventions
 
-- Python 3.13+.
+- Python 3.11+.
 - **Never use local/inline imports.** All imports at module top — no `import` inside functions/methods/`if TYPE_CHECKING` exception aside. Tests included. If `# noqa: PLC0415` is the only way to keep an import inline, hoist it instead.
 - `ruff` runs `select = ["ALL"]` with documented ignores in `pyproject.toml`; many `# noqa` are intentional.
 - Type checker is `ty`. Use `# ty: ignore[<rule>]` for intentional escapes.

architecture/subscriber.md

@@ -28,7 +28,7 @@ A subscriber can hold up to `fetch_batch_size + max_workers` leases at once, not
 
 Each subscriber holds `max_workers + 1` SQLAlchemy pool connections steady-state, plus one raw asyncpg connection for `LISTEN`.
 
-- Size the pool for `Σ subscribers × (max_workers + 1)` or startup blocks on checkout. The asyncpg `LISTEN` connection lives outside the pool, so it does not count toward pool sizing.
+- Size the pool for `Σ subscribers × (max_workers + 1)`. Undersizing does **not** block `broker.start()` — `start()` only schedules the loop tasks via `add_task` and returns; instead the fetch/worker loops stall on pool checkout at runtime. The asyncpg `LISTEN` connection lives outside the pool, so it does not count toward pool sizing.
 - Per process, Postgres `max_connections` must cover `replicas × Σ subscribers × (max_workers + 2)`: the `max_workers + 1` pool connections plus the out-of-pool asyncpg `LISTEN` connection. Undersize it and rolling deploys hit `FATAL: too many connections`.
 
 ## NOTIFY semantics

docs/concepts/instrumentation-seams.md

@@ -61,12 +61,14 @@ Three events fire **outside** the handler invocation, with no
 ## What the recorder seam observes naturally
 
 The recorder is a `Callable[[str, Mapping[str, Any]], None]` invoked at
-six core subscriber events (`fetched`, `dispatched`, `acked`,
-`nacked_retried`, `nacked_terminal`, `lease_lost`), a conditional
-`dlq_written` when the DLQ is configured, and one producer event
+seven core subscriber events (`fetched`, `dispatched`, `acked`,
+`nacked_retried`, `nacked_terminal`, `lease_lost`, `drain_timeout`), a
+conditional `dlq_written` when the DLQ is configured, and one producer event
 (`published`). It fires whether or not a handler is in scope:
 
-- All three bus-invisible events above.
+- All four bus-invisible events above: `fetched`, `lease_lost`, the
+  pre-consume `nacked_terminal(reason="max_deliveries")`, and `drain_timeout`
+  (a shutdown-drain overrun, which has no handler scope at all).
 - Plus `acked` / `nacked_retried` / `nacked_terminal` / `dispatched` /
   `published` from inside the handler-execution paths, with explicit
   `subscriber` and `queue` tags.
@@ -91,6 +93,7 @@ physically cannot observe.
 | `fetched` ticks (including empty) | ❌ (no `StreamMessage` at fetch time) | ✅ |
 | `lease_lost` after `consume_scope` exits | ❌ | ✅ |
 | `nacked_terminal(reason="max_deliveries")` before consume opens | ❌ | ✅ |
+| `drain_timeout` during `stop()` shutdown drain | ❌ (no `consume_scope`) | ✅ |
 
 ## Operator implication
 

docs/index.md

@@ -43,12 +43,15 @@ process, no Kafka.
 
 | If you want to… | Start at |
 |---|---|
+| Install and write the first publisher / subscriber | [Installation](introduction/installation.md) → [Tutorial: Your first outbox app](tutorials/first-outbox-app.md) |
 | See it work end-to-end on a FastAPI app | [FastAPI integration](usage/fastapi.md) |
 | Relay outbox rows to Kafka / RabbitMQ / NATS / Redis | [Relay to Kafka / RabbitMQ / NATS](usage/relay.md) |
 | Understand the architecture before adopting | [How it works](introduction/how-it-works.md) |
 | Compare against CDC / Kafka transactions / a hand-rolled outbox | [Comparison](concepts/comparison.md) |
 | Deploy to production safely | [Production checklist](operations/checklist.md) |
-| Install and write the first publisher / subscriber | [Installation](introduction/installation.md) → [Tutorial: Your first outbox app](tutorials/first-outbox-app.md) |
+
+The categorized index below (also the left-hand nav) lists every page with a
+one-line summary.
 
 ## Documentation
 

docs/introduction/how-it-works.md

@@ -189,9 +189,14 @@ 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. If the foreign publish fails, the row is **not**
-nacked through the `retry_strategy`; its lease simply expires and a later
-fetch retries the relay, so a transient bus outage never loses the row.
+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`) — the in-handler publish blocks until the broker returns,
+which the subscriber sees as one slow *successful* publish, 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](../usage/relay.md#what-not-to-do).)
 
 > **Worked end-to-end example → [Relay tutorial](../tutorials/add-kafka-relay.md).**
 

docs/introduction/installation.md

@@ -22,7 +22,7 @@
 
 ## Requirements
 
-- Python 3.13+
+- Python 3.11+
 - PostgreSQL 12+ — the features used (partial indexes, `FOR UPDATE SKIP
   LOCKED`, `make_interval`, `pg_notify`) all predate 12. The examples and
   CI run on 17; that is what's exercised, so 17 is the safest choice if

docs/operations/alembic.md

@@ -30,6 +30,7 @@ op.create_table('outbox',
     sa.Column('acquired_at', sa.DateTime(timezone=True), nullable=True),
     sa.Column('acquired_token', sa.Uuid(), nullable=True),
     sa.Column('timer_id', sa.String(length=255), nullable=True),
+    sa.CheckConstraint('(acquired_token IS NULL) = (acquired_at IS NULL)', name='outbox_lease_ck'),
     sa.PrimaryKeyConstraint('id')
 )
 op.create_index('outbox_lease_idx', 'outbox', ['queue', 'acquired_at'], unique=False,
@@ -63,9 +64,18 @@ Columns the operator can mostly ignore: `attempts_count` and
 `last_attempt_at` (debugging aids on retry-heavy rows). All four are
 maintained by the broker; no application code touches them.
 
+The `outbox_lease_ck` CHECK (`(acquired_token IS NULL) = (acquired_at
+IS NULL)`) is equally load-bearing: it enforces that a row's lease token
+and lease timestamp are always set or cleared together, so a half-written
+lease can never exist. Autogenerate renders it here **only because this is
+a fresh `create_table`** — on an incremental migration onto a pre-existing
+table Alembic has no check-constraint comparator and would ship a missing
+or drifted CHECK silently (that gap is what
+[`validate_schema()`](../usage/schema-validation.md) backstops).
+
 The `# please adjust!` comment from Alembic is misleading here —
-**don't adjust**. The column types, predicates, and indexes are
-exactly what the broker depends on. The
+**don't adjust**. The column types, the CHECK, the predicates, and the
+indexes are exactly what the broker depends on. The
 [`validate_schema()`](../usage/schema-validation.md) check — when you wire
 it into a `/health` probe or CI gate — fails when the live DB drifts from
 this declaration. (It is opt-in; it never runs at `broker.start()`.)

docs/operations/checklist.md

@@ -64,11 +64,14 @@ story.
   `[validate]` extra. Do **not** call at `broker.start()` — that
   would crash-loop on a pending migration. See [Schema validation §
   Where to call it](../usage/schema-validation.md#where-to-call-it).
-- [ ] **Outbox `table_name` short enough for the NOTIFY channel** — the
-  channel name is `outbox_<table_name>`, and `make_outbox_table` raises
-  `ValueError` at table-build time when that exceeds Postgres' 63-**byte**
-  identifier limit. There is no silent truncation or polling fallback — the
-  guard makes an over-long name impossible to ship.
+- [ ] **Outbox `table_name` short enough for every derived identifier** —
+  `make_outbox_table` raises `ValueError` at table-build time when the
+  *longest* identifier it derives exceeds Postgres' 63-**byte** limit. That
+  longest identifier is usually an index/constraint name
+  (`<table_name>_pending_idx`, `<table_name>_timer_id_uq`), which is longer
+  than the NOTIFY channel `outbox_<table_name>` — so a name that fits the
+  channel can still overflow an index name. There is no silent truncation or
+  polling fallback; the guard makes an over-long name impossible to ship.
 
 ## Observability
 

docs/tutorials/add-kafka-relay.md

@@ -241,7 +241,9 @@ the durability boundary, and it stays in the table for the duration of the
 retry budget (the default `ExponentialRetry` allows 10 attempts). Once the
 budget is exhausted the row is deleted — the default configures no DLQ — so
 configure a longer `retry_strategy` or a `dlq_table` to survive outages
-beyond that (with the default schedule, ~13–14 minutes).
+beyond that (the default schedule spans roughly 8-9 minutes: nine backoffs
+of 1, 2, 4, … 256 seconds sum to ~8.5 minutes before the 10th attempt is
+terminal).
 
 In practice, `aiokafka`'s producer has its own client-side reconnect
 and retry logic, so a short Kafka outage usually completes from the

docs/tutorials/first-outbox-app.md

@@ -9,7 +9,7 @@ print it.
 
 ## Before you start
 
-- Python 3.13+
+- Python 3.11+
 - Docker (for a one-line Postgres container)
 - [uv](https://docs.astral.sh/uv/) for project setup
 - Roughly ten minutes

docs/usage/fastapi.md

@@ -151,6 +151,8 @@ The [DLQ](./dlq.md) and the [metrics-recorder seam](./observability.md)
 `OutboxBroker(...)` — they forward to the inner broker.
 
 ```python
+from faststream_outbox import make_dlq_table  # alongside make_outbox_table
+
 outbox_router = OutboxRouter(
     engine,
     outbox_table=outbox_table,

docs/usage/messaging-service.md

@@ -55,12 +55,18 @@ The broker and routers are wired with a real DI container
 
 ```python title="ioc.py"
 from modern_di import Group, Scope, providers
+from sqlalchemy.ext.asyncio import create_async_engine
 from faststream_outbox import OutboxBroker
 
 from app.tables import OUTBOX_TABLE
 
 
 class Resources(Group):
+    # The caller owns the engine; the broker never disposes it.
+    database_engine = providers.Factory(
+        scope=Scope.APP,
+        creator=lambda: create_async_engine("postgresql+asyncpg://app:app@localhost/app"),
+    )
     outbox_broker = providers.Factory(
         scope=Scope.APP,
         creator=lambda engine: OutboxBroker(engine, outbox_table=OUTBOX_TABLE),
@@ -167,7 +173,9 @@ Register the router on the broker (the `OutboxBroker` built in `ioc.py`) with
 ## Pattern 2 — Fire-unless-cancelled timer
 
 The unread notification is a **delayed** outbox row, armed in the same create
-transaction. `timer_id` makes it idempotent; `activate_in` defers it:
+transaction. `timer_id` deduplicates while a row is live (at most one live
+row per `(queue, timer_id)`, not a global idempotency key); `activate_in`
+defers it:
 
 These two methods live on the same `OutboxEventProducer` from Pattern 1 (shown
 here as a continuation of the class):

docs/usage/observability.md

@@ -17,8 +17,11 @@ MetricsRecorder = Callable[[str, Mapping[str, Any]], None]
 The default (`_noop_recorder`) lets instrumentation sites call
 unconditionally. The recorder threads through `OutboxBrokerConfig` to:
 
-- The subscriber's seven emission points via `OutboxSubscriber._emit_metric`
-- The producer's single emission point via `OutboxProducer._emit_metric`
+- The subscriber's seven core events via `OutboxSubscriber._emit_metric`
+  (`fetched`, `dispatched`, `acked`, `nacked_retried`, `nacked_terminal`,
+  `lease_lost`, `drain_timeout`), plus a conditional `dlq_written` when a DLQ
+  is configured
+- The producer's single event (`published`) via `OutboxProducer._emit_metric`
 
 ### Bare seam
 

docs/usage/publisher.md

@@ -15,6 +15,9 @@ For "consume from A → enqueue to B" relay flows, a fourth path is
 available: returning `OutboxResponse(...)` from a handler. See [Chained
 publishing](#chained-publishing) below.
 
+There is no request/reply: the outbox is fire-and-forget, so `broker.request(...)`
+raises `NotImplementedError`.
+
 ## `broker.publish`
 
 ```python
@@ -144,15 +147,18 @@ flow routes the returned value through the producer; the same
 transactional contract applies (you provide the session, the row commits
 with your domain writes):
 
-!!! note "This pattern is FastAPI-specific"
+!!! note "The `session` must outlive the handler return"
     The returned `OutboxResponse` is published **after** the handler
-    returns, so its `session` must outlive the handler call. FastAPI's
-    `Depends(get_session)` provides exactly that — a session torn down by
-    the dependency after the response flow. Opening your own `async with
-    session_factory() as session:` inside the handler does **not** work
-    here: the session closes on `return`, before the row is inserted.
-    Outside FastAPI, call `broker.publish(..., session=session)` directly
-    inside your handler instead (see [§ `broker.publisher`](#not-a-relay-decorator)).
+    returns, so its `session` must still be open at that point. The
+    requirement is about session *lifetime*, not any particular framework:
+    provide the session through a dependency that the framework tears down
+    *after* the response flow — FastAPI's `Depends(get_session)` or
+    FastStream's own `Depends` / `Context` session both do this. Opening
+    your own `async with session_factory() as session:` inside the handler
+    does **not** work here: that session closes on `return`, before the row
+    is inserted — in that case call `broker.publish(..., session=session)`
+    directly inside the `async with` instead (see
+    [§ Not a relay decorator](#not-a-relay-decorator)).
 
 ```python
 from fastapi import Depends

docs/usage/relay.md

@@ -43,7 +43,7 @@ async def relay(body: dict) -> dict:
     return body
 ```
 
-That's the whole thing. `broker_outbox.publish(body, queue="outbox_queue", session=session)`
+That's the whole thing. `await broker_outbox.publish(body, queue="outbox_queue", session=session)`
 in your domain transaction writes a row; the subscriber dispatches it; the
 handler returns it; the Kafka publisher decorator picks it up and publishes
 to `kafka_topic`. Failure handling, retries, and DLQ are unchanged from
@@ -161,6 +161,8 @@ construction (before lifespan), so it's automatic.
 For the standalone (non-FastAPI) lifecycle, the order is:
 
 ```python
+# kafka_router / outbox_router are constructed exactly as in the FastAPI
+# example above (KafkaRouter(...) / OutboxRouter(...)).
 broker_kafka.include_router(kafka_router)
 broker_outbox.include_router(outbox_router)
 # then start

docs/usage/schema-validation.md

@@ -83,10 +83,16 @@ Or as a one-shot CI check after running migrations:
 
 ```python
 import asyncio
+
+from sqlalchemy import MetaData
+from sqlalchemy.ext.asyncio import create_async_engine
+
 from faststream_outbox import OutboxBroker, make_outbox_table
 
 
 async def main() -> None:
+    engine = create_async_engine("postgresql+asyncpg://outbox:outbox@localhost/outbox")
+    outbox_table = make_outbox_table(MetaData(), table_name="outbox")
     broker = OutboxBroker(engine, outbox_table=outbox_table)
     await broker.validate_schema()
     print("schema OK")

docs/usage/setup-prometheus-opentelemetry.md

@@ -126,10 +126,11 @@ and drop the `/metrics` route.
 Instrument names match `faststream.opentelemetry.TelemetryMiddleware` for
 the bus-scope metrics — `messaging.process.duration`,
 `messaging.publish.duration`, and (when `include_messages_counters=True`)
-`messaging.process.messages` / `messaging.publish.messages` — plus three
+`messaging.process.messages` / `messaging.publish.messages` — plus four
 outbox-specific counters the middleware can't emit:
-`messaging.outbox.fetch.batches`, `messaging.outbox.lease_lost`, and
-`messaging.outbox.dlq_written`. Units and constructor args
+`messaging.outbox.fetch.batches`, `messaging.outbox.lease_lost`,
+`messaging.outbox.dlq_written`, and `messaging.outbox.drain_timeout`. Units
+and constructor args
 (`meter_provider`, `meter`, `include_messages_counters`) follow upstream.
 The `messaging.system="outbox"` attribute disambiguates outbox traffic
 from Kafka / Rabbit data on the same instruments.
@@ -154,17 +155,21 @@ observability stack:
 
 ```bash
 pip install 'faststream-outbox[opentelemetry,prometheus]' \
-    opentelemetry-exporter-otlp opentelemetry-exporter-prometheus uvicorn
+    opentelemetry-exporter-otlp uvicorn
 ```
 
+Here OpenTelemetry supplies **spans** (exported to OTLP) and Prometheus
+supplies **all metrics** (two registries scraped over HTTP). The OTel
+middleware runs span-only — its meters would otherwise land on the global
+Prometheus registry, which neither endpoint below exposes, so they are left
+off to avoid a dead, unscraped meter path.
+
 ```python
 # app.py — run with `OTEL_EXPORTER_OTLP_ENDPOINT=http://localhost:4317 \
 #                    uvicorn app:app --host 0.0.0.0 --port 8000`
 from faststream.asgi import AsgiFastStream
-from opentelemetry import metrics, trace
+from opentelemetry import trace
 from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
-from opentelemetry.exporter.prometheus import PrometheusMetricReader
-from opentelemetry.sdk.metrics import MeterProvider
 from opentelemetry.sdk.resources import Resource
 from opentelemetry.sdk.trace import TracerProvider
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
@@ -184,9 +189,6 @@ tracer_provider = TracerProvider(resource=resource)
 tracer_provider.add_span_processor(BatchSpanProcessor(OTLPSpanExporter()))
 trace.set_tracer_provider(tracer_provider)
 
-meter_provider = MeterProvider(resource=resource, metric_readers=[PrometheusMetricReader()])
-metrics.set_meter_provider(meter_provider)
-
 # Two registries: the middleware and the recorder both define the same
 # faststream_* consume/publish collectors, so sharing one registry raises
 # "Duplicated timeseries in CollectorRegistry" at broker construction.
@@ -202,8 +204,8 @@ broker = OutboxBroker(
     engine,
     outbox_table=outbox_table,
     middlewares=[
-        # Bus-scope spans + meters around consume_scope / publish_scope.
-        OutboxTelemetryMiddleware(tracer_provider=tracer_provider, meter_provider=meter_provider),
+        # Bus-scope spans around consume_scope / publish_scope (span-only; see note above).
+        OutboxTelemetryMiddleware(tracer_provider=tracer_provider),
         OutboxPrometheusMiddleware(registry=MIDDLEWARE_REGISTRY, app_name="my-outbox-service"),
     ],
     # Outbox-internal events (fetched, lease_lost, terminal reasons, dlq_written)
@@ -226,9 +228,10 @@ app = AsgiFastStream(
 ```
 
 Traces flow to OTLP (Jaeger / Tempo / Honeycomb / collector); the
-middleware's meters land on `/metrics` and the recorder's outbox-internal
-counters on `/metrics/outbox` for Prometheus to scrape — two scrape targets,
-one process.
+Prometheus **middleware's** consume/publish meters land on `/metrics` and the
+**recorder's** outbox-internal counters on `/metrics/outbox` for Prometheus to
+scrape — two scrape targets, one process. (The OTel middleware here
+contributes spans only, per the note above.)
 
 **The two seams overlap on consume/publish series.** Both the middleware
 and the recorder emit the same `faststream_received_*` / `faststream_published_*`

docs/usage/subscriber.md

@@ -103,9 +103,12 @@ Per-subscriber knobs, passed to `@broker.subscriber("…", …)`:
 async def handle_urgent(body: dict) -> None: ...
 ```
 
-The factory in `subscriber/factory.py` warns or raises on likely-wrong
-combinations (`lease_ttl_seconds <= max_fetch_interval`, `max_deliveries`
-without retry, `min_fetch_interval > max_fetch_interval`, etc.).
+`OutboxSubscriberConfig.__post_init__` (in `subscriber/config.py`) warns or
+raises on likely-wrong combinations (`lease_ttl_seconds <= max_fetch_interval`,
+`max_deliveries` without retry, `min_fetch_interval > max_fetch_interval`,
+etc.). Validation lives on the config, not the factory, so **every**
+construction path — `@broker.subscriber`, `@router.subscriber`, direct
+construction — is checked.
 
 The table above lists the outbox-specific knobs. The standard FastStream
 subscriber kwargs pass through unchanged too: `dependencies`, `parser`,