refactor: address audit improvements P1–P35 (33 implemented, 2 deferred)

faststream_outbox/broker.py

@@ -10,6 +10,7 @@
 import datetime as _dt
 import logging
 import typing
+import warnings
 from collections.abc import Iterable, Sequence
 from types import TracebackType
 
@@ -42,6 +43,8 @@
 
 
 if typing.TYPE_CHECKING:
+    from weakref import WeakSet
+
     from fast_depends.dependencies import Dependant
     from fast_depends.library.serializer import SerializerProto
     from faststream._internal.context.repository import ContextRepo
@@ -121,7 +124,9 @@ class OutboxBroker(
 ):
     """FastStream broker backed by a Postgres outbox table."""
 
-    _subscribers: list["OutboxSubscriber"]
+    # P25: the runtime container is a WeakSet (set by the upstream Registrator), not a
+    # list — annotate it accurately so the _subscribers/subscribers distinction is clear.
+    _subscribers: "WeakSet[OutboxSubscriber]"
 
     def __init__(  # noqa: PLR0913
         self,
@@ -212,6 +217,11 @@ async def _connect(self) -> "AsyncEngine":
 
     @typing.override
     async def __aenter__(self) -> typing.Self:
+        # Upstream equivalent (replaced):
+        #   BrokerUsecase.__aenter__ -> faststream/_internal/broker/broker.py
+        # Upstream's __aenter__ only connects; we upgrade to a full start() so the
+        # subscriber loops spin up under `async with broker:`. Re-check this if upstream
+        # changes __aenter__/__aexit__ pairing (P26).
         await self.start()
         return self
 
@@ -220,6 +230,32 @@ async def start(self) -> None:
         await self.connect()
         await super().start()
         self._warn_on_unstarted_foreign_publishers()
+        self._warn_on_duplicate_queues()
+
+    def _warn_on_duplicate_queues(self) -> None:
+        # P22: the registration-time overlap warning only sees the broker's own
+        # _subscribers, so duplicates introduced via include_router slip through. Re-check
+        # at start() over the full ``subscribers`` property — but warn only for queues a
+        # router contributed to, since same-broker overlaps were already flagged at
+        # registration time (avoids double-warning one mistake).
+        direct = set(self._subscribers)
+        counts: dict[str, int] = {}
+        router_queues: set[str] = set()
+        for sub in self.subscribers:
+            from_router = sub not in direct
+            for q in getattr(sub, "_queues", []):
+                counts[q] = counts.get(q, 0) + 1
+                if from_router:
+                    router_queues.add(q)
+        duplicated = sorted(q for q, n in counts.items() if n > 1 and q in router_queues)
+        if duplicated:
+            warnings.warn(
+                f"Multiple subscribers serve queue(s) {duplicated}: their workers compete "
+                f"for the same rows (nondeterministic via SKIP LOCKED). Use one subscriber "
+                f"per queue, or attach multiple handlers to a single subscriber.",
+                UserWarning,
+                stacklevel=2,
+            )
 
     def _warn_on_unstarted_foreign_publishers(self) -> None:
         """
@@ -251,7 +287,9 @@ def _warn_on_unstarted_foreign_publishers(self) -> None:
                     if key in self._warned_foreign_config_ids:
                         continue
                     self._warned_foreign_config_ids.add(key)
-                    queues = sorted({q for s in self.subscribers for q in getattr(s, "_queues", [])})
+                    # P21: name the queue(s) of the subscriber actually decorated by this
+                    # foreign publisher, not every subscriber on the broker.
+                    queues = sorted(getattr(sub, "_queues", []))
                     _logger.warning(
                         "Foreign publisher %r is decorated on outbox subscriber(s) for "
                         "queue(s) %s, but its broker has not been started yet. The first "
@@ -276,11 +314,15 @@ async def stop(self, *_args: object, **_kwargs: object) -> None:
         #
         # Upstream equivalent (replaced):
         #   BrokerUsecase.stop -> faststream/_internal/broker/broker.py
+        # P20: snapshot once — re-evaluating the ``subscribers`` property after the await
+        # (e.g. a mid-shutdown include_router) would desync the strict zip and raise out of
+        # stop(), defeating its never-raise contract.
+        subs = list(self.subscribers)
         results = await asyncio.gather(
-            *(sub.stop() for sub in self.subscribers),
+            *(sub.stop() for sub in subs),
             return_exceptions=True,
         )
-        for sub, result in zip(self.subscribers, results, strict=True):
+        for sub, result in zip(subs, results, strict=True):
             if isinstance(result, BaseException):
                 self._log_subscriber_stop_error(sub, result)
         self.running = False
@@ -384,6 +426,11 @@ async def publish_batch(  # ty: ignore[invalid-method-override]
         every row in the batch identically — per-row timer dedup is not supported,
         use :meth:`publish` for that.
         """
+        # P1: validate the session type up front so an empty batch fails the same way a
+        # non-empty one does (the command constructor that checks it is only built below).
+        if not isinstance(session, AsyncSession):
+            msg = "broker.publish_batch requires an sqlalchemy.ext.asyncio.AsyncSession"
+            raise TypeError(msg)
         if not bodies:
             # Validate the activate args even when there's no work so callers
             # get the same misuse error on empty batches as on real ones.

faststream_outbox/client.py

@@ -21,9 +21,12 @@
 from typing import TYPE_CHECKING
 
 from sqlalchemy import (
+    ARRAY,
     Float,
     MetaData,
+    String,
     and_,
+    any_,
     bindparam,
     delete,
     func,
@@ -182,7 +185,11 @@ async def fetch(
             select(t.c.id)
             .where(
                 t.c.next_attempt_at <= func.now(),
-                or_(*(t.c.queue == q for q in queues)),
+                # P11: ``queue = ANY(:queues)`` binds a single array param, so the SQL
+                # text is stable regardless of the number of queues — asyncpg can reuse
+                # the prepared statement across ticks instead of recompiling an
+                # OR-of-N-equalities whose text changes with the queue count.
+                t.c.queue == any_(bindparam("queues", list(queues), type_=ARRAY(String))),
                 # The OR is split into two index-implying disjuncts so Postgres'
                 # partial-index inference picks up both `_pending_idx` (Branch A,
                 # `acquired_token IS NULL`) and `_lease_idx` (Branch B,
@@ -260,7 +267,16 @@ async def delete_with_lease(
         if conn is None:
             msg = "OutboxClient.delete_with_lease requires a live AsyncConnection (got None)"
             raise TypeError(msg)
-        if dlq_payload is not None and self._dlq_table is not None:
+        if dlq_payload is not None:
+            if self._dlq_table is None:
+                # P10: a dlq_payload with no dlq_table would silently degrade to a plain
+                # DELETE — losing the audit row. If the broker/client wiring ever desyncs,
+                # fail loudly rather than drop forensics.
+                msg = (
+                    "delete_with_lease received a dlq_payload but the client has no dlq_table; "
+                    "refusing to drop the audit row silently"
+                )
+                raise RuntimeError(msg)
             stmt, params = self._build_dlq_cte_stmt(message_id, acquired_token, dlq_payload)
             result = await conn.execute(stmt, params)
             return (result.rowcount or 0) > 0
@@ -303,14 +319,14 @@ def _build_dlq_cte_stmt(
             f"WITH deleted AS ("  # noqa: S608
             f"DELETE FROM {outbox_name} "
             f"WHERE id = :message_id AND acquired_token = :acquired_token "
-            f"RETURNING id, queue, payload, headers, deliveries_count, created_at"
+            f"RETURNING id, queue, payload, headers, deliveries_count, created_at, timer_id"
             f") "
             f"INSERT INTO {dlq_name} ("
             f"original_id, queue, payload, headers, deliveries_count, created_at, "
-            f"failure_reason, last_exception"
+            f"failure_reason, last_exception, timer_id"
             f") "
             f"SELECT id, queue, payload, headers, deliveries_count, created_at, "
-            f":failure_reason, :last_exception "
+            f":failure_reason, :last_exception, timer_id "
             f"FROM deleted"
         )
         sql = text(cte_sql)
@@ -405,6 +421,7 @@ def _row_to_message(row: dict) -> OutboxInnerMessage:
         last_attempt_at=row["last_attempt_at"],
         acquired_at=row["acquired_at"],
         acquired_token=row["acquired_token"],
+        timer_id=row["timer_id"],
     )
 
 

faststream_outbox/configs.py

@@ -30,11 +30,6 @@ class OutboxBrokerConfig(BrokerConfig):
     # copy audit data into this table in the same statement as the outbox DELETE.
     # See ``OutboxClient.delete_with_lease`` for the CTE shape.
     dlq_table: "Table | None" = None
-
-    async def connect(self) -> None:
-        # Engine and client are wired up by the broker's constructor; nothing to do here.
-        pass
-
-    async def disconnect(self) -> None:
-        # Caller owns the engine — never dispose it here.
-        pass
+    # P24: the former connect()/disconnect() overrides were dead code — BrokerConfig has
+    # no such hooks and nothing in the package or upstream called them. The broker's own
+    # start()/connect() wire the engine/client; the caller owns the engine lifecycle.

faststream_outbox/envelope.py

@@ -38,5 +38,16 @@ def _encode_payload(
         raise ValueError(msg)
     if content_type:
         out_headers["content-type"] = content_type
-    out_headers.setdefault("correlation_id", correlation_id or gen_cor_id())
+    # An explicit ``correlation_id`` kwarg used to lose silently to a
+    # ``headers["correlation_id"]`` of a different value (the kwarg was dropped).
+    # Treat a genuine mismatch as a conflict (like content-type above); otherwise the
+    # kwarg wins when set, falling back to the header, then a fresh id (P2).
+    header_cid = out_headers.get("correlation_id")
+    if correlation_id is not None and header_cid is not None and correlation_id != header_cid:
+        msg = (
+            f"correlation_id={correlation_id!r} conflicts with headers['correlation_id']="
+            f"{header_cid!r}. Pass correlation_id one way, not both."
+        )
+        raise ValueError(msg)
+    out_headers["correlation_id"] = correlation_id or header_cid or gen_cor_id()
     return payload, out_headers

faststream_outbox/message.py

@@ -62,6 +62,10 @@ class OutboxInnerMessage:
     acquired_at: _dt.datetime | None
     acquired_token: uuid.UUID | None
 
+    # P9: the originating timer_id (single-publish dedup key), so a terminally-failed
+    # timer keeps it in the DLQ audit trail. None for non-timer rows.
+    timer_id: str | None = None
+
     retry_strategy: "RetryStrategyProto | None" = None
     last_exception: BaseException | None = None
 
@@ -109,7 +113,9 @@ async def _nack(self) -> 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]
+                    # _record_attempt() above always sets first_attempt_at, so the
+                    # former ``or self.last_attempt_at`` fallback was dead code (P19).
+                    first_attempt_at=self.first_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,

faststream_outbox/metrics/__init__.py

@@ -21,6 +21,10 @@
   ran). ``exception_type`` is present when ``last_exception`` was set (post-handler raises;
   manual ``msg.reject()`` may omit it).
 * ``lease_lost`` — terminal flush found a foreign lease. Tags include ``phase`` (``terminal`` | ``retry``).
+  ``acked`` / ``nacked_retried`` / ``nacked_terminal`` are emitted only **after** the
+  terminal/retry write lands (P17); a row whose lease was reclaimed emits ``lease_lost``
+  *instead* (never a paired acked/nacked), so the row isn't counted twice when it
+  redelivers.
 * ``dlq_written`` — emitted from ``_flush_terminal`` after the DELETE+INSERT CTE commits.
   Fires only when ``OutboxBroker`` was constructed with ``dlq_table=...`` AND the row was
   terminal-by-failure (any ``nacked_terminal`` reason). Skipped on lease-lost. Tags:
@@ -31,7 +35,10 @@
 * ``published`` — producer-side insert. Tags include ``status`` (``success`` | ``error``),
   ``count, size_bytes, duration_seconds``. No ``subscriber`` tag.
   ``count`` is **messages landed**, not publish attempts — errors and ``timer_id``
-  no-ops carry ``count=0``. Counter-style adapters should `inc(count)` so totals
+  no-ops both carry ``count=0`` (P6). ``count=0`` alone does not distinguish them:
+  a successful ``timer_id`` conflict has ``status="success"`` with no ``exception_type``,
+  while a failed publish has ``status="error"`` with an ``exception_type``.
+  Counter-style adapters should `inc(count)` so totals
   reflect messages-on-the-wire; duration histograms record every attempt
   (including failures) so failed-publish latency stays observable.
 

faststream_outbox/metrics/prometheus.py

@@ -292,7 +292,12 @@ def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None:  # noqa:
             # every attempt (with the status label) so failed-publish latency stays
             # observable.
             count = tags.get("count", 1)
-            if count > 0:
+            if status == "error":
+                # P28: an error lands 0 messages but is one failed publish. Increment the
+                # status="error" series (the exact label dashboards alert on) by 1, rather
+                # than gating on count > 0 — which left it permanently at zero.
+                self._published_total.labels(*publish_base, status).inc()
+            elif count > 0:
                 self._published_total.labels(*publish_base, status).inc(count)
             duration = tags.get("duration_seconds")
             if duration is not None:

faststream_outbox/publisher/producer.py

@@ -79,15 +79,18 @@ def disconnect(self) -> None:
         pass
 
     async def publish(self, cmd: OutboxPublishCommand) -> int | None:
-        payload, hdrs = _encode_payload(
-            cmd.body,
-            headers=cmd.headers,
-            correlation_id=cmd.correlation_id,
-            serializer=self.serializer,
-        )
         start_perf = time.perf_counter()
-        size_bytes = len(payload)
+        size_bytes = 0
         try:
+            # P3: encode inside the try so a serialization / content-type / correlation_id
+            # failure still emits the ``published`` error metric instead of bypassing it.
+            payload, hdrs = _encode_payload(
+                cmd.body,
+                headers=cmd.headers,
+                correlation_id=cmd.correlation_id,
+                serializer=self.serializer,
+            )
+            size_bytes = len(payload)
             row_id = await self._do_publish(cmd, payload, hdrs)
         except Exception as exc:
             self._emit_metric(
@@ -173,15 +176,17 @@ async def publish_batch(self, cmd: OutboxPublishCommand) -> None:
             next_at = cmd.activate_at
         rows: list[dict[str, typing.Any]] = []
         total_size = 0
-        for body in bodies:
-            payload, hdrs = _encode_payload(body, headers=cmd.headers, serializer=self.serializer)
-            total_size += len(payload)
-            row: dict[str, typing.Any] = {"queue": cmd.queue, "payload": payload, "headers": hdrs}
-            if next_at is not None:
-                row["next_attempt_at"] = next_at
-            rows.append(row)
         start_perf = time.perf_counter()
         try:
+            # P3: encode inside the try so a serialization failure on any body still
+            # emits the ``published`` error metric.
+            for body in bodies:
+                payload, hdrs = _encode_payload(body, headers=cmd.headers, serializer=self.serializer)
+                total_size += len(payload)
+                row: dict[str, typing.Any] = {"queue": cmd.queue, "payload": payload, "headers": hdrs}
+                if next_at is not None:
+                    row["next_attempt_at"] = next_at
+                rows.append(row)
             await cmd.session.execute(insert(self._table), rows)
             # Skip NOTIFY only when genuinely future-dated; past times are eligible.
             if next_at is None or next_at <= now:

faststream_outbox/response.py

@@ -20,6 +20,10 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 
+# Matches the ``queue`` column width in ``make_outbox_table`` (``String(255)``).
+_MAX_QUEUE_LENGTH = 255
+
+
 class OutboxPublishCommand(BatchPublishCommand):
     """Outbox-specific publish command: carries session + scheduling fields end-to-end."""
 
@@ -46,6 +50,24 @@ def __init__(
         if activate_at is not None and activate_at.tzinfo is None:
             msg = "OutboxPublishCommand requires activate_at to be timezone-aware"
             raise ValueError(msg)
+        # P5: ``queue`` reaches SQL unvalidated — empty / non-str / over the
+        # ``String(255)`` column would surface as an opaque DB error (or silent
+        # truncation). Reject at the single-source-of-truth constructor instead.
+        if not isinstance(queue, str):
+            msg = f"queue must be a str, got {type(queue).__name__}"
+            raise TypeError(msg)
+        if not queue:
+            msg = "queue must be a non-empty string"
+            raise ValueError(msg)
+        if len(queue) > _MAX_QUEUE_LENGTH:
+            msg = f"queue must be at most {_MAX_QUEUE_LENGTH} characters (got {len(queue)})"
+            raise ValueError(msg)
+        # P4: timer_id / correlation_id are per-row single-publish concepts the batch
+        # path silently drops (each batched row gets its own auto correlation_id and
+        # no dedup key). Reject them on a batch command rather than accept-and-ignore.
+        if bodies and (timer_id is not None or correlation_id is not None):
+            msg = "timer_id / correlation_id are not supported for batch publishes (multiple bodies)"
+            raise ValueError(msg)
         super().__init__(
             body,
             *bodies,