docs: refresh architecture/ deep-dives for the 0.10.0 surfaces

architecture/dlq.md

@@ -24,7 +24,7 @@ User-facing: `docs/usage/dlq.md`. Invariant summary: `CLAUDE.md` § Opt-in DLQ.
 
 ## `last_exception` bounds
 
-`last_exception` is serialized via `repr()` and bounded by `_LAST_EXCEPTION_MAX_CHARS=8192` in `subscriber/usecase.py`. Some exceptions carry MB-scale payloads (validation errors with the full request body, `asyncpg.DataError` with the rejected row); an unbounded `repr` would extend the writer round-trip on a poison row and bloat the DLQ. Truncation appends `…[truncated]`. The DLQ `failure_reason` column is `String(64)` (current literals fit in 14 bytes; the breathing room lets the canonical set grow without a column-widening migration).
+`last_exception` is rendered by `_render_last_exception` (`subscriber/usecase.py`) and bounded by `_LAST_EXCEPTION_MAX_CHARS=8192`. The default render is `repr(exc)`; some exceptions carry MB-scale payloads (validation errors with the full request body, `asyncpg.DataError` with the rejected row), so an unbounded `repr` would extend the writer round-trip on a poison row and bloat the DLQ — truncation appends `…[truncated]`. Because that `repr` can embed payloads / PII / credentials, `OutboxBroker(..., last_exception_renderer=...)` (a `Callable[[BaseException], str | None]`, read from `OutboxBrokerConfig.last_exception_renderer`) lets a deployment redact (`type(exc).__name__`) or drop it (`None`); a custom renderer's output is still length-capped. The DLQ `failure_reason` column is `String(64)` (current literals fit in 14 bytes; the breathing room lets the canonical set grow without a column-widening migration).
 
 ## Retention
 
@@ -33,3 +33,5 @@ There is no built-in retention/pruning. Operators are responsible for archival 
 ## `validate_schema()` mechanics
 
 `validate_schema()` delegates to `alembic.autogenerate.compare_metadata` against a throwaway `MetaData` populated by `make_outbox_table(...)` — so the canonical `Table` is the single source of truth and the validator never duplicates the schema declaration. It only flags **missing** schema (`add_*` / `modify_*` ops); `remove_*` ops are intentionally ignored so users may attach extras (audit columns, their own indexes). Alembic is an **optional dependency** (`faststream-outbox[validate]`); without it, `validate_schema()` raises `ImportError`, but every other code path works (the import lives at the top of `client.py` inside a try/except, with module-level sentinels `_alembic_compare_metadata` / `_AlembicMigrationContext` set to `None` on failure).
+
+Alembic's diff is **blind to three things the producer's `ON CONFLICT` arbiter and the lease invariant depend on**, so `validate_schema()` runs extra `pg_catalog` probes alongside it: the partial-index **WHERE predicates** (alembic ignores `postgresql_where`), the **uniqueness** of `timer_id_uq` (`pg_index.indisunique` — a same-named non-unique index passes the predicate check yet breaks `ON CONFLICT` at publish time), and the **`<table>_lease_ck` CHECK** definition (alembic has no check-constraint comparator). Each surfaces a drifted/non-partial/non-unique index or a missing/altered CHECK that the diff alone would miss.

architecture/drain.md

@@ -13,6 +13,7 @@ The subscriber carries two flags during shutdown: `self.running` (FastStream's e
 - `_fetch_inner`'s loop guard checks both: `while self.running and not self._stopping:`
 - The worker loop only checks `running`.
 - `stop()` flips `_stopping`, kicks the fetch loop awake via `_notify_event` (in case it's parked in an idle `_wait_for_notify_or_timeout`), waits up to `graceful_timeout` for `_inflight.join()`, then flips `running=False` and cancels the spawned tasks.
+- **A timed-out drain is observable.** If `_inflight.join()` doesn't finish within the budget (`move_on_after`'s `cancelled_caught`), `stop()` emits a `WARNING` and a `drain_timeout` recorder metric (`faststream_outbox_drain_timeout_total` / `messaging.outbox.drain_timeout`) before cancelling — abandoned in-flight rows are left to lease-expiry retry, and an operator can now tell a timed-out drain from a clean one.
 - **`graceful_timeout=None`** stays unbounded where FastStream uses it that way (e.g. `ping()`), but the drain wait clamps `None` to a finite fallback (`_DEFAULT_DRAIN_TIMEOUT_SECONDS = 15.0`). `anyio.move_on_after(None)` has deadline `inf`, so without the clamp a single wedged handler would make `_inflight.join()` — and thus `stop()` — never return.
 
 **Why we skip `super().stop()`.** Its `MultiLock.wait_release(graceful_timeout)` would either return instantly (healthy path; `_inflight.join()` already waited a stricter condition) or re-wait the same stuck handlers for another full budget (wedged path; **2× shutdown regression**). The subscriber inlines `TasksMixin.stop`'s cleanup body instead. Per-subscriber shutdown bound: `graceful_timeout`.
@@ -25,6 +26,8 @@ The subscriber carries two flags during shutdown: `self.running` (FastStream's e
 
 `return_exceptions=True` (not `TaskGroup`) so a stuck subscriber doesn't cancel the others mid-drain. Exception results are logged via `_log_subscriber_stop_error` and never re-raised — shutdown must complete even when individual subscribers misbehave.
 
+`OutboxBroker.stop` sets `self.running = False` **before** the gather (shutdown is irreversible at that point), so an external cancellation of `stop()` mid-gather can't leave the broker advertising `running=True` over already-stopped subscribers (which `ping()` reads).
+
 ## Phase interaction with `dispatch_one` guard
 
 During drain `self.running` stays True, so the `dispatch_one` guard (`not row.state_set and not self.running`) is dormant. After drain completes (or times out), `stop()` sets `running=False` before `task.cancel()`; any worker mid-`dispatch_one` at that instant then benefits from the guard against the silent-DELETE race.
@@ -37,6 +40,7 @@ Both overrides replace upstream FastStream methods. Stable for years upstream, b
 
 - `tests/test_fake.py::test_drain_finishes_inflight_rows_before_returning_in_fake_mode` — drain waits for in-flight rows (off-Postgres)
 - `tests/test_fake.py::test_broker_stop_cancels_wedged_handler_within_graceful_timeout_in_fake_mode` — graceful-timeout bound (off-Postgres)
+- `tests/test_fake.py::test_drain_timeout_emits_warning_and_metric_in_fake_mode` — timed-out drain surfaces a WARNING + `drain_timeout` metric (off-Postgres)
 - `tests/test_integration.py` — the Postgres-backed drain + parallel-gather coverage
 
 ## Test-broker gotcha

architecture/test-broker.md

@@ -8,7 +8,9 @@ User-facing: `docs/usage/testing.md`. Invariant summary: `CLAUDE.md` § Test bro
 
 ### Sync (default, `run_loops=False`)
 
-`broker.publish` synchronously routes through `OutboxSubscriber.dispatch_one` — matches the FastStream test-broker idiom (`TestKafkaBroker` / `TestRabbitBroker`). The handler runs before `publish` returns; no background loops. `broker.publish_batch`, `cancel_timer`, and `fetch_unprocessed` are also patched to operate on the fake client (the `session` argument is ignored).
+`broker.publish` synchronously routes through `OutboxSubscriber.dispatch_one` — matches the FastStream test-broker idiom (`TestKafkaBroker` / `TestRabbitBroker`). The handler runs before `publish` returns; no background loops. `broker.publish_batch`, `cancel_timer`, and `fetch_unprocessed` are also patched to operate on the fake client (the `session` argument is **ignored** — `del session` — which diverges from production's `isinstance(session, AsyncSession)` `TypeError`; tests needing the session contract must use a real `OutboxClient`). `OutboxResponse` is *not* faked, so its eager session/queue/activate validation still fires under the test broker.
+
+`_sync_dispatch` claims the just-fed row via the shared `_claim_fake_row` (the same lease + `deliveries_count++` mechanics `FakeOutboxClient.fetch` uses), so the `max_deliveries` boundary runs on one path; only the eligibility gate (`next_attempt_at <= now`) lives in `fetch`, which is why sync mode fires future-dated rows immediately.
 
 The broker's `producer` slot is swapped for `FakeOutboxProducer` (`testing.py`) so `publisher.publish()` lands rows in the same fake store via the FastStream `_basic_publish` flow — tests using `broker.publisher("q").publish(...)` work identically to `broker.publish(queue="q", ...)`.
 

architecture/timers.md

@@ -6,13 +6,16 @@ User-facing: `docs/usage/timers.md`. Invariant summary: `CLAUDE.md` § Timers.
 
 `activate_in: timedelta` / `activate_at: datetime` (mutually exclusive) set `next_attempt_at` so the row is invisible to fetch until the gate opens — the `next_attempt_at <= now()` predicate in the fetch CTE is what gates eligibility, so no subscriber-side change is needed for scheduling.
 
-- `publish`: `next_attempt_at` is computed **server-side** via `now() + make_interval(secs => :s)` to stay clock-skew-safe.
-- `publish_batch`: client-side (`datetime.now(UTC) + activate_in`) because executemany doesn't compose cleanly with column-level SQL expressions. The few-ms drift is harmless for user-supplied scheduling.
+- `publish` + `activate_in`: `next_attempt_at` is computed **server-side** via `now() + make_interval(secs => :s)` to stay clock-skew-safe.
+- `publish` + `activate_at`: bound as the caller's **absolute literal** datetime (worker clock) — there's nothing to make skew-safe, it's an exact instant the user supplied. Note the NOTIFY future-dating decision for `activate_at` therefore compares against the worker clock, not the DB's.
+- `publish_batch`: fully client-side (`datetime.now(UTC) + activate_in`, or the `activate_at` literal) because executemany doesn't compose cleanly with column-level SQL expressions. The few-ms drift is harmless for user-supplied scheduling.
 
 ## `timer_id` dedup
 
 `timer_id` (single `publish` only) flows into a `String(255)` column with a partial unique index on `(queue, timer_id) WHERE timer_id IS NOT NULL`. The producer switches to `pg_insert(...).on_conflict_do_nothing(index_elements=[queue, timer_id], index_where=timer_id IS NOT NULL)` so re-publishing the same id is a silent no-op (returns `None`).
 
+The dedup window is **one *live* row per `(queue, timer_id)`** — the partial unique index constrains only rows still in the outbox. Once a timer row is delivered (DELETEd) or terminally fails, a later `publish` with the same id inserts a fresh row. So `timer_id` is "at most one in flight", not a global once-ever idempotency key; the DLQ keeps `timer_id` non-unique, so audit consumers can see repeats.
+
 ## NOTIFY-skip conditions
 
 NOTIFY is skipped when the row is **genuinely future-dated** (`activate_in > 0`, or `activate_at` resolves to a time after `now()`) OR the `on_conflict_do_nothing` path returned no row — both cases would either wake listeners that find nothing, or wake them prematurely. A past/zero `activate_at`/`activate_in` is immediately eligible, so it **does** fire NOTIFY.