bench: sample table health over an extend run and compare autovacuum variants (#258)

The extend sweep ran each point for 60 s, so it saw one autovacuum cycle and could not tell whether table size, index size or fetch time settle. A sampler now records them every --sample-interval seconds, and --vacuum runs each point untuned, with outbox_autovacuum_ddl defaults, or with the cost throttle off.

Author: lesnik512

benchmarks/__main__.py

@@ -11,8 +11,9 @@
 import os
 import pathlib
 import sys
+import typing
 
-from benchmarks.config import DEFAULT_DSN, ExtendConfig, RunConfig
+from benchmarks.config import DEFAULT_DSN, ExtendConfig, RunConfig, VacuumVariant
 from benchmarks.extend import ExtendResult, format_extend, run_extend, to_json
 from benchmarks.report import compare, format_markdown, format_table, to_baseline
 from benchmarks.workload import RunResult, make_engine, run_consumer, run_producer
@@ -126,17 +127,35 @@ async def _run_extend_sweep(dsn: str, points: list[ExtendConfig]) -> list[Extend
     results = []
     for cfg in points:
         result = await run_extend(dsn, cfg)
-        sys.stdout.write(f"{cfg.name}: {result.committed} pings in {result.wall_seconds:.1f}s\n")
+        sys.stdout.write(f"{cfg.name}/{cfg.vacuum}: {result.committed} pings in {result.wall_seconds:.1f}s\n")
         sys.stdout.flush()
         results.append(result)
     return results
 
 
-def _extend(dsn: str, *, seconds: float | None, only: list[str] | None, json_path: str | None) -> int:
-    """Run the extend sweep and print its tables; nothing is gated."""
-    points = [p for p in _EXTEND_POINTS if not only or p.name in only]
+def _extend_points(
+    *,
+    only: list[str] | None,
+    seconds: float | None,
+    vacuum: list[VacuumVariant] | None,
+    sample_interval: float | None,
+) -> list[ExtendConfig]:
+    """Return the selected points, each crossed with every requested vacuum variant."""
+    overrides: dict[str, float] = {}
     if seconds is not None:
-        points = [dataclasses.replace(p, seconds=seconds) for p in points]
+        overrides["seconds"] = seconds
+    if sample_interval is not None:
+        overrides["sample_interval"] = sample_interval
+    return [
+        dataclasses.replace(p, **overrides, vacuum=variant)
+        for p in _EXTEND_POINTS
+        if not only or p.name in only
+        for variant in (vacuum or [p.vacuum])
+    ]
+
+
+def _extend(dsn: str, *, points: list[ExtendConfig], json_path: str | None) -> int:
+    """Run the extend sweep and print its tables; nothing is gated."""
     results = asyncio.run(_run_extend_sweep(dsn, points))
     sys.stdout.write("\n" + format_extend(results) + "\n")
     if json_path is not None:
@@ -168,6 +187,8 @@ def _build_parser() -> argparse.ArgumentParser:
     extend.add_argument("--seconds", type=float)
     extend.add_argument("--only", nargs="+", choices=[p.name for p in _EXTEND_POINTS])
     extend.add_argument("--json")
+    extend.add_argument("--vacuum", nargs="+", choices=typing.get_args(VacuumVariant))
+    extend.add_argument("--sample-interval", type=float)
     return parser
 
 
@@ -179,7 +200,10 @@ def main(argv: list[str] | None = None) -> int:
     if args.command == "check":
         return _check(dsn, markdown=args.markdown)
     if args.command == "extend":
-        return _extend(dsn, seconds=args.seconds, only=args.only, json_path=args.json)
+        points = _extend_points(
+            only=args.only, seconds=args.seconds, vacuum=args.vacuum, sample_interval=args.sample_interval
+        )
+        return _extend(dsn, points=points, json_path=args.json)
     return _run(dsn, args.messages, args.repeats, write_baseline=args.write_baseline)
 
 

benchmarks/config.py

@@ -1,6 +1,7 @@
 """Workload configuration for a single benchmark run."""
 
 import dataclasses
+import typing
 
 
 # Matches the payload size in Tsvettsikh's «50 оттенков Transactional Outbox»
@@ -10,6 +11,10 @@
 # The DSN the harness runs against. Overridden by POSTGRES_DSN in compose.
 DEFAULT_DSN = "postgresql+asyncpg://outbox:outbox@localhost:5432/outbox"
 
+# How an extend point's table is vacuumed: no reloptions, outbox_autovacuum_ddl's defaults,
+# or those defaults with autovacuum's cost throttle turned off.
+VacuumVariant = typing.Literal["untuned", "tuned", "unthrottled"]
+
 # Tags the harness' own connections so the probe can assert it owns the database.
 APPLICATION_NAME = "faststream-outbox-bench"
 
@@ -49,8 +54,17 @@ class ExtendConfig:
     pool_size: int = 20
     max_fetch_interval: float = 1.0
     queue: str = "timeouts"
+    vacuum: VacuumVariant = "tuned"
+    # Seconds between table-health samples taken while the load runs.
+    sample_interval: float = 10.0
 
     def __post_init__(self) -> None:
+        if self.vacuum not in typing.get_args(VacuumVariant):
+            msg = f"{self.name}: vacuum must be one of {typing.get_args(VacuumVariant)}, got {self.vacuum!r}"
+            raise ValueError(msg)
+        if self.sample_interval <= 0:
+            msg = f"{self.name}: sample_interval must be positive, got {self.sample_interval}"
+            raise ValueError(msg)
         if self.ping_interval >= self.timeout_seconds:
             msg = f"{self.name}: each identity pings every {self.ping_interval:.2f}s, not before its timeout"
             raise ValueError(msg)

benchmarks/extend.py

@@ -10,9 +10,12 @@
 task on a pool of ``pool_size`` connections, so pool waits show up once Postgres falls behind. A point
 stops offering load once the backlog passes five seconds of pings and is reported as saturated.
 
-The table carries the recommended autovacuum settings, unlike the gated sweep's tables, because dead
-tuples and their cleanup are part of what this measures. Statement, tuple and WAL counters come from
-the same probe as the gated sweep. Nothing here is gated: every number depends on timing.
+The table's autovacuum settings follow the point's ``vacuum`` variant, ``outbox_autovacuum_ddl``'s
+defaults unless it says otherwise, because dead tuples and their cleanup are part of what this measures.
+Every ``sample_interval`` seconds a separate connection records table and index size, dead tuples,
+autovacuum runs and the fetch statement's cumulative counters, so a long run shows whether they settle.
+Statement, tuple and WAL counters come from the same probe as the gated sweep. Nothing here is gated:
+every number depends on timing.
 """
 
 import asyncio
@@ -50,6 +53,24 @@
     """,
 )
 
+# Mentions pg_stat, so the probe's NOT ILIKE '%pg_stat%' filter drops it from the statement counters.
+_SAMPLE_SQL = text(
+    """
+    SELECT pg_relation_size(t.relid) AS table_bytes,
+           pg_indexes_size(t.relid) AS index_bytes,
+           t.n_dead_tup, t.n_live_tup, t.autovacuum_count,
+           coalesce(f.calls, 0) AS fetch_calls,
+           coalesce(f.total_ms, 0) AS fetch_ms
+    FROM pg_stat_user_tables t,
+         LATERAL (
+           SELECT sum(s.calls) AS calls, sum(s.total_exec_time) AS total_ms
+           FROM pg_stat_statements s
+           WHERE s.query NOT ILIKE '%pg_stat%' AND lower(split_part(btrim(s.query), ' ', 1)) = 'with'
+         ) f
+    WHERE t.relname = :table AND t.schemaname = current_schema()
+    """,
+)
+
 _VACUUM_SQL = text(
     """
     SELECT t.autovacuum_count, t.n_live_tup
@@ -84,6 +105,20 @@ class _Tally:
     silent_fired: asyncio.Event = dataclasses.field(default_factory=asyncio.Event)
 
 
+@dataclasses.dataclass(frozen=True)
+class Sample:
+    """Table health *elapsed* seconds after sampling began; the fetch counters are cumulative."""
+
+    elapsed: float
+    table_bytes: int
+    index_bytes: int
+    dead_tuples: int
+    live_tuples: int
+    autovacuum_count: int
+    fetch_calls: int
+    fetch_ms: float
+
+
 @dataclasses.dataclass(frozen=True)
 class ExtendResult:
     """One extend point: client-side timings, expiry outcomes and the probe's catalog deltas."""
@@ -106,6 +141,7 @@ class ExtendResult:
     autovacuum_count: int
     live_tuples: int
     probe: ProbeResult
+    samples: tuple[Sample, ...] = ()
 
 
 def percentiles(values: typing.Sequence[float]) -> dict[str, float]:
@@ -120,6 +156,27 @@ def _rank(q: float) -> float:
     return {"p50": _rank(0.50), "p95": _rank(0.95), "p99": _rank(0.99), "max": ordered[-1]}
 
 
+def autovacuum_ddl(cfg: ExtendConfig, table_name: str) -> str | None:
+    """Return the ``ALTER TABLE`` that sets up *cfg*'s vacuum variant, or ``None`` to leave the table untuned."""
+    if cfg.vacuum == "untuned":
+        return None
+    return outbox_autovacuum_ddl(table_name, vacuum_cost_delay=0 if cfg.vacuum == "unthrottled" else None)
+
+
+def _sample_at(samples: typing.Sequence[Sample], elapsed: float) -> Sample:
+    """Return the last sample taken at or before *elapsed*, or the first one."""
+    return next((s for s in reversed(samples) if s.elapsed <= elapsed), samples[0])
+
+
+def fetch_mean_ms(samples: typing.Sequence[Sample], start: float, end: float) -> float:
+    """Mean fetch statement time between the samples at *start* and *end*; 0 when nothing ran."""
+    if not samples:
+        return 0.0
+    first, last = _sample_at(samples, start), _sample_at(samples, end)
+    calls = last.fetch_calls - first.fetch_calls
+    return (last.fetch_ms - first.fetch_ms) / calls if calls else 0.0
+
+
 def _body(identity: int, deadline: dt.datetime) -> dict[str, typing.Any]:
     return {"identity": identity, "deadline": deadline.isoformat()}
 
@@ -129,7 +186,8 @@ async def _create_table(engine: AsyncEngine, cfg: ExtendConfig) -> tuple[MetaDat
     table = make_outbox_table(metadata, table_name=f"bench_extend_{uuid.uuid4().hex[:10]}")
     async with engine.begin() as conn:
         await conn.run_sync(metadata.create_all)
-        await conn.execute(text(outbox_autovacuum_ddl(table.name)))
+        if (ddl := autovacuum_ddl(cfg, table.name)) is not None:
+            await conn.execute(text(ddl))
         start = utcnow()
         rows = []
         for identity in range(cfg.timers):
@@ -227,25 +285,52 @@ def _spawn() -> None:
     return started
 
 
+async def _sample_until(stop: asyncio.Event, engine: AsyncEngine, table: Table, interval: float) -> list[Sample]:
+    """Sample table health every *interval* seconds, plus once more after *stop* is set."""
+    samples: list[Sample] = []
+    started = time.perf_counter()
+    async with _autocommit(engine) as conn:
+        while True:
+            row = (await conn.execute(_SAMPLE_SQL, {"table": table.name})).one()
+            samples.append(
+                Sample(
+                    elapsed=time.perf_counter() - started,
+                    table_bytes=int(row.table_bytes),
+                    index_bytes=int(row.index_bytes),
+                    dead_tuples=int(row.n_dead_tup),
+                    live_tuples=int(row.n_live_tup),
+                    autovacuum_count=int(row.autovacuum_count),
+                    fetch_calls=int(row.fetch_calls),
+                    fetch_ms=float(row.fetch_ms),
+                ),
+            )
+            if stop.is_set():
+                return samples
+            with contextlib.suppress(TimeoutError):
+                await asyncio.wait_for(stop.wait(), timeout=interval)
+
+
 async def run_extend(dsn: str, cfg: ExtendConfig) -> ExtendResult:
     """Run one extend point against a fresh table and return its measurements."""
     engine = make_engine(dsn, pool_size=cfg.pool_size)
     subscriber_engine = make_engine(dsn)
+    sampler_engine = make_engine(dsn, pool_size=1)
     try:
         await ensure_extension(engine)
         metadata, table = await _create_table(engine, cfg)
         try:
-            return await _measure(engine, subscriber_engine, table, cfg)
+            return await _measure(engine, subscriber_engine, sampler_engine, table, cfg)
         finally:
             async with engine.begin() as conn:
                 await conn.run_sync(metadata.drop_all)
     finally:
+        await sampler_engine.dispose()
         await subscriber_engine.dispose()
         await engine.dispose()
 
 
 async def _measure(
-    engine: AsyncEngine, subscriber_engine: AsyncEngine, table: Table, cfg: ExtendConfig
+    engine: AsyncEngine, subscriber_engine: AsyncEngine, sampler_engine: AsyncEngine, table: Table, cfg: ExtendConfig
 ) -> ExtendResult:
     tally = _Tally()
     publisher = OutboxBroker(outbox_table=table, logger=None)
@@ -263,12 +348,16 @@ async def _expire(body: dict[str, typing.Any]) -> None:
 
     await _checkpoint(engine)
     await assert_owns_database(engine)
+    stop_sampling = asyncio.Event()
     async with probe(engine, table.name) as sink:
         await expirer.start()
+        sampler = asyncio.create_task(_sample_until(stop_sampling, sampler_engine, table, cfg.sample_interval))
         try:
             started = await _offer_load(engine=engine, publisher=publisher, cfg=cfg, tally=tally)
             await _wait_for_silent_expiries(cfg, tally, started)
         finally:
+            stop_sampling.set()
+            samples = await sampler
             await expirer.stop()
             await subscriber_engine.dispose()
         wall = (tally.last_commit or time.perf_counter()) - started
@@ -296,6 +385,7 @@ async def _expire(body: dict[str, typing.Any]) -> None:
         autovacuum_count=int(vacuum.autovacuum_count),
         live_tuples=int(vacuum.n_live_tup),
         probe=sink[0],
+        samples=tuple(samples),
     )
 
 
@@ -306,8 +396,27 @@ async def _wait_for_silent_expiries(cfg: ExtendConfig, tally: _Tally, started: f
         await asyncio.wait_for(tally.silent_fired.wait(), timeout=max(0.0, give_up - time.perf_counter()))
 
 
+def _label(cfg: ExtendConfig) -> str:
+    return f"{cfg.name}/{cfg.vacuum}"
+
+
+def _drift_row(r: ExtendResult) -> str:
+    samples = r.samples
+    first, last = samples[0], samples[-1]
+    mb = 2**20
+    table_peak = max(s.table_bytes for s in samples)
+    index_peak = max(s.index_bytes for s in samples)
+    early = fetch_mean_ms(samples, first.elapsed, first.elapsed + 60.0)
+    late = fetch_mean_ms(samples, last.elapsed - 60.0, last.elapsed)
+    return (
+        f"| {_label(r.config)} | {first.table_bytes / mb:.1f} / {last.table_bytes / mb:.1f} / {table_peak / mb:.1f} "
+        f"| {first.index_bytes / mb:.1f} / {last.index_bytes / mb:.1f} / {index_peak / mb:.1f} "
+        f"| {max(s.dead_tuples for s in samples):,} | {last.autovacuum_count} | {early:.3f} / {late:.3f} |"
+    )
+
+
 def format_extend(results: typing.Sequence[ExtendResult]) -> str:
-    """Render the extend points as two markdown tables: the write path, then table and expiry health."""
+    """Render the extend points as markdown tables: the write path, table and expiry health, then drift."""
     write = [
         (
             "| point | offered/s | committed/s | commits/s | ping p50 / p95 / p99 ms | pool wait p99 ms "
@@ -329,7 +438,7 @@ def format_extend(results: typing.Sequence[ExtendResult]) -> str:
         insert_ms = r.statement_ms.get("insert", (0.0, 0.0))
         fetch_ms = r.statement_ms.get("with", (0.0, 0.0))
         errors = ", ".join(f"{k} {v}" for k, v in sorted(r.errors.items())) or "0"
-        label = f"{cfg.name}{' (saturated)' if r.saturated else ''}"
+        label = f"{_label(cfg)}{' (saturated)' if r.saturated else ''}"
         offered_rate = r.offered / max(r.offered_seconds, 1e-9)
         write.append(
             f"| {label} | {offered_rate:,.0f} | {r.committed / span:,.0f} | {r.transactions / span:,.0f} "
@@ -338,12 +447,24 @@ def format_extend(results: typing.Sequence[ExtendResult]) -> str:
         )
         hot = 100.0 * p.tup_hot_upd / p.tup_upd if p.tup_upd else 0.0
         health.append(
-            f"| {cfg.name} | {p.wal_bytes / pings:,.0f} | {p.wal_records / pings:.2f} | {p.wal_fpi:,} | {hot:.0f} "
+            f"| {_label(cfg)} | {p.wal_bytes / pings:,.0f} | {p.wal_records / pings:.2f} | {p.wal_fpi:,} | {hot:.0f} "
             f"| {p.dead_tup:,} | {r.autovacuum_count} | {p.index_bytes / 2**20:.1f} "
             f"| {fetch_ms[0]:.3f} / {fetch_ms[1]:.1f} | {r.expired_silent} / {cfg.silent_timers} "
             f"| {r.false_expiries} | {r.lateness['p50']:.2f} / {r.lateness['max']:.2f} | {r.leased} |",
         )
-    return "\n".join([*write, "", *health])
+    tables = [*write, "", *health]
+    sampled = [r for r in results if r.samples]
+    if sampled:
+        tables += [
+            "",
+            (
+                "| point | table MB start / end / peak | index MB start / end / peak | dead tuples peak "
+                "| autovacuums | fetch mean ms first / last minute |"
+            ),
+            "|---|---|---|---|---|---|",
+            *(_drift_row(r) for r in sampled),
+        ]
+    return "\n".join(tables)
 
 
 def to_json(results: typing.Sequence[ExtendResult]) -> str: