From 68c182090ed60415402f8953d38eb62f465353b5 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sun, 14 Jun 2026 15:53:04 +0300 Subject: [PATCH] feat: forward dlq_table + metrics_recorder through the FastAPI OutboxRouter (F8-01) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The FastAPI OutboxRouter built its inner broker internally and didn't expose dlq_table or metrics_recorder, so FastAPI deployments could not enable dead-letter archival or the recorder seam (fetched/lease_lost/nacked_terminal/dlq_written) at all — the operability features a production service needs. Both are plain OutboxBroker.__init__ kwargs that flow through StreamRouter's **connection_kwars (same mechanism as outbox_table), so add them to OutboxRouter.__init__ and forward them through super().__init__. routers stays unforwarded pending a design call (its lifespan/AsyncAPI semantics through the FastAPI router are unsettled) — now the sole open FastAPI item in planning/deferred.md. Test asserts both reach the inner broker and that the forwarded recorder actually fires on the dispatch path under the FastAPI router. Docs corrected (docs/usage/fastapi.md) — this also clears F8-02 (the docs wrongly said observability was unavailable via the router). Pass-3 findings doc + deferred.md updated. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/usage/fastapi.md | 35 +++++++++++-------- faststream_outbox/fastapi/router.py | 6 ++++ .../2026-06-14-deep-audit-pass3-findings.md | 3 +- planning/deferred.md | 25 ++++++------- tests/test_fastapi.py | 33 +++++++++++++++-- 5 files changed, 70 insertions(+), 32 deletions(-) diff --git a/docs/usage/fastapi.md b/docs/usage/fastapi.md index 38e4529..7847e71 100644 --- a/docs/usage/fastapi.md +++ b/docs/usage/fastapi.md @@ -141,20 +141,27 @@ on `OutboxRouter.__init__`: - `dependencies` — on the router signature this means FastAPI `Depends(...)` only; the broker's FastStream `Dependant` list is the wrong shape for this flow. -- `dlq_table`, `metrics_recorder`, and `routers` — simply not forwarded - through the router today. - -The router builds the broker for you and gives you no handle to a -pre-constructed `OutboxBroker`, and all three of those arguments are -constructor-only on the broker (no setters). So a FastAPI user **cannot** -enable the [DLQ](./dlq.md) or the [metrics-recorder -seam](./observability.md) through `OutboxRouter` today — there is no -router-based workaround. Use the standalone `OutboxBroker` for those. - -If you need broker-level FastStream middlewares, pass them to -`OutboxRouter(middlewares=[...])` — the router builds the broker for you, so -there is no separate broker to configure. Use the FastAPI `Depends(...)` -mechanism in handlers for dependencies. +- `routers` — not forwarded through the router; its semantics through the + FastAPI lifespan are unsettled. Register subscribers directly on the + `OutboxRouter` instead. + +The [DLQ](./dlq.md) and the [metrics-recorder seam](./observability.md) +**are** available through the router: pass `dlq_table=` and +`metrics_recorder=` to `OutboxRouter(...)` exactly as you would to +`OutboxBroker(...)` — they forward to the inner broker. + +```python +outbox_router = OutboxRouter( + engine, + outbox_table=outbox_table, + dlq_table=make_dlq_table(metadata), + metrics_recorder=my_recorder, +) +``` + +Native Prometheus/OpenTelemetry middleware also works via +`OutboxRouter(middlewares=[...])`. Use the FastAPI `Depends(...)` mechanism +in handlers for dependencies. ## Engine ownership diff --git a/faststream_outbox/fastapi/router.py b/faststream_outbox/fastapi/router.py index da6faf2..a97167f 100644 --- a/faststream_outbox/fastapi/router.py +++ b/faststream_outbox/fastapi/router.py @@ -47,6 +47,7 @@ from starlette.routing import BaseRoute from starlette.types import ASGIApp, Lifespan + from faststream_outbox.metrics import MetricsRecorder from faststream_outbox.publisher.usecase import OutboxPublisher from faststream_outbox.retry import RetryStrategyProto from faststream_outbox.subscriber.usecase import OutboxSubscriber @@ -69,6 +70,7 @@ def __init__( # noqa: PLR0913 engine: "AsyncEngine | None" = None, *, outbox_table: "Table", + dlq_table: "Table | None" = None, # Outbox broker kwargs (mirror ``OutboxBroker.__init__``). Note: ``apply_types`` # is fixed to False by ``StreamRouter`` (FastAPI's FastDepends is used instead), # and ``dependencies`` here means FastAPI dependencies — the broker's @@ -78,6 +80,8 @@ def __init__( # noqa: PLR0913 middlewares: Sequence["type[BaseMiddleware] | BrokerMiddleware[OutboxInnerMessage]"] = (), graceful_timeout: float | None = 15.0, serializer: "SerializerProto | None" = EMPTY, + # Metrics (recorder seam — mirrors ``OutboxBroker.__init__``) + metrics_recorder: "MetricsRecorder | None" = None, # AsyncAPI / Specification specification: typing.Optional["SpecificationFactory"] = None, description: str | None = None, @@ -114,6 +118,8 @@ def __init__( # noqa: PLR0913 # Outbox-broker kwargs (flow through StreamRouter's **connection_kwars # into ``OutboxBroker(...)``) outbox_table=outbox_table, + dlq_table=dlq_table, + metrics_recorder=metrics_recorder, decoder=decoder, parser=parser, graceful_timeout=graceful_timeout, diff --git a/planning/audits/2026-06-14-deep-audit-pass3-findings.md b/planning/audits/2026-06-14-deep-audit-pass3-findings.md index ca7e2c0..dc30b3e 100644 --- a/planning/audits/2026-06-14-deep-audit-pass3-findings.md +++ b/planning/audits/2026-06-14-deep-audit-pass3-findings.md @@ -69,6 +69,7 @@ None. #### [F8-01][gap] FastAPI `OutboxRouter` cannot enable DLQ or the metrics-recorder seam - **Location:** `fastapi/router.py:67-149` vs `broker.py:154,160,162`. **Verdict:** CONFIRMED. **KNOWN/deferred** (`planning/deferred.md`; excluded from passes 1–2 by design — this pass audited it per request). +- > **RESOLVED (2026-06-14, PR #88)** — `OutboxRouter.__init__` now accepts `dlq_table` + `metrics_recorder` and forwards them to the inner broker (via `StreamRouter`'s `**connection_kwars`). DLQ archival and the recorder seam (`fetched`/`lease_lost`/`nacked_terminal`/`dlq_written`) are now reachable for FastAPI users. Test: `tests/test_fastapi.py::test_outbox_router_forwards_dlq_table_and_metrics_recorder_to_inner_broker`. `routers` remains unforwarded pending a design call (now the sole open item in `planning/deferred.md`). Docs updated (`docs/usage/fastapi.md`), which also clears **F8-02** (it wrongly said observability was unavailable via the router). - **Problem.** `OutboxRouter.__init__` omits `dlq_table`, `metrics_recorder`, and `routers` from its signature and `super().__init__` passthrough. The inner broker is built during `super().__init__` with constructor-frozen config and no post-hoc injection handle, so a FastAPI deployment **cannot** enable dead-letter archival or the recorder-based signals (`fetched`, `lease_lost`, `nacked_terminal`, `dlq_written`) at all — exactly the operability features a production service needs. - **Verification refinement.** The fix is *more mechanical* than `deferred.md` frames it: `dlq_table` and `metrics_recorder` are plain `OutboxBroker` kwargs that already flow through `StreamRouter`'s `**connection_kwars` (this is how `outbox_table` reaches the broker today). Adding them to the router signature + passthrough would just work. Only `routers` warrants a design decision (its semantics through the FastAPI lifespan / AsyncAPI are questionable). - **Doc correction (see F8-02):** native Prometheus/OTel **middleware** observability *does* work via `OutboxRouter(middlewares=[...])`; only the *recorder* seam is blocked. The docs conflate the two. @@ -135,7 +136,7 @@ None. - **[F6-13]** CLAUDE.md says `publish` "computes server-side via `make_interval`" without qualification — only `activate_in` does; `activate_at` is a client literal. Tighten the line. - **[F6-15]** `fetch_unprocessed` docstring says "Intended for test assertions" but `get_one()`/`__aiter__()` point operators to it as the canonical lease-free read and it has an OOM guard "for backlogged production tables." Reword to include operator inspection. - **[F6-04]** `_validate_index_predicates_sync` docstrings phrase the predicate probes as if they run "through `_run_validate`"; they're siblings. Reword to "the alembic diff (`_run_validate`) doesn't catch these." (`client.py:468-557`) -- **[F8-02]** `docs/usage/fastapi.md` says FastAPI users can't get observability; in fact native Prometheus/OTel **middleware** works via `OutboxRouter(middlewares=[...])` — only the recorder seam is blocked (see F8-01). Narrow the claim. +- **[F8-02]** *(RESOLVED 2026-06-14, PR #88)* `docs/usage/fastapi.md` said FastAPI users can't get observability; corrected as part of F8-01 (the recorder seam now forwards, and native Prometheus/OTel middleware was always available via `OutboxRouter(middlewares=[...])`). - **[F6-06]** *(downgraded Medium→Low)* Two public classes named `OutboxRouter` (`router.py` include-router vs `fastapi/router.py` APIRouter subclass). Namespaced (different importable modules, standard Python convention), so no runtime shadowing — a docstring cross-reference suffices. (`router.py:65`, `fastapi/router.py:61`) **Tests:** diff --git a/planning/deferred.md b/planning/deferred.md index cc6d959..a87e542 100644 --- a/planning/deferred.md +++ b/planning/deferred.md @@ -14,18 +14,13 @@ but deliberately unscheduled pending a trigger. ### FastAPI integration -- **`OutboxRouter` doesn't forward `dlq_table` / `metrics_recorder` / - `routers`.** These three `OutboxBroker.__init__` arguments are not - exposed on `OutboxRouter.__init__`, and the router constructs the broker - internally with no handle to inject a pre-built one — so a FastAPI user - **cannot** enable the dead-letter queue or the metrics-recorder seam - through the router at all. The only path today is a standalone - `OutboxBroker`. This is documented as a limitation in - [`docs/usage/fastapi.md`](../docs/usage/fastapi.md) (audit improvement - P18, #72). Forwarding the kwargs is small in `OutboxRouter.__init__` + - the `super().__init__` passthrough, but it's a real feature with a - design surface (defaults, AsyncAPI/typing implications, whether - `routers` even makes sense through the FastAPI lifespan), not a - mechanical fix — hence a spec, not a drive-by. Revisit when a concrete - "FastAPI + DLQ" or "FastAPI + recorder seam" demand surfaces. - (`faststream_outbox/fastapi/router.py`) +- **`OutboxRouter` doesn't forward `routers`.** `dlq_table` and + `metrics_recorder` now forward to the inner broker (pass-3 audit F8-01, + PR #88) — a FastAPI user can enable the DLQ and the recorder seam through + the router. The remaining unforwarded `OutboxBroker.__init__` argument is + `routers`: its semantics through the FastAPI lifespan are unsettled + (the router contributes its own subscribers via `app.include_router`, so + a separate `routers` sequence's start/AsyncAPI behavior needs a design + call). Subscribers can be registered directly on the `OutboxRouter` + instead. Revisit if a concrete "include a sub-router under the FastAPI + outbox router" demand surfaces. (`faststream_outbox/fastapi/router.py`) diff --git a/tests/test_fastapi.py b/tests/test_fastapi.py index 8c3a440..0d2278f 100644 --- a/tests/test_fastapi.py +++ b/tests/test_fastapi.py @@ -9,7 +9,7 @@ yielding a session-shaped mock from a normal FastAPI dependency. """ -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Mapping from contextlib import asynccontextmanager from typing import Any from unittest.mock import AsyncMock @@ -21,7 +21,7 @@ from sqlalchemy import MetaData from sqlalchemy.ext.asyncio import AsyncSession -from faststream_outbox import NoRetry, OutboxMessage, OutboxResponse, make_outbox_table +from faststream_outbox import NoRetry, OutboxMessage, OutboxResponse, make_dlq_table, make_outbox_table from faststream_outbox.client import AbstractOutboxClient from faststream_outbox.fastapi import ( OutboxBroker as AnnotatedOutboxBroker, @@ -245,3 +245,32 @@ def test_outbox_router_forwards_broker_kwargs_to_inner_broker() -> None: """End-to-end forwarding: an outbox-broker kwarg passed to OutboxRouter reaches the broker.""" router = OutboxRouter(outbox_table=_make_outbox_table(), graceful_timeout=3.5) assert router.broker.config.graceful_timeout == 3.5 + + +async def test_outbox_router_forwards_dlq_table_and_metrics_recorder_to_inner_broker() -> None: + """F8-01: dlq_table + metrics_recorder reach the inner broker (DLQ + recorder seam usable under FastAPI).""" + metadata = MetaData() + t = make_outbox_table(metadata) + dlq = make_dlq_table(metadata) + events: list[str] = [] + + def recorder(event: str, tags: Mapping[str, Any]) -> None: + del tags + events.append(event) + + router = OutboxRouter(outbox_table=t, dlq_table=dlq, metrics_recorder=recorder) + + @router.subscriber("orders") + async def handle(body: dict) -> None: + del body + + # Both kwargs reached the inner broker... + assert router.broker._dlq_table is dlq # noqa: SLF001 + assert router.broker.config.metrics_recorder is recorder + + # ...and the forwarded recorder actually fires on the inner broker's dispatch path. + app = _make_app_with_router(router) + with TestClient(app): + await router.broker.publish({"x": 1}, queue="orders") # ty: ignore[missing-argument] + + assert events # the recorder seam is live under the FastAPI router