feat: surface drain_timeout on the bundled Prometheus/OTel adapters

CLAUDE.md

@@ -174,7 +174,7 @@ Caller owns the `AsyncEngine` — the broker never disposes it. The engine lives
 
 Two complementary seams — **don't collapse them.**
 
-- **Recorder seam** (`OutboxBroker(..., metrics_recorder=...)`): `Callable[[str, Mapping[str, Any]], None]`. Subscriber emits `fetched`, `dispatched`, `acked`, `nacked_retried`, `nacked_terminal`, `lease_lost`, plus `dlq_written` when `dlq_table` is set. Producer emits `published`. Default `_noop_recorder` lets sites fire unconditionally. Every call site is wrapped in `try/except` + DEBUG log. **Recorder must not block** (sync `Counter.inc()` fine; HTTP/StatsD not). `dlq_written` vs `nacked_terminal` divergence detects DLQ misconfiguration.
+- **Recorder seam** (`OutboxBroker(..., metrics_recorder=...)`): `Callable[[str, Mapping[str, Any]], None]`. Subscriber emits `fetched`, `dispatched`, `acked`, `nacked_retried`, `nacked_terminal`, `lease_lost`, `drain_timeout` (on a timed-out `stop()` drain), plus `dlq_written` when `dlq_table` is set. Producer emits `published`. The bundled Prometheus/OTel adapters translate every one of these (`drain_timeout` → `_outbox_drain_timeout_total` / `messaging.outbox.drain_timeout`). Default `_noop_recorder` lets sites fire unconditionally. Every call site is wrapped in `try/except` + DEBUG log. **Recorder must not block** (sync `Counter.inc()` fine; HTTP/StatsD not). `dlq_written` vs `nacked_terminal` divergence detects DLQ misconfiguration.
 - **Native middleware** (`opentelemetry/`, `prometheus/`): thin subclasses of upstream's `TelemetryMiddleware[OutboxPublishCommand]` and `PrometheusMiddleware[OutboxInnerMessage, OutboxPublishCommand]`. Register via the public `OutboxBroker(..., middlewares=[...])` constructor kwarg (forwarded internally as `broker_middlewares`). Fire on `consume_scope` (via `dispatch_one → self.consume(row)`) and `publish_scope` (via `_basic_publish`).
 
 Why two: middleware owns `consume_scope` / `publish_scope` (spans, durations, status, size). Recorder owns events **outside** the bus — `fetched` (no `StreamMessage` at fetch time), `lease_lost` (after `consume_scope` exits), `nacked_terminal(reason="max_deliveries")` (before consume opens). Each fires for events the other physically cannot observe.

architecture/metrics.md

@@ -6,7 +6,7 @@ User-facing: `docs/usage/observability.md`. Invariant summary: `CLAUDE.md` § Me
 
 `OutboxBroker(..., metrics_recorder=...)` accepts a `MetricsRecorder = Callable[[str, Mapping[str, Any]], None]`. The default (`_noop_recorder`) lets instrumentation sites call unconditionally. The recorder threads through `OutboxBrokerConfig.metrics_recorder` to two places:
 
-- **Subscriber emission points** (`OutboxSubscriber._emit_metric`): `fetched`, `dispatched`, `acked`, `nacked_retried`, `nacked_terminal`, `lease_lost`, plus `dlq_written` when `dlq_table` is configured.
+- **Subscriber emission points** (`OutboxSubscriber._emit_metric`): `fetched`, `dispatched`, `acked`, `nacked_retried`, `nacked_terminal`, `lease_lost`, `drain_timeout` (a `stop()` drain that exceeded `graceful_timeout`), plus `dlq_written` when `dlq_table` is configured.
 - **Producer emission point** (`OutboxProducer._emit_metric`): `published`.
 
 The producer reads the recorder from its own constructor kwarg (passed in alongside the config field) so the canonical insert path doesn't have to reach through the broker config at call time.

docs/usage/observability.md

@@ -28,7 +28,7 @@ from faststream_outbox import MetricsRecorder, OutboxBroker
 
 def recorder(event: str, tags: dict) -> None:
     # event ∈ {fetched, dispatched, acked, nacked_retried, nacked_terminal,
-    #          lease_lost, dlq_written, published}
+    #          lease_lost, dlq_written, drain_timeout, published}
     # tags always include "queue"; subscriber-side events also include "subscriber"
     print(event, tags)
 
@@ -82,6 +82,7 @@ broken recorder never poisons the dispatch loop.
 | `lease_lost` | `queue`, `subscriber`, `phase`, `row_id`, `deliveries_count` | | Terminal or retry write found `rowcount == 0` (`phase` = `terminal` \| `retry`) |
 | `published` | `queue`, `status`, `count`, `size_bytes`, `duration_seconds` | `exception_type` | Producer, after the INSERT executes (pre-commit; also fires on error with `status="error"`) |
 | `dlq_written` | `queue`, `subscriber`, `deliveries_count`, `failure_reason` | `exception_type` | DLQ CTE wrote an audit row. `exception_type` is **omitted** — not set to `None` — when the terminal had no exception (`max_deliveries`, or a manual `reject()` without one) |
+| `drain_timeout` | `queue`, `subscriber`, `drain_timeout_seconds` | | A `stop()` drain exceeded `graceful_timeout`; in-flight rows were abandoned to lease-expiry retry. `queue` is the subscriber's **first** queue |
 
 `reason` on `nacked_terminal` is one of `max_deliveries`,
 `retry_terminal`, `rejected`. The same value lands in the DLQ

faststream_outbox/metrics/opentelemetry.py

@@ -149,6 +149,11 @@ def __init__(
             unit="event",
             description="DLQ audit rows written by terminal flush, broken down by reason",
         )
+        self._drain_timeout = self._meter.create_counter(
+            name="messaging.outbox.drain_timeout",
+            unit="event",
+            description="Drains that exceeded graceful_timeout, abandoning in-flight rows to lease-expiry retry",
+        )
 
     def _attrs(self, tags: Mapping[str, typing.Any], *, operation: str) -> dict[str, typing.Any]:
         attrs: dict[str, typing.Any] = {
@@ -161,7 +166,7 @@ def _attrs(self, tags: Mapping[str, typing.Any], *, operation: str) -> dict[str,
             attrs[_ATTR_HANDLER] = handler
         return attrs
 
-    def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None:  # noqa: C901, PLR0912
+    def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None:  # noqa: C901, PLR0911, PLR0912
         if event == "fetched":
             self._fetch_batches.add(1, self._attrs(tags, operation="receive"))
             return
@@ -204,6 +209,10 @@ def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None:  # noqa:
             self._dlq_written.add(1, attrs)
             return
 
+        if event == "drain_timeout":
+            self._drain_timeout.add(1, self._attrs(tags, operation="process"))
+            return
+
         if event == "published":
             attrs = self._attrs(tags, operation="publish")
             attrs[_ATTR_STATUS] = tags.get("status", "success")

faststream_outbox/metrics/prometheus.py

@@ -212,6 +212,12 @@ def __init__(
             [*consume_labels, "reason"],
             registry=registry,
         )
+        self._drain_timeout = Counter(
+            f"{p}_outbox_drain_timeout_total",
+            "Drains that exceeded graceful_timeout, abandoning in-flight rows to lease-expiry retry.",
+            consume_labels,
+            registry=registry,
+        )
 
     def _resolve_custom_values(self, tags: Mapping[str, typing.Any]) -> tuple[str, ...]:
         return tuple(
@@ -234,7 +240,7 @@ def _publish_values(self, tags: Mapping[str, typing.Any]) -> tuple[str, ...]:
         destination = tags.get("queue", "")
         return (self._app_name, BROKER_SYSTEM, destination, *self._resolve_custom_values(tags))
 
-    def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None:  # noqa: C901, PLR0912
+    def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None:  # noqa: C901, PLR0911, PLR0912
         consume_base = self._consume_values(tags)
 
         if event == "fetched":
@@ -282,6 +288,10 @@ def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None:  # noqa:
             self._dlq_written.labels(*consume_base, tags["failure_reason"]).inc()
             return
 
+        if event == "drain_timeout":
+            self._drain_timeout.labels(*consume_base).inc()
+            return
+
         if event == "published":
             publish_base = self._publish_values(tags)
             status = tags.get("status", "success")

tests/test_metrics_opentelemetry.py

@@ -123,6 +123,15 @@ def test_otel_unknown_event_is_silently_ignored() -> None:
     assert _collect_metrics(reader) == {}
 
 
+def test_otel_drain_timeout_emits_counter() -> None:
+    """The subscriber-emitted drain_timeout event increments its dedicated meter counter."""
+    reader, rec = _reader_and_recorder()
+    rec("drain_timeout", {"queue": "q", "subscriber": "h", "drain_timeout_seconds": 0.2})
+    metrics = _collect_metrics(reader)
+    assert "messaging.outbox.drain_timeout" in metrics
+    assert sum(p.value for p in metrics["messaging.outbox.drain_timeout"].data_points) == 1
+
+
 def test_otel_dlq_written_emits_counter_with_reason_attr() -> None:
     reader, rec = _reader_and_recorder()
     rec(

tests/test_metrics_prometheus.py

@@ -219,6 +219,13 @@ def test_prometheus_unknown_event_is_silently_ignored() -> None:
     rec("future_event_not_yet_added", {"queue": "q", "subscriber": "h"})  # forward-compat
 
 
+def test_prometheus_drain_timeout_increments() -> None:
+    """The subscriber-emitted drain_timeout event increments its dedicated counter."""
+    reg, rec = _make_recorder()
+    rec("drain_timeout", {"queue": "q", "subscriber": "h", "drain_timeout_seconds": 0.2})
+    assert _sample(reg, "faststream_outbox_drain_timeout_total", _base_labels()) == 1.0
+
+
 def test_prometheus_dlq_written_records_reason_label() -> None:
     reg, rec = _make_recorder()
     rec(