#!/usr/bin/env python
"""How many SQL round trips each hr use case costs, and how long it takes — on SQLite (counts) or a real DataK3
bucket (counts AND time). Re-run it after any change to a repository, the configuration cache or the tick.
python scripts/bench_live.py # SQLite: round trips only
BUCKET=erp-platform-test DODIL_ORG=dodil python scripts/bench_live.py # live: + timings, p50/p95
--people 50 --open 10 --repeat 10 --skip-migrate
Why round trips: on DataK3 a statement costs ~70-100 ms from a laptop (measured 2026-10-17 on erp-platform-test:
SELECT on a small table ~80 ms, a commit ~70 ms, psycopg's BEGIN one more), and almost nothing on SQLite — so the
COUNT is the portable measure and the time is what the count costs on the wire. A round trip here is one statement
(engine `before_cursor_execute`), plus one BEGIN and one COMMIT/ROLLBACK per transaction.
What it does, through the doors (HR.app() in this process, gateway-shaped identities):
migrate (live) → install the gb pack → the rest of the seed → hire --people → cancel all but --open of their runs
→ (--settle seconds) → one unmeasured call of each read (a warm app) → show_person / my_hr_tasks / list_people /
run_board / headcount × --repeat → tick × --repeat (the first acts on the open runs; the rest are the steady
tick) → every hr table dropped (live, unless --keep). --reuse measures the reads and the tick on what a --keep
run left. Each hire is a new person, so a hire's time is a real hire, not an idempotent replay.
"""
import argparse
import json
import os
import pathlib
import re
import statistics
import sys
import tempfile
import time
from collections import Counter, defaultdict
from contextlib import contextmanager
from datetime import date, timedelta
ROOT = pathlib.Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "src"))
sys.path.insert(0, str(ROOT / "tests"))
LIVE = bool(os.environ.get("BUCKET"))
class Meter:
"""Statements and transactions per label, from engine events — no change to the code under test."""
def __init__(self):
self.label = None
self.calls: dict = defaultdict(list) # label -> [(round_trips, seconds, statements, txns)]
self.tables: dict = defaultdict(Counter) # label -> table -> statements
self._st = self._tx = self._saved = 0
self._t: Counter = Counter()
self._t0 = 0.0
self._slow: list = []
self.stmt_time: dict = defaultdict(float) # label -> seconds spent inside statements
def attach(self, eng):
from sqlalchemy import event
@event.listens_for(eng, "before_cursor_execute")
def _stmt(conn, cursor, statement, params, context, executemany):
self._st += 1
self._t0 = time.monotonic()
m = re.search(r"\b(?:FROM|INTO|UPDATE|TABLE(?: IF (?:NOT )?EXISTS)?)\s+\"?(\w+)", statement, re.I)
self._t[m.group(1) if m else "?"] += 1
@event.listens_for(eng, "after_cursor_execute")
def _done(conn, cursor, statement, params, context, executemany):
self._slow.append((time.monotonic() - self._t0, statement[:90]))
@event.listens_for(eng, "begin")
def _begin(conn):
self._tx += 1
from brick_hr.infrastructure import batching
def _batch(n, pipelined, seconds=0.0, sql=""):
# a batch is one wire round trip: pipelined statements fire no cursor event (count them); sequential
# ones (SQLite) fired n events, of which n - 1 would not be a round trip on the pg wire
self._st += n if pipelined else 0
self._saved += (n - 1)
if pipelined:
self._slow.append((seconds, f"[pipeline x{n}] {sql}"))
batching.observer = _batch
@contextmanager
def measure(self, label):
self._st = self._tx = self._saved = 0
self._t = Counter()
self._slow = []
t0 = time.monotonic()
try:
yield
finally:
dt = time.monotonic() - t0
# one BEGIN and one COMMIT/ROLLBACK per transaction: what psycopg sends around the statements
self.calls[label].append((self._st - self._saved + 2 * self._tx, dt, self._st, self._tx))
self.tables[label].update(self._t)
self.stmt_time[label] += sum(x[0] for x in self._slow)
if os.environ.get("BENCH_TRACE") and self._slow:
print(f"-- {label}: {len(self._slow)} statements, {sum(x[0] for x in self._slow):.2f}s in them, "
f"{dt:.2f}s wall; slowest:", file=sys.stderr)
for d, sql in sorted(self._slow, reverse=True)[:6]:
print(f" {d*1000:6.0f} ms {' '.join(sql.split())}", file=sys.stderr)
def report(self) -> dict:
out = {}
for label, xs in self.calls.items():
rts = [x[0] for x in xs]
ts = sorted(x[1] for x in xs)
row = {"n": len(xs), "round_trips": int(statistics.median(rts)), "statements": int(statistics.median(x[2] for x in xs)),
"transactions": int(statistics.median(x[3] for x in xs))}
if LIVE:
row.update(p50_s=round(statistics.median(ts), 2),
p95_s=round(ts[min(len(ts) - 1, int(round(0.95 * (len(ts) - 1))))], 2),
max_s=round(ts[-1], 2))
if LIVE:
row["in_statements_s_per_call"] = round(self.stmt_time[label] / len(xs), 2)
if len(xs) <= 12:
row["each"] = [x[0] for x in xs]
row["top_tables"] = dict(self.tables[label].most_common(5))
out[label] = row
return out
def main() -> int:
ap = argparse.ArgumentParser()
ap.add_argument("--people", type=int, default=50)
ap.add_argument("--open", type=int, default=10, help="open runs left after cancelling the rest")
ap.add_argument("--repeat", type=int, default=10)
ap.add_argument("--skip-migrate", action="store_true", help="live: create the tables directly (no erp-migrate)")
ap.add_argument("--keep", action="store_true", help="live: do not drop the tables at the end")
ap.add_argument("--reuse", action="store_true",
help="live: the tables and people a --keep run left — measure the reads and the tick only")
ap.add_argument("--settle", type=float, default=0.0,
help="seconds to wait after the writes, before the reads (DataK3 drains its write-log in the "
"background; reads right after a burst of writes are slower than a settled bucket's)")
ap.add_argument("--json", help="write the report here as well")
a = ap.parse_args()
import anyio
import httpx
from erp_platform import capabilities, public, seam
from brick_hr.application import use_cases
from brick_hr.infrastructure.tables import HrRepository, md
from conftest import headers
from seed import hire_body, seed
meter = Meter()
tables = [t.name for t in md.sorted_tables]
if LIVE:
from erp_platform import livecheck as lc
sys.path.insert(0, str(ROOT / "scripts"))
if a.reuse:
pass
elif a.skip_migrate:
lc.create_schema(ROOT / "src" / "brick_hr" / "schema.sql", tables)
t0 = time.monotonic()
meter.calls["migrate"].append((0, 0.0, 0, 0))
else:
import live_check
t0 = time.monotonic()
ok = live_check.migrate(tables)
meter.calls["migrate"].append((0, time.monotonic() - t0, 0, 0))
if not ok:
print("migrate failed")
return 1
eng = lc.engine()
else:
from sqlalchemy import create_engine
path = pathlib.Path(tempfile.mkdtemp()) / "bench.db"
eng = create_engine(f"sqlite:///{path}", connect_args={"check_same_thread": False, "timeout": 30})
md.create_all(eng)
public.md.create_all(eng)
public.configure(eng.begin)
capabilities.reset_providers()
seam.unregister_local("business-partner")
meter.attach(eng)
@contextmanager
def repo():
with eng.begin() as conn:
yield HrRepository(conn)
use_cases.repository = repo
from brick_hr.app import app
def call(method, path, who, body=None, params=None):
async def go():
async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url="http://hr", timeout=600) as c:
return await c.request(method, path, headers=who, json=body, params=params)
r = anyio.run(go)
try:
return r.status_code, r.json()
except ValueError:
return r.status_code, r.text
def must(label, method, path, who, body=None):
with meter.measure(label):
s, out = call(method, path, who, body)
assert 200 <= s < 300, (label, s, out)
return out
HR, TICK = headers("hr_admin"), headers("clock")
try:
def post(p, b):
label = "install gb pack" if p.startswith("/localisation-packs") else "seed: " + p.split("/")[1]
with meter.measure(label):
return call("POST", p, HR, b)
people, runs = [], []
if a.reuse:
from sqlalchemy import select
from brick_hr.infrastructure import tables as T
with eng.connect() as c:
people = [str(r[0]) for r in c.execute(select(T.person.c.person_id).order_by(T.person.c.person_id))]
else:
seed(post)
start = date.today() + timedelta(days=21)
for i in range(0 if a.reuse else a.people):
h = must("hire", "POST", "/hires", HR, hire_body(f"Bench{i}", f"Person{i}", start.isoformat(),
rate_amount="42000.0000", rate_basis="year",
currency="GBP"))
people.append(h["person_id"])
runs.append(h["run_id"])
for rid in runs[a.open:]:
must("cancel_run", "POST", f"/runs/{rid}/cancel", HR, {"reason": "bench: only --open runs stay open"})
if a.settle:
time.sleep(a.settle)
# a warm app: one unmeasured call of each read first (a new process reads the configuration once)
for path, who in ((f"/people/{people[-1]}", HR), ("/my-tasks", HR), ("/people", HR), ("/runs", HR),
("/reports/headcount", HR)):
call("GET", path, who)
for i in range(a.repeat):
pid = people[i % len(people)]
must("show_person", "GET", f"/people/{pid}", HR)
must("my_hr_tasks (hr)", "GET", "/my-tasks", HR)
must("my_hr_tasks (it)", "GET", "/my-tasks", headers("it"))
must("list_people", "GET", "/people", HR)
must("run_board", "GET", "/runs", HR)
must("headcount", "GET", "/reports/headcount", HR)
for i in range(a.repeat):
must("tick", "POST", "/run-due-steps", TICK, {})
must("keep_configuration (1 row)", "POST", "/configuration/job",
HR, {"rows": [{"job_code": "bench_job", "title": f"Bench {time.time()}"}]})
finally:
rep = meter.report()
print(json.dumps({"mode": "live" if LIVE else "sqlite", "people": a.people, "open_runs": a.open, "report": rep},
indent=1, default=str))
if a.json:
pathlib.Path(a.json).write_text(json.dumps(rep, indent=1, default=str))
eng.dispose()
if LIVE and not a.keep:
left = lc.drop_all(tables)
print("tables dropped" if not left else f"COULD NOT DROP {left}")
return 0
if __name__ == "__main__":
raise SystemExit(main())