diff --git a/CLAUDE.md b/CLAUDE.md
index ee2e45a..02fac73 100644
--- a/CLAUDE.md
+++ b/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.
diff --git a/architecture/metrics.md b/architecture/metrics.md
index 45fe97b..3c5d8ed 100644
--- a/architecture/metrics.md
+++ b/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.
diff --git a/docs/usage/observability.md b/docs/usage/observability.md
index 75d9d7f..ab2b90b 100644
--- a/docs/usage/observability.md
+++ b/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
diff --git a/faststream_outbox/metrics/opentelemetry.py b/faststream_outbox/metrics/opentelemetry.py
index 70e5de9..2e6b96a 100644
--- a/faststream_outbox/metrics/opentelemetry.py
+++ b/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")
diff --git a/faststream_outbox/metrics/prometheus.py b/faststream_outbox/metrics/prometheus.py
index add2495..2707858 100644
--- a/faststream_outbox/metrics/prometheus.py
+++ b/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")
diff --git a/tests/test_metrics_opentelemetry.py b/tests/test_metrics_opentelemetry.py
index f6552b5..63d6f2b 100644
--- a/tests/test_metrics_opentelemetry.py
+++ b/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(
diff --git a/tests/test_metrics_prometheus.py b/tests/test_metrics_prometheus.py
index 959930f..9c9712d 100644
--- a/tests/test_metrics_prometheus.py
+++ b/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(