fix: resolve 16 confirmed bugs from the 2026-06-12 code audit (B1–B16)

faststream_outbox/_import_checker.py

@@ -21,10 +21,21 @@
 is_prometheus_client_installed = find_spec("prometheus_client") is not None
 
 
+def missing_extra_message(component: str, extra: str) -> str:
+    """
+    Build the friendly "this needs an optional extra" install hint.
+
+    Single source of truth for the message text so the import-time guard and the
+    ``__init__`` probe guard in each middleware module stay in sync (B13).
+    """
+    return f"{component} requires the '{extra}' extra: pip install 'faststream-outbox[{extra}]'"
+
+
 __all__ = [
     "is_alembic_installed",
     "is_asyncpg_installed",
     "is_fastapi_installed",
     "is_opentelemetry_installed",
     "is_prometheus_client_installed",
+    "missing_extra_message",
 ]

faststream_outbox/broker.py

@@ -13,6 +13,7 @@
 from collections.abc import Iterable, Sequence
 from types import TracebackType
 
+import anyio
 from faststream import BaseMiddleware
 from faststream._internal.basic_types import LoggerProto
 from faststream._internal.broker import BrokerUsecase
@@ -298,16 +299,28 @@ def _log_subscriber_stop_error(self, sub: object, exc: BaseException) -> None:
 
     @typing.override
     async def ping(self, timeout: float | None = None) -> bool:
-        client = self.config.broker_config.client
-        if client is None:
-            return False
-        if not await client.ping():
-            return False
-        for subscriber in self._subscribers:
-            for task in subscriber.tasks:
-                if task.done():
-                    return False
-        return True
+        # ``move_on_after(None)`` is an unbounded scope, so threading the caller's
+        # timeout keeps the historical "wait forever" default while honoring a bound
+        # when given — a black-holed TCP or a stuck pool checkout can no longer hang
+        # the probe past *timeout*, which is the exact partition ``ping`` exists to
+        # detect (upstream brokers wrap their probe the same way).
+        with anyio.move_on_after(timeout):
+            client = self.config.broker_config.client
+            if client is None:
+                return False
+            if not await client.ping():
+                return False
+            # Walk the ``subscribers`` property, not the ``_subscribers`` list, so
+            # router-registered subscribers (the FastAPI pattern) are health-checked
+            # too — a dead worker task on a router subscriber must fail the probe.
+            for subscriber in self.subscribers:
+                outbox_sub = typing.cast("OutboxSubscriber", subscriber)
+                for task in outbox_sub.tasks:
+                    if task.done():
+                        return False
+            return True
+        # Only reached when the timeout scope cancelled the probe.
+        return False
 
     async def validate_schema(self) -> None:
         """Validate the user's table matches what the package expects. Opt-in."""

faststream_outbox/client.py

@@ -289,8 +289,13 @@ def _build_dlq_cte_stmt(
         dlq_table = self._dlq_table
         assert dlq_table is not None  # noqa: S101
         preparer = self._engine.dialect.identifier_preparer
-        outbox_name = preparer.quote(self._table.name)
-        dlq_name = preparer.quote(dlq_table.name)
+        # ``format_table`` renders the schema-qualified, quoted name (``"app"."outbox"``)
+        # when the Table carries a non-default ``schema=``. ``quote(table.name)`` dropped
+        # the schema, so a ``MetaData(schema="app")`` deployment hit ``UndefinedTable`` on
+        # every terminal failure (poison rows retry forever, the outbox grows) or silently
+        # wrote to a same-named search_path table (B10).
+        outbox_name = preparer.format_table(self._table)
+        dlq_name = preparer.format_table(dlq_table)
         # S608: outbox_name / dlq_name come from application-defined SQLAlchemy
         # Table objects (not request input) and are quoted via the dialect's
         # identifier preparer — values flow through :bindparam placeholders.
@@ -445,8 +450,11 @@ def _run_validate(
         raise ImportError(msg)
 
     # Isolated MetaData containing ONLY the canonical table, so the user's
-    # domain tables (in their own MetaData) don't show up in the diff.
-    canonical_metadata = MetaData()
+    # domain tables (in their own MetaData) don't show up in the diff. Carry the
+    # user's schema onto the canonical copy so the autogenerate diff compares
+    # ``app.outbox`` against ``app.outbox`` rather than the default search_path —
+    # matching the schema-qualified DLQ CTE in ``_build_dlq_cte`` (B10).
+    canonical_metadata = MetaData(schema=table.schema)
     canonical_factory(canonical_metadata, table.name)
 
     def _include_name(name: str | None, type_: str, parent_names: "Mapping[str, str | None]") -> bool:

faststream_outbox/message.py

@@ -14,7 +14,7 @@
 import uuid
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Any, Literal
 
 from faststream.message.message import StreamMessage
 
@@ -25,6 +25,9 @@
     from faststream_outbox.retry import RetryStrategyProto
 
 
+_logger = logging.getLogger("faststream_outbox.message")
+
+
 # Public contract for the DLQ ``failure_reason`` column and the ``reason`` tag on
 # ``nacked_terminal`` / ``dlq_written`` metric events. Operators query against these
 # literals (and dashboards key labels off them) — adding a new value is a public
@@ -74,13 +77,20 @@ class OutboxInnerMessage:
     # never touches the DLQ.
     terminal_failure_reason: "DLQFailureReason | None" = field(default=None, init=False)
 
-    async def ack(self) -> None:
+    # ``**_options`` are accepted-and-ignored: FastStream's AcknowledgementMiddleware
+    # forwards ``message.nack(**extra_options)`` for native idioms like
+    # ``raise NackMessage(delay=5)``. Those options map to broker-native ack
+    # semantics that don't apply to the outbox (reschedule timing is owned by the
+    # retry strategy, not a per-call delay). Rejecting them with a ``TypeError``
+    # here would be swallowed by the middleware and silently fall through to the
+    # destructive reject fallback — so we ignore them instead.
+    async def ack(self, **_options: Any) -> None:
         await self._update_state_if_not_set(self._ack)
 
-    async def nack(self) -> None:
+    async def nack(self, **_options: Any) -> None:
         await self._update_state_if_not_set(self._nack)
 
-    async def reject(self) -> None:
+    async def reject(self, **_options: Any) -> None:
         await self._update_state_if_not_set(self._reject)
 
     async def _update_state_if_not_set(self, fn: Callable[[], Awaitable[None]]) -> None:
@@ -95,16 +105,29 @@ async def _ack(self) -> None:
 
     async def _nack(self) -> None:
         self._record_attempt()
-        delay = (
-            self.retry_strategy.get_next_attempt_delay(
-                first_attempt_at=self.first_attempt_at or self.last_attempt_at,  # ty: ignore[invalid-argument-type]
-                last_attempt_at=self.last_attempt_at,  # ty: ignore[invalid-argument-type]
-                attempts_count=self.attempts_count,
-                exception=self.last_exception,
-            )
-            if self.retry_strategy is not None
-            else None
-        )
+        delay: float | None = None
+        if self.retry_strategy is not None:
+            try:
+                delay = self.retry_strategy.get_next_attempt_delay(
+                    first_attempt_at=self.first_attempt_at or self.last_attempt_at,  # ty: ignore[invalid-argument-type]
+                    last_attempt_at=self.last_attempt_at,  # ty: ignore[invalid-argument-type]
+                    attempts_count=self.attempts_count,
+                    exception=self.last_exception,
+                )
+            except Exception:
+                # A retry strategy that raises (a user bug, or an unclamped
+                # ExponentialRetry overflowing at very high attempt counts) must
+                # not destroy the row as ``"rejected"``. Degrade to terminal-by-
+                # retry so the row is deleted (and DLQ'd if configured) with a
+                # reason that reflects what happened, and surface the bug.
+                _logger.exception(
+                    "Retry strategy %r raised computing the next attempt delay for %r; treating delivery as terminal",
+                    self.retry_strategy,
+                    self,
+                )
+                self.to_delete = True
+                self.terminal_failure_reason = "retry_terminal"
+                return
         if delay is None:
             self.to_delete = True
             self.terminal_failure_reason = "retry_terminal"
@@ -138,14 +161,39 @@ def allow_delivery(self, *, max_deliveries: int | None, logger: "LoggerProto | N
         return True
 
     async def assert_state_set(self, logger: "LoggerProto | None") -> None:
-        """Manual-ack fallback: if the handler returned without ack/nack/reject, reject."""
-        if not self.state_set:
+        """
+        Fallback when the consume pipeline returned without recording ack/nack/reject intent.
+
+        Two distinct shapes land here:
+
+        * **The handler raised but nothing recorded intent** — ``AckPolicy.MANUAL``
+          disables the ack middleware entirely, and even under NACK/REJECT policies
+          the middleware swallows any error raised from ``nack()`` itself. In both
+          cases ``last_exception`` is set (the broker-wide capture middleware). A
+          failed delivery must not be destroyed: honor the retry strategy via
+          ``nack()`` so the row reschedules (or goes terminal-by-retry), matching
+          every native FastStream broker's "unacked failure -> redeliver" semantics.
+        * **The handler returned cleanly without acking** (``last_exception is None``)
+          — a genuinely forgetful MANUAL handler. Preserve the historical reject
+          fallback so the row doesn't redeliver forever.
+        """
+        if self.state_set:
+            return
+        if self.last_exception is not None:
             if logger is not None:
                 logger.log(
                     logging.ERROR,
-                    f"Outbox message {self} state not set after handler returned; rejecting as fallback",
+                    f"Outbox message {self} handler raised without recording ack state; "
+                    f"nacking to honor the retry strategy",
                 )
-            await self.reject()
+            await self.nack()
+            return
+        if logger is not None:
+            logger.log(
+                logging.ERROR,
+                f"Outbox message {self} state not set after handler returned; rejecting as fallback",
+            )
+        await self.reject()
 
     def __repr__(self) -> str:
         return f"OutboxInnerMessage(id={self.id}, queue={self.queue!r})"
@@ -154,14 +202,14 @@ def __repr__(self) -> str:
 class OutboxMessage(StreamMessage[OutboxInnerMessage]):
     """FastStream stream-message wrapper. Forwards ack/nack/reject to the inner row."""
 
-    async def ack(self) -> None:
-        await self.raw_message.ack()
+    async def ack(self, **options: Any) -> None:
+        await self.raw_message.ack(**options)
         await super().ack()
 
-    async def nack(self) -> None:
-        await self.raw_message.nack()
+    async def nack(self, **options: Any) -> None:
+        await self.raw_message.nack(**options)
         await super().nack()
 
-    async def reject(self) -> None:
-        await self.raw_message.reject()
+    async def reject(self, **options: Any) -> None:
+        await self.raw_message.reject(**options)
         await super().reject()

faststream_outbox/metrics/prometheus.py

@@ -38,20 +38,20 @@
 if typing.TYPE_CHECKING:
     from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram
 
+# ``faststream.prometheus`` imports ``prometheus_client`` at its module top, so its
+# import must be guarded behind the same probe as the third-party import. Importing
+# ``faststream.prometheus.container`` unconditionally raised ``ModuleNotFoundError`` at
+# import time for users without the ``[prometheus]`` extra — defeating the friendly
+# ``ImportError`` in ``PrometheusRecorder.__init__`` (B13). ``DEFAULT_SIZE_BUCKETS`` is a
+# class attribute on FastStream's MetricsContainer; re-exporting keeps bucket spacing in
+# sync with upstream when the extra is present.
 if is_prometheus_client_installed:
+    from faststream.prometheus.container import MetricsContainer as _UpstreamContainer
     from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram
 
-# ``DEFAULT_SIZE_BUCKETS`` is a class attribute on FastStream's MetricsContainer,
-# not a module-level constant. Re-export through the attribute access so any
-# future upstream tweak (different bucket spacing, additional buckets) flows
-# in automatically. Upstream ``faststream.prometheus`` is part of the dev group;
-# the import is unconditional rather than guarded behind an extra probe because
-# the canonical source of buckets lives in upstream FastStream, not in a separate
-# package.
-from faststream.prometheus.container import MetricsContainer as _UpstreamContainer
-
-
-_UPSTREAM_SIZE_BUCKETS: tuple[float, ...] = tuple(_UpstreamContainer.DEFAULT_SIZE_BUCKETS)
+    _UPSTREAM_SIZE_BUCKETS: tuple[float, ...] = tuple(_UpstreamContainer.DEFAULT_SIZE_BUCKETS)
+else:  # pragma: no cover - exercised only when the [prometheus] extra is absent
+    _UPSTREAM_SIZE_BUCKETS = ()
 
 # Mirror FastStream's PrometheusMiddleware duration histogram boundaries verbatim
 # so dashboards comparing process duration across brokers use the same buckets.
@@ -254,9 +254,14 @@ def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None:  # noqa:
             # Map outbox event names to upstream ``ProcessingStatus`` values.
             status = "acked" if event == "acked" else "nacked"
             self._processed_total.labels(*consume_base, status).inc()
-            self._in_process.labels(*consume_base).dec()
             duration = tags.get("duration_seconds")
             if duration is not None:
+                # ``duration_seconds`` is present exactly for terminals that followed a
+                # ``dispatched`` (which carries the matching in-process ``.inc()``).
+                # ``nacked_terminal(reason="max_deliveries")`` fires WITHOUT a preceding
+                # ``dispatched`` (the handler never ran) and carries no duration — dec'ing
+                # the gauge there drives ``..._in_process`` negative (B9).
+                self._in_process.labels(*consume_base).dec()
                 self._processed_duration.labels(*consume_base).observe(duration)
             if event == "nacked_terminal":
                 self._terminal_reason.labels(*consume_base, tags["reason"]).inc()

faststream_outbox/opentelemetry/middleware.py

@@ -9,9 +9,18 @@
 
 import typing
 
-from faststream.opentelemetry.middleware import TelemetryMiddleware
+from faststream_outbox._import_checker import is_opentelemetry_installed, missing_extra_message
+
+
+try:
+    from faststream.opentelemetry.middleware import TelemetryMiddleware
+except ImportError as _exc:  # pragma: no cover - only without the [opentelemetry] extra
+    # ``faststream.opentelemetry`` imports ``opentelemetry`` at module top, and the
+    # upstream class is needed at class-definition time so it can't be probe-guarded.
+    # Surface the friendly message at import time instead of a raw ModuleNotFoundError
+    # (B13); the __init__ probe guard below stays as defense for a falsified probe.
+    raise ImportError(missing_extra_message("OutboxTelemetryMiddleware", "opentelemetry")) from _exc
 
-from faststream_outbox._import_checker import is_opentelemetry_installed
 from faststream_outbox.opentelemetry.provider import OutboxTelemetrySettingsProvider
 from faststream_outbox.response import OutboxPublishCommand
 
@@ -33,11 +42,7 @@ def __init__(
         include_messages_counters: bool = False,
     ) -> None:
         if not is_opentelemetry_installed:
-            msg = (
-                "OutboxTelemetryMiddleware requires the 'opentelemetry' extra: "
-                "pip install 'faststream-outbox[opentelemetry]'"
-            )
-            raise ImportError(msg)
+            raise ImportError(missing_extra_message("OutboxTelemetryMiddleware", "opentelemetry"))
         super().__init__(
             settings_provider_factory=lambda _: OutboxTelemetrySettingsProvider(),
             tracer_provider=tracer_provider,

faststream_outbox/prometheus/middleware.py

@@ -12,9 +12,19 @@
 from collections.abc import Callable, Sequence
 
 from faststream._internal.constants import EMPTY
-from faststream.prometheus.middleware import PrometheusMiddleware
 
-from faststream_outbox._import_checker import is_prometheus_client_installed
+from faststream_outbox._import_checker import is_prometheus_client_installed, missing_extra_message
+
+
+try:
+    from faststream.prometheus.middleware import PrometheusMiddleware
+except ImportError as _exc:  # pragma: no cover - only without the [prometheus] extra
+    # ``faststream.prometheus`` imports ``prometheus_client`` at module top, and the
+    # upstream class is needed at class-definition time so it can't be probe-guarded.
+    # Surface the friendly message at import time instead of a raw ModuleNotFoundError
+    # (B13); the __init__ probe guard below stays as defense for a falsified probe.
+    raise ImportError(missing_extra_message("OutboxPrometheusMiddleware", "prometheus")) from _exc
+
 from faststream_outbox.message import OutboxInnerMessage
 from faststream_outbox.prometheus.provider import OutboxMetricsSettingsProvider
 from faststream_outbox.response import OutboxPublishCommand
@@ -39,11 +49,7 @@ def __init__(
         custom_labels: dict[str, str | Callable[[typing.Any], str]] | None = None,
     ) -> None:
         if not is_prometheus_client_installed:
-            msg = (
-                "OutboxPrometheusMiddleware requires the 'prometheus' extra: "
-                "pip install 'faststream-outbox[prometheus]'"
-            )
-            raise ImportError(msg)
+            raise ImportError(missing_extra_message("OutboxPrometheusMiddleware", "prometheus"))
         super().__init__(
             settings_provider_factory=lambda _: OutboxMetricsSettingsProvider(),
             registry=registry,

faststream_outbox/publisher/producer.py

@@ -163,8 +163,6 @@ async def _do_publish(
 
     async def publish_batch(self, cmd: OutboxPublishCommand) -> None:
         bodies = cmd.batch_bodies
-        if not bodies:
-            return
         # Client-side time for batch: executemany doesn't compose cleanly with
         # column-level SQL expressions, and a few-ms drift versus the DB is
         # harmless for user-supplied scheduling. Retries still use server time.

faststream_outbox/response.py

@@ -63,6 +63,17 @@ def __init__(
     def queue(self) -> str:
         return self.destination
 
+    @property
+    def batch_bodies(self) -> tuple[typing.Any, ...]:
+        # Upstream's PublishCommand.batch_bodies drops ``self.body`` when it is
+        # None, so a leading (or sole) None body silently vanishes from the batch
+        # — a lost row with no error and no metric. The outbox treats None as a
+        # valid body (``publish(None)`` inserts ``b""``), so every positional body
+        # must survive, in order. ``OutboxPublishCommand`` is the single source of
+        # truth, so overriding here fixes the producer, the fake producer, and the
+        # OpenTelemetry batch-count attribute in one place.
+        return (self.body, *self.extra_bodies)
+
     @classmethod
     def from_cmd(
         cls,