From 0d5477ad9b1846faf0e0e881b50230962ff18f50 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 16:21:40 +0300 Subject: [PATCH] docs: correct cancel_timer, autovacuum and fetch_unprocessed docstrings, document ping --- docs/usage/setup-prometheus-opentelemetry.md | 6 +++++- faststream_outbox/autovacuum.py | 10 +++++----- faststream_outbox/broker.py | 16 ++++++++++++++-- 3 files changed, 24 insertions(+), 8 deletions(-) diff --git a/docs/usage/setup-prometheus-opentelemetry.md b/docs/usage/setup-prometheus-opentelemetry.md index 48c601d..bb26372 100644 --- a/docs/usage/setup-prometheus-opentelemetry.md +++ b/docs/usage/setup-prometheus-opentelemetry.md @@ -55,7 +55,11 @@ app = AsgiFastStream( `AsgiFastStream` accepts any ASGI sub-app under `asgi_routes`; mount `make_asgi_app(REGISTRY)` to expose Prometheus exposition without pulling FastAPI in. `make_ping_asgi(broker)` is FastStream's built-in -liveness probe, handy for Kubernetes. +health probe, handy for Kubernetes. It calls `broker.ping()`, which +takes a connection from the engine's pool, so it also fails while the +application's own sessions hold every pooled connection. Use it as a +readiness probe, or leave the pool room for it if it backs a liveness +probe. The `broker` label is always `"outbox"`; add `broker="outbox"` to the PromQL filter to scope dashboards to outbox traffic. diff --git a/faststream_outbox/autovacuum.py b/faststream_outbox/autovacuum.py index 6afc26f..5e024c3 100644 --- a/faststream_outbox/autovacuum.py +++ b/faststream_outbox/autovacuum.py @@ -2,11 +2,11 @@ The outbox is a high-churn queue table: every message is one INSERT + one lease UPDATE + one terminal DELETE, so dead tuples accumulate at ~2x the message rate. -Postgres' default ``autovacuum_vacuum_scale_factor = 0.2`` fires vacuum only after a -*fraction of the table* is dead -- on a queue table that lets bloat accumulate, and -if the table ever bloats the fraction is of the bloated size, so vacuum fires ever -less often (a death spiral). Setting the scale factor to 0 disables that trigger so -vacuum frequency tracks *churn* via a constant threshold, not table size. +Postgres' default ``autovacuum_vacuum_scale_factor = 0.2`` fires vacuum only once dead +tuples reach 20% of the table's *live* row count (``reltuples``). While the outbox holds +a large backlog, that is a large number of dead tuples, so vacuum runs rarely exactly +when churn is highest. Setting the scale factor to 0 disables that trigger so vacuum +frequency tracks *churn* via a constant threshold, not table size. SQLAlchemy's ``Table`` cannot carry reloptions (the PG dialect accepts no such kwarg), so Alembic autogenerate can never emit them. These settings must be applied diff --git a/faststream_outbox/broker.py b/faststream_outbox/broker.py index ef50347..29e56aa 100644 --- a/faststream_outbox/broker.py +++ b/faststream_outbox/broker.py @@ -368,6 +368,13 @@ def _log_subscriber_stop_error(self, sub: object, exc: BaseException) -> None: @override async def ping(self, timeout: float | None = None) -> bool: + """Return ``True`` when the database answers and every subscriber task is alive. + + The database check takes a connection from the engine's pool, so a pool the + application's own sessions have exhausted fails it too, after up to 5 seconds or + *timeout*, whichever is shorter. That suits a readiness probe. Behind a liveness + probe, leave the pool room for the check, or a busy pod gets restarted. + """ # ``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 @@ -501,9 +508,11 @@ async def cancel_timer( Returns ``True`` if *this call* deleted the row. ``False`` means the row is no longer cancelable by you, which covers three cases: - 1. The row's handler is already in flight (``acquired_token IS NOT NULL``). The + 1. A worker has claimed the row (``acquired_token IS NOT NULL``). The ``acquired_token IS NULL`` guard prevents this call from clobbering the - in-flight lease; the delivery completes normally. + lease. Normally the delivery then completes. If that worker died, the lease + stays set until ``lease_ttl_seconds`` passes and the row is delivered again; + ``cancel_timer`` can't stop that redelivery. 2. Another caller already canceled the same ``(queue, timer_id)`` in a concurrent transaction. 3. No row exists with that ``(queue, timer_id)`` — either the timer was never @@ -538,6 +547,9 @@ async def fetch_unprocessed( does not OOM the process. Runs on the caller's session (same transactional contract as :meth:`publish`); does not acquire a lease and does not mutate row state, so it is safe to call alongside running subscribers. + + No index covers the *queue* filter, so for a rare or empty queue Postgres reads + the whole table. Keep it for tests and occasional inspection, not a hot path. """ return await self.client.fetch_unprocessed(session=session, queue=queue, limit=limit)