Benchmark harness for the outbox transport

.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:

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

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())

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
+    }
+  }
+}

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"