diff --git a/CLAUDE.md b/CLAUDE.md index a2c8a03..b6399db 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -46,7 +46,7 @@ For chained publishing, handlers can `return OutboxResponse(body=..., queue=..., `OutboxSubscriber` can source a FastStream-native cross-broker chain: `@kafka_pub @broker_outbox.subscriber("q")` (Kafka/Rabbit/NATS/Redis/Confluent). Upstream's `SubscriberUsecase.process_message` walks the publisher chain — no dispatch override is needed for the chain itself. Three guardrails on top: -- **Bad chain composition is refused.** `OutboxResponse(...)` + a non-`OutboxFakePublisher` in the chain raises `_OutboxConfigError` (private `RuntimeError` subclass) via `process_message` / `consume()` / `dispatch_one` overrides; rides `AcknowledgementMiddleware` to the normal nack path so the row is retried (and logged) until fixed. +- **Bad chain composition is refused.** `OutboxResponse(...)` + a non-`OutboxFakePublisher` in the chain raises `_OutboxConfigError` (private `RuntimeError` subclass) via `process_message` / `consume()` / `dispatch_one` overrides; the worker loop catches it, logs it at ERROR, and leaves the row — the lease expires and another fetch reclaims it (retry via lease expiry, **not** the `retry_strategy`) until the config is fixed (P18). - **WARNING for unstarted foreign brokers at `start()`** — one per broker, deduped via `_warned_foreign_config_ids: set[int]`. - **`propagate_inbound_headers: bool = False`** — when True, inbound headers fill `Response.headers` only if the handler returned a `Response` with empty headers (user-set wins). Default False matches FastStream convention. @@ -70,11 +70,13 @@ Deep dive: `architecture/timers.md`. User-facing: `docs/usage/timers.md`. - `(queue, acquired_at) WHERE acquired_token IS NOT NULL` — fetch CTE Branch B (expired-lease reclaim). - unique `(queue, timer_id) WHERE timer_id IS NOT NULL` — `timer_id` dedup. +Plus a `CHECK ((acquired_token IS NULL) = (acquired_at IS NULL))` (the `_lease_ck` constraint) so a half-set lease is unrepresentable. + The fetch CTE's OR is written so each disjunct **explicitly carries its partial-index predicate as a conjunct** — Postgres only uses a partial index when the query implies its WHERE clause; the naive form falls back to seq-scan. Both fetch indexes pay write amplification on every claim. There is **no `state` column**: a row is "available" iff `acquired_token IS NULL` or `acquired_at < now() - lease_ttl_seconds`. Terminal failures `DELETE` by default; opt in to audit via `dlq_table=make_dlq_table(metadata)`. -`validate_schema()` is **opt-in** (call from `/health` or startup hook, not `broker.start()`) so migrations can run against the same DB without a loop. Alembic is optional (`faststream-outbox[validate]`); without it `validate_schema()` raises `ImportError` but every other path works. +`validate_schema()` is **opt-in** (call from `/health` or startup hook, not `broker.start()`) so migrations can run against the same DB without a loop. Beyond the alembic column/index diff it also probes the live partial-index **predicates** (alembic ignores `postgresql_where`), catching a drifted or non-partial `timer_id_uq` that would otherwise break `ON CONFLICT` at publish time (S2). Alembic is optional (`faststream-outbox[validate]`); without it `validate_schema()` raises `ImportError` but every other path works. ### Opt-in DLQ on terminal failure @@ -95,13 +97,13 @@ Deep dive: `architecture/dlq.md`. User-facing: `docs/usage/dlq.md`. Per subscriber: 1. **`_fetch_loop`** — long-lived `AsyncConnection` for the fetch CTE + separate raw asyncpg connection for `LISTEN outbox_
`. Single CTE: `SELECT … FOR UPDATE SKIP LOCKED → UPDATE acquired_token=:uuid, acquired_at=now() RETURNING *`. WHERE reclaims unleased rows **and** expired leases (`acquired_at < now() - make_interval(secs => :lease_ttl)`) — no separate reaper. NOTIFY shortcircuits idle sleep via `asyncio.Event` (idle latency from `max_fetch_interval` to ~10ms). LISTEN failures log once and fall back to polling. DB error → connections close, exponential backoff (`_BACKOFF_EXP_CAP=30`), reopen. -2. **`_worker_loop`** × `max_workers` — pulls from `asyncio.Queue(maxsize=fetch_batch_size)`, dispatches via `consume()`, flushes terminal state. Each worker owns a long-lived `AsyncConnection` (held across reconnect) and routes terminal writes through `delete_with_lease_with_conn` / `mark_pending_with_lease_with_conn` — drain of N rows costs O(workers) pool checkouts. Flush exceptions propagate (outer loop rebuilds the connection); inflight slot still releases in `finally`. Default `AckPolicy.NACK_ON_ERROR`; `REJECT_ON_ERROR` and `MANUAL` allowed. **`AckPolicy.ACK_FIRST` is rejected at registration with `ValueError`** — it would delete before the handler runs, defeating the outbox contract. `subscriber/factory.py` raises or warns on other footguns (`lease_ttl_seconds <= max_fetch_interval`, `max_deliveries` without retry, etc.). +2. **`_worker_loop`** × `max_workers` — pulls from `asyncio.Queue(maxsize=fetch_batch_size)`, dispatches via `consume()`, flushes terminal state. Each worker owns a long-lived `AsyncConnection` (held across reconnect) and routes terminal writes through `delete_with_lease(conn, …)` / `mark_pending_with_lease(conn, …)` — drain of N rows costs O(workers) pool checkouts. Flush exceptions propagate (outer loop rebuilds the connection); inflight slot still releases in `finally`. Default `AckPolicy.NACK_ON_ERROR`; `REJECT_ON_ERROR` and `MANUAL` allowed. **`AckPolicy.ACK_FIRST` is rejected at registration with `ValueError`** — it would delete before the handler runs, defeating the outbox contract. `subscriber/factory.py` raises or warns on other footguns (`lease_ttl_seconds <= max_fetch_interval`, `max_deliveries` without retry, etc.). `OutboxSubscriber.get_one()` and `__aiter__()` are explicit `NotImplementedError`s — point operators at `broker.fetch_unprocessed(session=..., queue=...)`. A peek that acquires a lease has surprising `deliveries_count` semantics; lease-free reads belong on `fetch_unprocessed`. **Connection budget.** Each subscriber holds `max_workers + 1` SQLAlchemy pool connections steady-state + one raw asyncpg connection for LISTEN. Size the pool for `Σ subscribers × (max_workers + 1)` or startup blocks on checkout. **Per process** — Postgres `max_connections` must cover `replicas × Σ subscribers × (max_workers + 1)` or rolling deploys hit `FATAL: too many connections`. -**NOTIFY semantics.** `broker.publish` / `publish_batch` emit `SELECT pg_notify('outbox_
', queue)` on the caller's session right after the INSERT, **except** when future-dated or `timer_id` conflict no-op'd the insert. NOTIFY is transactional — atomicity with the row is automatic; rolled-back transactions silently drop it. Channel naming is `outbox_`; Postgres limits identifiers to 63 chars, so table names longer than ~56 chars silently lose NOTIFY and degrade to polling. +**NOTIFY semantics.** `broker.publish` / `publish_batch` emit `SELECT pg_notify('outbox_
', queue)` on the caller's session right after the INSERT, **except** when future-dated or `timer_id` conflict no-op'd the insert. NOTIFY is transactional — atomicity with the row is automatic; rolled-back transactions silently drop it. Channel naming is `outbox_`. Postgres limits identifiers to 63 bytes; `make_outbox_table` **raises `ValueError`** when the longest derived identifier — an index name like `
_pending_idx`, longer than the NOTIFY channel itself — would exceed it — so over-long table names (~>51 bytes) are rejected at construction, not silently degraded to polling. ### Lease-token invariant — load-bearing @@ -122,7 +124,7 @@ Both `OutboxSubscriber.stop()` and `OutboxBroker.stop()` override FastStream par - **Subscriber: two flags during drain.** `self.running` (FastStream's "actively dispatching") stays True for the duration of drain; `self._stopping` (new) signals "no new claims". `_fetch_inner` checks both; the worker loop only `running`. `stop()` flips `_stopping`, kicks `_notify_event`, waits up to `graceful_timeout` for `_inflight.join()`, then flips `running=False` and cancels tasks. `super().stop()` is **not** called — its `MultiLock.wait_release` would re-wait stuck handlers for another full budget (2× shutdown regression). - **Broker: parallel-gather subscriber stop** via `asyncio.gather(..., return_exceptions=True)` — sequential N × `graceful_timeout` exceeds K8s default `terminationGracePeriodSeconds=30s` once a service has 2+ subscribers. Exceptions logged via `_log_subscriber_stop_error`, never re-raised. - **Phase interaction.** During drain `running` stays True so the `dispatch_one` guard is dormant; after drain `running=False` is set before `task.cancel()` so workers mid-`dispatch_one` benefit from the guard. The two changes are complementary. -- **Upstream divergence flag.** If FastStream adds cleanup to `BrokerUsecase.stop`, `SubscriberUsecase.stop`, or `TasksMixin.stop`, we silently miss it. **Re-check both overrides when touching shutdown.** Regression tests in `tests/test_fake.py` (`grep test_drain_timeout` / `test_broker_stop`). +- **Upstream divergence flag.** If FastStream adds cleanup to `BrokerUsecase.stop`, `SubscriberUsecase.stop`, or `TasksMixin.stop`, we silently miss it. **Re-check both overrides when touching shutdown.** Regression tests in `tests/test_fake.py` (`test_drain_finishes_inflight_rows_before_returning_in_fake_mode`, `test_broker_stop_cancels_wedged_handler_within_graceful_timeout_in_fake_mode`) and the Postgres-backed `tests/test_integration.py`. - **Test-broker gotcha.** `_fake_close` sets `sub.running = False` and bypasses `subscriber.stop()` / `broker.stop()` entirely — drain tests must `await broker.stop()` explicitly inside the `async with` block. Deep dive: `architecture/drain.md`. @@ -158,7 +160,7 @@ Critical for the transactional contract: `wrap_callable_to_fastapi_compatible` ( ### Engine ownership -Caller owns the `AsyncEngine`. `OutboxBrokerConfig.disconnect()` deliberately does nothing; `EngineState` is a lazy holder so the broker can be constructed before the engine is wired (used by the test broker). +Caller owns the `AsyncEngine` — the broker never disposes it. The engine lives on `OutboxBrokerConfig` (set by the broker constructor) and may be `None` until wired, so the broker can be constructed before the engine exists (used by the test broker). ### Metrics + native middleware diff --git a/architecture/drain.md b/architecture/drain.md index 859216e..0f42a62 100644 --- a/architecture/drain.md +++ b/architecture/drain.md @@ -34,8 +34,9 @@ The two changes are complementary — the `dispatch_one` guard covers correctnes Both overrides replace upstream FastStream methods. Stable for years upstream, but if FastStream adds new cleanup to `BrokerUsecase.stop`, `SubscriberUsecase.stop`, or `TasksMixin.stop`, we silently miss it. **Reviewers touching shutdown must re-check both overrides.** Regression tests pin both behaviors: -- `tests/test_fake.py::test_drain_timeout_strict_bound_per_subscriber` — per-subscriber strict bound -- `tests/test_fake.py::test_broker_stop_runs_subscribers_in_parallel` — gather shape +- `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_integration.py` — the Postgres-backed drain + parallel-gather coverage ## Test-broker gotcha diff --git a/architecture/relay.md b/architecture/relay.md index ff18c1a..b1f8ac0 100644 --- a/architecture/relay.md +++ b/architecture/relay.md @@ -16,7 +16,7 @@ c. `AcknowledgementMiddleware.__aexit__` turns publisher-chain exceptions into o ### `OutboxResponse` + foreign publisher refused -`OutboxSubscriber` overrides `process_message` to check chain composition. If a handler returns `OutboxResponse(...)` while having a non-`OutboxFakePublisher` entry in `handler._publishers`, the override raises `_OutboxConfigError` (a private `RuntimeError` subclass). The subscriber also overrides `consume()` and extends `dispatch_one`'s exception handler to re-raise `_OutboxConfigError` rather than swallowing it via upstream's `except Exception: pass`. The exception propagates through `AcknowledgementMiddleware` and triggers the outbox's normal nack path so the row is retried (and the operator sees the error log) until the handler is fixed. +`OutboxSubscriber` overrides `process_message` to check chain composition. If a handler returns `OutboxResponse(...)` while having a non-`OutboxFakePublisher` entry in `handler._publishers`, the override raises `_OutboxConfigError` (a private `RuntimeError` subclass). The subscriber also overrides `consume()` and extends `dispatch_one`'s exception handler to re-raise `_OutboxConfigError` rather than swallowing it via upstream's `except Exception: pass`. The re-raised error unwinds out of `dispatch_one` **before** any terminal flush, so the worker loop catches it, logs it at ERROR, and moves on — it does **not** route the row through the reconnect/backoff path (which would throttle unrelated rows) and no nack is ever flushed. The row's lease simply expires and a later fetch reclaims it (retry via lease expiry, **not** the `retry_strategy`) until the handler is fixed (P18). ### WARNING for unstarted foreign brokers at `start()` diff --git a/architecture/timers.md b/architecture/timers.md index e245a81..a468ee7 100644 --- a/architecture/timers.md +++ b/architecture/timers.md @@ -15,7 +15,7 @@ User-facing: `docs/usage/timers.md`. Invariant summary: `CLAUDE.md` § Timers. ## NOTIFY-skip conditions -NOTIFY is skipped when `activate_in` / `activate_at` is set OR the conflict path returned no row — both cases would either wake listeners that find nothing, or wake them prematurely. +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. ## `cancel_timer` lease guard diff --git a/planning/active/2026-06-12-code-audit-findings.md b/planning/archived/2026-06-12-code-audit-findings.md similarity index 98% rename from planning/active/2026-06-12-code-audit-findings.md rename to planning/archived/2026-06-12-code-audit-findings.md index 89dfe4a..a6ca584 100644 --- a/planning/active/2026-06-12-code-audit-findings.md +++ b/planning/archived/2026-06-12-code-audit-findings.md @@ -1,9 +1,16 @@ --- +status: shipped date: 2026-06-12 +slug: 2026-06-12-code-audit-findings scope: faststream_outbox/ (package) + tests/ (test quality) -status: bugs-remediated # B1-B16 fixed in PR #61; suspected/test-holes/improvements still open -bugs_pr: 61 -bugs_remediated: 2026-06-13 +prs: [61, 66, 67, 68, 69, 70] +findings_doc_prs: [62, 65] +releases: ["0.9.0", "0.9.1"] +outcome: > + All findings remediated, nothing deferred. Bugs B1–B16 (#61) + test-holes + T1–T8 (#66) + improvements P1–P35 (#67) shipped in 0.9.0; suspected S1–S5 + (#68; S3 was already resolved by P17) + warning-attribution P27 (#69) + + test-broker dedup/NOTIFY P29/P30 (#70) shipped in 0.9.1. --- # Code audit findings — 2026-06-12 diff --git a/planning/releases/0.9.0.md b/planning/releases/0.9.0.md index 52bc9f4..464d98d 100644 --- a/planning/releases/0.9.0.md +++ b/planning/releases/0.9.0.md @@ -107,5 +107,5 @@ No change is needed for the common path: a publish with a valid queue, a handler ## See also -- Audit findings: [`planning/active/2026-06-12-code-audit-findings.md`](../active/2026-06-12-code-audit-findings.md). +- Audit findings: [`planning/archived/2026-06-12-code-audit-findings.md`](../archived/2026-06-12-code-audit-findings.md). - PRs: [#61](https://github.com/modern-python/faststream-outbox/pull/61) (bugs B1–B16), [#66](https://github.com/modern-python/faststream-outbox/pull/66) (test-holes T1–T8), [#67](https://github.com/modern-python/faststream-outbox/pull/67) (improvements P1–P35). diff --git a/planning/releases/0.9.1.md b/planning/releases/0.9.1.md index cd7579a..b8ce4b2 100644 --- a/planning/releases/0.9.1.md +++ b/planning/releases/0.9.1.md @@ -31,5 +31,5 @@ No other behavior change. Producers, subscribers, the lease/terminal-write paths ## See also -- Audit findings + resolution: [`planning/active/2026-06-12-code-audit-findings.md`](../active/2026-06-12-code-audit-findings.md). +- Audit findings + resolution: [`planning/archived/2026-06-12-code-audit-findings.md`](../archived/2026-06-12-code-audit-findings.md). - PRs: [#68](https://github.com/modern-python/faststream-outbox/pull/68) (S1–S5), [#69](https://github.com/modern-python/faststream-outbox/pull/69) (P27), [#70](https://github.com/modern-python/faststream-outbox/pull/70) (P29/P30).