diff --git a/.github/workflows/_checks.yml b/.github/workflows/_checks.yml index 12a3f71..536972f 100644 --- a/.github/workflows/_checks.yml +++ b/.github/workflows/_checks.yml @@ -25,6 +25,20 @@ jobs: - uses: astral-sh/setup-uv@v8.2.0 - run: just docs-build + 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: runs-on: ubuntu-latest strategy: 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/__init__.py b/benchmarks/__init__.py new file mode 100644 index 0000000..e69de29 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 + } + } +} 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/benchmarks/probes.py b/benchmarks/probes.py new file mode 100644 index 0000000..9b4cb9d --- /dev/null +++ b/benchmarks/probes.py @@ -0,0 +1,297 @@ +"""Postgres catalog probes: snapshot the server's counters around a workload. + +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`` 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 contextlib +import dataclasses +from collections.abc import AsyncIterator + +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncConnection, AsyncEngine + +from benchmarks.config import APPLICATION_NAME + + +# 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. 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, + 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. schemaname is pinned: pg_stat_user_tables covers +# every schema, so a bare relname match could return two rows. +_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 AND t.schemaname = :schema + """, +) + +# 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() " + "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") + +# 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 + 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 + + +@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 _autocommit(engine) 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 _autocommit(engine) 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) + + +@contextlib.asynccontextmanager +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. ``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 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] = [] + + # 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 + + 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: + row = ( + await conn.execute( + _SNAPSHOT_SQL, + {"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), + # 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), + ), + ) 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/benchmarks/workload.py b/benchmarks/workload.py new file mode 100644 index 0000000..bb105be --- /dev/null +++ b/benchmarks/workload.py @@ -0,0 +1,186 @@ +"""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, _autocommit, 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, + 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) + # 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 + + +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 _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: + async with engine.begin() as conn: + await conn.execute( + insert(table), + [ + {"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) + + # 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() + # 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: + seen_ids.add(msg) + if len(seen_ids) >= 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 + + # 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) + + +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, logger=None) + 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) 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/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..87a91e7 --- /dev/null +++ b/planning/changes/2026-07-14.01-benchmark-harness.md @@ -0,0 +1,258 @@ +--- +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 + +## 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 (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. + +## Design + +### Layout + +``` +benchmarks/ + __init__.py + 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 +``` + +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 + +`probe()` is an async context manager yielding a `ProbeResult` of catalog deltas: + +- **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`. +- **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: + +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 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}`. + + 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. + +### Gate + +> **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 + +`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` 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 **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 +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. 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:"] 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