From dd98e23859c920e4c82f03828fc06847371ee680 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sun, 14 Jun 2026 17:50:38 +0300 Subject: [PATCH] docs: refresh architecture/ deep-dives for the 0.10.0 surfaces Promote the pass-3 / 0.10.0 conclusions into the architecture/*.md deep-dives so they match the shipped code (the audit updated CLAUDE.md inline; these lagged): - dlq.md: last_exception_renderer redaction hook (F3-01); validate_schema's extra pg_catalog probes (partial-index predicate, timer_id_uq indisunique, lease CHECK). - drain.md: timed-out drain emits WARNING + drain_timeout metric (F1-04); broker.stop sets running=False before the gather (F1-03); new drain_timeout regression test. - timers.md: activate_at on single publish is a client literal (not make_interval), worker-clock NOTIFY (F2-04/F6-13); timer_id dedup is one-live-row, not once-ever (F2-06). - test-broker.md: fake session leniency diverges from production (F4-09), OutboxResponse validation still fires; shared _claim_fake_row claim mechanics (F1-08). relay.md (F5-01) and metrics.md (drain_timeout) were already current. Docs-only; just lint clean. Co-Authored-By: Claude Opus 4.8 (1M context) --- architecture/dlq.md | 4 +++- architecture/drain.md | 4 ++++ architecture/test-broker.md | 4 +++- architecture/timers.md | 7 +++++-- 4 files changed, 15 insertions(+), 4 deletions(-) diff --git a/architecture/dlq.md b/architecture/dlq.md index 59fe7b4..e7fe7f0 100644 --- a/architecture/dlq.md +++ b/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 **`_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. diff --git a/architecture/drain.md b/architecture/drain.md index 620e66a..8a6605f 100644 --- a/architecture/drain.md +++ b/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 diff --git a/architecture/test-broker.md b/architecture/test-broker.md index 8b82fcc..9a6dcba 100644 --- a/architecture/test-broker.md +++ b/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", ...)`. diff --git a/architecture/timers.md b/architecture/timers.md index a468ee7..1ebdfe9 100644 --- a/architecture/timers.md +++ b/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.