diff --git a/faststream_outbox/_import_checker.py b/faststream_outbox/_import_checker.py
index a70d049..797ee46 100644
--- a/faststream_outbox/_import_checker.py
+++ b/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",
 ]
diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py
index 0bb8f32..1441f80 100644
--- a/faststream_outbox/broker.py
+++ b/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."""
diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py
index ec727ad..77d8a87 100644
--- a/faststream_outbox/client.py
+++ b/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:
diff --git a/faststream_outbox/message.py b/faststream_outbox/message.py
index edbdfe8..9295b99 100644
--- a/faststream_outbox/message.py
+++ b/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()
diff --git a/faststream_outbox/metrics/prometheus.py b/faststream_outbox/metrics/prometheus.py
index eab02f6..17f8f38 100644
--- a/faststream_outbox/metrics/prometheus.py
+++ b/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()
diff --git a/faststream_outbox/opentelemetry/middleware.py b/faststream_outbox/opentelemetry/middleware.py
index c3d607b..955f0dd 100644
--- a/faststream_outbox/opentelemetry/middleware.py
+++ b/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,
diff --git a/faststream_outbox/prometheus/middleware.py b/faststream_outbox/prometheus/middleware.py
index 227b082..0abdd4a 100644
--- a/faststream_outbox/prometheus/middleware.py
+++ b/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,
diff --git a/faststream_outbox/publisher/producer.py b/faststream_outbox/publisher/producer.py
index bba46c1..5b6a652 100644
--- a/faststream_outbox/publisher/producer.py
+++ b/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.
diff --git a/faststream_outbox/response.py b/faststream_outbox/response.py
index 0303ae3..42e401c 100644
--- a/faststream_outbox/response.py
+++ b/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,
diff --git a/faststream_outbox/retry.py b/faststream_outbox/retry.py
index caf962c..8c51181 100644
--- a/faststream_outbox/retry.py
+++ b/faststream_outbox/retry.py
@@ -5,6 +5,15 @@
 from typing import Protocol
 
 
+# Absolute ceiling on a single computed delay. Guards two failure modes for an
+# ExponentialRetry left unbounded (no ``max_attempts`` and no ``max_delay_seconds``):
+# (1) ``multiplier ** attempts`` raising ``OverflowError`` at very high attempt
+# counts (``2.0 ** 1024``), and (2) producing a delay larger than Postgres'
+# ``make_interval(secs => …)`` can represent. ~100 years is far beyond any sane
+# retry horizon, so this never changes behavior for a real configuration.
+_MAX_DELAY_SECONDS = 100.0 * 365.0 * 24.0 * 60.0 * 60.0
+
+
 class RetryStrategyProto(Protocol):
     """
     Decides whether a Nack'ed row gets another attempt and how long to wait.
@@ -104,10 +113,18 @@ class ExponentialRetry(_RetryStrategyTemplate):
     _random: random.Random = field(default_factory=random.Random)
 
     def _delay_seconds(self, *, attempts_count: int) -> float:
-        delay = self.initial_delay_seconds * (self.multiplier ** max(0, attempts_count - 1))
+        try:
+            delay = self.initial_delay_seconds * (self.multiplier ** max(0, attempts_count - 1))
+        except OverflowError:
+            # An unbounded exponential eventually overflows float (``2.0 ** 1024``).
+            # Saturate at the absolute ceiling rather than letting the strategy
+            # raise into the destructive reject fallback.
+            delay = _MAX_DELAY_SECONDS
         # Jitter before clamp so max_delay_seconds is the true ceiling.
         if self.jitter_factor:
             delay *= 1.0 + self._random.uniform(-self.jitter_factor / 2, self.jitter_factor / 2)
         if self.max_delay_seconds is not None:
             delay = min(delay, self.max_delay_seconds)
-        return delay
+        # Absolute backstop so an unbounded config can't emit a delay Postgres'
+        # make_interval() can't represent.
+        return min(delay, _MAX_DELAY_SECONDS)
diff --git a/faststream_outbox/subscriber/usecase.py b/faststream_outbox/subscriber/usecase.py
index 1490f16..018bcbf 100644
--- a/faststream_outbox/subscriber/usecase.py
+++ b/faststream_outbox/subscriber/usecase.py
@@ -53,6 +53,10 @@
 
 _BACKOFF_EXP_CAP = 30
 _BACKOFF_MAX_SECONDS = 30.0
+# A reconnect loop whose connection stayed healthy at least this long before
+# failing is treated as recovered: the next failure starts a fresh backoff
+# sequence instead of inheriting the lifetime error count (B3).
+_BACKOFF_RESET_THRESHOLD_SECONDS = 60.0
 
 
 # Marker exception raised by programming guards inside ``process_message`` (e.g.
@@ -197,6 +201,11 @@ def _emit_metric(self, event: str, tags: Mapping[str, typing.Any]) -> None:
     @typing.override
     async def start(self) -> None:
         await super().start()
+        # Clear the drain flag so a stop()->start() cycle fetches again. Without
+        # this, _stopping stays True from the previous drain and the fetch loop's
+        # reconnect predicate exits immediately while ping() still reports healthy
+        # — the subscriber hot-spins connect/close and consumes nothing (B2).
+        self._stopping = False
         self._post_start()
         if not self.calls:
             return
@@ -249,6 +258,7 @@ async def _fetch_loop(self) -> None:
             name="fetch",
             open_resources=self._open_fetch_resources,
             inner=self._fetch_inner,
+            halt_on_drain=True,
         )
 
     @asynccontextmanager
@@ -359,17 +369,27 @@ async def _open_listen_connection(self, engine: "AsyncEngine") -> "_asyncpg.Conn
         _, opts = engine.dialect.create_connect_args(engine.url)
         for sa_only_key in ("prepared_statement_cache_size", "async_fallback", "async_creator_fn"):
             opts.pop(sa_only_key, None)
+        conn: _asyncpg.Connection | None = None
+        listening = False
         try:
             conn = await _asyncpg.connect(**opts)
             await conn.add_listener(self._notify_channel, self._on_notify)
+            listening = True
         except Exception as e:  # noqa: BLE001
             self._log(
                 log_level=logging.WARNING,
                 message=f"LISTEN setup failed; falling back to polling: {e!r}",
                 exc_info=e,
             )
-            return None
-        return conn
+        finally:
+            if conn is not None and not listening:
+                # connect() opened the socket but add_listener failed (PgBouncer txn
+                # pooling, a drop between awaits) or the task was cancelled — close the
+                # orphaned connection so a reconnect storm can't leak one raw asyncpg
+                # connection per cycle (B4). suppress: the socket may already be dead.
+                with suppress(Exception):
+                    await conn.close()
+        return conn if listening else None
 
     def _on_notify(self, *_args: object) -> None:
         """
@@ -419,6 +439,7 @@ async def _run_with_reconnect(
         name: str,
         open_resources: Callable[["AsyncEngine | None"], AbstractAsyncContextManager[Mapping[str, object]]],
         inner: Callable[..., Awaitable[None]],
+        halt_on_drain: bool = False,
     ) -> None:
         """
         Reconnect-with-backoff scaffold shared by ``_fetch_loop`` and ``_worker_loop``.
@@ -430,14 +451,27 @@ async def _run_with_reconnect(
         capped at ``_BACKOFF_MAX_SECONDS``), and reopens.
         """
         error_attempt = 0
-        while self.running:
+        # The fetch loop passes halt_on_drain=True so stop()'s _stopping flag breaks
+        # this loop instead of re-entering _fetch_inner (which returns immediately on
+        # _stopping) in a tight connect/close churn — a production reconnect storm and
+        # an event-loop-starving livelock under the test broker (B1). The worker loop
+        # keeps running through drain to flush in-flight rows.
+        while self.running and not (halt_on_drain and self._stopping):
             client = self._outer_config.client
             if client is None:  # pragma: no cover  # defensive teardown race
                 return
+            started = time.monotonic()
             try:
                 async with open_resources(client.engine) as kwargs:
                     await inner(**kwargs)
             except Exception as e:  # noqa: BLE001
+                # Reset the backoff counter when the connection was healthy for a
+                # sustained window before failing — otherwise the lifetime error count
+                # accrues and every transient blip after the first handful costs the
+                # full capped delay, while min(..., cap) annihilates the jitter and
+                # synchronizes a reconnect herd (B3).
+                if time.monotonic() - started >= _BACKOFF_RESET_THRESHOLD_SECONDS:
+                    error_attempt = 0
                 self._log(
                     log_level=logging.ERROR,
                     message=f"Outbox {name} loop error: {e!r}; reconnecting",
diff --git a/faststream_outbox/testing.py b/faststream_outbox/testing.py
index 720b906..acd630f 100644
--- a/faststream_outbox/testing.py
+++ b/faststream_outbox/testing.py
@@ -332,8 +332,6 @@ async def publish(self, cmd: OutboxPublishCommand) -> int | None:
     async def publish_batch(self, cmd: OutboxPublishCommand) -> None:
         _validate_activate_args("broker.publish_batch", cmd.activate_in, cmd.activate_at)
         bodies = cmd.batch_bodies
-        if not bodies:
-            return
         next_at = _compute_next_at_client_side(cmd.activate_in, cmd.activate_at)
         recorder = self._broker.config.broker_config.metrics_recorder
         total_size = 0
@@ -508,12 +506,16 @@ async def fake_fetch_unprocessed(
         *,
         session: typing.Any = None,
         queue: str | None = None,
+        limit: int = 1000,
     ) -> list[OutboxInnerMessage]:
         del session
         rows = sorted(fake_client.rows, key=lambda r: r.id)
         if queue is not None:
             rows = [r for r in rows if r.queue == queue]
-        return [_to_inner(r) for r in rows]
+        # Mirror production's ``limit`` (default 1000) so a valid
+        # ``broker.fetch_unprocessed(..., limit=N)`` call doesn't TypeError under
+        # the test broker (B16).
+        return [_to_inner(r) for r in rows[:limit]]
 
     return fake_fetch_unprocessed
 
@@ -540,6 +542,9 @@ def __init__(self, broker: OutboxBroker, *, run_loops: bool = False, **kwargs: t
         super().__init__(broker, **kwargs)
         self.fake_client = FakeOutboxClient()
         self.run_loops = run_loops
+        # Guards against the upstream harness spawning the loops twice (B14); reset
+        # by _fake_close so a re-entered context can spawn again.
+        self._loops_spawned = False
 
     def feed(
         self,
@@ -630,6 +635,13 @@ def _fake_start(self, broker: OutboxBroker, *args: typing.Any, **kwargs: typing.
         # In sync mode, publish drives dispatch directly — don't spawn the loops.
         if not self.run_loops:
             return
+        # The upstream TestBroker harness calls the patched start() twice (once via
+        # ``async with broker`` -> __aenter__ -> start, once via _do_start ->
+        # broker.start()). Spawn the loops only once or a max_workers=1 test silently
+        # runs two workers (B14).
+        if self._loops_spawned:
+            return
+        self._loops_spawned = True
         # Loop mode: spin up the real subscriber loops against the in-memory fake client.
         # Skip subscribers without a registered handler — matches OutboxSubscriber.start()'s
         # ``if not self.calls: return`` behavior so the test broker doesn't access ``_client``
@@ -642,5 +654,21 @@ def _fake_start(self, broker: OutboxBroker, *args: typing.Any, **kwargs: typing.
                 sub.add_task(sub._worker_loop)  # noqa: SLF001
             sub.add_task(sub._fetch_loop)  # noqa: SLF001
 
+    def _fake_close(self, broker: OutboxBroker, *args: typing.Any, **kwargs: typing.Any) -> None:
+        # Upstream's _fake_close only flips ``sub.running = False``; in loop mode the
+        # spawned fetch/worker tasks are left pending (workers parked on
+        # ``_inflight.get()`` never wake), leaking "Task was destroyed but it is pending!"
+        # noise and stale workers that fire on a re-entered context (B15). Cancel and
+        # clear them, and reset the spawn guard so a re-entered context starts fresh.
+        if self.run_loops:
+            for raw_subscriber in broker.subscribers:
+                sub = typing.cast("OutboxSubscriber", raw_subscriber)
+                for task in sub.tasks:
+                    if not task.done():
+                        task.cancel()
+                sub.tasks.clear()
+            self._loops_spawned = False
+        super()._fake_close(broker, *args, **kwargs)
+
     async def _fake_connect(self, broker: OutboxBroker, *args: typing.Any, **kwargs: typing.Any) -> None:  # noqa: ARG002
         return
diff --git a/tests/test_fake.py b/tests/test_fake.py
index 6a2c8c0..34bf354 100644
--- a/tests/test_fake.py
+++ b/tests/test_fake.py
@@ -9,6 +9,7 @@
 import pytest
 from faststream import Context as _Context
 from faststream._internal.producer import ProducerProto
+from faststream.exceptions import NackMessage
 from faststream.middlewares import AckPolicy
 from sqlalchemy import MetaData
 from sqlalchemy.ext.asyncio import AsyncSession
@@ -396,6 +397,19 @@ async def test_fake_broker_fetch_unprocessed_reads_fake_client() -> None:
         assert [r.queue for r in q1_only] == ["q1"]
 
 
+async def test_fake_broker_fetch_unprocessed_respects_limit() -> None:
+    """B16: limit= (production signature) must be accepted and cap the result set under the test broker."""
+    broker = _make_broker()
+    test_broker = TestOutboxBroker(broker)
+    async with test_broker:
+        for i in range(5):
+            await broker.publish(str(i), queue="q1")  # ty: ignore[missing-argument]
+        limited = await broker.fetch_unprocessed(limit=2)  # ty: ignore[missing-argument]
+        assert len(limited) == 2
+        all_rows = await broker.fetch_unprocessed()  # ty: ignore[missing-argument]
+        assert len(all_rows) == 5
+
+
 async def test_fake_broker_router_subscriber_receives_publish() -> None:
     received: list[str] = []
     router = OutboxRouter()
@@ -850,6 +864,35 @@ async def test_subscriber_with_no_handler_skips_loop_setup() -> None:
         assert sub.tasks == [] or all(t.done() for t in sub.tasks)
 
 
+async def test_loop_mode_spawns_each_loop_once() -> None:
+    """B14: the upstream harness calls the patched start() twice; loops must spawn once, not 2x."""
+    broker = _make_broker()
+
+    @broker.subscriber("orders", max_workers=1, min_fetch_interval=0.01, max_fetch_interval=0.05)
+    async def handle(body: str) -> None: ...
+
+    test_broker = TestOutboxBroker(broker, run_loops=True)
+    async with test_broker:
+        sub = next(iter(broker._subscribers))  # noqa: SLF001
+        # max_workers=1 → exactly 1 worker loop + 1 fetch loop = 2 tasks (the bug spawned 4).
+        assert len(sub.tasks) == 2
+
+
+async def test_loop_mode_cancels_loop_tasks_on_context_exit() -> None:
+    """B15: spawned fetch/worker tasks must be cancelled and cleared on exit, not leaked pending."""
+    broker = _make_broker()
+
+    @broker.subscriber("orders", max_workers=1, min_fetch_interval=0.01, max_fetch_interval=0.05)
+    async def handle(body: str) -> None: ...
+
+    test_broker = TestOutboxBroker(broker, run_loops=True)
+    async with test_broker:
+        sub = next(iter(broker._subscribers))  # noqa: SLF001
+        assert sub.tasks  # loops are running
+
+    assert sub.tasks == []  # _fake_close cancelled + cleared them
+
+
 async def test_sync_publish_skips_callless_subscriber() -> None:
     """``_find_subscriber_for_queue`` must skip subscribers that have no registered call."""
     from faststream_outbox.subscriber.factory import create_subscriber  # noqa: PLC0415
@@ -1001,6 +1044,48 @@ async def handle(body: str) -> None:
     assert row.next_attempt_at > _dt.datetime.now(tz=_dt.UTC)
 
 
+async def test_fake_broker_manual_policy_handler_exception_retries_not_deletes() -> None:
+    """B5: AckPolicy.MANUAL + handler exception must redeliver (honor retry), not DELETE the row."""
+    broker = _make_broker()
+
+    @broker.subscriber("orders", ack_policy=AckPolicy.MANUAL)
+    async def handle(body: str) -> None:
+        del body  # MANUAL handler raises before any manual ack/nack
+        boom = "db blip before ack"
+        raise RuntimeError(boom)
+
+    test_broker = TestOutboxBroker(broker)
+    async with test_broker:
+        await broker.publish("x", queue="orders")  # ty: ignore[missing-argument]
+
+    # The row survived: the default ExponentialRetry rescheduled it instead of
+    # the destructive reject fallback deleting it.
+    assert len(test_broker.fake_client.rows) == 1
+    row = test_broker.fake_client.rows[0]
+    assert row.attempts_count == 1
+    assert row.next_attempt_at > _dt.datetime.now(tz=_dt.UTC)
+
+
+async def test_fake_broker_nack_message_exception_retries_not_deletes() -> None:
+    """B6: ``raise NackMessage(delay=…)`` (native idiom) must reschedule, not DELETE under NACK_ON_ERROR."""
+    broker = _make_broker()
+
+    @broker.subscriber("orders", ack_policy=AckPolicy.NACK_ON_ERROR)
+    async def handle(body: str) -> None:
+        del body
+        raise NackMessage(delay=5)
+
+    test_broker = TestOutboxBroker(broker)
+    async with test_broker:
+        await broker.publish("x", queue="orders")  # ty: ignore[missing-argument]
+
+    # The kwargs no longer TypeError inside the ack middleware -> no reject fallback.
+    assert len(test_broker.fake_client.rows) == 1
+    row = test_broker.fake_client.rows[0]
+    assert row.attempts_count == 1
+    assert row.next_attempt_at > _dt.datetime.now(tz=_dt.UTC)
+
+
 # --- Publisher tests --------------------------------------------------------------------
 
 
diff --git a/tests/test_integration.py b/tests/test_integration.py
index f2fdf7c..64bb1ce 100644
--- a/tests/test_integration.py
+++ b/tests/test_integration.py
@@ -1288,6 +1288,48 @@ async def handle(body: dict) -> None:
     assert row["payload"] is not None
 
 
+async def test_dlq_writes_to_schema_qualified_tables(pg_engine: AsyncEngine) -> None:
+    """
+    B10: with a non-default ``MetaData(schema=...)`` the DLQ CTE must target ``schema.table``.
+
+    The buggy ``quote(table.name)`` dropped the schema, so the raw DELETE+INSERT CTE
+    referenced a bare ``outbox`` / ``outbox_dlq`` not on the search_path → ``UndefinedTable``
+    on every terminal failure and no audit row ever landed.
+    """
+    schema = f"sch_{uuid.uuid4().hex[:8]}"
+    metadata = MetaData(schema=schema)
+    outbox = make_outbox_table(metadata, table_name="outbox")
+    dlq = make_dlq_table(metadata, table_name="outbox_dlq")
+    async with pg_engine.begin() as conn:
+        await conn.execute(text(f'CREATE SCHEMA "{schema}"'))
+        await conn.run_sync(metadata.create_all)
+    try:
+        broker = OutboxBroker(pg_engine, outbox_table=outbox, dlq_table=dlq)
+        handled = asyncio.Event()
+
+        @broker.subscriber("orders", retry_strategy=NoRetry(), min_fetch_interval=0.02)
+        async def handle(body: dict) -> None:
+            del body
+            handled.set()
+            boom = "always fails"
+            raise RuntimeError(boom)
+
+        session_factory = async_sessionmaker(pg_engine, expire_on_commit=False)
+        async with broker:
+            async with session_factory() as session, session.begin():
+                await broker.publish({"audit": True}, queue="orders", session=session)
+            await asyncio.wait_for(handled.wait(), timeout=5.0)
+
+        assert await _row_count(pg_engine, outbox) == 0
+        rows = await _dlq_rows(pg_engine, dlq)
+        assert len(rows) == 1
+        assert rows[0]["failure_reason"] == "retry_terminal"
+    finally:
+        async with pg_engine.begin() as conn:
+            await conn.run_sync(metadata.drop_all)
+            await conn.execute(text(f'DROP SCHEMA IF EXISTS "{schema}" CASCADE'))
+
+
 async def test_dlq_insert_failure_rolls_back_delete(
     pg_engine: AsyncEngine,
     outbox_table: Table,
diff --git a/tests/test_metrics_prometheus.py b/tests/test_metrics_prometheus.py
index 4ec182a..2e7ba02 100644
--- a/tests/test_metrics_prometheus.py
+++ b/tests/test_metrics_prometheus.py
@@ -94,6 +94,18 @@ def test_prometheus_nacked_terminal_records_reason_label() -> None:
     assert _sample(reg, "faststream_received_processed_messages_total", {**_base_labels(), "status": "nacked"}) == 1.0
 
 
+def test_prometheus_max_deliveries_terminal_does_not_drive_in_process_negative() -> None:
+    """B9: nacked_terminal(reason=max_deliveries) has no preceding dispatched/duration → gauge must not go negative."""
+    reg, rec = _make_recorder()
+    # No 'dispatched' — the max_deliveries path short-circuits before the handler runs.
+    rec("nacked_terminal", {"queue": "q", "subscriber": "h", "deliveries_count": 6, "reason": "max_deliveries"})
+    sample = _sample(reg, "faststream_received_messages_in_process", _base_labels())
+    assert sample is None or sample >= 0.0  # the unconditional .dec() bug produced -1.0
+    # The terminal-reason and processed counters still fire (only the gauge dec is gated).
+    assert _sample(reg, "faststream_outbox_terminal_total", {**_base_labels(), "reason": "max_deliveries"}) == 1.0
+    assert _sample(reg, "faststream_received_processed_messages_total", {**_base_labels(), "status": "nacked"}) == 1.0
+
+
 def test_prometheus_lease_lost_increments_error_status_and_phase_counter() -> None:
     reg, rec = _make_recorder()
     rec("lease_lost", {"queue": "q", "subscriber": "h", "phase": "terminal"})
diff --git a/tests/test_unit.py b/tests/test_unit.py
index d266dde..cc2a060 100644
--- a/tests/test_unit.py
+++ b/tests/test_unit.py
@@ -2,9 +2,11 @@
 import datetime as _dt
 import json
 import logging
+import math
 import typing
 import uuid
 import warnings
+from contextlib import asynccontextmanager
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import faststream.asgi.factories.asyncapi.try_it_out
@@ -18,7 +20,7 @@
 from pydantic import BaseModel
 from sqlalchemy import MetaData
 from sqlalchemy.dialects import postgresql
-from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 
 from faststream_outbox import (
     ConstantRetry,
@@ -384,6 +386,74 @@ async def test_assert_state_set_rejects_when_not_set() -> None:
     assert msg.to_delete  # reject path → terminal
 
 
+async def test_assert_state_set_with_exception_nacks_for_retry() -> None:
+    """B5: a handler that raised without acking must honor the retry strategy, not reject-delete."""
+    msg = _make_msg(retry_strategy=ConstantRetry(delay_seconds=60), last_exception=RuntimeError("boom"))
+    await msg.assert_state_set(logger=None)
+    assert msg.state_set
+    assert not msg.to_delete  # nack scheduled a retry, did not delete
+    assert msg.pending_delay_seconds == 60.0
+    assert msg.terminal_failure_reason is None
+
+
+async def test_assert_state_set_with_exception_no_strategy_is_terminal_retry() -> None:
+    """B5: handler raised, no retry strategy -> terminal via nack (retry_terminal), not rejected."""
+    msg = _make_msg(last_exception=RuntimeError("boom"))
+    await msg.assert_state_set(logger=None)
+    assert msg.state_set
+    assert msg.to_delete
+    assert msg.terminal_failure_reason == "retry_terminal"
+
+
+async def test_inner_message_nack_accepts_and_ignores_kwargs() -> None:
+    """B6: NackMessage(delay=5) forwards delay= to nack(); we accept-and-ignore it."""
+    msg = _make_msg(retry_strategy=ConstantRetry(delay_seconds=60))
+    await msg.nack(delay=5)  # native-broker idiom; the kwarg must not raise
+    assert msg.pending_delay_seconds == 60.0  # our strategy owns timing, not the kwarg
+
+
+async def test_outbox_message_nack_accepts_and_ignores_kwargs() -> None:
+    """B6: the StreamMessage wrapper the ack middleware calls must accept **options too."""
+    inner = _make_msg(retry_strategy=ConstantRetry(delay_seconds=60))
+    msg = OutboxMessage(
+        raw_message=inner,
+        body=b"",
+        headers={},
+        content_type=None,
+        message_id="1",
+        correlation_id="1",
+    )
+    await msg.nack(delay=5)
+    assert inner.pending_delay_seconds == 60.0
+
+
+class _RaisingRetryStrategy:
+    """A buggy retry strategy that raises when computing the next delay."""
+
+    def get_next_attempt_delay(self, **_kwargs: object) -> float | None:
+        msg = "strategy boom"
+        raise RuntimeError(msg)
+
+
+async def test_nack_with_raising_strategy_degrades_to_retry_terminal() -> None:
+    """B7: a strategy that raises must degrade to retry_terminal, never reject-delete."""
+    msg = _make_msg(retry_strategy=_RaisingRetryStrategy())
+    await msg.nack()
+    assert msg.state_set
+    assert msg.to_delete
+    assert msg.terminal_failure_reason == "retry_terminal"
+
+
+def test_exponential_retry_does_not_overflow_at_extreme_attempts() -> None:
+    """B7: ExponentialRetry with no max_attempts/max_delay must not raise OverflowError."""
+    strategy = ExponentialRetry(initial_delay_seconds=1.0, multiplier=2.0)
+    now = _dt.datetime.now(tz=_dt.UTC)
+    delay = strategy.get_next_attempt_delay(first_attempt_at=now, last_attempt_at=now, attempts_count=2000)
+    assert delay is not None
+    assert math.isfinite(delay)
+    assert delay <= 100.0 * 365.0 * 24.0 * 60.0 * 60.0  # clamped at the absolute ceiling
+
+
 # --- parser ---
 
 
@@ -581,6 +651,14 @@ async def test_publish_command_rejects_naive_activate_at() -> None:
         OutboxPublishCommand(b"x", queue="orders", session=session, activate_at=naive)
 
 
+async def test_publish_command_batch_bodies_preserves_none_body() -> None:
+    """B8: batch_bodies must keep a leading/sole None body; upstream's getter drops body=None."""
+    session = _make_session_mock()
+    assert OutboxPublishCommand(None, b"x", queue="q", session=session).batch_bodies == (None, b"x")
+    assert OutboxPublishCommand(None, queue="q", session=session).batch_bodies == (None,)
+    assert OutboxPublishCommand(b"a", b"b", queue="q", session=session).batch_bodies == (b"a", b"b")
+
+
 async def test_publish_command_rejects_non_async_session() -> None:
     with pytest.raises(TypeError, match="AsyncSession"):
         OutboxPublishCommand(b"x", queue="orders", session=object())  # ty: ignore[invalid-argument-type]
@@ -945,16 +1023,18 @@ async def test_publisher_request_raises_not_implemented() -> None:
         await pub.request(b"x")
 
 
-async def test_outbox_producer_publish_batch_empty_bodies_is_noop() -> None:
-    """Empty ``batch_bodies`` returns before any SQL fires (the real broker also short-circuits)."""
+async def test_outbox_producer_publish_batch_with_none_body_inserts_row() -> None:
+    """B8: a sole None body must insert one b"" row, not silently vanish (upstream drops body=None)."""
     metadata = MetaData()
     t = make_outbox_table(metadata)
     producer = OutboxProducer(table=t, parser=None, decoder=None)
     session = _make_session_mock()
     cmd = OutboxPublishCommand(None, queue="orders", session=session)
-    cmd.batch_bodies = ()  # PublishCommand carries body=None, but be explicit
     await producer.publish_batch(cmd)
-    session.execute.assert_not_called()
+    # First execute is the multi-row INSERT; its rows arg carries one b"" payload.
+    insert_rows = session.execute.call_args_list[0].args[1]
+    assert len(insert_rows) == 1
+    assert insert_rows[0]["payload"] == b""
 
 
 def _make_outbox_publisher_spec(*, title: str | None = None, queue: str = "orders") -> OutboxPublisherSpecification:
@@ -1000,14 +1080,26 @@ async def test_fake_outbox_producer_publish_batch_inserts_rows() -> None:
     assert {r.payload for r in fake_client.rows} == {b"a", b"b"}
 
 
-async def test_fake_outbox_producer_publish_batch_empty_is_noop() -> None:
+async def test_fake_outbox_producer_publish_batch_with_none_body_inserts_row() -> None:
+    """B8: a sole None body inserts one b"" row via the fake producer too."""
     broker = _make_broker()
     fake_client = FakeOutboxClient()
     producer = FakeOutboxProducer(fake_client, broker, serializer=None, run_loops=False)
     cmd = OutboxPublishCommand(None, queue="orders", session=_make_session_mock())
-    cmd.batch_bodies = ()
     await producer.publish_batch(cmd)
-    assert fake_client.rows == []
+    assert len(fake_client.rows) == 1
+    assert fake_client.rows[0].payload == b""
+
+
+async def test_fake_outbox_producer_publish_batch_leading_none_inserts_all() -> None:
+    """B8: (None, b"x") inserts 2 rows — the leading None is not dropped."""
+    broker = _make_broker()
+    fake_client = FakeOutboxClient()
+    producer = FakeOutboxProducer(fake_client, broker, serializer=None, run_loops=False)
+    cmd = OutboxPublishCommand(None, b"x", queue="orders", session=_make_session_mock())
+    await producer.publish_batch(cmd)
+    assert len(fake_client.rows) == 2
+    assert {r.payload for r in fake_client.rows} == {b"", b"x"}
 
 
 async def test_fake_outbox_producer_request_raises() -> None:
@@ -1102,6 +1194,46 @@ async def handle(body: dict) -> None: ...
     assert await broker.ping() is True
 
 
+async def test_broker_ping_checks_router_registered_subscribers() -> None:
+    """B11: a dead task on a router-registered subscriber must fail the probe."""
+    metadata = MetaData()
+    t = make_outbox_table(metadata)
+    engine = AsyncMock()
+    broker = OutboxBroker(engine, outbox_table=t)
+    router = OutboxRouter()
+
+    @router.subscriber("orders")
+    async def handle(body: dict) -> None: ...
+
+    broker.include_router(router)
+    broker.config.broker_config.client.ping = AsyncMock(return_value=True)  # type: ignore[union-attr]
+    # Router subscribers live on the router, not in broker._subscribers — the old
+    # ping() iterated _subscribers and never saw this one.
+    subs = list(broker.subscribers)
+    assert subs
+    assert all(s not in broker._subscribers for s in subs)  # noqa: SLF001
+    done_task = MagicMock()
+    done_task.done.return_value = True
+    subs[0].tasks = [done_task]  # ty: ignore[unresolved-attribute]
+    assert await broker.ping() is False
+
+
+async def test_broker_ping_honors_timeout_when_probe_hangs() -> None:
+    """B12: ping(timeout) must bound a hanging client.ping() and return False, not hang."""
+    metadata = MetaData()
+    t = make_outbox_table(metadata)
+    engine = AsyncMock()
+    broker = OutboxBroker(engine, outbox_table=t)
+
+    async def _hang() -> bool:
+        await asyncio.sleep(10)
+        return True  # pragma: no cover - move_on_after cancels the sleep before this returns
+
+    broker.config.broker_config.client.ping = _hang  # type: ignore[union-attr]
+    result = await broker.ping(timeout=0.05)
+    assert result is False
+
+
 def test_outbox_params_storage_caches_logger() -> None:
     from unittest.mock import MagicMock  # noqa: PLC0415
 
@@ -1896,6 +2028,116 @@ async def _raise_then_stop(*, writer_conn: object) -> None:  # noqa: ARG001
     assert sleeps[0] > 0
 
 
+async def test_open_listen_connection_closes_conn_when_add_listener_fails() -> None:
+    """B4: connect() succeeded but add_listener() failed → close the orphaned conn, don't leak it."""
+    sub = _make_subscriber_for_listener_test()
+    engine = MagicMock()
+    engine.url.drivername = "postgresql+asyncpg"
+    engine.dialect.create_connect_args.return_value = (
+        [],
+        {"host": "h", "user": "u", "password": "p", "database": "db"},
+    )
+    fake_conn = MagicMock()
+    fake_conn.add_listener = AsyncMock(side_effect=OSError("listen rejected"))
+    fake_conn.close = AsyncMock()
+    with (
+        patch("faststream_outbox.subscriber.usecase._asyncpg.connect", new=AsyncMock(return_value=fake_conn)),
+        patch.object(OutboxSubscriber, "_notify_channel", new="outbox_orders"),
+        patch.object(sub, "_log"),
+    ):
+        result = await sub._open_listen_connection(engine)  # noqa: SLF001
+    assert result is None
+    fake_conn.close.assert_awaited_once()
+
+
+async def test_subscriber_start_resets_stopping_flag() -> None:
+    """B2: start() clears _stopping so a stop()->start() cycle fetches again instead of hot-spinning."""
+    sub = _make_subscriber_for_listener_test()
+    sub._stopping = True  # noqa: SLF001  # simulate a completed drain
+    with (
+        patch("faststream._internal.endpoint.subscriber.SubscriberUsecase.start", new=AsyncMock()),
+        patch.object(sub, "_post_start"),
+        patch.object(sub, "add_task"),
+    ):
+        await sub.start()
+    assert sub._stopping is False  # noqa: SLF001
+
+
+async def test_fetch_reconnect_loop_exits_on_drain_without_churning() -> None:
+    """B1: with _stopping set, the fetch reconnect loop must exit, not re-open connections in a tight spin."""
+    fake = FakeOutboxClient()
+    broker, test_broker = _make_broker_for_dispatch(fake)
+    opens = {"n": 0}
+
+    @asynccontextmanager
+    async def _open(_engine: object) -> typing.AsyncIterator[dict[str, object]]:
+        opens["n"] += 1  # pragma: no cover - the drain guard exits before resources open
+        yield {}  # pragma: no cover
+
+    async def _inner_returns_immediately() -> None:
+        return  # pragma: no cover - inner is never entered during drain
+
+    async with test_broker:
+        sub = next(iter(broker._subscribers))  # noqa: SLF001
+        sub.running = True
+        sub._stopping = True  # noqa: SLF001
+        # Without the halt_on_drain guard this loop re-opens resources forever and
+        # asyncio.wait_for would time out.
+        await asyncio.wait_for(
+            sub._run_with_reconnect(  # noqa: SLF001
+                name="fetch",
+                open_resources=_open,
+                inner=_inner_returns_immediately,
+                halt_on_drain=True,
+            ),
+            timeout=1.0,
+        )
+    assert opens["n"] == 0  # never opened resources during drain
+
+
+async def test_run_with_reconnect_resets_backoff_after_sustained_uptime(monkeypatch: pytest.MonkeyPatch) -> None:
+    """B3: error_attempt resets when the connection was healthy longer than the reset threshold."""
+    fake = FakeOutboxClient()
+    broker, test_broker = _make_broker_for_dispatch(fake)
+    attempts: list[int] = []
+
+    async def _no_sleep(_delay: float) -> None: ...
+
+    def _capture_backoff(attempt: int, ceiling: float, *, base: float = 1.0) -> float:
+        del ceiling, base
+        attempts.append(attempt)
+        return 0.0
+
+    monkeypatch.setattr("faststream_outbox.subscriber.usecase.anyio.sleep", _no_sleep)
+    monkeypatch.setattr("faststream_outbox.subscriber.usecase._compute_backoff", _capture_backoff)
+    # threshold 0 → every failure counts as "sustained uptime" → reset before each increment
+    monkeypatch.setattr("faststream_outbox.subscriber.usecase._BACKOFF_RESET_THRESHOLD_SECONDS", 0.0)
+
+    @asynccontextmanager
+    async def _open(_engine: object) -> typing.AsyncIterator[dict[str, object]]:
+        yield {}
+
+    async with test_broker:
+        sub = next(iter(broker._subscribers))  # noqa: SLF001
+        sub.running = True
+        calls = {"n": 0}
+
+        async def _inner() -> None:
+            calls["n"] += 1
+            if calls["n"] >= 4:
+                sub.running = False
+                return
+            boom = "blip"
+            raise RuntimeError(boom)
+
+        with patch.object(sub, "_log"):
+            await sub._run_with_reconnect(name="t", open_resources=_open, inner=_inner)  # noqa: SLF001
+
+    # 3 failures, each preceded by a reset → _compute_backoff always sees attempt 1
+    # (the buggy lifetime-accumulating counter would produce [1, 2, 3]).
+    assert attempts == [1, 1, 1]
+
+
 async def test_outbox_client_delete_with_lease_uses_caller_conn() -> None:
     """``delete_with_lease`` runs its statement on the supplied conn with no explicit transaction."""
     metadata = MetaData()
@@ -2440,6 +2682,28 @@ def raising_recorder(event: str, tags: typing.Any) -> None:  # noqa: ARG001
 # --- make_dlq_table -----------------------------------------------------------
 
 
+async def test_dlq_cte_uses_schema_qualified_table_names() -> None:
+    """B10: the DLQ CTE must render schema-qualified names when the tables carry a non-default schema."""
+    engine = create_async_engine("postgresql+asyncpg://u:p@localhost/db")  # constructed, never connected
+    try:
+        metadata = MetaData(schema="app")
+        outbox = make_outbox_table(metadata)
+        dlq = make_dlq_table(metadata)
+        client = OutboxClient(engine, outbox, dlq_table=dlq)
+        stmt, _params = client._build_dlq_cte_stmt(  # noqa: SLF001
+            1,
+            uuid.uuid4(),
+            {"failure_reason": "rejected", "last_exception": None},
+        )
+        sql = str(stmt)
+        # The buggy quote(table.name) rendered bare "outbox" / "outbox_dlq"; format_table
+        # carries the schema → "app.outbox" / "app.outbox_dlq" (B10).
+        assert "DELETE FROM app.outbox " in sql
+        assert "INSERT INTO app.outbox_dlq " in sql
+    finally:
+        await engine.dispose()
+
+
 def test_make_dlq_table_columns_present() -> None:
     metadata = MetaData()
     t = make_dlq_table(metadata, table_name="my_dlq")