diff --git a/benchmarks/__main__.py b/benchmarks/__main__.py index 9cbb55f..c019ae9 100644 --- a/benchmarks/__main__.py +++ b/benchmarks/__main__.py @@ -1,17 +1,19 @@ -"""CLI: `python -m benchmarks run` / `python -m benchmarks check`. +"""CLI: `python -m benchmarks run` / `check` / `extend`. -Invoked through `just bench` / `just bench-check`, which run it inside docker compose -so the postgres service has pg_stat_statements preloaded. +Invoked through `just bench` / `just bench-check` / `just bench-extend`, which run it inside +docker compose so the postgres service has pg_stat_statements preloaded. """ import argparse import asyncio +import dataclasses import json import os import pathlib import sys -from benchmarks.config import DEFAULT_DSN, RunConfig +from benchmarks.config import DEFAULT_DSN, ExtendConfig, RunConfig +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 @@ -23,6 +25,19 @@ _WORKER_COUNTS = (1, 2, 4) _BATCH_SIZES = (10, 100) +# The extend sweep: commit per ping against bounded batches at rising rates, with each identity +# pinged every 2 s, plus two points where 60 identities share the load so pings to one timer collide. +_EXTEND_POINTS = ( + ExtendConfig(name="r1000-b1", rate=1_000, timers=2_000), + ExtendConfig(name="r3000-b1", rate=3_000, timers=6_000), + ExtendConfig(name="r3000-b10", rate=3_000, timers=6_000, pings_per_transaction=10), + ExtendConfig(name="r5000-b1", rate=5_000, timers=10_000), + ExtendConfig(name="r10000-b1", rate=10_000, timers=20_000), + ExtendConfig(name="r10000-b50", rate=10_000, timers=20_000, pings_per_transaction=50), + ExtendConfig(name="hot-r3000-b1", rate=3_000, timers=60), + ExtendConfig(name="hot-r3000-b10", rate=3_000, timers=60, pings_per_transaction=10), +) + def _median_by_wall(runs: list[RunResult]) -> RunResult: """Return the run with the median wall-clock. @@ -107,6 +122,28 @@ def _run(dsn: str, messages: int, repeats: int, *, write_baseline: bool) -> int: return 0 +async def _run_extend_sweep(dsn: str, points: list[ExtendConfig]) -> list[ExtendResult]: + 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.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] + if seconds is not None: + points = [dataclasses.replace(p, seconds=seconds) for p in points] + results = asyncio.run(_run_extend_sweep(dsn, points)) + sys.stdout.write("\n" + format_extend(results) + "\n") + if json_path is not None: + pathlib.Path(json_path).write_text(to_json(results) + "\n") + return 0 + + def _build_parser() -> argparse.ArgumentParser: """Two subcommands with their own flags, so a misplaced flag errors at parse. @@ -126,6 +163,11 @@ def _build_parser() -> argparse.ArgumentParser: # only burns CI time -- it takes no --repeats/--messages (pinned to the baseline). check = subparsers.add_parser("check", help="gate the counters against baseline.json") check.add_argument("--markdown", action="store_true") + + extend = subparsers.add_parser("extend", help="run the timer extend sweep and print the tables") + extend.add_argument("--seconds", type=float) + extend.add_argument("--only", nargs="+", choices=[p.name for p in _EXTEND_POINTS]) + extend.add_argument("--json") return parser @@ -136,6 +178,8 @@ 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) return _run(dsn, args.messages, args.repeats, write_baseline=args.write_baseline) diff --git a/benchmarks/config.py b/benchmarks/config.py index 0f4afc0..9fa7bf8 100644 --- a/benchmarks/config.py +++ b/benchmarks/config.py @@ -30,3 +30,42 @@ class RunConfig: fetch_interval: tuple[float, float] = (0.001, 0.001) # Producer publishes per transaction; None puts every publish in one transaction. publishes_per_transaction: int | None = None + + +@dataclasses.dataclass(frozen=True) +class ExtendConfig: + """One point of the timer extend sweep: heartbeats at *rate* pings/s over *timers* identities.""" + + name: str + rate: int + timers: int + timeout_seconds: float = 6.0 + pings_per_transaction: int = 1 + # Longest a ping waits for its transaction to fill before it is flushed anyway. + flush_interval: float = 0.01 + seconds: float = 30.0 + # This share of identities stops pinging at silence_at, so their timers must fire. + silent_fraction: float = 0.05 + pool_size: int = 20 + max_fetch_interval: float = 1.0 + queue: str = "timeouts" + + def __post_init__(self) -> None: + 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) + if self.silence_at + self.timeout_seconds + self.max_fetch_interval + 2.0 > self.seconds: + msg = f"{self.name}: {self.seconds}s is too short for the silent timers to fire inside the run" + raise ValueError(msg) + + @property + def ping_interval(self) -> float: + return self.timers / self.rate + + @property + def silence_at(self) -> float: + return self.seconds / 4 + + @property + def silent_timers(self) -> int: + return int(self.timers * self.silent_fraction) diff --git a/benchmarks/extend.py b/benchmarks/extend.py new file mode 100644 index 0000000..70cf005 --- /dev/null +++ b/benchmarks/extend.py @@ -0,0 +1,351 @@ +"""The timer extend workload: heartbeats extend per-identity timers while a subscriber fires the silent ones. + +Pings arrive open loop at ``rate`` per second, round-robin over ``timers`` identities, so each identity is +pinged once per ``timers / rate`` seconds and its deadline is pushed to ``ping time + timeout``. At +``silence_at`` a ``silent_fraction`` of identities stop pinging; their timers must fire about ``timeout`` +later. Any other identity firing is a false expiry: the pings fell behind their deadlines. + +Pings are grouped into transactions of up to ``pings_per_transaction``, sorted by identity so two +transactions extending the same timers lock them in the same order. Each transaction runs in its own +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. +""" + +import asyncio +import contextlib +import dataclasses +import datetime as dt +import json +import time +import typing +import uuid + +from sqlalchemy import MetaData, Table, insert, text +from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession + +from benchmarks.config import ExtendConfig +from benchmarks.probes import ProbeResult, _autocommit, assert_owns_database, ensure_extension, probe +from benchmarks.workload import _checkpoint, make_engine +from faststream_outbox import OutboxBroker, make_outbox_table, outbox_autovacuum_ddl +from faststream_outbox._time import utcnow +from faststream_outbox.envelope import _encode_payload + + +_BACKLOG_SECONDS = 5.0 +_TICK_SECONDS = 0.001 + +_STATEMENT_TIMES_SQL = text( + """ + SELECT lower(split_part(btrim(s.query), ' ', 1)) AS op, + sum(s.calls) AS calls, + sum(s.total_exec_time) / nullif(sum(s.calls), 0) AS mean_ms, + max(s.max_exec_time) AS max_ms + FROM pg_stat_statements s + WHERE s.query NOT ILIKE '%pg_stat%' + GROUP BY 1 + """, +) + +_VACUUM_SQL = text( + """ + SELECT t.autovacuum_count, t.n_live_tup + FROM pg_stat_user_tables t + WHERE t.relname = :table AND t.schemaname = current_schema() + """, +) + + +@dataclasses.dataclass +class _Ping: + identity: int + arrived: float + deadline: dt.datetime + + +@dataclasses.dataclass +class _Tally: + latencies: list[float] = dataclasses.field(default_factory=list) + pool_waits: list[float] = dataclasses.field(default_factory=list) + lateness: list[float] = dataclasses.field(default_factory=list) + offered: int = 0 + committed: int = 0 + transactions: int = 0 + leased: int = 0 + errors: dict[str, int] = dataclasses.field(default_factory=dict) + expired_silent: int = 0 + false_expiries: int = 0 + saturated: bool = False + offered_seconds: float = 0.0 + last_commit: float = 0.0 + silent_fired: asyncio.Event = dataclasses.field(default_factory=asyncio.Event) + + +@dataclasses.dataclass(frozen=True) +class ExtendResult: + """One extend point: client-side timings, expiry outcomes and the probe's catalog deltas.""" + + config: ExtendConfig + wall_seconds: float + offered: int + offered_seconds: float + committed: int + transactions: int + leased: int + errors: dict[str, int] + saturated: bool + latency: dict[str, float] + pool_wait: dict[str, float] + lateness: dict[str, float] + expired_silent: int + false_expiries: int + statement_ms: dict[str, tuple[float, float]] + autovacuum_count: int + live_tuples: int + probe: ProbeResult + + +def percentiles(values: typing.Sequence[float]) -> dict[str, float]: + """p50, p95, p99 and max of *values* (nearest rank); zeros for an empty sequence.""" + if not values: + return {"p50": 0.0, "p95": 0.0, "p99": 0.0, "max": 0.0} + ordered = sorted(values) + + def _rank(q: float) -> float: + return ordered[min(len(ordered) - 1, max(0, round(q * len(ordered)) - 1))] + + return {"p50": _rank(0.50), "p95": _rank(0.95), "p99": _rank(0.99), "max": ordered[-1]} + + +def _body(identity: int, deadline: dt.datetime) -> dict[str, typing.Any]: + return {"identity": identity, "deadline": deadline.isoformat()} + + +async def _create_table(engine: AsyncEngine, cfg: ExtendConfig) -> tuple[MetaData, Table]: + metadata = MetaData() + 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))) + start = utcnow() + rows = [] + for identity in range(cfg.timers): + deadline = start + dt.timedelta(seconds=cfg.timeout_seconds + identity / cfg.rate) + payload, headers = _encode_payload(_body(identity, deadline)) + rows.append( + { + "queue": cfg.queue, + "payload": payload, + "headers": headers, + "timer_id": f"t{identity}", + "next_attempt_at": deadline, + }, + ) + await conn.execute(insert(table), rows) + return metadata, table + + +async def _flush( + batch: list[_Ping], + *, + engine: AsyncEngine, + publisher: OutboxBroker, + cfg: ExtendConfig, + tally: _Tally, +) -> None: + started = time.perf_counter() + try: + async with AsyncSession(engine) as session, session.begin(): + await session.connection() + tally.pool_waits.append(time.perf_counter() - started) + for ping in sorted(batch, key=lambda p: p.identity): + rid = await publisher.publish( + _body(ping.identity, ping.deadline), + queue=cfg.queue, + session=session, + activate_at=ping.deadline, + timer_id=f"t{ping.identity}", + timer_conflict="extend", + ) + if rid is None: + tally.leased += 1 + except Exception as exc: # noqa: BLE001 # a failed transaction is a measurement, not a harness bug + name = type(exc).__name__ + tally.errors[name] = tally.errors.get(name, 0) + len(batch) + return + done = time.perf_counter() + tally.latencies.extend(done - ping.arrived for ping in batch) + tally.committed += len(batch) + tally.transactions += 1 + tally.last_commit = done + + +async def _offer_load( + *, + engine: AsyncEngine, + publisher: OutboxBroker, + cfg: ExtendConfig, + tally: _Tally, +) -> float: + """Generate pings for ``cfg.seconds`` and wait for every transaction; return when the load started.""" + tasks: set[asyncio.Task[None]] = set() + buffer: list[_Ping] = [] + sequence = 0 + backlog_limit = cfg.rate * _BACKLOG_SECONDS + + def _spawn() -> None: + task = asyncio.create_task(_flush(list(buffer), engine=engine, publisher=publisher, cfg=cfg, tally=tally)) + tasks.add(task) + task.add_done_callback(tasks.discard) + buffer.clear() + + started = time.perf_counter() + while (elapsed := time.perf_counter() - started) < cfg.seconds: + if tally.offered - tally.committed - sum(tally.errors.values()) > backlog_limit: + tally.saturated = True + break + while sequence < int(elapsed * cfg.rate): + identity = sequence % cfg.timers + sequence += 1 + if identity < cfg.silent_timers and elapsed >= cfg.silence_at: + continue + now = time.perf_counter() + buffer.append(_Ping(identity, now, utcnow() + dt.timedelta(seconds=cfg.timeout_seconds))) + tally.offered += 1 + if len(buffer) >= cfg.pings_per_transaction: + _spawn() + if buffer and time.perf_counter() - buffer[0].arrived >= cfg.flush_interval: + _spawn() + await asyncio.sleep(_TICK_SECONDS) + tally.offered_seconds = time.perf_counter() - started + if buffer: + _spawn() + await asyncio.gather(*tasks) + return started + + +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) + try: + await ensure_extension(engine) + metadata, table = await _create_table(engine, cfg) + try: + return await _measure(engine, subscriber_engine, table, cfg) + finally: + async with engine.begin() as conn: + await conn.run_sync(metadata.drop_all) + finally: + await subscriber_engine.dispose() + await engine.dispose() + + +async def _measure( + engine: AsyncEngine, subscriber_engine: AsyncEngine, table: Table, cfg: ExtendConfig +) -> ExtendResult: + tally = _Tally() + publisher = OutboxBroker(outbox_table=table, logger=None) + expirer = OutboxBroker(subscriber_engine, outbox_table=table, logger=None) + + @expirer.subscriber(cfg.queue, max_fetch_interval=cfg.max_fetch_interval) + async def _expire(body: dict[str, typing.Any]) -> None: + tally.lateness.append((utcnow() - dt.datetime.fromisoformat(body["deadline"])).total_seconds()) + if body["identity"] < cfg.silent_timers: + tally.expired_silent += 1 + if tally.expired_silent >= cfg.silent_timers: + tally.silent_fired.set() + else: + tally.false_expiries += 1 + + await _checkpoint(engine) + await assert_owns_database(engine) + async with probe(engine, table.name) as sink: + await expirer.start() + try: + started = await _offer_load(engine=engine, publisher=publisher, cfg=cfg, tally=tally) + await _wait_for_silent_expiries(cfg, tally, started) + finally: + await expirer.stop() + await subscriber_engine.dispose() + wall = (tally.last_commit or time.perf_counter()) - started + + async with _autocommit(engine) as conn: + times = (await conn.execute(_STATEMENT_TIMES_SQL)).all() + vacuum = (await conn.execute(_VACUUM_SQL, {"table": table.name})).one() + + return ExtendResult( + config=cfg, + wall_seconds=wall, + offered=tally.offered, + offered_seconds=tally.offered_seconds, + committed=tally.committed, + transactions=tally.transactions, + leased=tally.leased, + errors=dict(tally.errors), + saturated=tally.saturated, + latency=percentiles(tally.latencies), + pool_wait=percentiles(tally.pool_waits), + lateness=percentiles(tally.lateness), + expired_silent=tally.expired_silent, + false_expiries=tally.false_expiries, + statement_ms={row.op: (float(row.mean_ms or 0.0), float(row.max_ms or 0.0)) for row in times}, + autovacuum_count=int(vacuum.autovacuum_count), + live_tuples=int(vacuum.n_live_tup), + probe=sink[0], + ) + + +async def _wait_for_silent_expiries(cfg: ExtendConfig, tally: _Tally, started: float) -> None: + """Wait until every silent timer fired, or until it is clearly late.""" + give_up = started + cfg.seconds + cfg.max_fetch_interval + 5.0 + with contextlib.suppress(TimeoutError): + await asyncio.wait_for(tally.silent_fired.wait(), timeout=max(0.0, give_up - time.perf_counter())) + + +def format_extend(results: typing.Sequence[ExtendResult]) -> str: + """Render the extend points as two markdown tables: the write path, then table and expiry health.""" + write = [ + ( + "| point | offered/s | committed/s | commits/s | ping p50 / p95 / p99 ms | pool wait p99 ms " + "| extend stmt mean / max ms | errors |" + ), + "|---|---|---|---|---|---|---|---|", + ] + health = [ + ( + "| point | WAL B/ping | WAL rec/ping | FPI | HOT % | dead tuples | autovacuums | index MB " + "| fetch mean / max ms | silent fired | false expiries | lateness p50 / max s | leased |" + ), + "|---|---|---|---|---|---|---|---|---|---|---|---|---|", + ] + for r in results: + cfg, p = r.config, r.probe + span = max(r.wall_seconds, 1e-9) + pings = max(r.committed, 1) + 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 ''}" + 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} " + f"| {r.latency['p50'] * 1e3:.1f} / {r.latency['p95'] * 1e3:.1f} / {r.latency['p99'] * 1e3:.1f} " + f"| {r.pool_wait['p99'] * 1e3:.1f} | {insert_ms[0]:.3f} / {insert_ms[1]:.1f} | {errors} |", + ) + 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"| {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]) + + +def to_json(results: typing.Sequence[ExtendResult]) -> str: + """Every field of every point, for anyone who wants more than the tables show.""" + return json.dumps([dataclasses.asdict(r) for r in results], indent=2, default=str) diff --git a/benchmarks/workload.py b/benchmarks/workload.py index 8fe4a0d..5cd052b 100644 --- a/benchmarks/workload.py +++ b/benchmarks/workload.py @@ -48,11 +48,13 @@ class RunResult: probe: ProbeResult -def make_engine(dsn: str) -> AsyncEngine: +def make_engine(dsn: str, *, pool_size: int | None = None) -> AsyncEngine: """Engine tagged with APPLICATION_NAME so the ownership check can recognize us.""" + pool = {} if pool_size is None else {"pool_size": pool_size, "max_overflow": 0} return create_async_engine( dsn, connect_args={"server_settings": {"application_name": APPLICATION_NAME}}, + **pool, ) diff --git a/docs/concepts/performance.md b/docs/concepts/performance.md index aad1856..4b55721 100644 --- a/docs/concepts/performance.md +++ b/docs/concepts/performance.md @@ -89,6 +89,41 @@ settings are applied. Full guidance, the eligibility-vs-throughput split, and th I/O caution on `vacuum_cost_delay=0` are in [Autovacuum tuning](../operations/alembic.md#autovacuum-tuning-recommended). +## Timer extends under heartbeat load + +Each `publish(..., timer_conflict="extend")` is one statement, and Postgres +runs it in about 0.02 ms. The cost is what comes with it: + +- a commit, if every ping has its own transaction; +- about 550 to 650 bytes and 5 to 7 records of WAL; +- one dead tuple. These are never HOT updates, because `next_attempt_at` is + indexed. + +So the ceiling is set by commits and round trips, not by the statement. + +One run of `just bench-extend`, on an Apple M2 with OrbStack, Postgres 17 at +default settings, one Python client process and a 20-connection pool. Each +identity pings every 2 seconds and has a 6-second timeout: + +| Offered pings/s | Pings per transaction | Kept up | Ping p99 | Expiry lateness max | +|---|---|---|---|---| +| 1,000 | 1 | yes | 12 ms | 0.2 s | +| 3,000 | 1 | yes | 193 ms | 0.7 s | +| 3,000 | 10 | yes | 14 ms | 1.2 s | +| 5,000 | 1 | no | 17 s | 12.8 s | +| 10,000 | 50 | yes | 618 ms | 3.2 s | + +Read this as a shape, not as capacity numbers for your setup. Once pings fall +behind, live timers start to fire: thousands of false expiries in the +saturated runs. Plan for a wide margin. Grouping pings into short transactions +(sorted by `timer_id`, so concurrent transactions lock rows in the same order) +raises the ceiling a lot, at the cost of latency. + +Dead tuples pile up at the ping rate, about 150,000 in a minute at 10,000 +pings/s, and autovacuum only runs once per `autovacuum_naptime`, which +defaults to 60 seconds. The claim query slowed from about 0.2 to 1 ms over +such a minute. + ## Measuring it yourself The gated numbers above come from the repository's benchmark harness. Contributors @@ -98,6 +133,8 @@ can run it against a local Postgres: DB counters (split by leading SQL keyword). - `just bench-check` gates the deterministic counts against `benchmarks/baseline.json`; this is what CI runs. +- `just bench-extend` runs the timer extend sweep above and prints its tables. + Nothing in it is gated: every number depends on timing. The statement-count reductions (`select_calls`, `delete_calls`, `insert_calls`, …) are machine-independent and exact. Wall-clock throughput is not: it scales diff --git a/justfile b/justfile index c2d52f7..a8c09b2 100644 --- a/justfile +++ b/justfile @@ -64,6 +64,10 @@ bench-check: down && down bench-report: down && down docker compose run -T application uv run python -m benchmarks check --markdown +# Heartbeat load on timer extends: latency, WAL, dead tuples, expiry lateness. Not gated. +bench-extend *args: down && down + docker compose run application uv run python -m benchmarks extend {{ args }} + # 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/tests/test_benchmarks.py b/tests/test_benchmarks.py index 17a587d..1125e52 100644 --- a/tests/test_benchmarks.py +++ b/tests/test_benchmarks.py @@ -10,7 +10,8 @@ import pytest from benchmarks.__main__ import _build_parser -from benchmarks.config import RunConfig +from benchmarks.config import ExtendConfig, RunConfig +from benchmarks.extend import ExtendResult, format_extend, percentiles from benchmarks.probes import ProbeResult from benchmarks.report import EXACT_KEYS, compare, format_markdown, format_table, normalize, to_baseline from benchmarks.workload import RunResult @@ -334,3 +335,59 @@ def test_baseline_keys_name_the_interval_and_transaction_variants() -> None: "consumer/w1/b100/interval1-10", "producer/w1/b100/ppt1", } + + +def test_percentiles_use_nearest_rank() -> None: + assert percentiles([float(n) for n in range(1, 101)]) == {"p50": 50.0, "p95": 95.0, "p99": 99.0, "max": 100.0} + + +def test_percentiles_of_nothing_are_zero() -> None: + assert percentiles([]) == {"p50": 0.0, "p95": 0.0, "p99": 0.0, "max": 0.0} + + +def test_extend_config_derives_the_ping_interval_and_silent_share() -> None: + cfg = ExtendConfig(name="p", rate=1_000, timers=2_000) + assert (cfg.ping_interval, cfg.silent_timers, cfg.silence_at) == (2.0, 100, 7.5) + + +def test_extend_config_rejects_pings_slower_than_the_timeout() -> None: + with pytest.raises(ValueError, match=r"pings every 6\.00s"): + ExtendConfig(name="p", rate=1_000, timers=6_000) + + +def test_extend_config_rejects_a_run_too_short_for_silent_timers_to_fire() -> None: + with pytest.raises(ValueError, match="too short"): + ExtendConfig(name="p", rate=1_000, timers=2_000, seconds=10.0) + + +def test_format_extend_reports_saturation_and_per_ping_wal() -> None: + result = ExtendResult( + config=ExtendConfig(name="r1000-b1", rate=1_000, timers=2_000), + wall_seconds=30.0, + offered=30_000, + offered_seconds=30.0, + committed=30_000, + transactions=30_000, + leased=2, + errors={"TimeoutError": 5}, + saturated=True, + latency={"p50": 0.002, "p95": 0.004, "p99": 0.017, "max": 0.05}, + pool_wait={"p50": 0.0, "p95": 0.0, "p99": 0.0003, "max": 0.001}, + lateness={"p50": 0.98, "p95": 1.0, "p99": 1.0, "max": 1.02}, + expired_silent=100, + false_expiries=0, + statement_ms={"insert": (0.024, 0.5), "with": (0.1, 0.4)}, + autovacuum_count=1, + live_tuples=2_000, + probe=_probe(insert_calls=30_000, wal_bytes=18_000_000, wal_records=199_000, tup_upd=30_000, tup_hot_upd=0), + ) + table = format_extend([result]) + assert "| r1000-b1 (saturated) | 1,000 | 1,000 | 1,000 | 2.0 / 4.0 / 17.0 |" in table + assert "TimeoutError 5" in table + assert "| r1000-b1 | 600 | 6.63 |" in table + assert "| 100 / 100 | 0 | 0.98 / 1.02 | 2 |" in table + + +def test_extend_accepts_its_flags() -> None: + args = _build_parser().parse_args(["extend", "--seconds", "12", "--only", "r1000-b1", "--json", "out.json"]) + assert (args.seconds, args.only, args.json) == (12.0, ["r1000-b1"], "out.json")