Observability

Setting it up: Setup Prometheus and OpenTelemetry. Why two seams: Concepts § Instrumentation seams.

This page is the reference for the recorder-seam API, the event catalog, and the operator PromQL playbook.

The recorder seam

OutboxBroker(..., metrics_recorder=...) accepts a callable:

MetricsRecorder = Callable[[str, Mapping[str, Any]], None]

The default is a no-op recorder. The broker passes the recorder to:

• Every subscriber, which emits seven core events (fetched, dispatched, acked, nacked_retried, nacked_terminal, lease_lost, drain_timeout), plus a conditional dlq_written when a DLQ is configured
• The producer, which emits a single event (published)

Bare seam

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, drain_timeout, published}
    # tags always include "queue"; subscriber-side events also include "subscriber"
    print(event, tags)


broker = OutboxBroker(engine, outbox_table=outbox_table, metrics_recorder=recorder)

Bundled adapters

You rarely need to hand-write the callable. Two ready-made recorders ship as optional extras and emit the faststream_outbox_* series the PromQL playbook below keys off:

from prometheus_client import CollectorRegistry
from faststream_outbox.metrics.prometheus import PrometheusRecorder

registry = CollectorRegistry()
broker = OutboxBroker(
    engine,
    outbox_table=outbox_table,
    metrics_recorder=PrometheusRecorder(app_name="checkout", registry=registry),
)

OpenTelemetryRecorder (faststream_outbox.metrics.opentelemetry) is the OTel equivalent. Full wiring, including running the recorder seam and the native middleware together, is in Setup Prometheus and OpenTelemetry.

Recorder must not block

The recorder is called from the event loop. Do not block in it. Synchronous prometheus_client.Counter.inc() is fine (microseconds); a blocking HTTP / StatsD call is not. The library does not wrap recorders in asyncio.to_thread, because that would destroy ordering and explode the task graph.

Every call site wraps the recorder in try/except and logs at DEBUG, so a broken recorder never poisons the dispatch loop.

Event catalog

Event	Tags (always present)	Tags (situational)	Fired by
fetched	queue, subscriber, count		Fetch loop, once per fetch attempt (count=0 on an empty fetch). Skipped when the in-flight queue is full (no fetch is issued). queue is tagged with the subscriber's first queue only; multi-queue subscribers should break down by queue using the row-level events instead
dispatched	queue, subscriber, deliveries_count, size_bytes		Worker loop, before handler runs
acked	queue, subscriber, deliveries_count, duration_seconds		Handler returned successfully
nacked_retried	queue, subscriber, deliveries_count, duration_seconds, next_delay_seconds	exception_type	Retry scheduled
nacked_terminal	queue, subscriber, deliveries_count, reason	duration_seconds, exception_type	Row terminally failed (duration_seconds absent for max_deliveries, which never ran the handler)
lease_lost	queue, subscriber, phase, row_id, deliveries_count	duration_seconds	Terminal or retry write found rowcount == 0 (phase = terminal | retry). duration_seconds is present when the handler ran
dispatch_abandoned	queue, subscriber, deliveries_count, duration_seconds		A dispatched row was left to lease-expiry retry with no outcome written: an error escaped the handler pipeline, the subscriber stopped mid-dispatch, or the outcome write failed
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

Every dispatched is closed by exactly one later event that carries duration_seconds: acked, nacked_retried, nacked_terminal, lease_lost or dispatch_abandoned. An in-flight gauge goes up on dispatched and down on that tag.

reason on nacked_terminal is one of max_deliveries, retry_terminal, rejected. The same value lands in the DLQ failure_reason column when the DLQ is configured.

PromQL playbook

Operator queries that key off the recorder-side metrics. The faststream_outbox_* series below (_lease_lost_total, _terminal_total, _dlq_written_total) are emitted by PrometheusRecorder (faststream_outbox.metrics.prometheus), wired via metrics_recorder=… (see Setup). The native OutboxPrometheusMiddleware does not emit them. The broker label is always "outbox"; add the filter to disambiguate from upstream FastStream services.

# Handler throughput (acked / sec)
rate(faststream_received_processed_messages_total{broker="outbox",status="acked"}[1m])

# Handler error rate. NB: status="error" also counts lease losses (the
# recorder maps lease_lost onto the error status), so this includes an
# operational, not handler, failure mode. To isolate handler failures use
# status="nacked"; track lease loss separately via the query below.
rate(faststream_received_processed_messages_total{broker="outbox",status!="acked"}[5m])
  /
rate(faststream_received_processed_messages_total{broker="outbox"}[5m])

# P99 handler latency
histogram_quantile(0.99,
  rate(faststream_received_processed_messages_duration_seconds_bucket{broker="outbox"}[5m]))

# In-flight gauge
faststream_received_messages_in_process{broker="outbox"}

# Operator playbook: lease_ttl_seconds is too low for this handler's P99
rate(faststream_outbox_lease_lost_total[5m]) > 0

# Publish throughput per queue (publish metrics are tagged by `destination`)
sum by (destination) (
  rate(faststream_published_messages_total{broker="outbox",status="success"}[1m]))

# P99 publish (INSERT) latency per queue
histogram_quantile(0.99,
  sum by (destination, le) (
    rate(faststream_published_messages_duration_seconds_bucket{broker="outbox"}[5m])))

# DLQ misconfiguration: terminal-failure rate diverges from DLQ-write rate
rate(faststream_outbox_terminal_total[5m])
  -
rate(faststream_outbox_dlq_written_total[5m])
  > 0

The first seven are direct ports of the recorder-side metrics into operator-actionable PromQL. The last one is the DLQ-misconfiguration-detection alert covered in DLQ § Metric: dlq_written.

Test broker note

TestOutboxBroker publishes through the production producer and middleware stack, so publish-scope middleware metrics and the recorder published event fire in test mode with the same tags as in production. 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.