From 07bf5828b39a4cb36958518714431d55c49b55eb Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Tue, 14 Jul 2026 23:19:47 +0300 Subject: [PATCH 01/13] docs(planning): design for the benchmark harness Places the library in Variant 2 of Tsvettsikh's outbox taxonomy and records the measurements that shape the harness: per-row terminal DELETE is the entire round-trip budget (1.044 rt/msg), WAL and wall-clock disagree by two orders of magnitude on whether batching matters, and fillfactor cannot help the claim UPDATE (HOT is structurally impossible). Gate tiers are set by measured determinism, not assumption: calls is exact (0.0% spread), wal_records tolerant (3.3%), wal_bytes ungated (65.8%, FPI-dominated). Co-Authored-By: Claude Opus 4.8 (1M context) --- .../2026-07-14.01-benchmark-harness.md | 185 ++++++++++++++++++ 1 file changed, 185 insertions(+) create mode 100644 planning/changes/2026-07-14.01-benchmark-harness.md diff --git a/planning/changes/2026-07-14.01-benchmark-harness.md b/planning/changes/2026-07-14.01-benchmark-harness.md new file mode 100644 index 0000000..945524d --- /dev/null +++ b/planning/changes/2026-07-14.01-benchmark-harness.md @@ -0,0 +1,185 @@ +--- +summary: Add a `benchmarks/` harness that drives the real broker and reports per-message DB counters + throughput, with a CI gate on the deterministic subset (round-trips, tuples, WAL records); establishes the baseline for the outbox-performance work. +--- + +# Design: Benchmark harness + +## Summary + +Add a top-level `benchmarks/` package that drives the **real broker** end-to-end +against Postgres and reports two classes of number: wall-clock throughput (for +humans, comparing architectures) and per-message database counters (for CI, +catching regressions). A `just bench` recipe prints both; a `just bench-check` +recipe compares the counters against a committed `baseline.json` and fails on +drift. Today's code is the only thing benchmarked — this change ships the +instrument, not a fix. + +It exists because we are about to argue about outbox performance, and we +currently have no numbers at all. + +## Motivation + +A review of the library against Denis Tsvettsikh's talk *«50 оттенков +Transactional Outbox»* placed `faststream-outbox` squarely in his **Variant 2** +("two short transactions") — the design he benchmarks as the slowest per worker +(1.13×) and explicitly recommends against, on the grounds of 2× row mutations +driving 2× autovacuum. That prompted a list of candidate fixes (batched terminal +DELETE, table storage params, NOTIFY dedup). Every one of them was speculation. + +Prototyping against a live Postgres 17 immediately showed why speculation is not +good enough here — **the obvious metrics contradict each other**. For per-row +versus batched terminal DELETE over 2000 messages: + +| Metric | Per-row (today) | Batched | Implication | +| --- | --- | --- | --- | +| WAL bytes | 891 KB | 797 KB | 1.12× — batching is worthless | +| Client wall-clock | 1679 ms | 15 ms | 110× — batching is enormous | + +Both are true. The DB barely notices (a commit record is small next to a tuple +record); the *client* notices enormously, because each terminal DELETE is a +network round-trip. A harness measuring only DB load would have told us to skip +the single biggest win available. + +The same prototyping falsified a proposed fix outright: `fillfactor` cannot help +the claim UPDATE, because `n_tup_hot_upd` is **0 of 10,000 at any fillfactor** — +HOT is structurally impossible when the UPDATE mutates `acquired_token` and +`acquired_at`, the very columns both partial indexes are keyed on. + +Measured baseline for today's code (`max_workers=1`, `fetch_batch_size=100`, +256-byte payloads, 500 messages, 5 runs): + +- **1.044 round-trips per message** — i.e. one terminal DELETE per message plus + an amortized 0.01 fetch. The per-row flush *is* the entire round-trip budget. +- **11.0 WAL records per message** — a number for the write-amplification cost. +- ~840 µs per message spent purely in the DELETE round-trip on loopback, which + caps single-worker throughput at ~1200 msg/s regardless of handler speed. + +## Design + +### Layout + +``` +benchmarks/ + __init__.py + probes.py # DbProbe: catalog snapshot -> deltas + workload.py # producer + consumer scenarios against the real broker + report.py # human table + JSON, normalized per-message + baseline.json # committed; bench-check gates against it +``` + +Top-level, never collected by pytest. `--cov=.` only reports *imported* modules, +so this does not disturb the `--cov-fail-under=100` gate — `planning/index.py` is +the existing precedent. + +### Probe + +`DbProbe` is an async context manager yielding a `ProbeResult` of catalog deltas: + +- **Round-trips** — `pg_stat_statements.calls`, summed. +- **WAL** — `pg_stat_wal.wal_records`, `wal_fpi`, plus byte count via + `pg_wal_lsn_diff(pg_current_wal_lsn(), start_lsn)`. +- **Tuples** — `pg_stat_user_tables` for the outbox table: `n_tup_ins/upd/del`, + `n_tup_hot_upd`, `n_tup_newpage_upd`, `n_dead_tup`, `autovacuum_count`. +- **IO / size** — `shared_blks_hit`/`read`; `pg_relation_size` for heap and each + index. +- **DB CPU proxy** — `pg_stat_statements.total_exec_time`. + +Snapshot-diff throughout, except one `pg_stat_statements_reset()` at run start. +Two hazards, both hit while prototyping and both required to be handled +explicitly: + +1. `pg_stat_user_tables` **lags** — a bare read after the workload returned + `n_tup_upd = 0` for 10,000 updates that had definitely happened. The final + read polls until the counters settle. +2. `pg_stat_statements` is **database-wide** — the probe asserts it owns the + database (no other backends in `pg_stat_activity`) or the round-trip count is + meaningless. + +### Scenarios + +Against a fresh per-run table (UUID suffix, mirroring the `outbox_table` +fixture), 256-byte payloads to stay commensurable with the talk: + +- **Producer** — N `publish()` calls in one session; separately, one + `publish_batch()`. Surfaces the two-statements-per-publish cost + (`INSERT … RETURNING` + `SELECT pg_notify`). +- **Consumer** — bulk-seed N rows directly, start the broker with a no-op + handler, wait for drain, stop. A no-op handler is the point: what is being + measured is transport cost, not handler cost. Sweeps + `max_workers ∈ {1,2,4}` × `fetch_batch_size ∈ {10,100}`. + +### Gate + +Metrics are tiered by **measured** determinism (5 real-broker runs): + +| Tier | Metric | Observed spread | Treatment | +| --- | --- | --- | --- | +| 1 | `pg_stat_statements.calls`, `n_tup_ins/upd/del` | **0.0%** | gate on exact equality | +| 2 | `wal_records` / msg | 3.3% | gate with ±10% tolerance | +| 3 | `wal_bytes` / msg, `wal_fpi`, wall-clock | **65.8%** | report only, never gate | + +The WAL-*bytes* swing (771 → 1278 B/msg) is entirely full-page images — `fpi=34` +on the noisy run versus `fpi=5` on the settled ones. Gating it would flake +constantly. `wal_records` counts the same work without the FPI noise, which is +why it is the WAL gate. `wal_fpi` is reported alongside the byte count precisely +so a future reader does not misread FPI noise as a regression. + +Everything is normalized **per message**; that is what makes the counters +machine-independent and therefore gateable in CI at all. + +### Infrastructure + +`docker-compose.yml`'s `postgres` service gains +`command: postgres -c shared_preload_libraries=pg_stat_statements -c pg_stat_statements.track=all` +(the extension is available in the `postgres:17` image but not currently loaded). +Harmless to the existing suite. + +CI runs `just bench-check` as a **separate job** with small N (~5k messages, +seconds). The full throughput sweep stays manual. + +### Noise control + +Fresh table per run; a discarded warm-up pass; median of 3 for wall-clock; and a +`CHECKPOINT` before each measured run — done *consistently* so the FPI cost is at +least reproducible, and reported separately so it is legible. + +## Non-goals + +- **No fix ships here.** Batched DELETE, storage params, and NOTIFY dedup each + land as their own PR, each showing its delta against this baseline. +- **No variant seam.** Benchmarking alternatives would mean an abstraction in the + subscriber hot path existing only for benchmarks, and it likely would not + survive contact with the real fix. +- **No latency percentiles.** The ~10 ms LISTEN/NOTIFY idle-latency claim stays + unverified for now; adding it later does not invalidate the baseline. +- **No DB CPU graphs.** `total_exec_time` is the machine-comparable stand-in. +- **Wall-clock is never gated.** Docker-on-macOS IO noise would make it a flake + generator. + +## Testing + +- `just bench --messages 500` runs green and prints both tables. +- `just bench-check` passes against the committed `baseline.json` on unmodified + `main`. +- **The gate actually bites:** artificially perturb the terminal flush (e.g. an + extra no-op statement per message) and confirm `bench-check` fails on the + round-trip count. A gate never observed failing is not a gate. +- Determinism holds: 5 consecutive `just bench` runs report identical Tier-1 + counters (verified in prototype: `calls = 522`, spread 0.0%). + +## Risk + +**The gate is only as strong as its determinism.** This was the design's central +risk and it has been retired empirically: `calls` was identical across 5 +real-broker runs (spread 0.0%). Residual exposure is that a *future* change +introduces pool churn or reconnect logic that makes it nondeterministic — which +is arguably the gate working, not failing, but it will present as a flake and +should be diagnosed as such before the tolerance is loosened. + +**Docker/macOS wall-clock is indicative, not authoritative.** Mitigated by never +gating on it and by saying so in the report output, so the numbers are not quoted +as absolutes in an argument they cannot settle. + +**`pg_stat_statements` being database-wide** makes the round-trip count sensitive +to any stray backend. Mitigated by the explicit ownership assertion; it fails +loudly rather than reporting a quietly wrong number. From d669fdf2daa026087b979aaaa862bb2010d5d302 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Tue, 14 Jul 2026 23:47:24 +0300 Subject: [PATCH 02/13] build(bench): preload pg_stat_statements, add benchmarks package skeleton --- benchmarks/__init__.py | 5 +++++ benchmarks/config.py | 25 +++++++++++++++++++++++++ docker-compose.yml | 4 ++++ pyproject.toml | 3 +++ 4 files changed, 37 insertions(+) create mode 100644 benchmarks/__init__.py create mode 100644 benchmarks/config.py diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py new file mode 100644 index 0000000..0598077 --- /dev/null +++ b/benchmarks/__init__.py @@ -0,0 +1,5 @@ +"""Benchmark harness for faststream-outbox. + +Dev tooling, not shipped. Run via `just bench` / `just bench-check`; never +collected by pytest. See planning/changes/2026-07-14.01-benchmark-harness.md. +""" diff --git a/benchmarks/config.py b/benchmarks/config.py new file mode 100644 index 0000000..5f33745 --- /dev/null +++ b/benchmarks/config.py @@ -0,0 +1,25 @@ +"""Workload configuration for a single benchmark run.""" + +import dataclasses + + +# Matches the payload size in Tsvettsikh's «50 оттенков Transactional Outbox» +# benchmarks, so our numbers stay commensurable with his. +DEFAULT_PAYLOAD_BYTES = 256 + +# The DSN the harness runs against. Overridden by POSTGRES_DSN in compose. +DEFAULT_DSN = "postgresql+asyncpg://outbox:outbox@localhost:5432/outbox" + +# Tags the harness' own connections so the probe can assert it owns the database. +APPLICATION_NAME = "faststream-outbox-bench" + + +@dataclasses.dataclass(frozen=True) +class RunConfig: + """One point in the benchmark sweep.""" + + messages: int = 5_000 + max_workers: int = 1 + fetch_batch_size: int = 100 + payload_bytes: int = DEFAULT_PAYLOAD_BYTES + queue: str = "bench" diff --git a/docker-compose.yml b/docker-compose.yml index 41958f3..41a41d6 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -18,6 +18,10 @@ services: postgres: image: postgres:17 restart: always + command: >- + postgres + -c shared_preload_libraries=pg_stat_statements + -c pg_stat_statements.track=all environment: - POSTGRES_USER=outbox - POSTGRES_PASSWORD=outbox diff --git a/pyproject.toml b/pyproject.toml index a8bff85..a453cea 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -109,5 +109,8 @@ Repository = "https://github.com/modern-python/faststream-outbox" Issues = "https://github.com/modern-python/faststream-outbox/issues" Changelog = "https://github.com/modern-python/faststream-outbox/releases" +[tool.coverage.run] +omit = ["benchmarks/*"] + [tool.coverage.report] exclude_also = ["if typing.TYPE_CHECKING:"] From 3fd14a471327bce3b6115dd391aad5bb6363f4a2 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Tue, 14 Jul 2026 23:56:38 +0300 Subject: [PATCH 03/13] feat(bench): add Postgres catalog probe Snapshots pg_stat_statements/pg_stat_wal/pg_stat_user_tables around a workload and returns a ProbeResult delta. Filters its own queries out of the round-trip count by SQL text (NOT ILIKE '%pg_stat%'), subtracts cumulative table/WAL counters from a start snapshot, and settles on pg_stat_user_tables before reading (the stats collector flushes lazily). assert_owns_database guards against a foreign backend polluting the database-wide statement counters. --- benchmarks/probes.py | 216 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 216 insertions(+) create mode 100644 benchmarks/probes.py diff --git a/benchmarks/probes.py b/benchmarks/probes.py new file mode 100644 index 0000000..66bc351 --- /dev/null +++ b/benchmarks/probes.py @@ -0,0 +1,216 @@ +"""Postgres catalog probes: snapshot the server's counters around a workload. + +Every statement issued here contains the substring ``pg_stat`` in its SQL text, +and the round-trip aggregation filters those out (``query NOT ILIKE '%pg_stat%'``). +That is how the probe avoids counting itself. The workload's own statements +(INSERT/UPDATE/DELETE/SELECT on the outbox table, ``pg_notify``) never match the +filter, so they are all counted. + +Two hazards, both hit while prototyping this design: + +* ``pg_stat_user_tables`` **lags**. A bare read straight after the workload + returned ``n_tup_upd = 0`` for 10,000 updates that had definitely happened. + :func:`_settle` polls until the counters stop moving. +* ``pg_stat_statements`` is **database-wide**. A stray psql session would silently + corrupt the round-trip count, so :func:`assert_owns_database` fails loudly rather + than reporting a quietly wrong number. +""" + +import asyncio +import contextlib +import dataclasses +from collections.abc import AsyncIterator + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncEngine + +from benchmarks.config import APPLICATION_NAME + + +_SETTLE_POLL_SECONDS = 0.25 +_SETTLE_STABLE_READS = 3 +_SETTLE_TIMEOUT_SECONDS = 15.0 + +# Resets the statement counters and captures every cumulative origin in ONE round-trip, +# so exactly one probe statement lands in pg_stat_statements before the workload runs +# (and it is filtered out again by the NOT ILIKE '%pg_stat%' predicate below). +# +# The table counters MUST be captured here, not assumed zero: pg_stat_user_tables is +# cumulative from table creation, and the consumer scenario seeds its rows *before* the +# probe opens. Treating the end-of-run snapshot as the delta would report the seed's +# 5,000 inserts as workload inserts. coalesce+LEFT JOIN because a freshly created table +# may have no pg_stat_user_tables row at all until it is first touched. +_START_SQL = text( + """ + SELECT + pg_stat_statements_reset() IS NOT NULL AS reset, + pg_current_wal_lsn() AS lsn, + (SELECT w.wal_records FROM pg_stat_wal w) AS wal_records, + (SELECT w.wal_fpi FROM pg_stat_wal w) AS wal_fpi, + coalesce((SELECT t.n_tup_ins FROM pg_stat_user_tables t WHERE t.relname = :table), 0) AS tup_ins, + coalesce((SELECT t.n_tup_upd FROM pg_stat_user_tables t WHERE t.relname = :table), 0) AS tup_upd, + coalesce((SELECT t.n_tup_del FROM pg_stat_user_tables t WHERE t.relname = :table), 0) AS tup_del, + coalesce((SELECT t.n_tup_hot_upd FROM pg_stat_user_tables t WHERE t.relname = :table), 0) AS tup_hot_upd, + coalesce((SELECT t.n_tup_newpage_upd FROM pg_stat_user_tables t WHERE t.relname = :table), 0) + AS tup_newpage_upd + """, +) + +# One statement for the whole end-of-run snapshot. Every aggregate excludes the +# probe's own queries by SQL text. +_SNAPSHOT_SQL = text( + """ + SELECT + (SELECT coalesce(sum(s.calls), 0) + FROM pg_stat_statements s WHERE s.query NOT ILIKE '%pg_stat%') AS calls, + (SELECT coalesce(sum(s.total_exec_time), 0.0) + FROM pg_stat_statements s WHERE s.query NOT ILIKE '%pg_stat%') AS exec_ms, + (SELECT coalesce(sum(s.shared_blks_hit), 0) + FROM pg_stat_statements s WHERE s.query NOT ILIKE '%pg_stat%') AS blks_hit, + (SELECT coalesce(sum(s.shared_blks_read), 0) + FROM pg_stat_statements s WHERE s.query NOT ILIKE '%pg_stat%') AS blks_read, + w.wal_records AS wal_records, + w.wal_fpi AS wal_fpi, + pg_wal_lsn_diff(pg_current_wal_lsn(), CAST(:lsn0 AS pg_lsn)) AS wal_bytes, + t.n_tup_ins, t.n_tup_upd, t.n_tup_del, + t.n_tup_hot_upd, t.n_tup_newpage_upd, t.n_dead_tup, + pg_relation_size(t.relid) AS heap_bytes, + pg_indexes_size(t.relid) AS index_bytes + FROM pg_stat_wal w, pg_stat_user_tables t + WHERE t.relname = :table + """, +) + +# Contains 'pg_stat', so the settle polls are excluded from the round-trip count too. +_SETTLE_SQL = text( + "SELECT coalesce(n_tup_upd, 0) + coalesce(n_tup_del, 0) AS mutations " + "FROM pg_stat_user_tables WHERE relname = :table", +) + +_OWNERSHIP_SQL = text( + "SELECT count(*) FROM pg_stat_activity " + "WHERE datname = current_database() " + "AND backend_type = 'client backend' " + "AND pid <> pg_backend_pid() " + "AND coalesce(application_name, '') <> :app", +) + +_EXTENSION_SQL = text("CREATE EXTENSION IF NOT EXISTS pg_stat_statements") + + +@dataclasses.dataclass(frozen=True) +class ProbeResult: + """Catalog deltas across one workload.""" + + calls: int + exec_ms: float + blks_hit: int + blks_read: int + wal_records: int + wal_fpi: int + wal_bytes: int + tup_ins: int + tup_upd: int + tup_del: int + tup_hot_upd: int + tup_newpage_upd: int + dead_tup: int + heap_bytes: int + index_bytes: int + + +async def ensure_extension(engine: AsyncEngine) -> None: + """Create pg_stat_statements if the server preloaded it; raise a legible error if not.""" + async with engine.begin() as conn: + try: + await conn.execute(_EXTENSION_SQL) + except Exception as exc: + msg = ( + "pg_stat_statements is unavailable. The postgres service must run with " + "`-c shared_preload_libraries=pg_stat_statements` (see docker-compose.yml). " + f"Underlying error: {exc}" + ) + raise RuntimeError(msg) from exc + + +async def assert_owns_database(engine: AsyncEngine) -> None: + """Fail loudly if a foreign backend shares the database. + + pg_stat_statements is database-wide, so any other session's queries land in the + round-trip count. Checked once, *before* the broker starts — at that point the + only connections are the harness' own (tagged with APPLICATION_NAME). + """ + async with engine.connect() as conn: + foreign = (await conn.execute(_OWNERSHIP_SQL, {"app": APPLICATION_NAME})).scalar_one() + if foreign: + msg = ( + f"{foreign} foreign backend(s) are connected to the benchmark database. " + "pg_stat_statements is database-wide, so the round-trip count would be wrong. " + "Close other sessions (psql, an IDE, a stray app) and re-run." + ) + raise RuntimeError(msg) + + +async def _settle(engine: AsyncEngine, table_name: str) -> None: + """Poll until pg_stat_user_tables stops moving. + + The stats collector flushes lazily; a bare read right after the workload can + report zero for mutations that have definitely happened. + """ + deadline = asyncio.get_running_loop().time() + _SETTLE_TIMEOUT_SECONDS + previous = -1 + stable = 0 + while stable < _SETTLE_STABLE_READS: + await asyncio.sleep(_SETTLE_POLL_SECONDS) + async with engine.connect() as conn: + current = (await conn.execute(_SETTLE_SQL, {"table": table_name})).scalar_one_or_none() or 0 + stable = stable + 1 if current == previous else 0 + previous = current + if asyncio.get_running_loop().time() > deadline: + msg = f"pg_stat_user_tables for {table_name!r} never settled within {_SETTLE_TIMEOUT_SECONDS}s" + raise RuntimeError(msg) + + +@contextlib.asynccontextmanager +async def probe(engine: AsyncEngine, table_name: str) -> AsyncIterator[list[ProbeResult]]: + """Snapshot the catalogs around the wrapped workload. + + Yields a list that holds exactly one :class:`ProbeResult` once the block exits. + A list (rather than a return value) is the only way an async context manager can + hand a result back to its caller. + """ + sink: list[ProbeResult] = [] + # Settle first: the seed's inserts must be visible in pg_stat_user_tables before we + # capture the origin, or they leak into the delta. + await _settle(engine, table_name) + async with engine.begin() as conn: + start = (await conn.execute(_START_SQL, {"table": table_name})).one() + + yield sink + + await _settle(engine, table_name) + async with engine.connect() as conn: + row = (await conn.execute(_SNAPSHOT_SQL, {"lsn0": start.lsn, "table": table_name})).one() + + sink.append( + ProbeResult( + # pg_stat_statements was reset at start, so these are already deltas. + calls=int(row.calls), + exec_ms=float(row.exec_ms), + blks_hit=int(row.blks_hit), + blks_read=int(row.blks_read), + # Server-wide and table-level counters are cumulative -- subtract the origin. + wal_records=int(row.wal_records) - int(start.wal_records), + wal_fpi=int(row.wal_fpi) - int(start.wal_fpi), + wal_bytes=int(row.wal_bytes), + tup_ins=int(row.n_tup_ins) - int(start.tup_ins), + tup_upd=int(row.n_tup_upd) - int(start.tup_upd), + tup_del=int(row.n_tup_del) - int(start.tup_del), + tup_hot_upd=int(row.n_tup_hot_upd) - int(start.tup_hot_upd), + tup_newpage_upd=int(row.n_tup_newpage_upd) - int(start.tup_newpage_upd), + # Point-in-time, not a delta. + dead_tup=int(row.n_dead_tup), + heap_bytes=int(row.heap_bytes), + index_bytes=int(row.index_bytes), + ), + ) From 142211df81ba5783e192bfec9348e5edcd08552c Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 00:29:20 +0300 Subject: [PATCH 04/13] fix(bench): make the probe deterministic and self-excluding The settle loop polled n_tup_upd + n_tup_del, which never moves for an insert-only workload, and its 0.75s stability window was shorter than Postgres' 1s PGSTAT_MIN_INTERVAL, so the seed's stats could still be unflushed when the origin was captured and leaked into the delta. Replace it with engine.dispose() before each snapshot: Postgres force-flushes a backend's pending stats when the backend exits, so the flush is exact rather than eventual. Run every probe statement on an AUTOCOMMIT connection. SQLAlchemy's implicit BEGIN/COMMIT/ROLLBACK are tracked by pg_stat_statements and contain no 'pg_stat' substring, so they slipped past the self-exclusion filter and inflated the exactly-gated 'calls' metric by a run-dependent amount. The workload's own transaction control is still counted. Also pin schemaname in the catalog lookups and fail fast at probe open on a missing table, instead of dying with NoResultFound at the end of a long run. --- benchmarks/probes.py | 154 +++++++++++++++++++++++++------------------ 1 file changed, 91 insertions(+), 63 deletions(-) diff --git a/benchmarks/probes.py b/benchmarks/probes.py index 66bc351..239bb29 100644 --- a/benchmarks/probes.py +++ b/benchmarks/probes.py @@ -1,36 +1,37 @@ """Postgres catalog probes: snapshot the server's counters around a workload. -Every statement issued here contains the substring ``pg_stat`` in its SQL text, -and the round-trip aggregation filters those out (``query NOT ILIKE '%pg_stat%'``). -That is how the probe avoids counting itself. The workload's own statements -(INSERT/UPDATE/DELETE/SELECT on the outbox table, ``pg_notify``) never match the -filter, so they are all counted. +Every statement the probe issues is a plain ``SELECT`` whose SQL text contains the +substring ``pg_stat``, and it runs on an **AUTOCOMMIT** connection so SQLAlchemy emits +no implicit ``BEGIN``/``COMMIT``/``ROLLBACK`` around it. The round-trip aggregation +filters those SELECTs out (``query NOT ILIKE '%pg_stat%'``); with no implicit +transaction control to leak, that filter is enough for the probe to fully exclude +itself. The workload's statements -- including *its* transaction control, which is real +round-trip work -- never match the filter, so they are all counted, by design. Two hazards, both hit while prototyping this design: -* ``pg_stat_user_tables`` **lags**. A bare read straight after the workload - returned ``n_tup_upd = 0`` for 10,000 updates that had definitely happened. - :func:`_settle` polls until the counters stop moving. +* ``pg_stat_user_tables`` and ``pg_stat_wal`` **lag**: a backend accumulates its stats + locally and flushes them to shared memory no more often than ``PGSTAT_MIN_INTERVAL`` + (1s). Reading straight after a workload can report zero for mutations that definitely + happened, and reading straight after a seed can miss the seed -- which then leaks into + the delta. Postgres force-flushes a backend's pending stats when the backend *exits*, + so the probe calls :meth:`AsyncEngine.dispose` (which closes every pooled connection) + immediately before each snapshot. That is deterministic, unlike waiting. * ``pg_stat_statements`` is **database-wide**. A stray psql session would silently corrupt the round-trip count, so :func:`assert_owns_database` fails loudly rather than reporting a quietly wrong number. """ -import asyncio import contextlib import dataclasses from collections.abc import AsyncIterator from sqlalchemy import text -from sqlalchemy.ext.asyncio import AsyncEngine +from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine from benchmarks.config import APPLICATION_NAME -_SETTLE_POLL_SECONDS = 0.25 -_SETTLE_STABLE_READS = 3 -_SETTLE_TIMEOUT_SECONDS = 15.0 - # Resets the statement counters and captures every cumulative origin in ONE round-trip, # so exactly one probe statement lands in pg_stat_statements before the workload runs # (and it is filtered out again by the NOT ILIKE '%pg_stat%' predicate below). @@ -38,26 +39,43 @@ # The table counters MUST be captured here, not assumed zero: pg_stat_user_tables is # cumulative from table creation, and the consumer scenario seeds its rows *before* the # probe opens. Treating the end-of-run snapshot as the delta would report the seed's -# 5,000 inserts as workload inserts. coalesce+LEFT JOIN because a freshly created table -# may have no pg_stat_user_tables row at all until it is first touched. +# 5,000 inserts as workload inserts. The `present` flag lets probe() fail fast on a +# missing table instead of dying with NoResultFound at the end of a long run; the +# coalesce()s keep this statement returning a row either way so that check can run. +# +# :schema is NULL by default and resolves to current_schema(); it is echoed back so the +# end snapshot pins the same schema. pg_stat_user_tables spans every schema, so matching +# on a bare relname could hit two same-named tables. _START_SQL = text( """ + WITH scope AS (SELECT coalesce(CAST(:schema AS text), current_schema()::text) AS schemaname) SELECT pg_stat_statements_reset() IS NOT NULL AS reset, pg_current_wal_lsn() AS lsn, + scope.schemaname AS schemaname, (SELECT w.wal_records FROM pg_stat_wal w) AS wal_records, (SELECT w.wal_fpi FROM pg_stat_wal w) AS wal_fpi, - coalesce((SELECT t.n_tup_ins FROM pg_stat_user_tables t WHERE t.relname = :table), 0) AS tup_ins, - coalesce((SELECT t.n_tup_upd FROM pg_stat_user_tables t WHERE t.relname = :table), 0) AS tup_upd, - coalesce((SELECT t.n_tup_del FROM pg_stat_user_tables t WHERE t.relname = :table), 0) AS tup_del, - coalesce((SELECT t.n_tup_hot_upd FROM pg_stat_user_tables t WHERE t.relname = :table), 0) AS tup_hot_upd, - coalesce((SELECT t.n_tup_newpage_upd FROM pg_stat_user_tables t WHERE t.relname = :table), 0) - AS tup_newpage_upd + EXISTS ( + SELECT 1 FROM pg_stat_user_tables t + WHERE t.relname = :table AND t.schemaname = scope.schemaname + ) AS present, + coalesce((SELECT t.n_tup_ins FROM pg_stat_user_tables t + WHERE t.relname = :table AND t.schemaname = scope.schemaname), 0) AS tup_ins, + coalesce((SELECT t.n_tup_upd FROM pg_stat_user_tables t + WHERE t.relname = :table AND t.schemaname = scope.schemaname), 0) AS tup_upd, + coalesce((SELECT t.n_tup_del FROM pg_stat_user_tables t + WHERE t.relname = :table AND t.schemaname = scope.schemaname), 0) AS tup_del, + coalesce((SELECT t.n_tup_hot_upd FROM pg_stat_user_tables t + WHERE t.relname = :table AND t.schemaname = scope.schemaname), 0) AS tup_hot_upd, + coalesce((SELECT t.n_tup_newpage_upd FROM pg_stat_user_tables t + WHERE t.relname = :table AND t.schemaname = scope.schemaname), 0) AS tup_newpage_upd + FROM scope """, ) # One statement for the whole end-of-run snapshot. Every aggregate excludes the -# probe's own queries by SQL text. +# probe's own queries by SQL text. schemaname is pinned: pg_stat_user_tables covers +# every schema, so a bare relname match could return two rows. _SNAPSHOT_SQL = text( """ SELECT @@ -77,16 +95,10 @@ pg_relation_size(t.relid) AS heap_bytes, pg_indexes_size(t.relid) AS index_bytes FROM pg_stat_wal w, pg_stat_user_tables t - WHERE t.relname = :table + WHERE t.relname = :table AND t.schemaname = :schema """, ) -# Contains 'pg_stat', so the settle polls are excluded from the round-trip count too. -_SETTLE_SQL = text( - "SELECT coalesce(n_tup_upd, 0) + coalesce(n_tup_del, 0) AS mutations " - "FROM pg_stat_user_tables WHERE relname = :table", -) - _OWNERSHIP_SQL = text( "SELECT count(*) FROM pg_stat_activity " "WHERE datname = current_database() " @@ -119,9 +131,22 @@ class ProbeResult: index_bytes: int +@contextlib.asynccontextmanager +async def _autocommit(engine: AsyncEngine) -> AsyncIterator[AsyncConnection]: + """Open a connection that emits no implicit BEGIN/COMMIT/ROLLBACK. + + pg_stat_statements tracks utility statements, and none of BEGIN/COMMIT/ROLLBACK + contains 'pg_stat', so SQLAlchemy's implicit transaction control would slip past + the probe's self-exclusion filter and inflate `calls` by a run-dependent amount. + AUTOCOMMIT removes it at the source. + """ + async with engine.connect() as conn: + yield await conn.execution_options(isolation_level="AUTOCOMMIT") + + async def ensure_extension(engine: AsyncEngine) -> None: """Create pg_stat_statements if the server preloaded it; raise a legible error if not.""" - async with engine.begin() as conn: + async with _autocommit(engine) as conn: try: await conn.execute(_EXTENSION_SQL) except Exception as exc: @@ -137,10 +162,10 @@ async def assert_owns_database(engine: AsyncEngine) -> None: """Fail loudly if a foreign backend shares the database. pg_stat_statements is database-wide, so any other session's queries land in the - round-trip count. Checked once, *before* the broker starts — at that point the + round-trip count. Checked once, *before* the broker starts -- at that point the only connections are the harness' own (tagged with APPLICATION_NAME). """ - async with engine.connect() as conn: + async with _autocommit(engine) as conn: foreign = (await conn.execute(_OWNERSHIP_SQL, {"app": APPLICATION_NAME})).scalar_one() if foreign: msg = ( @@ -151,46 +176,49 @@ async def assert_owns_database(engine: AsyncEngine) -> None: raise RuntimeError(msg) -async def _settle(engine: AsyncEngine, table_name: str) -> None: - """Poll until pg_stat_user_tables stops moving. - - The stats collector flushes lazily; a bare read right after the workload can - report zero for mutations that have definitely happened. - """ - deadline = asyncio.get_running_loop().time() + _SETTLE_TIMEOUT_SECONDS - previous = -1 - stable = 0 - while stable < _SETTLE_STABLE_READS: - await asyncio.sleep(_SETTLE_POLL_SECONDS) - async with engine.connect() as conn: - current = (await conn.execute(_SETTLE_SQL, {"table": table_name})).scalar_one_or_none() or 0 - stable = stable + 1 if current == previous else 0 - previous = current - if asyncio.get_running_loop().time() > deadline: - msg = f"pg_stat_user_tables for {table_name!r} never settled within {_SETTLE_TIMEOUT_SECONDS}s" - raise RuntimeError(msg) - - @contextlib.asynccontextmanager -async def probe(engine: AsyncEngine, table_name: str) -> AsyncIterator[list[ProbeResult]]: +async def probe(engine: AsyncEngine, table_name: str, schema: str | None = None) -> AsyncIterator[list[ProbeResult]]: """Snapshot the catalogs around the wrapped workload. Yields a list that holds exactly one :class:`ProbeResult` once the block exits. A list (rather than a return value) is the only way an async context manager can - hand a result back to its caller. + hand a result back to its caller. ``schema`` defaults to ``current_schema()``. + + If the workload raises, the exception propagates and the list stays **empty** -- + the snapshot is deliberately not taken in a ``finally``, because a half-run + workload's counters are not a measurement. Callers must not assume ``sink[0]`` + without first checking that the block completed. + + ``engine`` is disposed (all pooled connections closed) immediately before each + snapshot: that is what force-flushes the pending backend stats. The engine stays + usable afterwards -- SQLAlchemy just opens fresh connections -- so a sweep can keep + sharing one engine across runs. Connections still *checked out* when the probe exits + are not disposed, so the workload must have released them before the block ends. """ sink: list[ProbeResult] = [] - # Settle first: the seed's inserts must be visible in pg_stat_user_tables before we - # capture the origin, or they leak into the delta. - await _settle(engine, table_name) - async with engine.begin() as conn: - start = (await conn.execute(_START_SQL, {"table": table_name})).one() + + # Flush the seed's stats before reading the origin, or they leak into the delta. + await engine.dispose() + async with _autocommit(engine) as conn: + start = (await conn.execute(_START_SQL, {"table": table_name, "schema": schema})).one() + if not start.present: + msg = ( + f"table {start.schemaname}.{table_name} has no pg_stat_user_tables row: it does not exist " + "(or is not a user table). Create it before opening the probe." + ) + raise RuntimeError(msg) yield sink - await _settle(engine, table_name) - async with engine.connect() as conn: - row = (await conn.execute(_SNAPSHOT_SQL, {"lsn0": start.lsn, "table": table_name})).one() + # Flush the workload's stats before reading the end snapshot. + await engine.dispose() + async with _autocommit(engine) as conn: + row = ( + await conn.execute( + _SNAPSHOT_SQL, + {"lsn0": start.lsn, "table": table_name, "schema": start.schemaname}, + ) + ).one() sink.append( ProbeResult( From 93f33bae0f4f16ea466f9a10e410b4f72b365c44 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 00:39:40 +0300 Subject: [PATCH 05/13] fix(bench): fail loudly when probe() exits with a connection checked out engine.dispose() only closes checked-in connections. A workload that still holds a connection when the probe block exits never flushes that backend's stats, so tup_ins/tup_upd/tup_del/wal_records silently under-report (or read zero) while calls stays correct and looks plausible. Assert the pool has zero checked-out connections on the exit path and raise RuntimeError naming the count, matching the existing assert_owns_database standard of failing loudly instead of reporting a quietly wrong number. Also document the other footgun review found: constructing an AsyncEngine inside the probe window inflates calls by 6 via SQLAlchemy's one-time dialect-init queries, none of which match the pg_stat filter. --- benchmarks/probes.py | 24 +++++++++++++++++++++++- 1 file changed, 23 insertions(+), 1 deletion(-) diff --git a/benchmarks/probes.py b/benchmarks/probes.py index 239bb29..f2f517c 100644 --- a/benchmarks/probes.py +++ b/benchmarks/probes.py @@ -193,7 +193,17 @@ async def probe(engine: AsyncEngine, table_name: str, schema: str | None = None) snapshot: that is what force-flushes the pending backend stats. The engine stays usable afterwards -- SQLAlchemy just opens fresh connections -- so a sweep can keep sharing one engine across runs. Connections still *checked out* when the probe exits - are not disposed, so the workload must have released them before the block ends. + are not disposed by :meth:`AsyncEngine.dispose`, so this raises ``RuntimeError`` + instead of silently under-reporting; the workload must release every connection + (stop the broker, exit all ``async with engine.begin()`` blocks) before the block ends. + + ``engine`` must also be constructed *before* the probe block opens. SQLAlchemy's + one-time dialect init (``select pg_catalog.version()``, ``select current_schema()``, + ``show standard_conforming_strings``, ``show transaction isolation level``, plus a + BEGIN/ROLLBACK) runs on an engine's first connection and inflates ``calls`` by 6; + none of those statements contain ``pg_stat``, so the self-exclusion filter cannot + catch them. It is deterministic, so it will not flap the gate -- it will just bake + a wrong baseline in. """ sink: list[ProbeResult] = [] @@ -210,6 +220,18 @@ async def probe(engine: AsyncEngine, table_name: str, schema: str | None = None) yield sink + checked_out = engine.pool.checkedout() # ty: ignore[unresolved-attribute] + if checked_out: + msg = ( + f"{checked_out} connection(s) are still checked out of the pool. " + "AsyncEngine.dispose() only closes checked-in connections, so a checked-out " + "backend never exits and its stats are never flushed -- the tuple and WAL " + "counters would silently under-report. Release/close all connections (stop " + "the broker, exit all `async with engine.begin()` blocks) before leaving " + "the probe() block." + ) + raise RuntimeError(msg) + # Flush the workload's stats before reading the end snapshot. await engine.dispose() async with _autocommit(engine) as conn: From 41b3cea2a6afd56083829b2e2e1512ccb6710539 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 00:43:43 +0300 Subject: [PATCH 06/13] feat(bench): add consumer and producer workloads --- benchmarks/workload.py | 150 +++++++++++++++++++++++++++++++++++++++++ 1 file changed, 150 insertions(+) create mode 100644 benchmarks/workload.py diff --git a/benchmarks/workload.py b/benchmarks/workload.py new file mode 100644 index 0000000..cb614f8 --- /dev/null +++ b/benchmarks/workload.py @@ -0,0 +1,150 @@ +"""The workloads: drive the real broker, measure what Postgres did. + +Two scenarios: + +* **consumer** -- seed N rows directly, then start the broker with a no-op handler + and drain them. A no-op handler is the point: what is measured is transport cost, + not handler cost. +* **producer** -- N ``publish()`` calls in one transaction. Surfaces the two + statements per publish (``INSERT ... RETURNING`` + ``SELECT pg_notify``). + +Each run gets a fresh table (UUID suffix, mirroring the ``outbox_table`` fixture in +tests/conftest.py) so pg_stat_user_tables counters start at zero. + +Two rules the probe imposes on everything below (see :func:`benchmarks.probes.probe`): + +* the ``AsyncEngine`` is built by :func:`make_engine` and passed in, never constructed + inside a probe block -- SQLAlchemy's one-time dialect init would bake 6 extra + statements into the baseline; +* every connection is released before the probe block exits (the broker is stopped and + every ``engine.begin()`` / ``AsyncSession`` block is closed inside the window), because + ``dispose()`` cannot flush the stats of a backend that is still checked out. +""" + +import asyncio +import dataclasses +import time +import uuid + +from sqlalchemy import MetaData, Table, insert, text +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine + +from benchmarks.config import APPLICATION_NAME, RunConfig +from benchmarks.probes import ProbeResult, assert_owns_database, ensure_extension, probe +from faststream_outbox import OutboxBroker, make_outbox_table + + +_DRAIN_TIMEOUT_SECONDS = 600.0 + + +@dataclasses.dataclass(frozen=True) +class RunResult: + """One scenario's wall time plus the catalog deltas it produced.""" + + scenario: str + config: RunConfig + wall_seconds: float + probe: ProbeResult + + +def make_engine(dsn: str) -> AsyncEngine: + """Engine tagged with APPLICATION_NAME so the ownership check can recognize us.""" + return create_async_engine( + dsn, + future=True, + connect_args={"server_settings": {"application_name": APPLICATION_NAME}}, + ) + + +async def _create_table(engine: AsyncEngine) -> tuple[MetaData, Table]: + metadata = MetaData() + table = make_outbox_table(metadata, table_name=f"bench_{uuid.uuid4().hex[:10]}") + async with engine.begin() as conn: + await conn.run_sync(metadata.create_all) + return metadata, table + + +async def _drop_table(engine: AsyncEngine, metadata: MetaData) -> None: + async with engine.begin() as conn: + await conn.run_sync(metadata.drop_all) + + +async def _checkpoint(engine: AsyncEngine) -> None: + """Force a checkpoint before measuring. + + Done consistently so the full-page-image cost is at least reproducible. WAL *bytes* + still swing run to run because of FPI; that is why the gate is on wal_records, not + wal_bytes. + """ + async with engine.connect() as conn: + await conn.execution_options(isolation_level="AUTOCOMMIT") + await conn.execute(text("CHECKPOINT")) + + +async def run_consumer(engine: AsyncEngine, cfg: RunConfig) -> RunResult: + """Seed N rows, drain them through the real broker with a no-op handler.""" + await ensure_extension(engine) + metadata, table = await _create_table(engine) + try: + payload = b"x" * cfg.payload_bytes + async with engine.begin() as conn: + await conn.execute( + insert(table), + [{"queue": cfg.queue, "payload": payload, "headers": {}} for _ in range(cfg.messages)], + ) + + await _checkpoint(engine) + await assert_owns_database(engine) + + broker = OutboxBroker(engine, outbox_table=table) + drained = asyncio.Event() + seen = 0 + + @broker.subscriber( + cfg.queue, + max_workers=cfg.max_workers, + fetch_batch_size=cfg.fetch_batch_size, + ) + async def _handler(msg: bytes) -> None: # noqa: ARG001 + nonlocal seen + seen += 1 + if seen >= cfg.messages: + drained.set() + + async with probe(engine, table.name) as sink: + started = time.perf_counter() + await broker.start() + try: + await asyncio.wait_for(drained.wait(), timeout=_DRAIN_TIMEOUT_SECONDS) + finally: + # Must complete inside the probe block: a still-checked-out connection + # never flushes its stats. + await broker.stop() + wall = time.perf_counter() - started + + return RunResult(scenario="consumer", config=cfg, wall_seconds=wall, probe=sink[0]) + finally: + await _drop_table(engine, metadata) + + +async def run_producer(engine: AsyncEngine, cfg: RunConfig) -> RunResult: + """N publish() calls in one transaction -- the write-path cost.""" + await ensure_extension(engine) + metadata, table = await _create_table(engine) + try: + await _checkpoint(engine) + await assert_owns_database(engine) + + broker = OutboxBroker(engine, outbox_table=table) + payload = b"x" * cfg.payload_bytes + + async with probe(engine, table.name) as sink: + started = time.perf_counter() + async with AsyncSession(engine) as session, session.begin(): + for _ in range(cfg.messages): + await broker.publish(payload, queue=cfg.queue, session=session) + wall = time.perf_counter() - started + + return RunResult(scenario="producer", config=cfg, wall_seconds=wall, probe=sink[0]) + finally: + await _drop_table(engine, metadata) From 146e4e3abd1380768eadecdecf0c549f50daa6f7 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 09:12:15 +0300 Subject: [PATCH 07/13] fix(bench): gate load-independent per-op counts; make drain DB-bound and exact The benchmark gate must assert structural, per-message quantities, not totals that scale with wall-clock. Five measurement defects, all reproduced under load: - probes: break `calls` down by SQL operation into `ProbeResult.calls_by_op` (a second aggregate grouped by the leading keyword, on the same AUTOCOMMIT connection, self-excluding via `pg_stat` in its text). The terminal DELETE is exactly 1/message on the happy path, so `calls_by_op['delete']` isolates a load-independent gate metric; the total `calls` stays informational. - workload: pass min/max_fetch_interval=0.001 so drain is DB-bound, not sleep-bound (the seed emits no pg_notify, so the fetch loop otherwise slept its full max_fetch_interval and max_workers had zero throughput effect). - workload: count distinct row identities (unique seed payload per row) for drain detection so a lease-expiry redelivery can't stop the broker early; assert count(*)==0 outside the probe block to catch a partial drain. - workload: disable autovacuum on the bench table so a mid-window autovacuum can't reap dead tuples; dead_tup becomes an exact, deterministic garbage count. - workload: silence the broker (logger=None) so per-message INFO logging stays out of the window; reuse probes._autocommit in _checkpoint; drop the no-op future=True from make_engine. Co-Authored-By: Claude Opus 4.8 (1M context) --- benchmarks/probes.py | 31 +++++++++++++++++++++ benchmarks/workload.py | 62 +++++++++++++++++++++++++++++++++--------- 2 files changed, 80 insertions(+), 13 deletions(-) diff --git a/benchmarks/probes.py b/benchmarks/probes.py index f2f517c..9b4cb9d 100644 --- a/benchmarks/probes.py +++ b/benchmarks/probes.py @@ -99,6 +99,21 @@ """, ) +# Per-operation statement counts. pg_stat_statements holds one row per normalized query; +# grouping sum(calls) by the lowercased leading SQL keyword isolates each operation so the +# gate can assert load-independent per-message quantities (the terminal DELETE is exactly 1 +# per message on the happy path) instead of the total `calls`, which scales with the fetch +# loop's wall-clock polling. Its own text contains 'pg_stat', so it self-excludes like the +# rest, and it runs on the same AUTOCOMMIT connection. +_CALLS_BY_OP_SQL = text( + """ + SELECT lower(split_part(btrim(s.query), ' ', 1)) AS op, sum(s.calls) AS calls + FROM pg_stat_statements s + WHERE s.query NOT ILIKE '%pg_stat%' + GROUP BY 1 + """, +) + _OWNERSHIP_SQL = text( "SELECT count(*) FROM pg_stat_activity " "WHERE datname = current_database() " @@ -109,12 +124,20 @@ _EXTENSION_SQL = text("CREATE EXTENSION IF NOT EXISTS pg_stat_statements") +# The leading SQL keywords we classify per-operation; anything else lands in ``other``. +_KNOWN_OPS = ("select", "insert", "update", "delete", "with") + @dataclasses.dataclass(frozen=True) class ProbeResult: """Catalog deltas across one workload.""" calls: int + # Per-operation statement counts (a delta; pg_stat_statements was reset at start), + # keyed by the lowercased leading SQL keyword plus a catch-all ``other``. The gate + # asserts these load-independent per-message counts (notably ``delete`` == messages + # on the happy path), not the load-dependent ``calls`` total. + calls_by_op: dict[str, int] exec_ms: float blks_hit: int blks_read: int @@ -241,11 +264,19 @@ async def probe(engine: AsyncEngine, table_name: str, schema: str | None = None) {"lsn0": start.lsn, "table": table_name, "schema": start.schemaname}, ) ).one() + # Same AUTOCOMMIT connection; self-excludes via 'pg_stat' in its own text. + op_rows = (await conn.execute(_CALLS_BY_OP_SQL)).all() + + calls_by_op = dict.fromkeys((*_KNOWN_OPS, "other"), 0) + for op_row in op_rows: + key = op_row.op if op_row.op in _KNOWN_OPS else "other" + calls_by_op[key] += int(op_row.calls) sink.append( ProbeResult( # pg_stat_statements was reset at start, so these are already deltas. calls=int(row.calls), + calls_by_op=calls_by_op, exec_ms=float(row.exec_ms), blks_hit=int(row.blks_hit), blks_read=int(row.blks_read), diff --git a/benchmarks/workload.py b/benchmarks/workload.py index cb614f8..bb105be 100644 --- a/benchmarks/workload.py +++ b/benchmarks/workload.py @@ -30,7 +30,7 @@ from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine from benchmarks.config import APPLICATION_NAME, RunConfig -from benchmarks.probes import ProbeResult, assert_owns_database, ensure_extension, probe +from benchmarks.probes import ProbeResult, _autocommit, assert_owns_database, ensure_extension, probe from faststream_outbox import OutboxBroker, make_outbox_table @@ -51,7 +51,6 @@ def make_engine(dsn: str) -> AsyncEngine: """Engine tagged with APPLICATION_NAME so the ownership check can recognize us.""" return create_async_engine( dsn, - future=True, connect_args={"server_settings": {"application_name": APPLICATION_NAME}}, ) @@ -61,6 +60,10 @@ async def _create_table(engine: AsyncEngine) -> tuple[MetaData, Table]: table = make_outbox_table(metadata, table_name=f"bench_{uuid.uuid4().hex[:10]}") async with engine.begin() as conn: await conn.run_sync(metadata.create_all) + # Disable autovacuum on the bench table so a mid-window autovacuum can't reap dead + # tuples (dead_tup swung 686/486/306/106 across identical runs) or perturb WAL. With + # it off, dead_tup is an exact count of the garbage the outbox generated. + await conn.execute(text(f'ALTER TABLE "{table.name}" SET (autovacuum_enabled = off)')) return metadata, table @@ -76,39 +79,60 @@ async def _checkpoint(engine: AsyncEngine) -> None: still swing run to run because of FPI; that is why the gate is on wal_records, not wal_bytes. """ - async with engine.connect() as conn: - await conn.execution_options(isolation_level="AUTOCOMMIT") + async with _autocommit(engine) as conn: await conn.execute(text("CHECKPOINT")) +def _seed_payload(index: int, size: int) -> bytes: + """Unique-per-row payload of at least *size* bytes. + + The row index is embedded as a prefix so drain detection can count distinct rows + (via the decoded body the handler receives), not handler invocations: a single + lease-expiry redelivery must not be miscounted as progress and stop the broker early. + """ + prefix = f"{index}:".encode() + return prefix + b"x" * max(0, size - len(prefix)) + + async def run_consumer(engine: AsyncEngine, cfg: RunConfig) -> RunResult: """Seed N rows, drain them through the real broker with a no-op handler.""" await ensure_extension(engine) metadata, table = await _create_table(engine) try: - payload = b"x" * cfg.payload_bytes async with engine.begin() as conn: await conn.execute( insert(table), - [{"queue": cfg.queue, "payload": payload, "headers": {}} for _ in range(cfg.messages)], + [ + {"queue": cfg.queue, "payload": _seed_payload(i, cfg.payload_bytes), "headers": {}} + for i in range(cfg.messages) + ], ) await _checkpoint(engine) await assert_owns_database(engine) - broker = OutboxBroker(engine, outbox_table=table) + # logger=None silences the broker's per-message INFO logging, which would + # otherwise run inside the measured window (2 lines per message). + broker = OutboxBroker(engine, outbox_table=table, logger=None) drained = asyncio.Event() - seen = 0 + # Collect distinct row identities, not invocation count, so a lease-expiry + # redelivery can't drive the counter to N while rows are still unprocessed. + seen_ids: set[bytes] = set() @broker.subscriber( cfg.queue, max_workers=cfg.max_workers, fetch_batch_size=cfg.fetch_batch_size, + # Tiny fetch interval so drain is DB-bound, not sleep-bound: the seed's raw + # INSERT emits no pg_notify and the broker isn't running during it, so the + # fetch loop never gets a NOTIFY wake and would otherwise sleep its full + # max_fetch_interval whenever _inflight drains. min must be <= max. + min_fetch_interval=0.001, + max_fetch_interval=0.001, ) - async def _handler(msg: bytes) -> None: # noqa: ARG001 - nonlocal seen - seen += 1 - if seen >= cfg.messages: + async def _handler(msg: bytes) -> None: + seen_ids.add(msg) + if len(seen_ids) >= cfg.messages: drained.set() async with probe(engine, table.name) as sink: @@ -122,6 +146,18 @@ async def _handler(msg: bytes) -> None: # noqa: ARG001 await broker.stop() wall = time.perf_counter() - started + # Assert emptiness OUTSIDE the probe block (stop() had to run inside it): a partial + # drain leaves rows behind, which this catches instead of baking a silently wrong + # baseline. Distinct-identity drain detection makes an early stop possible only if a + # row was genuinely never delivered. + async with _autocommit(engine) as conn: + # S608: table.name is the harness-generated ``bench_`` identifier, not + # request input, and is quoted; no injection surface. + remaining = (await conn.execute(text(f'SELECT count(*) FROM "{table.name}"'))).scalar_one() # noqa: S608 + if remaining: + msg = f"consumer left {remaining} undrained row(s) in {table.name}: measurement is not a clean drain" + raise RuntimeError(msg) + return RunResult(scenario="consumer", config=cfg, wall_seconds=wall, probe=sink[0]) finally: await _drop_table(engine, metadata) @@ -135,7 +171,7 @@ async def run_producer(engine: AsyncEngine, cfg: RunConfig) -> RunResult: await _checkpoint(engine) await assert_owns_database(engine) - broker = OutboxBroker(engine, outbox_table=table) + broker = OutboxBroker(engine, outbox_table=table, logger=None) payload = b"x" * cfg.payload_bytes async with probe(engine, table.name) as sink: From c50cd4600323e97bcc047fb89e22c4a45aed0927 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 09:28:23 +0300 Subject: [PATCH 08/13] feat(bench): normalize per-message metrics and gate against a baseline Gate raw structural totals (delete_calls + tuple counters exact, wal_records within a 10% band), not the load-dependent total calls or FPI-noisy WAL bytes. Adds pure-function unit tests that import benchmarks.* to prove the coverage omit keeps the Postgres-only modules out of the 100% gate. --- benchmarks/report.py | 134 +++++++++++++++++++++++++++ tests/test_benchmarks.py | 195 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 329 insertions(+) create mode 100644 benchmarks/report.py create mode 100644 tests/test_benchmarks.py diff --git a/benchmarks/report.py b/benchmarks/report.py new file mode 100644 index 0000000..48772fc --- /dev/null +++ b/benchmarks/report.py @@ -0,0 +1,134 @@ +"""Normalize runs to per-message metrics, render them, and gate against a baseline. + +Metric tiers are set by Task 3's *measured* determinism, not by assumption. The total +statement count is NOT structural: the subscriber's fetch loop polls on a wall-clock +timer, so ``calls`` was 519 idle but 894 under CPU load. What IS load-independent and +was verified exact both idle and under load: + +* ``tup_upd`` / ``tup_del`` / ``tup_ins`` -- tuple counters (rows actually changed). + These are the PRIMARY structural invariant. +* ``delete_calls`` -- the terminal DELETE execution count (corroborating). The fetch + CTE compiles to ``WITH ready AS (...) UPDATE`` and classifies under the ``with`` key, + so ``calls_by_op['update']`` is ~0 on the happy path -- gate ``delete``, never that. +* ``insert_calls`` / ``select_calls`` -- the producer's INSERT and pg_notify SELECT. + +``wal_records`` is near-structural but carries ~5% server-wide spread, so it is gated +with a 10% upper-bound band, never exact. The total ``calls``, ``wal_bytes`` (full-page +images), ``wal_fpi`` and wall-clock / ``msgs_per_second`` are reported for humans but +NEVER gated -- they are timing- or FPI-noisy and would flake the build. +""" + +import typing + +from benchmarks.workload import RunResult + + +# Raw structural totals, gated on EXACT equality. Load-independent across idle/loaded runs. +# For a consumer run insert/select are 0; for a producer run delete is 0. Exact-0 == +# exact-0 passes, so the same key set gates both scenarios. +EXACT_KEYS: tuple[str, ...] = ("delete_calls", "tup_upd", "tup_del", "tup_ins", "insert_calls", "select_calls") + +# Near-deterministic: ~5% observed spread, so a 10% upper-bound band leaves headroom. +TOLERANT_KEYS: dict[str, float] = {"wal_records": 0.10} + + +def normalize(result: RunResult) -> dict[str, float]: + """Per-message metrics for humans, plus the raw totals the gate compares. + + The gate reads the raw structural totals (``delete_calls``, the tuple counters, + ``wal_records``); the ``*_per_msg`` ratios and ``msgs_per_second`` exist only for the + human table. Task 5 pins the message count to the baseline's, so the raw totals are + directly comparable and free of float-rounding ambiguity in the gate. + """ + n = result.config.messages + p = result.probe + delete_calls = p.calls_by_op.get("delete", 0) + insert_calls = p.calls_by_op.get("insert", 0) + select_calls = p.calls_by_op.get("select", 0) + return { + "messages": n, + "wall_seconds": result.wall_seconds, + "msgs_per_second": n / result.wall_seconds if result.wall_seconds else 0.0, + # Raw structural totals -- the gate compares exactly these. + "delete_calls": delete_calls, + "insert_calls": insert_calls, + "select_calls": select_calls, + "tup_ins": p.tup_ins, + "tup_upd": p.tup_upd, + "tup_del": p.tup_del, + "wal_records": p.wal_records, + # Per-message ratios -- informational only. + "round_trips_per_msg": p.calls / n, + "delete_calls_per_msg": delete_calls / n, + "wal_records_per_msg": p.wal_records / n, + "wal_bytes_per_msg": p.wal_bytes / n, + # Reported context, never gated: timing/FPI noise and post-mortem bloat signals. + "calls": p.calls, + "wal_bytes": p.wal_bytes, + "wal_fpi": p.wal_fpi, + "tup_hot_upd": p.tup_hot_upd, + "dead_tup": p.dead_tup, + "heap_bytes": p.heap_bytes, + "index_bytes": p.index_bytes, + } + + +def _key(result: RunResult) -> str: + cfg = result.config + return f"{result.scenario}/w{cfg.max_workers}/b{cfg.fetch_batch_size}" + + +def to_baseline(results: list[RunResult]) -> dict[str, typing.Any]: + """Build the JSON payload: each run keyed by scenario + workers + batch size.""" + return {"runs": {_key(r): normalize(r) for r in results}} + + +def format_table(results: list[RunResult]) -> str: + """Render the runs as a human table with a gated-vs-informational footer.""" + header = ( + f"{'scenario':<22} {'msg/s':>9} {'delete/msg':>11} {'WALrec/msg':>11} " + f"{'WALB/msg':>9} {'fpi':>5} {'upd':>7} {'del':>7} {'dead_tup':>9}" + ) + lines = [header, "-" * len(header)] + for r in results: + m = normalize(r) + lines.append( + f"{_key(r):<22} {m['msgs_per_second']:>9.0f} {m['delete_calls_per_msg']:>11.3f} " + f"{m['wal_records_per_msg']:>11.2f} {m['wal_bytes_per_msg']:>9.0f} " + f"{m['wal_fpi']:>5.0f} {m['tup_upd']:>7.0f} {m['tup_del']:>7.0f} {m['dead_tup']:>9.0f}", + ) + lines.append("") + lines.append("GATED (fail the build): delete_calls + tuple counters (upd/del/ins) + the") + lines.append("producer's insert_calls/select_calls, all exact; wal_records within a 10%") + lines.append("band. Columns upd/del are those gated raw counters.") + lines.append("INFORMATIONAL (never gated): total calls, WAL bytes, msg/s -- FPI/timing noise.") + lines.append("delete/msg, WALrec/msg and WALB/msg are per-message views, not the gated totals.") + return "\n".join(lines) + + +def compare(current: dict[str, typing.Any], baseline: dict[str, typing.Any]) -> list[str]: + """Diff current raw totals against the baseline. Empty list means pass. + + EXACT_KEYS must match exactly; TOLERANT_KEYS may only exceed an upper bound (a + regression is MORE work, so a lower value never fails). A baseline run key absent + from ``current`` is itself a failure. + """ + failures: list[str] = [] + for key, want in baseline["runs"].items(): + got = current["runs"].get(key) + if got is None: + failures.append(f"{key}: missing from the current run") + continue + failures.extend( + f"{key}: {metric} changed (exact-gated): baseline {want[metric]:.0f} -> current {got[metric]:.0f}" + for metric in EXACT_KEYS + if got[metric] != want[metric] + ) + for metric, tolerance in TOLERANT_KEYS.items(): + limit = want[metric] * (1 + tolerance) + if got[metric] > limit: + failures.append( + f"{key}: {metric} exceeded the {tolerance:.0%} band: " + f"baseline {want[metric]:.0f} -> current {got[metric]:.0f} (limit {limit:.0f})", + ) + return failures diff --git a/tests/test_benchmarks.py b/tests/test_benchmarks.py new file mode 100644 index 0000000..9cfb2d8 --- /dev/null +++ b/tests/test_benchmarks.py @@ -0,0 +1,195 @@ +"""Unit tests for the benchmark harness' pure logic. + +Deliberately no Postgres: the pytest CI job's postgres service cannot preload +pg_stat_statements, so anything catalog-dependent is verified by `just bench-check`. +The gate compares RAW STRUCTURAL TOTALS (delete_calls, the tuple counters, wal_records), +which Task 3 proved load-independent -- NOT the total `calls`, which scales with the +fetch loop's wall-clock polling. +""" + +import pytest + +from benchmarks.config import RunConfig +from benchmarks.probes import ProbeResult +from benchmarks.report import EXACT_KEYS, compare, format_table, normalize, to_baseline +from benchmarks.workload import RunResult + + +def _probe( + *, + delete_calls: int = 500, + insert_calls: int = 0, + select_calls: int = 0, + **overrides: float, +) -> ProbeResult: + calls_by_op = { + "select": select_calls, + "insert": insert_calls, + "update": 0, + "delete": delete_calls, + "with": 12, + "other": 10, + } + defaults: dict[str, object] = { + "calls": 522, + "calls_by_op": calls_by_op, + "exec_ms": 100.0, + "blks_hit": 1000, + "blks_read": 10, + "wal_records": 5501, + "wal_fpi": 5, + "wal_bytes": 385408, + "tup_ins": 0, + "tup_upd": 500, + "tup_del": 500, + "tup_hot_upd": 0, + "tup_newpage_upd": 500, + "dead_tup": 0, + "heap_bytes": 65536, + "index_bytes": 32768, + } + defaults.update(overrides) + return ProbeResult(**defaults) + + +def _result( + scenario: str = "consumer", *, messages: int = 500, wall_seconds: float = 2.0, **probe_kwargs: float +) -> RunResult: + return RunResult( + scenario=scenario, + config=RunConfig(messages=messages), + wall_seconds=wall_seconds, + probe=_probe(**probe_kwargs), # ty: ignore[invalid-argument-type] + ) + + +def _producer(**probe_kwargs: float) -> RunResult: + """Build a producer run: inserts + pg_notify selects, no deletes or updates.""" + return _result( + "producer", + delete_calls=0, + insert_calls=500, + select_calls=500, + tup_ins=500, + tup_upd=0, + tup_del=0, + **probe_kwargs, # ty: ignore[invalid-argument-type] + ) + + +def test_normalize_divides_by_message_count() -> None: + metrics = normalize(_result()) + assert metrics["round_trips_per_msg"] == pytest.approx(1.044) + assert metrics["wal_records_per_msg"] == pytest.approx(11.002) + assert metrics["delete_calls_per_msg"] == pytest.approx(1.0) + assert metrics["msgs_per_second"] == pytest.approx(250.0) + # delete_calls is surfaced from calls_by_op, not from the `calls` total. + assert metrics["delete_calls"] == 500 + # Raw totals survive normalization -- the gate compares these. + assert metrics["tup_del"] == 500 + assert metrics["tup_upd"] == 500 + assert metrics["messages"] == 500 + assert metrics["wall_seconds"] == pytest.approx(2.0) + + +def test_compare_passes_on_identical_results() -> None: + baseline = to_baseline([_result()]) + assert compare(to_baseline([_result()]), baseline) == [] + + +def test_compare_fails_exactly_on_extra_delete_call() -> None: + # One extra terminal DELETE per run is the core regression the gate catches. + baseline = to_baseline([_result()]) + current = to_baseline([_result(delete_calls=501)]) + failures = compare(current, baseline) + assert len(failures) == 1 + assert "delete_calls" in failures[0] + assert "501" in failures[0] + + +def test_compare_fails_on_tup_del_drift() -> None: + baseline = to_baseline([_result()]) + failures = compare(to_baseline([_result(tup_del=499)]), baseline) + assert len(failures) == 1 + assert "tup_del" in failures[0] + + +def test_compare_fails_on_tup_upd_drift() -> None: + baseline = to_baseline([_result()]) + failures = compare(to_baseline([_result(tup_upd=501)]), baseline) + assert len(failures) == 1 + assert "tup_upd" in failures[0] + + +def test_compare_tolerates_small_wal_record_drift() -> None: + baseline = to_baseline([_result(wal_records=5501)]) + # +3.3%, within the 10% band. + assert compare(to_baseline([_result(wal_records=5683)]), baseline) == [] + + +def test_compare_fails_on_large_wal_record_drift() -> None: + baseline = to_baseline([_result(wal_records=5501)]) + # +20%, well past the 10% band. + failures = compare(to_baseline([_result(wal_records=6601)]), baseline) + assert len(failures) == 1 + assert "wal_records" in failures[0] + + +def test_compare_tolerates_lower_wal_records() -> None: + # Fewer WAL records is less work, never a regression: only the upper bound fails. + baseline = to_baseline([_result(wal_records=5501)]) + assert compare(to_baseline([_result(wal_records=4000)]), baseline) == [] + + +def test_compare_never_gates_calls_wal_bytes_or_wall_clock() -> None: + baseline = to_baseline([_result(calls=522, wal_bytes=385408, wall_seconds=2.0)]) + # Total calls (fetch-loop polling), WAL bytes (full-page images) and wall time all + # swing wildly run to run -- none of them is gated. + current = to_baseline([_result(calls=894, wal_bytes=639104, wall_seconds=9.0)]) + assert compare(current, baseline) == [] + + +def test_compare_reports_a_missing_scenario() -> None: + baseline = to_baseline([_result()]) + failures = compare({"runs": {}}, baseline) + assert len(failures) == 1 + assert "missing" in failures[0].lower() + + +def test_consumer_and_producer_pass_against_themselves_with_same_exact_keys() -> None: + # The same EXACT_KEYS set gates both: consumer deletes (insert/select 0), producer + # inserts (delete 0). Exact-0 == exact-0 passes for the empty buckets. + consumer = to_baseline([_result()]) + producer = to_baseline([_producer()]) + assert compare(consumer, consumer) == [] + assert compare(producer, producer) == [] + # And they are distinct runs, so a combined baseline keeps both. + combined = to_baseline([_result(), _producer()]) + assert len(combined["runs"]) == 2 + assert compare(combined, combined) == [] + + +def test_exact_keys_cover_the_structural_totals() -> None: + assert set(EXACT_KEYS) == { + "delete_calls", + "tup_upd", + "tup_del", + "tup_ins", + "insert_calls", + "select_calls", + } + + +def test_format_table_labels_gated_vs_informational() -> None: + table = format_table([_result(), _producer()]) + # Both runs are rendered. + assert "consumer/w1/b100" in table + assert "producer/w1/b100" in table + # Required columns are present. + for column in ("msg/s", "delete/msg", "WALrec/msg", "WALB/msg", "fpi", "upd", "del", "dead_tup"): + assert column in table + # The gated vs informational split is legible in the footer. + assert "GATED" in table + assert "INFORMATIONAL" in table + assert "wal_records" in table + assert "delete_calls" in table From 8bfd8cdadac2443bcde9cdc92f4a85f787af68a0 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 09:41:26 +0300 Subject: [PATCH 09/13] feat(bench): add bench/bench-check recipes and commit the baseline --- Justfile | 8 ++ benchmarks/__main__.py | 113 +++++++++++++++++++++++++++ benchmarks/baseline.json | 165 +++++++++++++++++++++++++++++++++++++++ 3 files changed, 286 insertions(+) create mode 100644 benchmarks/__main__.py create mode 100644 benchmarks/baseline.json diff --git a/Justfile b/Justfile index bf7aec9..7c1272b 100644 --- a/Justfile +++ b/Justfile @@ -44,6 +44,14 @@ publish: uv build uv publish +# Run the benchmark sweep and print the report. Args forward unquoted (same caveat as `test`). +bench *args: down && down + docker compose run application uv run python -m benchmarks run {{ args }} + +# Gate the deterministic counters against benchmarks/baseline.json. CI runs this. +bench-check: down && down + docker compose run application uv run python -m benchmarks check + # Serve docs at http://127.0.0.1:8000 with hot-reload on save. docs-serve: uvx --with-requirements docs/requirements.txt mkdocs serve diff --git a/benchmarks/__main__.py b/benchmarks/__main__.py new file mode 100644 index 0000000..861ce1d --- /dev/null +++ b/benchmarks/__main__.py @@ -0,0 +1,113 @@ +"""CLI: `python -m benchmarks run` / `python -m benchmarks check`. + +Invoked through `just bench` / `just bench-check`, which run it inside docker compose +so the postgres service has pg_stat_statements preloaded. +""" + +import argparse +import asyncio +import json +import os +import pathlib +import sys + +from benchmarks.config import DEFAULT_DSN, RunConfig +from benchmarks.report import compare, format_table, to_baseline +from benchmarks.workload import RunResult, make_engine, run_consumer, run_producer + + +BASELINE_PATH = pathlib.Path(__file__).parent / "baseline.json" + +# The sweep. Consumer points cross workers x batch size; the producer point measures +# the write path (two statements per publish). +_WORKER_COUNTS = (1, 2, 4) +_BATCH_SIZES = (10, 100) + + +def _median_by_wall(runs: list[RunResult]) -> RunResult: + """Return the run with the median wall-clock. + + Wall-clock is the only noisy thing we report (Docker IO), so repeat it and take the + middle. Returning the whole RunResult -- rather than a median of each metric + independently -- keeps every number in a row internally consistent: they all come + from one real run. The gated counters are deterministic anyway, so which run wins + does not affect the gate. + """ + return sorted(runs, key=lambda r: r.wall_seconds)[len(runs) // 2] + + +async def _run_sweep(dsn: str, messages: int, repeats: int) -> list[RunResult]: + """Build the engine once, then run every sweep point through it. + + NEVER construct an engine inside the loop: SQLAlchemy's one-time dialect init emits + 6 extra statements on an engine's first connection and would inflate the baseline. + """ + engine = make_engine(dsn) + results: list[RunResult] = [] + try: + # Warm-up, discarded: the first run pays connection setup and page faults. + await run_consumer(engine, RunConfig(messages=min(messages, 200))) + for workers in _WORKER_COUNTS: + for batch in _BATCH_SIZES: + cfg = RunConfig(messages=messages, max_workers=workers, fetch_batch_size=batch) + runs = [await run_consumer(engine, cfg) for _ in range(repeats)] + results.append(_median_by_wall(runs)) + producer_cfg = RunConfig(messages=messages) + producer_runs = [await run_producer(engine, producer_cfg) for _ in range(repeats)] + results.append(_median_by_wall(producer_runs)) + finally: + await engine.dispose() + return results + + +def _check(dsn: str) -> int: + """Run the sweep at the baseline's message count and gate the counters.""" + if not BASELINE_PATH.exists(): + sys.stdout.write(f"no baseline at {BASELINE_PATH}; run `just bench --write-baseline` first\n") + return 1 + baseline = json.loads(BASELINE_PATH.read_text()) + # Pin the message count to the baseline's: raw totals are only comparable at the same + # N (there is a small fixed statement overhead per run). Counters are deterministic, + # so repeats=1 -- repeating them only burns CI time. + messages = int(next(iter(baseline["runs"].values()))["messages"]) + results = asyncio.run(_run_sweep(dsn, messages, repeats=1)) + sys.stdout.write(format_table(results) + "\n") + failures = compare(to_baseline(results), baseline) + if failures: + sys.stdout.write("\nBENCHMARK GATE FAILED:\n") + for failure in failures: + sys.stdout.write(f" - {failure}\n") + return 1 + sys.stdout.write("\nbenchmark gate: OK\n") + return 0 + + +def _run(dsn: str, messages: int, repeats: int, *, write_baseline: bool) -> int: + """Run the sweep, print the table, and optionally overwrite the committed baseline.""" + results = asyncio.run(_run_sweep(dsn, messages, repeats=repeats)) + sys.stdout.write(format_table(results) + "\n") + if write_baseline: + BASELINE_PATH.write_text(json.dumps(to_baseline(results), indent=2, sort_keys=True) + "\n") + sys.stdout.write(f"\nwrote {BASELINE_PATH}\n") + return 0 + + +def main() -> int: + parser = argparse.ArgumentParser(prog="benchmarks") + parser.add_argument("command", choices=("run", "check")) + parser.add_argument("--messages", type=int, default=5_000) + # Wall-clock is IO-noisy, so `run` repeats each point and reports the median. `check` + # uses 1: the gated counters are deterministic, so repeating them only burns CI time. + parser.add_argument("--repeats", type=int, default=3) + parser.add_argument("--write-baseline", action="store_true") + args = parser.parse_args() + + dsn = os.environ.get("POSTGRES_DSN", DEFAULT_DSN) + + if args.command == "check": + return _check(dsn) + return _run(dsn, args.messages, args.repeats, write_baseline=args.write_baseline) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/benchmarks/baseline.json b/benchmarks/baseline.json new file mode 100644 index 0000000..6606e3d --- /dev/null +++ b/benchmarks/baseline.json @@ -0,0 +1,165 @@ +{ + "runs": { + "consumer/w1/b10": { + "calls": 14928, + "dead_tup": 10000, + "delete_calls": 5000, + "delete_calls_per_msg": 1.0, + "heap_bytes": 1875968, + "index_bytes": 344064, + "insert_calls": 0, + "messages": 5000, + "msgs_per_second": 460.47656165416345, + "round_trips_per_msg": 2.9856, + "select_calls": 0, + "tup_del": 5000, + "tup_hot_upd": 0, + "tup_ins": 0, + "tup_upd": 5000, + "wal_bytes": 4493088, + "wal_bytes_per_msg": 898.6176, + "wal_fpi": 242, + "wal_records": 36677, + "wal_records_per_msg": 7.3354, + "wall_seconds": 10.858315963007044 + }, + "consumer/w1/b100": { + "calls": 14598, + "dead_tup": 10000, + "delete_calls": 5000, + "delete_calls_per_msg": 1.0, + "heap_bytes": 2048000, + "index_bytes": 344064, + "insert_calls": 0, + "messages": 5000, + "msgs_per_second": 447.63515436129717, + "round_trips_per_msg": 2.9196, + "select_calls": 0, + "tup_del": 5000, + "tup_hot_upd": 0, + "tup_ins": 0, + "tup_upd": 5000, + "wal_bytes": 4797960, + "wal_bytes_per_msg": 959.592, + "wal_fpi": 262, + "wal_records": 37645, + "wal_records_per_msg": 7.529, + "wall_seconds": 11.169810841005528 + }, + "consumer/w2/b10": { + "calls": 8565, + "dead_tup": 10000, + "delete_calls": 5000, + "delete_calls_per_msg": 1.0, + "heap_bytes": 2097152, + "index_bytes": 278528, + "insert_calls": 0, + "messages": 5000, + "msgs_per_second": 1209.8879048216263, + "round_trips_per_msg": 1.713, + "select_calls": 0, + "tup_del": 5000, + "tup_hot_upd": 0, + "tup_ins": 0, + "tup_upd": 5000, + "wal_bytes": 4738120, + "wal_bytes_per_msg": 947.624, + "wal_fpi": 249, + "wal_records": 34059, + "wal_records_per_msg": 6.8118, + "wall_seconds": 4.13261425300152 + }, + "consumer/w2/b100": { + "calls": 8904, + "dead_tup": 10000, + "delete_calls": 5000, + "delete_calls_per_msg": 1.0, + "heap_bytes": 2408448, + "index_bytes": 294912, + "insert_calls": 0, + "messages": 5000, + "msgs_per_second": 978.2605434470029, + "round_trips_per_msg": 1.7808, + "select_calls": 0, + "tup_del": 5000, + "tup_hot_upd": 0, + "tup_ins": 0, + "tup_upd": 5000, + "wal_bytes": 4660688, + "wal_bytes_per_msg": 932.1376, + "wal_fpi": 243, + "wal_records": 34218, + "wal_records_per_msg": 6.8436, + "wall_seconds": 5.111112814978696 + }, + "consumer/w4/b10": { + "calls": 8007, + "dead_tup": 10000, + "delete_calls": 5000, + "delete_calls_per_msg": 1.0, + "heap_bytes": 2097152, + "index_bytes": 278528, + "insert_calls": 0, + "messages": 5000, + "msgs_per_second": 1107.6370562205375, + "round_trips_per_msg": 1.6014, + "select_calls": 0, + "tup_del": 5000, + "tup_hot_upd": 0, + "tup_ins": 0, + "tup_upd": 5000, + "wal_bytes": 4589160, + "wal_bytes_per_msg": 917.832, + "wal_fpi": 242, + "wal_records": 33485, + "wal_records_per_msg": 6.697, + "wall_seconds": 4.514114051999059 + }, + "consumer/w4/b100": { + "calls": 6714, + "dead_tup": 10000, + "delete_calls": 5000, + "delete_calls_per_msg": 1.0, + "heap_bytes": 2760704, + "index_bytes": 270336, + "insert_calls": 0, + "messages": 5000, + "msgs_per_second": 1618.337011594222, + "round_trips_per_msg": 1.3428, + "select_calls": 0, + "tup_del": 5000, + "tup_hot_upd": 0, + "tup_ins": 0, + "tup_upd": 5000, + "wal_bytes": 5176184, + "wal_bytes_per_msg": 1035.2368, + "wal_fpi": 267, + "wal_records": 34364, + "wal_records_per_msg": 6.8728, + "wall_seconds": 3.0895913299755193 + }, + "producer/w1/b100": { + "calls": 10002, + "dead_tup": 0, + "delete_calls": 0, + "delete_calls_per_msg": 0.0, + "heap_bytes": 2162688, + "index_bytes": 196608, + "insert_calls": 5000, + "messages": 5000, + "msgs_per_second": 4784.773275886715, + "round_trips_per_msg": 2.0004, + "select_calls": 5000, + "tup_del": 0, + "tup_hot_upd": 0, + "tup_ins": 5000, + "tup_upd": 0, + "wal_bytes": 2918536, + "wal_bytes_per_msg": 583.7072, + "wal_fpi": 0, + "wal_records": 15219, + "wal_records_per_msg": 3.0438, + "wall_seconds": 1.044981593004195 + } + } +} From 4f0fb7d0930bad545b227f8eebd6ad3a009f713f Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 09:53:32 +0300 Subject: [PATCH 10/13] ci: gate per-message DB counters against the benchmark baseline Adds a bench job that runs just bench-check via docker compose (not a services: container, since GitHub Actions service containers can't take command args and therefore can't set shared_preload_libraries for pg_stat_statements). --- .github/workflows/_checks.yml | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/.github/workflows/_checks.yml b/.github/workflows/_checks.yml index 12a3f71..7649ec7 100644 --- a/.github/workflows/_checks.yml +++ b/.github/workflows/_checks.yml @@ -25,6 +25,13 @@ jobs: - uses: astral-sh/setup-uv@v8.2.0 - run: just docs-build + bench: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - uses: extractions/setup-just@v4 + - run: just bench-check + pytest: runs-on: ubuntu-latest strategy: From 5b6b6d5a8a6c07ecd4531b22f88c93dddbde5824 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 09:59:35 +0300 Subject: [PATCH 11/13] docs(planning): reconcile benchmark spec with the realized gate The plan proposed gating total pg_stat_statements.calls on exact equality; implementation falsified that (fetch-loop polling makes it load-dependent, 519 idle vs 894 loaded). Record the pivot to per-operation + tuple-counter gating, correct the discovery numbers (dead_tup=2x messages, ~7 WAL records/msg not 11), and document the force-flush probe and the descoped publish_batch. --- .../2026-07-14.01-benchmark-harness.md | 159 +++++++++++++----- 1 file changed, 116 insertions(+), 43 deletions(-) diff --git a/planning/changes/2026-07-14.01-benchmark-harness.md b/planning/changes/2026-07-14.01-benchmark-harness.md index 945524d..87a91e7 100644 --- a/planning/changes/2026-07-14.01-benchmark-harness.md +++ b/planning/changes/2026-07-14.01-benchmark-harness.md @@ -1,5 +1,5 @@ --- -summary: Add a `benchmarks/` harness that drives the real broker and reports per-message DB counters + throughput, with a CI gate on the deterministic subset (round-trips, tuples, WAL records); establishes the baseline for the outbox-performance work. +summary: Added a `benchmarks/` harness driving the real broker, reporting per-message DB counters + throughput, with a CI gate on the structural subset (per-op statement counts + tuple counters exact, `wal_records` banded). Implementation falsified the original plan to gate total `calls`: fetch-loop polling makes it load-dependent, so the gate moved to per-operation counts. Establishes the baseline for the outbox-performance work. --- # Design: Benchmark harness @@ -45,12 +45,21 @@ the claim UPDATE, because `n_tup_hot_upd` is **0 of 10,000 at any fillfactor** HOT is structurally impossible when the UPDATE mutates `acquired_token` and `acquired_at`, the very columns both partial indexes are keyed on. -Measured baseline for today's code (`max_workers=1`, `fetch_batch_size=100`, -256-byte payloads, 500 messages, 5 runs): - -- **1.044 round-trips per message** — i.e. one terminal DELETE per message plus - an amortized 0.01 fetch. The per-row flush *is* the entire round-trip budget. -- **11.0 WAL records per message** — a number for the write-amplification cost. +Measured baseline for today's code (256-byte payloads, committed at 5000 +messages; the discovery numbers below are from 500-message prototype runs): + +- **Exactly 1 terminal DELETE per message** (`delete_calls == messages`) — the + per-row flush *is* the entire structural round-trip budget on the consumer path. +- **`dead_tup == 2 × messages`, exact and stable** — one dead tuple from the + lease-acquire UPDATE plus one from the terminal DELETE. This is a *direct + measurement* of the talk's "2× mutations → 2× autovacuum" complaint about + Variant 2, and it is a headline number the harness exists to surface. +- **`tup_hot_upd == 0`** — HOT is structurally impossible; the claim UPDATE + mutates `acquired_token`/`acquired_at`, the columns both partial indexes key on. +- **~7 WAL records per message** (not the ~11 an early prototype guessed) — the + write-amplification cost. +- **Producer: ~2 statements per `publish()`** (`INSERT … RETURNING` + + `SELECT pg_notify`) — the NOTIFY-per-publish cost, a candidate for a later fix. - ~840 µs per message spent purely in the DELETE round-trip on loopback, which caps single-worker throughput at ~1200 msg/s regardless of handler speed. @@ -61,7 +70,7 @@ Measured baseline for today's code (`max_workers=1`, `fetch_batch_size=100`, ``` benchmarks/ __init__.py - probes.py # DbProbe: catalog snapshot -> deltas + probes.py # probe(): catalog snapshot -> deltas workload.py # producer + consumer scenarios against the real broker report.py # human table + JSON, normalized per-message baseline.json # committed; bench-check gates against it @@ -73,18 +82,41 @@ the existing precedent. ### Probe -`DbProbe` is an async context manager yielding a `ProbeResult` of catalog deltas: +`probe()` is an async context manager yielding a `ProbeResult` of catalog deltas: -- **Round-trips** — `pg_stat_statements.calls`, summed. +- **Round-trips** — `pg_stat_statements.calls` (informational total) **plus + `calls_by_op`**, the split-by-leading-keyword breakdown the gate actually uses. - **WAL** — `pg_stat_wal.wal_records`, `wal_fpi`, plus byte count via `pg_wal_lsn_diff(pg_current_wal_lsn(), start_lsn)`. - **Tuples** — `pg_stat_user_tables` for the outbox table: `n_tup_ins/upd/del`, - `n_tup_hot_upd`, `n_tup_newpage_upd`, `n_dead_tup`, `autovacuum_count`. + `n_tup_hot_upd`, `n_tup_newpage_upd`, `n_dead_tup`. - **IO / size** — `shared_blks_hit`/`read`; `pg_relation_size` for heap and each index. - **DB CPU proxy** — `pg_stat_statements.total_exec_time`. +Two hazards surfaced in implementation and shaped the probe: + +1. **Stats flush is lazy.** Postgres' `pg_stat_user_tables` and + `pg_stat_statements` lag by up to `PGSTAT_MIN_INTERVAL` (1s), so a bare read + after the workload can under-report. The original plan's settle-loop heuristic + was worse than nothing (it polled `upd+del`, which never moves for an + insert-only producer, and its window was shorter than the flush interval). It + was replaced with a **deterministic force-flush**: `await engine.dispose()` + before each snapshot exits the pooled backends, and Postgres flushes a + backend's pending stats on exit. The probe raises if any connection is still + checked out at that point (an un-flushed backend would silently drop counters). +2. **The probe must not count itself.** Every probe statement runs on an + `AUTOCOMMIT` connection and its SQL text contains `pg_stat`, so the + `query NOT ILIKE '%pg_stat%'` filter excludes it — including SQLAlchemy's + implicit `BEGIN`/`COMMIT`, which AUTOCOMMIT suppresses entirely. The + *workload's* own transaction control is real round-trip work and stays counted. + Verified: an empty workload measures `calls == 0`, zero floor. A corollary + constraint on callers: build every `AsyncEngine` **before** opening the probe, + or SQLAlchemy's one-time dialect init leaks 6 statements into the window. + Snapshot-diff throughout, except one `pg_stat_statements_reset()` at run start. +`pg_stat_statements` (reset) yields deltas directly; `pg_stat_wal` and +`pg_stat_user_tables` (cumulative) are subtracted against a start snapshot. Two hazards, both hit while prototyping and both required to be handled explicitly: @@ -100,32 +132,65 @@ explicitly: Against a fresh per-run table (UUID suffix, mirroring the `outbox_table` fixture), 256-byte payloads to stay commensurable with the talk: -- **Producer** — N `publish()` calls in one session; separately, one - `publish_batch()`. Surfaces the two-statements-per-publish cost - (`INSERT … RETURNING` + `SELECT pg_notify`). +- **Producer** — N `publish()` calls in one transaction. Surfaces the + two-statements-per-publish cost (`INSERT … RETURNING` + `SELECT pg_notify`). + (`publish_batch` was in the plan but descoped from this first harness; it can + be added without invalidating the baseline.) - **Consumer** — bulk-seed N rows directly, start the broker with a no-op handler, wait for drain, stop. A no-op handler is the point: what is being measured is transport cost, not handler cost. Sweeps `max_workers ∈ {1,2,4}` × `fetch_batch_size ∈ {10,100}`. -### Gate - -Metrics are tiered by **measured** determinism (5 real-broker runs): + Two realities the sweep had to account for: (a) the direct seed emits no + `pg_notify` and the broker is not yet running, so the fetch loop would sleep its + full `max_fetch_interval` between batches — the benchmark pins + `min/max_fetch_interval = 1 ms` so drain is DB-bound and `max_workers` actually + moves throughput (verified w=1 → w=4 gives a real speedup). This inflates the + *ungated, informational* total `calls` with polling, which is exactly why total + `calls` is not a gated metric. (b) Autovacuum is disabled on the bench table + (`autovacuum_enabled = off`) so it cannot fire mid-window and eat the dead + tuples we are trying to count — making `dead_tup` an exact measurement. -| Tier | Metric | Observed spread | Treatment | -| --- | --- | --- | --- | -| 1 | `pg_stat_statements.calls`, `n_tup_ins/upd/del` | **0.0%** | gate on exact equality | -| 2 | `wal_records` / msg | 3.3% | gate with ±10% tolerance | -| 3 | `wal_bytes` / msg, `wal_fpi`, wall-clock | **65.8%** | report only, never gate | - -The WAL-*bytes* swing (771 → 1278 B/msg) is entirely full-page images — `fpi=34` -on the noisy run versus `fpi=5` on the settled ones. Gating it would flake -constantly. `wal_records` counts the same work without the FPI noise, which is -why it is the WAL gate. `wal_fpi` is reported alongside the byte count precisely -so a future reader does not misread FPI noise as a regression. +### Gate -Everything is normalized **per message**; that is what makes the counters -machine-independent and therefore gateable in CI at all. +> **The plan's original gate did not survive implementation, and that is the +> most important thing this section records.** The plan proposed gating total +> `pg_stat_statements.calls` on exact equality, on the strength of a prototype +> that measured it at 0.0% spread across 5 idle runs. When the harness ran the +> *real* two-loop subscriber under CPU contention (a loaded CI runner), total +> `calls` swung from 519 to 894 — the subscriber's fetch loop polls on a +> wall-clock timer, so its statement count scales with elapsed time, not with +> message count. The idle prototype had simply never exercised that. Total +> `calls` is therefore **not gateable**, and the gate was re-architected to +> count work **per operation** instead of in aggregate. + +What survives is load-independent because it is per-message structural work, +verified exact both idle and under load: + +| Metric | Gate | Why | +| --- | --- | --- | +| `tup_upd`, `tup_del`, `tup_ins` (rows actually changed) | **exact** | table-scoped; the primary structural invariant | +| `delete_calls`, `insert_calls`, `select_calls` (`calls_by_op`) | **exact** | terminal DELETE / producer INSERT+`pg_notify`, one per message | +| `wal_records` / msg | **±10% band** | ~5% server-wide spread | +| total `calls`, `wal_bytes`, `wal_fpi`, wall-clock / msg-s | **never** | timing- or FPI-noisy | + +`calls_by_op` splits `pg_stat_statements` by the leading SQL keyword. The terminal +DELETE is a plain `DELETE`, but the fetch CTE is `WITH ready AS (…) UPDATE …`, so +it classifies under `with`, not `update` — which is why the gate keys on `delete`, +and why `calls_by_op['update']` is ~0 on the happy path. The tuple counters are +the load-independent bedrock (`delete_calls` counts DELETE *executions*, which a +pathological >lease-TTL stall could inflate via re-lease while `tup_del` stays +exact); `delete_calls` corroborates. + +The WAL-*bytes* swing (up to ~66% run-to-run) is entirely full-page images — high +`wal_fpi` on a post-checkpoint run versus low on a settled one. `wal_records` +counts the same work without the FPI noise, which is why it is the WAL gate. +`wal_fpi` is reported alongside the byte count precisely so a future reader does +not misread FPI noise as a regression. + +The gate compares **raw totals** at a message count pinned to the baseline's, not +per-message ratios — exact-integer comparison, no float-rounding ambiguity. The +per-message ratios exist only for the human table. ### Infrastructure @@ -158,23 +223,31 @@ least reproducible, and reported separately so it is legible. ## Testing -- `just bench --messages 500` runs green and prints both tables. -- `just bench-check` passes against the committed `baseline.json` on unmodified - `main`. -- **The gate actually bites:** artificially perturb the terminal flush (e.g. an - extra no-op statement per message) and confirm `bench-check` fails on the - round-trip count. A gate never observed failing is not a gate. -- Determinism holds: 5 consecutive `just bench` runs report identical Tier-1 - counters (verified in prototype: `calls = 522`, spread 0.0%). +- `just bench` runs green and prints both tables; `just bench-check` passes + against the committed `baseline.json`. +- **The gate actually bites** (verified): perturbing the terminal flush with an + extra `SELECT 1` per message made `bench-check` exit non-zero on all six + consumer points with `select_calls changed (exact-gated): baseline 0 → current + 5000`. Reverted, it returned to green. A gate never observed failing is not a + gate. +- Determinism of the *gated* metrics holds **under load**: the fixed gate + (`delete_calls` + tuple counters) is identical idle and under CPU contention, + where the original total-`calls` gate was not (519 vs 894). +- The full suite still passes at `--cov-fail-under=100` with a test importing + `benchmarks.*` — the `[tool.coverage.run] omit` keeps the package out of the + gate. ## Risk **The gate is only as strong as its determinism.** This was the design's central -risk and it has been retired empirically: `calls` was identical across 5 -real-broker runs (spread 0.0%). Residual exposure is that a *future* change -introduces pool churn or reconnect logic that makes it nondeterministic — which -is arguably the gate working, not failing, but it will present as a flake and -should be diagnosed as such before the tolerance is loosened. +risk and it **materialized** rather than being retired: the plan's exact-`calls` +gate was falsified when the real subscriber ran under load (519 idle vs 894 +loaded). It was addressed by re-architecting the gate onto per-operation and +tuple counts, which are per-message structural work and were verified identical +idle and under CPU contention. Residual exposure is that a *future* change adds a +new per-message statement on a gated path (caught — that is the gate working) or +makes a currently-gated count timing-dependent (would present as a flake and must +be diagnosed, not tolerance-widened, before the baseline is trusted again). **Docker/macOS wall-clock is indicative, not authoritative.** Mitigated by never gating on it and by saying so in the report output, so the numbers are not quoted From 4d66e77a4fa14f65ecbbe19a785f702ad0ff6a0b Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 10:09:27 +0300 Subject: [PATCH 12/13] ci(bench): generate uv.lock before the compose build, add timeout The bench job is the only CI job that builds the docker compose application image, whose Dockerfile does COPY uv.lock + uv sync --frozen. uv.lock is git-ignored, so a clean CI checkout lacks it and docker build fails on COPY uv.lock. Generate the lock in the job first. Verified by building from a clean git-archive checkout with and without the step. Also cap the job at 20 minutes so a pathological drain hang fails fast. --- .github/workflows/_checks.yml | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/.github/workflows/_checks.yml b/.github/workflows/_checks.yml index 7649ec7..536972f 100644 --- a/.github/workflows/_checks.yml +++ b/.github/workflows/_checks.yml @@ -27,9 +27,16 @@ jobs: bench: runs-on: ubuntu-latest + timeout-minutes: 20 steps: - uses: actions/checkout@v6 - uses: extractions/setup-just@v4 + - uses: astral-sh/setup-uv@v8.2.0 + # uv.lock is git-ignored (repo convention), so a clean checkout lacks it. + # `just bench-check` builds the compose `application` image, whose Dockerfile + # does `COPY uv.lock` + `uv sync --frozen`; regenerate the lock first so that + # build has it. (The pytest job runs `uv sync` directly and needs no compose.) + - run: uv lock - run: just bench-check pytest: From e0255d88cbf1e32d8d00a9551b5b055c6c471b4b Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Wed, 15 Jul 2026 12:07:00 +0300 Subject: [PATCH 13/13] chore(bench): empty benchmarks/__init__.py --- benchmarks/__init__.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/benchmarks/__init__.py b/benchmarks/__init__.py index 0598077..e69de29 100644 --- a/benchmarks/__init__.py +++ b/benchmarks/__init__.py @@ -1,5 +0,0 @@ -"""Benchmark harness for faststream-outbox. - -Dev tooling, not shipped. Run via `just bench` / `just bench-check`; never -collected by pytest. See planning/changes/2026-07-14.01-benchmark-harness.md. -"""