Add opt-in Prometheus outbox queue lag metrics

docs/usage/observability.md

@@ -158,3 +158,55 @@ middleware stack, so publish-scope middleware metrics and the recorder
 Consume-scope metrics fire too. In sync mode the handler runs inside
 `publish`, so a publish-scope duration includes the handler's run time;
 the recorder's `duration_seconds` covers only the in-memory insert.
+
+## Queue lag
+
+Set `queue_lag_interval_seconds` on your existing `PrometheusRecorder` to
+enable sampling. Omit it or pass `None` to disable sampling.
+
+```python
+broker = OutboxBroker(
+    engine,
+    outbox_table=outbox_table,
+    metrics_recorder=PrometheusRecorder(
+        app_name="checkout",
+        registry=registry,
+        queue_lag_interval_seconds=60,
+    ),
+)
+```
+
+The broker starts and stops the sampler. It samples immediately, then waits
+the configured interval between rounds. At startup it reads the lease TTL
+for each registered subscriber queue. If subscribers overlap on a queue,
+the sampler uses the shortest TTL.
+
+Both gauges have `table` and `queue` labels:
+
+- `faststream_outbox_oldest_due_message_age_seconds` is the time since the
+  oldest claimable row became due. Rows under active leases and future timers
+  and retries are excluded. A successful sample of an empty claimable queue is
+  zero.
+- `faststream_outbox_queue_lag_last_success_timestamp_seconds` is the database
+  time of the last successful sample. It starts at zero and stays unchanged
+  on errors, as does the age gauge, which starts at NaN. Alert separately on
+  stale samples and scrape target availability.
+
+Each queue adds one read per round with a
+five-second maximum timeout (or the sampling interval if shorter). The
+query uses the existing fetch indexes and can scan past active leases to
+find an expired lease. If several replicas share an outbox, enabling the
+sampler on one replica avoids redundant reads.
+
+For a queue with many active leases, consider this optional index on the
+actual outbox table. Run the statement outside a transaction.
+
+```sql
+CREATE INDEX CONCURRENTLY outbox_lag_lease_idx
+ON outbox (queue, acquired_at) INCLUDE (next_attempt_at)
+WHERE acquired_token IS NOT NULL;
+```
+
+The index lets Postgres narrow the leased branch by the expiry cutoff. It
+adds maintenance work on claims and terminal writes. If many leases have
+already expired, the query can still scan and sort the qualifying rows.

faststream_outbox/broker.py

@@ -11,6 +11,7 @@
 import typing
 import warnings
 from collections.abc import Iterable, Sequence
+from contextlib import AbstractAsyncContextManager
 from types import TracebackType
 
 import anyio
@@ -34,9 +35,11 @@
 from faststream_outbox.configs import LastExceptionRenderer, OutboxBrokerConfig
 from faststream_outbox.message import OutboxInnerMessage, TimerConflict
 from faststream_outbox.metrics import MetricsRecorder, _noop_recorder
+from faststream_outbox.metrics.prometheus import PrometheusRecorder
 from faststream_outbox.publisher.producer import OutboxProducer
 from faststream_outbox.registrator import OutboxRegistrator
 from faststream_outbox.response import _REQUEST_UNSUPPORTED_MSG, OutboxPublishCommand, _validate_publish_args
+from faststream_outbox.subscriber.usecase import OutboxSubscriber
 
 
 _logger = logging.getLogger(__name__)
@@ -52,8 +55,6 @@
     from sqlalchemy import Table
     from sqlalchemy.ext.asyncio import AsyncEngine
 
-    from faststream_outbox.subscriber.usecase import OutboxSubscriber
-
 
 def _spec_url(engine: "AsyncEngine | None", outbox_table: "Table") -> list[str]:
     """AsyncAPI server URL(s) for the broker spec.
@@ -155,6 +156,7 @@ def __init__(  # noqa: PLR0913
     ) -> None:
         self._outbox_table = outbox_table
         self._dlq_table = dlq_table
+        self._queue_lag: AbstractAsyncContextManager[None] | None = None
         client = OutboxClient(engine, outbox_table, dlq_table=dlq_table) if engine is not None else None
         fd_config = FastDependsConfig(use_fastdepends=apply_types, serializer=serializer)
         recorder: MetricsRecorder = metrics_recorder or _noop_recorder
@@ -235,6 +237,27 @@ async def start(self) -> None:
         await super().start()
         self._warn_on_unstarted_foreign_publishers()
         self._warn_on_duplicate_queues()
+        await self._start_queue_lag()
+
+    async def _start_queue_lag(self) -> None:
+        """Start the recorder's sampler for the queues registered on this broker."""
+        config = self.config.broker_config
+        recorder = config.metrics_recorder
+        if self._queue_lag is not None or not isinstance(recorder, PrometheusRecorder) or config.engine is None:
+            return
+        if recorder._queue_lag_interval_seconds is None:  # noqa: SLF001
+            return
+        lease_ttl_by_queue: dict[str, float] = {}
+        for sub in self.subscribers:
+            if not isinstance(sub, OutboxSubscriber):
+                continue
+            ttl = sub._config.lease_ttl_seconds  # noqa: SLF001
+            for queue in sub._queues:  # noqa: SLF001
+                lease_ttl_by_queue[queue] = min(ttl, lease_ttl_by_queue.get(queue, ttl))
+        context = recorder._queue_lag_context(config.engine, self._outbox_table, lease_ttl_by_queue)  # noqa: SLF001
+        if context is not None:
+            await context.__aenter__()
+            self._queue_lag = context
 
     def _check_pool_capacity(self) -> None:
         """Raise when a fixed-size pool can't hold every subscriber's connections at once.
@@ -347,6 +370,9 @@ async def stop(self, *_args: object, **_kwargs: object) -> None:
         # otherwise an external cancellation of stop() during the gather leaves
         # running=True over already-stopped subscribers (ping() would lie).
         self.running = False
+        queue_lag, self._queue_lag = self._queue_lag, None
+        if queue_lag is not None:
+            await queue_lag.__aexit__(None, None, None)
         results = await asyncio.gather(
             *(sub.stop() for sub in subs),
             return_exceptions=True,

faststream_outbox/client.py

@@ -197,6 +197,16 @@ def _seconds(name: str) -> "ColumnElement[typing.Any]":
     return func.make_interval(0, 0, 0, 0, 0, 0, bindparam(name, type_=Float))
 
 
+def _claimable_branches(
+    t: "Table", lease_cutoff: "ColumnElement[typing.Any]"
+) -> tuple[tuple["ColumnElement[bool]", ...], tuple["ColumnElement[bool]", ...]]:
+    """Return separate lease predicates for unleased rows and expired leases."""
+    return (
+        (t.c.acquired_token.is_(None),),
+        (t.c.acquired_token.is_not(None), t.c.acquired_at < lease_cutoff),
+    )
+
+
 def _build_fetch_stmt(outbox: "Table") -> "Update":
     """Build the claim statement once; queues, limit, lease TTL and the new token are bind parameters.
 
@@ -222,8 +232,9 @@ def branch(name: str, *lease_state: "ColumnElement[bool]") -> "CTE":
         )
         return select(per_queue.c.id, per_queue.c.next_attempt_at).select_from(queues.join(per_queue, true())).cte(name)
 
-    pending = branch("pending", t.c.acquired_token.is_(None))
-    expired = branch("expired", t.c.acquired_token.is_not(None), t.c.acquired_at < lease_cutoff)
+    unleased, expired_lease = _claimable_branches(t, lease_cutoff)
+    pending = branch("pending", *unleased)
+    expired = branch("expired", *expired_lease)
     candidates = union_all(
         select(pending.c.id, pending.c.next_attempt_at), select(expired.c.id, expired.c.next_attempt_at)
     ).subquery("candidates")

faststream_outbox/metrics/lag.py

@@ -0,0 +1,77 @@
+"""Opt-in Prometheus sampling of outbox queue lag."""
+
+import asyncio
+import logging
+import typing
+from collections.abc import AsyncGenerator, Mapping
+from contextlib import asynccontextmanager
+
+from sqlalchemy import bindparam, extract, func, select
+
+from faststream_outbox.client import _claimable_branches, _seconds
+
+
+if typing.TYPE_CHECKING:
+    from prometheus_client import Gauge
+    from sqlalchemy import Select, Table
+    from sqlalchemy.ext.asyncio import AsyncEngine
+
+_logger = logging.getLogger(__name__)
+
+
+def _lag_statement(table: "Table") -> "Select[int, int]":
+    """Select the oldest claimable row age and sample time from the database clock."""
+    now = func.now()
+    unleased, expired = _claimable_branches(table, now - _seconds("lease_ttl"))
+    due = (
+        select(table.c.next_attempt_at)
+        .where(table.c.queue == bindparam("queue"), table.c.next_attempt_at <= now)
+        .order_by(table.c.next_attempt_at)
+        .limit(1)
+    )
+    pending = due.where(*unleased).scalar_subquery()
+    overdue = due.where(*expired, table.c.next_attempt_at <= func.coalesce(pending, now)).scalar_subquery()
+    return select(extract("epoch", now - func.least(pending, overdue)), extract("epoch", now))
+
+
+@asynccontextmanager
+async def _sample_queue_lag(
+    engine: "AsyncEngine",
+    table: "Table",
+    *,
+    age: "Gauge",
+    freshness: "Gauge",
+    lease_ttl_by_queue: Mapping[str, float],
+    interval_seconds: float,
+) -> AsyncGenerator[None, None]:
+    """Sample the oldest claimable row for each queue until context exit."""
+    statement = _lag_statement(table)
+    for queue in lease_ttl_by_queue:
+        age.labels(table.fullname, queue).set(float("nan"))
+        freshness.labels(table.fullname, queue).set(0)
+
+    async def sample() -> None:
+        while True:
+            for queue, ttl in lease_ttl_by_queue.items():
+                try:
+                    async with asyncio.timeout(min(interval_seconds, 5.0)), engine.connect() as connection:
+                        delay, sampled_at = (
+                            await connection.execute(statement, {"queue": queue, "lease_ttl": ttl})
+                        ).one()
+                    age.labels(table.fullname, queue).set(float(delay or 0))
+                    freshness.labels(table.fullname, queue).set(float(sampled_at))
+                except Exception:
+                    _logger.warning("Failed to sample outbox lag for queue %r", queue, exc_info=True)
+            await asyncio.sleep(interval_seconds)
+
+    task = asyncio.create_task(sample())
+    try:
+        yield
+    finally:
+        task.cancel()
+        try:
+            await asyncio.gather(task, return_exceptions=True)
+        finally:
+            for queue in lease_ttl_by_queue:
+                age.remove(table.fullname, queue)
+                freshness.remove(table.fullname, queue)

faststream_outbox/metrics/prometheus.py

@@ -29,17 +29,21 @@
     )
 """
 
+import math
 import typing
 from collections.abc import Callable, Mapping, Sequence
 
 from faststream._internal.constants import EMPTY
 
 from faststream_outbox._import_checker import is_prometheus_client_installed
 from faststream_outbox.metrics import BROKER_SYSTEM
+from faststream_outbox.metrics.lag import _sample_queue_lag
 
 
 if typing.TYPE_CHECKING:
     from prometheus_client import CollectorRegistry, Counter, Gauge, Histogram
+    from sqlalchemy import Table
+    from sqlalchemy.ext.asyncio import AsyncEngine
 
 # ``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
@@ -89,13 +93,15 @@ class PrometheusRecorder:
             upstream so series fall onto the existing exposition.
         received_messages_size_buckets: bucket boundaries for the message-size
             histogram. Defaults to FastStream's ``DEFAULT_SIZE_BUCKETS``.
-        custom_labels: extra labels appended to every metric. Same shape as
+        custom_labels: extra labels appended to event metrics. Same shape as
             upstream: ``dict[str, str | Callable[[Mapping], str]]``. Callable
             values receive the event's ``tags`` mapping and return the
             label value. ``published`` tags carry no ``subscriber``.
         include_bus_metrics: if False, skip the consume/publish families that
             ``OutboxPrometheusMiddleware`` also records and keep only the
             ``{metrics_prefix}_outbox_*`` series. Set it when registering both.
+        queue_lag_interval_seconds: seconds between lag samples. Defaults to
+            ``None``, which disables sampling.
 
     """
 
@@ -108,10 +114,30 @@ def __init__(
         received_messages_size_buckets: Sequence[float] | None = None,
         custom_labels: dict[str, str | Callable[[Mapping[str, typing.Any]], str]] | None = None,
         include_bus_metrics: bool = True,
+        queue_lag_interval_seconds: float | None = None,
     ) -> None:
         if not is_prometheus_client_installed:
             msg = "PrometheusRecorder requires the 'prometheus' extra: pip install 'faststream-outbox[prometheus]'"
             raise ImportError(msg)
+        if queue_lag_interval_seconds is not None and (
+            not math.isfinite(queue_lag_interval_seconds) or queue_lag_interval_seconds <= 0
+        ):
+            msg = "queue_lag_interval_seconds must be finite and positive."
+            raise ValueError(msg)
+        self._queue_lag_interval_seconds = queue_lag_interval_seconds
+        if queue_lag_interval_seconds is not None:
+            self._queue_lag_age = Gauge(
+                "faststream_outbox_oldest_due_message_age_seconds",
+                "Age of the oldest claimable row, or zero for an empty claimable queue.",
+                ("table", "queue"),
+                registry=registry,
+            )
+            self._queue_lag_freshness = Gauge(
+                "faststream_outbox_queue_lag_last_success_timestamp_seconds",
+                "Database timestamp of the last successful queue lag sample.",
+                ("table", "queue"),
+                registry=registry,
+            )
         self._app_name = "" if app_name is EMPTY else app_name
         self._custom_label_keys = list((custom_labels or {}).keys())
         self._custom_label_resolvers = list((custom_labels or {}).values())
@@ -242,6 +268,19 @@ 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 _queue_lag_context(self, engine: "AsyncEngine", table: "Table", lease_ttl_by_queue: Mapping[str, float]):
+        interval = self._queue_lag_interval_seconds
+        if interval is None or not lease_ttl_by_queue:
+            return None
+        return _sample_queue_lag(
+            engine,
+            table,
+            age=self._queue_lag_age,
+            freshness=self._queue_lag_freshness,
+            lease_ttl_by_queue=lease_ttl_by_queue,
+            interval_seconds=interval,
+        )
+
     def __call__(self, event: str, tags: Mapping[str, typing.Any]) -> None:
         if event == "published":
             if self._include_bus_metrics: