From 9281a90779c4f6781c2027df3074131bdcc1478c Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Fri, 6 Nov 2026 12:21:06 +0900 Subject: [PATCH 1/2] benchmarks: halve the io figures of pre-fix mini rows in the frozen copy (#555) The io.stat double count (stacked 259:0 NVMe + 252:0 dm on the mini) left every io_* figure at 2x in rows written before the harness fix. The freeze now halves them in runs_paper.csv for rows of a stacked-device host whose runner started before io_stat_fix.json's cutoff, leaves runs.jsonl and the in-memory rows raw, and writes the raw values of every corrected row to generated/io_stat_correction.json. Rows carry no harness commit, so the cutoff is the runner's manifest stamp (null until the fixed harness has run on the host). Only the freeze imports the new module; no stage or runner path does. Co-Authored-By: Claude Sonnet 5.5 --- benchmarks/experiments/PUBLISHING.md | 3 +- benchmarks/experiments/io_stat_correction.py | 100 ++++++++++++++ benchmarks/experiments/io_stat_fix.json | 6 + benchmarks/experiments/make_paper_tables.py | 16 +++ .../experiments/test_io_stat_correction.py | 126 ++++++++++++++++++ 5 files changed, 250 insertions(+), 1 deletion(-) create mode 100644 benchmarks/experiments/io_stat_correction.py create mode 100644 benchmarks/experiments/io_stat_fix.json create mode 100644 benchmarks/experiments/test_io_stat_correction.py diff --git a/benchmarks/experiments/PUBLISHING.md b/benchmarks/experiments/PUBLISHING.md index 925683cd1fb..8ca9f5c6f94 100644 --- a/benchmarks/experiments/PUBLISHING.md +++ b/benchmarks/experiments/PUBLISHING.md @@ -26,7 +26,7 @@ The Engines page of the documentation site (humemai/dbbench, `docs/engines.md`) The publish side renders rows stamped `instrument: 2026-11`, and nothing else. The freeze, the exporter, the gates, `land_stage.py` and `publish_results_asset.py` refuse a row of any other instrument, or a row with no stamp, with a message that says the tooling publishes 2026-11 rows only (`publish_instrument.refuse_other_instruments`; DECISIONS #196). If `BENCH_INSTRUMENT` is set to anything else, they refuse too; unset and `2026-11` mean the same thing here. -There is one freeze (`results/runs_paper.csv`), one payload (`results/web_benchmarks.json`) and one generated set (`results/generated`: `tables/`, `figures/`, `withheld_recall.json`, `GATE_STATUS.txt`). `BENCH_RESULTS_DIR` moves the whole `results/` tree and exists for tests only; a publish never sets it. The proof that the publish side works on 2026-11 rows is the fixture `fixtures/publish_2611` and two tests: `test_publish_2611_e2e.py` freezes and exports it end to end, and `test_publish_gates_2611.py` runs every gate below on the payload that comes out and requires each to pass. The fixture rows are copies of September and October measurements, so `build_fixture.realistic_2611` gives them what a 2026-11 row has and they lack (the override stamps, one pin, the roster of arms, the deep10m campaign rows behind the overlay files); its docstring names each patch and its reason, and `python build_fixture.py --patch` re-applies them to the committed rows. A gate that fails on the fixture is either right (the fixture row is one no 2026-11 run could write: fix the fixture) or still encodes a retired instrument (fix the gate). +There is one freeze (`results/runs_paper.csv`), one payload (`results/web_benchmarks.json`) and one generated set (`results/generated`: `tables/`, `figures/`, `withheld_recall.json`, `io_stat_correction.json`, `GATE_STATUS.txt`). `BENCH_RESULTS_DIR` moves the whole `results/` tree and exists for tests only; a publish never sets it. The proof that the publish side works on 2026-11 rows is the fixture `fixtures/publish_2611` and two tests: `test_publish_2611_e2e.py` freezes and exports it end to end, and `test_publish_gates_2611.py` runs every gate below on the payload that comes out and requires each to pass. The fixture rows are copies of September and October measurements, so `build_fixture.realistic_2611` gives them what a 2026-11 row has and they lack (the override stamps, one pin, the roster of arms, the deep10m campaign rows behind the overlay files); its docstring names each patch and its reason, and `python build_fixture.py --patch` re-applies them to the committed rows. A gate that fails on the fixture is either right (the fixture row is one no 2026-11 run could write: fix the fixture) or still encodes a retired instrument (fix the gate). Every ArcadeDB row measured under another pin is superseded, embedded as well as served; no ArcadeDB row is carried forward, and a comparator row carried forward is stamped `carried_forward_from` and `carried_forward_reason`. @@ -117,6 +117,7 @@ The allowlist says WHICH files are tracked; this says why, and what to re-run to | `runs_paper.csv` | `make_paper_tables.py` (the freeze) | every gate, the exporter, the release asset | | `web_benchmarks.json` | `export_web.py` | `page_check`, `version_consistency_check`, `refresh_web_page` | | `generated/tables/*`, `withheld_recall.json` | `make_paper_tables.py` | `page_check`, `provenance_check`, the exporter's withheld-cell notes | +| `generated/io_stat_correction.json` | `make_paper_tables.py` (the freeze, via `io_stat_correction.py`) | nothing reads it; it records which frozen rows had their io figures halved (the mini's stacked block devices counted twice before #555) and their raw values. The cutoff is `io_stat_fix.json` (`fixed_from_manifest`, null until the fixed harness has run on the host) | | `generated/memo_bottlenecks.html` | `memo_bottlenecks.py` | the maintainers' memo, not the page | | `sparse_cliff.jsonl` | `sparse_cliff_probe.py`, one shot | `make_paper_figures.f3_sparse_perquery`; f3 is not built (PAGE-SPEC.md section 3) | | `e3_q17/*.json`, `ingest_ab/ab_r*.json`, `tentag/tentag_ab.json` | `e3_recovery.py`, `async_ingest_probe.py`, `ts_stride_probe.py`, one shot each | `claims_check.py` (the recovery table is planned, not built) | diff --git a/benchmarks/experiments/io_stat_correction.py b/benchmarks/experiments/io_stat_correction.py new file mode 100644 index 00000000000..4e8280b6009 --- /dev/null +++ b/benchmarks/experiments/io_stat_correction.py @@ -0,0 +1,100 @@ +"""Freeze-time correction for the io.stat double count (humemai/arcadedb-embedded-python#555). Pure functions, no file writes. + +THE DEFECT. Until the fix, runner.read_io_stat and memory_probe.parse_io_stat summed every device line of a container's cgroup v2 +io.stat. On the mini the file lists 259:0 (NVMe) and 252:0 (a device-mapper device stacked on it) with identical bytes, plus an +unrelated 8:0 with 0 bytes, so every figure derived from it was recorded at twice its value. The fixed harness counts a stacked +device once. + +THE CORRECTION. The raw rows (runs.jsonl) are never edited. The freeze (make_paper_tables.freeze_paper_rows) calls `correct_rows` +on the rows it writes to runs_paper.csv: for a row that ran on a host named in `hosts` and predates the fix, every io figure is +halved in the FROZEN copy, and `report` names each corrected row with its raw values, so the recorded value is recoverable from +the freeze report (results/generated/io_stat_correction.json) and from runs.jsonl. No column is added to the CSV. + +WHICH ROWS PREDATE THE FIX. A row does not record the harness commit (only `engine_commit`, the engine's), so "contains the fix" +cannot be read off the row and is not guessed. What a row does record is `manifest`, the start time of the runner invocation that +wrote it (UTC, `YYYYmmddTHHMMSSZ`), and the runner is the harness code for that whole invocation: a stage pulls main before the +runner starts and never mid-run. io_stat_fix.json holds `fixed_from_manifest`, the first runner start whose checkout contains the +fix (`git merge-base --is-ancestor HEAD` on the host, read when the first stage after the fix landed). A row with +`manifest` earlier than that predates the fix. While the value is null no row is fixed, which is the truth until the fix reaches +the host. A row with neither a parseable `manifest` nor `ts_utc` is treated as predating the fix. +""" +import json +import re +from datetime import datetime, timezone +from pathlib import Path + +HERE = Path(__file__).resolve().parent +FIX_FILE = HERE / "io_stat_fix.json" + +# Every field the harness derives from io.stat: io_read_mib, io_write_mib, io_{read,write}_mib_sum, +# {client,server,server_container,sidecar_}_io_{read,write}_mib, and the memory probe's io__{read,write}_mib. +IO_FIELD = re.compile(r"^(?:[a-z0-9_]+_)?io_(?:read|write)_mib(?:_sum)?$|^io_[a-z0-9_]+_(?:read|write)_mib$") + + +def is_io_field(name): + return bool(IO_FIELD.match(str(name))) + + +def load_fix(path=FIX_FILE): + """The fix record: {"hosts": [...], "fixed_from_manifest": "YYYYmmddTHHMMSSZ" or None, "fix_commits": [...]}.""" + with open(path) as fh: + fix = json.load(fh) + if fix.get("fixed_from_manifest") and parse_stamp(fix["fixed_from_manifest"]) is None: + raise ValueError(f"fixed_from_manifest {fix['fixed_from_manifest']!r} is not a UTC stamp") # fail loudly, never skip + return fix + + +def parse_stamp(text): + """A runner `manifest` stamp (20260920T225804Z) or an ISO `ts_utc` as an aware UTC datetime; None when it is neither.""" + s = str(text or "").strip() + try: + return datetime.strptime(s, "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc) + except ValueError: + pass + try: + d = datetime.fromisoformat(s) + except ValueError: + return None + return d if d.tzinfo else d.replace(tzinfo=timezone.utc) + + +def predates_fix(row, fixed_from): + """True when the row was written by a harness without the fix. `fixed_from` is the cutoff stamp, or None (no fixed run yet).""" + if not fixed_from: + return True + cut = parse_stamp(fixed_from) + if cut is None: + raise ValueError(f"fixed_from_manifest {fixed_from!r} is not a UTC stamp") + when = parse_stamp(row.get("manifest")) or parse_stamp(row.get("ts_utc")) + return when is None or when < cut + + +def correct_row(row, hosts, fixed_from): + """(row or a corrected copy, {field: raw value} of what was halved). The input dict is never changed.""" + if row.get("bench_host") not in hosts or not predates_fix(row, fixed_from): + return row, {} + out, raw = dict(row), {} + for k, v in row.items(): + if is_io_field(k) and isinstance(v, (int, float)) and not isinstance(v, bool): + raw[k] = v + out[k] = round(v / 2, 2) # a recorded value is 0.1 MiB-quantised, so its half is exact at 0.05 + return (out, raw) if raw else (row, {}) + + +def correct_rows(rows, fix=None): + """(rows, report) for the freeze: `rows` in input order with the io figures of stacked-device rows halved, `report` the record + of what changed, fit for json.dump (rule, cutoff, counts, and per corrected row its identity and raw values).""" + fix = load_fix() if fix is None else fix + hosts, fixed_from = tuple(fix["hosts"]), fix.get("fixed_from_manifest") + out, corrected = [], [] + for r in rows: + c, raw = correct_row(r, hosts, fixed_from) + out.append(c) + if raw: + corrected.append({"run_id": r.get("run_id"), "lane": r.get("lane"), "backend": r.get("backend"), + "scale": r.get("scale"), "rep": r.get("rep"), "raw": raw}) + report = {"issue": "humemai/arcadedb-embedded-python#555", + "rule": "io figures halved in the frozen rows of hosts that stack block devices, for rows written before the fix", + "hosts": list(hosts), "fixed_from_manifest": fixed_from, "fix_commits": fix.get("fix_commits", []), + "rows_frozen": len(rows), "rows_corrected": len(corrected), "corrected": corrected} + return out, report diff --git a/benchmarks/experiments/io_stat_fix.json b/benchmarks/experiments/io_stat_fix.json new file mode 100644 index 00000000000..0368e28d70f --- /dev/null +++ b/benchmarks/experiments/io_stat_fix.json @@ -0,0 +1,6 @@ +{ + "issue": "humemai/arcadedb-embedded-python#555", + "hosts": ["mini"], + "fix_commits": [], + "fixed_from_manifest": null +} diff --git a/benchmarks/experiments/make_paper_tables.py b/benchmarks/experiments/make_paper_tables.py index b7f41eb9f56..8a7d15735b2 100644 --- a/benchmarks/experiments/make_paper_tables.py +++ b/benchmarks/experiments/make_paper_tables.py @@ -88,6 +88,7 @@ def _commit_matches(row_commit): HERE = os.path.dirname(os.path.abspath(__file__)) import publish_instrument as _PI +import io_stat_correction as _IOC RESULTS = str(_PI.results_dir()) # Every published cell is N=5. A row outside 1..5 is a probe, not a repetition; @@ -229,6 +230,17 @@ def _write_withheld_recall(): fh.write("\n") +def _write_io_stat_correction(report): + """Sidecar beside the freeze: which rows had their io figures halved, with the raw values (io_stat_correction.py).""" + out = os.path.join(GENERATED, "io_stat_correction.json") + os.makedirs(os.path.dirname(out), exist_ok=True) + with open(out, "w") as fh: + json.dump(report, fh, indent=1) + fh.write("\n") # the end-of-file hook wants it; see _write_withheld_recall + print(f"io.stat correction: {report['rows_corrected']} of {report['rows_frozen']} frozen rows halved " + f"(fixed_from_manifest={report['fixed_from_manifest']})") + + def _assert_lanes_registered(): """Every lane the runner can run is registered here, or it says which. @@ -1325,6 +1337,10 @@ def freeze_paper_rows(rows): recompute every T2/T3 cell from this file alone. """ import csv as _csv + # THE IO.STAT DOUBLE COUNT (#555): rows of a host that stacks block devices, written before the fixed harness, carry io figures at + # twice their value. They are halved in the FROZEN copy only; runs.jsonl keeps the raw rows and the sidecar the raw values. + rows, io_report = _IOC.correct_rows(rows) + _write_io_stat_correction(io_report) cols = [] for r in rows: for k in r: diff --git a/benchmarks/experiments/test_io_stat_correction.py b/benchmarks/experiments/test_io_stat_correction.py new file mode 100644 index 00000000000..ca89d2f5ccb --- /dev/null +++ b/benchmarks/experiments/test_io_stat_correction.py @@ -0,0 +1,126 @@ +"""The freeze halves the io figures of pre-fix rows from a host that stacks block devices (#555), and changes nothing else. + +Run with `python -m pytest test_io_stat_correction.py -q` from this directory. What is held: + + * the field rule takes every io.stat-derived figure (io_*_mib, the _sum pair, the per-side and per-phase ones) and nothing else + that merely looks like it (a memory or disk figure, a thread count); + * a mini row before the cutoff is halved, a row at or after it is not, a row of another host is not, and while the cutoff is + null every mini row predates the fix; + * the input rows are never changed, so the tables built from them and runs.jsonl keep the raw values; + * the freeze writes the halved values to runs_paper.csv and the raw ones to the sidecar report, and the committed fix file parses. +""" +import copy +import csv +import json +import sys +import tempfile +from pathlib import Path + +HERE = Path(__file__).resolve().parent +sys.path.insert(0, str(HERE)) + +import io_stat_correction as C # noqa: E402 + +CUT = "20261107T000000Z" +MINI = {"run_id": "r1", "bench_host": "mini", "lane": "l1", "backend": "arcadedb", "scale": "tiny", "rep": 1, + "manifest": "20261101T000000Z", "ts_utc": "2026-11-01T10:25:13+00:00", + "io_read_mib": 0.2, "io_write_mib": 16590.4, "io_read_mib_sum": 0.2, "io_write_mib_sum": 16590.4, + "client_io_write_mib": 8.0, "server_io_read_mib": 1.0, "server_container_io_write_mib": 6.0, + "sidecar_x_io_write_mib": 4.0, "io_load_write_mib": 100.0, + "peak_mib_sum": 123.0, "disk_mb_sum": 50.0, "redis_io_threads": 4, "io_note": "text", "wall_s": 9.0} +FIX = {"hosts": ["mini"], "fixed_from_manifest": CUT, "fix_commits": []} + + +def test_field_rule(): + for k in ("io_read_mib", "io_write_mib", "io_read_mib_sum", "io_write_mib_sum", "client_io_read_mib", "server_io_write_mib", + "server_container_io_write_mib", "sidecar_x_io_read_mib", "io_load_write_mib", "io_query_read_mib"): + assert C.is_io_field(k), k + for k in ("peak_mib_sum", "disk_mb_sum", "server_disk_rw_mb", "redis_io_threads", "io_note", "ingest_mib", "wall_s", + "io_read_mib_extra"): + assert not C.is_io_field(k), k + + +def test_mini_row_before_cutoff_is_halved_and_the_rest_is_untouched(): + before = copy.deepcopy(MINI) + out, raw = C.correct_row(MINI, ("mini",), CUT) + assert MINI == before, "the input row must not change" + assert out["io_write_mib"] == 8295.2 and out["io_write_mib_sum"] == 8295.2 and out["io_read_mib"] == 0.1 + assert out["client_io_write_mib"] == 4.0 and out["server_container_io_write_mib"] == 3.0 and out["io_load_write_mib"] == 50.0 + assert raw["io_write_mib"] == 16590.4 and "peak_mib_sum" not in raw and len(raw) == 9 + assert {k: v for k, v in out.items() if not C.is_io_field(k)} == {k: v for k, v in MINI.items() if not C.is_io_field(k)} + + +def test_rows_not_corrected(): + assert C.correct_row(dict(MINI, manifest="20261107T000000Z"), ("mini",), CUT)[1] == {} # at the cutoff: fixed + assert C.correct_row(dict(MINI, manifest="20261201T000000Z"), ("mini",), CUT)[1] == {} + assert C.correct_row(dict(MINI, bench_host="laptop"), ("mini",), CUT)[1] == {} + assert C.correct_row({k: v for k, v in MINI.items() if k != "bench_host"}, ("mini",), CUT)[1] == {} + # None, text and bool pass through unhalved + odd = C.correct_row(dict(MINI, io_write_mib=None, io_read_mib="n/a", io_write_mib_sum=True), ("mini",), CUT)[1] + assert odd.keys() == {"io_read_mib_sum", "client_io_write_mib", "server_io_read_mib", "server_container_io_write_mib", + "sidecar_x_io_write_mib", "io_load_write_mib"} + + +def test_cutoff_null_means_every_mini_row_predates_the_fix(): + assert C.correct_row(dict(MINI, manifest="20301231T000000Z"), ("mini",), None)[1]["io_write_mib"] == 16590.4 + + +def test_row_without_a_stamp_predates_the_fix_and_ts_utc_is_the_fallback(): + bare = {k: v for k, v in MINI.items() if k not in ("manifest", "ts_utc")} + assert C.correct_row(bare, ("mini",), CUT)[1] + late = dict(bare, ts_utc="2026-11-08T00:00:00+00:00") + assert not C.correct_row(late, ("mini",), CUT)[1] + assert C.correct_row(dict(late, ts_utc="2026-11-06T23:59:59+00:00"), ("mini",), CUT)[1] + assert C.correct_row(dict(bare, ts_utc="garbage"), ("mini",), CUT)[1] + + +def test_correct_rows_report_names_each_corrected_row_with_its_raw_values(): + rows = [MINI, dict(MINI, run_id="r2", manifest="20261108T000000Z"), dict(MINI, run_id="r3", bench_host="laptop")] + out, report = C.correct_rows(rows, FIX) + assert [r["run_id"] for r in out] == ["r1", "r2", "r3"] + assert out[0]["io_write_mib"] == 8295.2 and out[1]["io_write_mib"] == 16590.4 and out[2]["io_write_mib"] == 16590.4 + assert report["rows_frozen"] == 3 and report["rows_corrected"] == 1 + assert report["corrected"][0]["run_id"] == "r1" and report["corrected"][0]["raw"]["io_write_mib"] == 16590.4 + json.dumps(report) + + +def test_malformed_cutoff_fails_loudly(): + with tempfile.TemporaryDirectory() as d: + p = Path(d) / "fix.json" + p.write_text(json.dumps({"hosts": ["mini"], "fixed_from_manifest": "yesterday"})) + try: + C.load_fix(p) + except ValueError: + return + raise AssertionError("a malformed cutoff must raise") + + +def test_committed_fix_file_parses(): + fix = C.load_fix() + assert fix["hosts"] == ["mini"] and "fixed_from_manifest" in fix + + +def test_freeze_writes_halved_rows_and_the_raw_values_to_the_sidecar(): + import make_paper_tables as M + rows = [dict(MINI), dict(MINI, run_id="r2", rep=2, bench_host="laptop")] + before = copy.deepcopy(rows) + with tempfile.TemporaryDirectory() as d: + saved = (M.RESULTS, M.GENERATED, C.load_fix) + M.RESULTS, M.GENERATED, C.load_fix = d, str(Path(d) / "generated"), lambda: FIX # not the committed cutoff, which moves + try: + M.freeze_paper_rows(rows) + finally: + M.RESULTS, M.GENERATED, C.load_fix = saved + frozen = {r["run_id"]: r for r in csv.DictReader(open(Path(d) / "runs_paper.csv"))} + report = json.load(open(Path(d) / "generated" / "io_stat_correction.json")) + assert rows == before, "freeze_paper_rows must not change the rows the tables are built from" + assert float(frozen["r1"]["io_write_mib"]) == 8295.2 and float(frozen["r2"]["io_write_mib"]) == 16590.4 + assert frozen["r1"]["peak_mib_sum"] == "123.0" and set(frozen["r1"]) == set(frozen["r2"]) + assert report["rows_corrected"] == 1 and report["corrected"][0]["raw"]["io_write_mib"] == 16590.4 + + +if __name__ == "__main__": + for name, fn in sorted(globals().items()): + if name.startswith("test_") and callable(fn): + fn() + print("ok", name) From 47ae40b4411df24ac1c3faedeae0607f107897f4 Mon Sep 17 00:00:00 2001 From: Taewoon Kim Date: Fri, 6 Nov 2026 12:25:59 +0900 Subject: [PATCH 2/2] benchmarks: freeze correction reads the io_stat_dedup stamp, not a time cutoff (#555) A row of a stacked-device host without io_stat_dedup == 1 predates the fix and is halved in the frozen copy (its frozen stamp reads 'halved'); a stamped row is never halved; any other stamp raises. The manifest cutoff is gone: the stamp is written by the commit that carries the fix, so no value has to be set later. page_check.NOT_PRINTED declares the column and provenance_check's schema-homogeneity check ignores it as provenance, so a cell whose reps straddle the fix passes. Co-Authored-By: Claude Sonnet 5.5 --- benchmarks/experiments/PUBLISHING.md | 2 +- benchmarks/experiments/io_stat_correction.py | 84 +++++++---------- benchmarks/experiments/io_stat_fix.json | 4 +- benchmarks/experiments/make_paper_tables.py | 4 +- benchmarks/experiments/page_check.py | 4 + benchmarks/experiments/provenance_check.py | 5 +- .../experiments/test_io_stat_correction.py | 93 ++++++++++--------- 7 files changed, 96 insertions(+), 100 deletions(-) diff --git a/benchmarks/experiments/PUBLISHING.md b/benchmarks/experiments/PUBLISHING.md index 8ca9f5c6f94..b0f6f14b4d1 100644 --- a/benchmarks/experiments/PUBLISHING.md +++ b/benchmarks/experiments/PUBLISHING.md @@ -117,7 +117,7 @@ The allowlist says WHICH files are tracked; this says why, and what to re-run to | `runs_paper.csv` | `make_paper_tables.py` (the freeze) | every gate, the exporter, the release asset | | `web_benchmarks.json` | `export_web.py` | `page_check`, `version_consistency_check`, `refresh_web_page` | | `generated/tables/*`, `withheld_recall.json` | `make_paper_tables.py` | `page_check`, `provenance_check`, the exporter's withheld-cell notes | -| `generated/io_stat_correction.json` | `make_paper_tables.py` (the freeze, via `io_stat_correction.py`) | nothing reads it; it records which frozen rows had their io figures halved (the mini's stacked block devices counted twice before #555) and their raw values. The cutoff is `io_stat_fix.json` (`fixed_from_manifest`, null until the fixed harness has run on the host) | +| `generated/io_stat_correction.json` | `make_paper_tables.py` (the freeze, via `io_stat_correction.py`) | nothing reads it; it records which frozen rows had their io figures halved (the mini's stacked block devices counted twice before #555) and their raw values. A row is corrected unless it carries `io_stat_dedup: 1`, which the fixed runner stamps; the hosts are in `io_stat_fix.json` | | `generated/memo_bottlenecks.html` | `memo_bottlenecks.py` | the maintainers' memo, not the page | | `sparse_cliff.jsonl` | `sparse_cliff_probe.py`, one shot | `make_paper_figures.f3_sparse_perquery`; f3 is not built (PAGE-SPEC.md section 3) | | `e3_q17/*.json`, `ingest_ab/ab_r*.json`, `tentag/tentag_ab.json` | `e3_recovery.py`, `async_ingest_probe.py`, `ts_stride_probe.py`, one shot each | `claims_check.py` (the recovery table is planned, not built) | diff --git a/benchmarks/experiments/io_stat_correction.py b/benchmarks/experiments/io_stat_correction.py index 4e8280b6009..b1803288dc6 100644 --- a/benchmarks/experiments/io_stat_correction.py +++ b/benchmarks/experiments/io_stat_correction.py @@ -3,28 +3,27 @@ THE DEFECT. Until the fix, runner.read_io_stat and memory_probe.parse_io_stat summed every device line of a container's cgroup v2 io.stat. On the mini the file lists 259:0 (NVMe) and 252:0 (a device-mapper device stacked on it) with identical bytes, plus an unrelated 8:0 with 0 bytes, so every figure derived from it was recorded at twice its value. The fixed harness counts a stacked -device once. +device once and says so: its runner stamps `io_stat_dedup: 1` into every row it writes. THE CORRECTION. The raw rows (runs.jsonl) are never edited. The freeze (make_paper_tables.freeze_paper_rows) calls `correct_rows` -on the rows it writes to runs_paper.csv: for a row that ran on a host named in `hosts` and predates the fix, every io figure is -halved in the FROZEN copy, and `report` names each corrected row with its raw values, so the recorded value is recoverable from -the freeze report (results/generated/io_stat_correction.json) and from runs.jsonl. No column is added to the CSV. - -WHICH ROWS PREDATE THE FIX. A row does not record the harness commit (only `engine_commit`, the engine's), so "contains the fix" -cannot be read off the row and is not guessed. What a row does record is `manifest`, the start time of the runner invocation that -wrote it (UTC, `YYYYmmddTHHMMSSZ`), and the runner is the harness code for that whole invocation: a stage pulls main before the -runner starts and never mid-run. io_stat_fix.json holds `fixed_from_manifest`, the first runner start whose checkout contains the -fix (`git merge-base --is-ancestor HEAD` on the host, read when the first stage after the fix landed). A row with -`manifest` earlier than that predates the fix. While the value is null no row is fixed, which is the truth until the fix reaches -the host. A row with neither a parseable `manifest` nor `ts_utc` is treated as predating the fix. +on the rows it writes to runs_paper.csv. A row that ran on a host named in `hosts` (the row's `bench_host`) and does NOT carry +`io_stat_dedup == 1` predates the fix: every io figure in the frozen copy is halved, and the frozen row's `io_stat_dedup` reads +"halved" (a stamped row keeps its 1, a row of another host keeps whatever it had, usually blank). `report` names each corrected +row with its raw values, so the recorded value is recoverable from the freeze report (results/generated/io_stat_correction.json) +and from runs.jsonl. + +WHICH ROWS PREDATE THE FIX. The row says so itself: the stamp is written by the same commit as the fix, so a row without it was +written without the fix, and no cutoff time has to be set later. A stamp that is present but is not 1 raises: a row the rule +cannot read is never silently counted either way. """ import json import re -from datetime import datetime, timezone from pathlib import Path HERE = Path(__file__).resolve().parent FIX_FILE = HERE / "io_stat_fix.json" +MARKER = "io_stat_dedup" +HALVED = "halved" # Every field the harness derives from io.stat: io_read_mib, io_write_mib, io_{read,write}_mib_sum, # {client,server,server_container,sidecar_}_io_{read,write}_mib, and the memory probe's io__{read,write}_mib. @@ -36,65 +35,54 @@ def is_io_field(name): def load_fix(path=FIX_FILE): - """The fix record: {"hosts": [...], "fixed_from_manifest": "YYYYmmddTHHMMSSZ" or None, "fix_commits": [...]}.""" + """The fix record: {"hosts": [hosts whose io.stat stacks block devices]}.""" with open(path) as fh: fix = json.load(fh) - if fix.get("fixed_from_manifest") and parse_stamp(fix["fixed_from_manifest"]) is None: - raise ValueError(f"fixed_from_manifest {fix['fixed_from_manifest']!r} is not a UTC stamp") # fail loudly, never skip + if not fix.get("hosts"): + raise ValueError(f"{path}: no hosts") return fix -def parse_stamp(text): - """A runner `manifest` stamp (20260920T225804Z) or an ISO `ts_utc` as an aware UTC datetime; None when it is neither.""" - s = str(text or "").strip() - try: - return datetime.strptime(s, "%Y%m%dT%H%M%SZ").replace(tzinfo=timezone.utc) - except ValueError: - pass - try: - d = datetime.fromisoformat(s) - except ValueError: - return None - return d if d.tzinfo else d.replace(tzinfo=timezone.utc) - - -def predates_fix(row, fixed_from): - """True when the row was written by a harness without the fix. `fixed_from` is the cutoff stamp, or None (no fixed run yet).""" - if not fixed_from: +def predates_fix(row): + """True when the row was written without the fix (no stamp), False when stamped 1; ValueError for any other stamp.""" + v = row.get(MARKER) + if v is None or v == "": return True - cut = parse_stamp(fixed_from) - if cut is None: - raise ValueError(f"fixed_from_manifest {fixed_from!r} is not a UTC stamp") - when = parse_stamp(row.get("manifest")) or parse_stamp(row.get("ts_utc")) - return when is None or when < cut + if v == 1 and not isinstance(v, (str, bool)): + return False + raise ValueError(f"{MARKER}={v!r} on run_id {row.get('run_id')!r}: the stamp is 1 or absent") -def correct_row(row, hosts, fixed_from): +def correct_row(row, hosts): """(row or a corrected copy, {field: raw value} of what was halved). The input dict is never changed.""" - if row.get("bench_host") not in hosts or not predates_fix(row, fixed_from): + if row.get("bench_host") not in hosts or not predates_fix(row): return row, {} out, raw = dict(row), {} for k, v in row.items(): if is_io_field(k) and isinstance(v, (int, float)) and not isinstance(v, bool): raw[k] = v out[k] = round(v / 2, 2) # a recorded value is 0.1 MiB-quantised, so its half is exact at 0.05 - return (out, raw) if raw else (row, {}) + if raw: + out[MARKER] = HALVED + return out, raw + return row, {} def correct_rows(rows, fix=None): """(rows, report) for the freeze: `rows` in input order with the io figures of stacked-device rows halved, `report` the record - of what changed, fit for json.dump (rule, cutoff, counts, and per corrected row its identity and raw values).""" + of what changed, fit for json.dump (rule, counts, and per corrected row its identity and raw values).""" fix = load_fix() if fix is None else fix - hosts, fixed_from = tuple(fix["hosts"]), fix.get("fixed_from_manifest") - out, corrected = [], [] + hosts = tuple(fix["hosts"]) + out, corrected, stamped = [], [], 0 for r in rows: - c, raw = correct_row(r, hosts, fixed_from) + c, raw = correct_row(r, hosts) out.append(c) + stamped += r.get("bench_host") in hosts and r.get(MARKER) == 1 if raw: corrected.append({"run_id": r.get("run_id"), "lane": r.get("lane"), "backend": r.get("backend"), "scale": r.get("scale"), "rep": r.get("rep"), "raw": raw}) report = {"issue": "humemai/arcadedb-embedded-python#555", - "rule": "io figures halved in the frozen rows of hosts that stack block devices, for rows written before the fix", - "hosts": list(hosts), "fixed_from_manifest": fixed_from, "fix_commits": fix.get("fix_commits", []), - "rows_frozen": len(rows), "rows_corrected": len(corrected), "corrected": corrected} + "rule": f"io figures halved in the frozen rows of hosts that stack block devices unless the row carries {MARKER} == 1", + "hosts": list(hosts), "rows_frozen": len(rows), "rows_stamped_untouched": stamped, + "rows_corrected": len(corrected), "corrected": corrected} return out, report diff --git a/benchmarks/experiments/io_stat_fix.json b/benchmarks/experiments/io_stat_fix.json index 0368e28d70f..b4b290064e2 100644 --- a/benchmarks/experiments/io_stat_fix.json +++ b/benchmarks/experiments/io_stat_fix.json @@ -1,6 +1,4 @@ { "issue": "humemai/arcadedb-embedded-python#555", - "hosts": ["mini"], - "fix_commits": [], - "fixed_from_manifest": null + "hosts": ["mini"] } diff --git a/benchmarks/experiments/make_paper_tables.py b/benchmarks/experiments/make_paper_tables.py index 8a7d15735b2..c8301f71405 100644 --- a/benchmarks/experiments/make_paper_tables.py +++ b/benchmarks/experiments/make_paper_tables.py @@ -237,8 +237,8 @@ def _write_io_stat_correction(report): with open(out, "w") as fh: json.dump(report, fh, indent=1) fh.write("\n") # the end-of-file hook wants it; see _write_withheld_recall - print(f"io.stat correction: {report['rows_corrected']} of {report['rows_frozen']} frozen rows halved " - f"(fixed_from_manifest={report['fixed_from_manifest']})") + print(f"io.stat correction: {report['rows_corrected']} of {report['rows_frozen']} frozen rows halved, " + f"{report['rows_stamped_untouched']} stamped io_stat_dedup=1 and left as recorded") def _assert_lanes_registered(): diff --git a/benchmarks/experiments/page_check.py b/benchmarks/experiments/page_check.py index d75d0b13828..03797df61a6 100644 --- a/benchmarks/experiments/page_check.py +++ b/benchmarks/experiments/page_check.py @@ -779,6 +779,10 @@ def main() -> int: "which HTTP client an ArcadeDB served arm ran with (CAMPAIGN 7 row 72, lean_http.py): a condition of the served arms, read from " "the session that ran, never a column. The e4 decomposition's row lists the two clients it measured (`arcadedb_http_clients`); its table " "names the client in each HTTP column instead (export_web._e4_arms)"), + (r"^io_stat_dedup$", + "provenance of the io figures (#555): 1 when the harness that wrote the row counted a stacked block device once, \"halved\" " + "when the freeze halved the recorded io figures of a row that predates the fix (io_stat_correction.py, raw values in " + "generated/io_stat_correction.json), blank otherwise; never a column"), (r"^jpype_version$", "which JPype the process that ran the cell had loaded (CAMPAIGN 7 row 73, bench_common.jpype_version): a condition of the embedded " "ArcadeDB arms, read from the process (`jpype.__version__`), empty for every arm that never imports it, never a column"), diff --git a/benchmarks/experiments/provenance_check.py b/benchmarks/experiments/provenance_check.py index 3ae1376c753..606029a1a49 100644 --- a/benchmarks/experiments/provenance_check.py +++ b/benchmarks/experiments/provenance_check.py @@ -690,7 +690,10 @@ def check_schema_homogeneity(rows): # logged, captured only once the capture worked (BUGS F1-F4). # A family split on these is a split in # instrumentation history, not in what was measured. - "graph_build_cache", "graph_build_access", "driver_") + "graph_build_cache", "graph_build_access", "driver_", + # PROVENANCE, not a measurement: the harness that wrote the row counted a stacked device once (#555), + # so a cell whose reps straddle the fix legitimately splits on it. + "io_stat_dedup") def measured(r): return {k for k, v in r.items() if isinstance(v, (int, float)) diff --git a/benchmarks/experiments/test_io_stat_correction.py b/benchmarks/experiments/test_io_stat_correction.py index ca89d2f5ccb..544a3b9b52f 100644 --- a/benchmarks/experiments/test_io_stat_correction.py +++ b/benchmarks/experiments/test_io_stat_correction.py @@ -4,10 +4,11 @@ * the field rule takes every io.stat-derived figure (io_*_mib, the _sum pair, the per-side and per-phase ones) and nothing else that merely looks like it (a memory or disk figure, a thread count); - * a mini row before the cutoff is halved, a row at or after it is not, a row of another host is not, and while the cutoff is - null every mini row predates the fix; + * a mini row without the `io_stat_dedup` stamp predates the fix and is halved (and says "halved"), a mini row stamped 1 is never + halved, a row of another host is not, and a stamp that is neither absent nor 1 raises; * the input rows are never changed, so the tables built from them and runs.jsonl keep the raw values; - * the freeze writes the halved values to runs_paper.csv and the raw ones to the sidecar report, and the committed fix file parses. + * the freeze writes the halved values to runs_paper.csv and the raw ones to the sidecar report, the committed fix file parses, + and the stamp is declared in page_check.NOT_PRINTED and ignored by provenance_check's schema-homogeneity check. """ import copy import csv @@ -21,14 +22,13 @@ import io_stat_correction as C # noqa: E402 -CUT = "20261107T000000Z" MINI = {"run_id": "r1", "bench_host": "mini", "lane": "l1", "backend": "arcadedb", "scale": "tiny", "rep": 1, - "manifest": "20261101T000000Z", "ts_utc": "2026-11-01T10:25:13+00:00", "io_read_mib": 0.2, "io_write_mib": 16590.4, "io_read_mib_sum": 0.2, "io_write_mib_sum": 16590.4, "client_io_write_mib": 8.0, "server_io_read_mib": 1.0, "server_container_io_write_mib": 6.0, "sidecar_x_io_write_mib": 4.0, "io_load_write_mib": 100.0, "peak_mib_sum": 123.0, "disk_mb_sum": 50.0, "redis_io_threads": 4, "io_note": "text", "wall_s": 9.0} -FIX = {"hosts": ["mini"], "fixed_from_manifest": CUT, "fix_commits": []} +STAMPED = dict(MINI, run_id="r2", io_stat_dedup=1) +FIX = {"hosts": ["mini"]} def test_field_rule(): @@ -36,85 +36,88 @@ def test_field_rule(): "server_container_io_write_mib", "sidecar_x_io_read_mib", "io_load_write_mib", "io_query_read_mib"): assert C.is_io_field(k), k for k in ("peak_mib_sum", "disk_mb_sum", "server_disk_rw_mb", "redis_io_threads", "io_note", "ingest_mib", "wall_s", - "io_read_mib_extra"): + "io_read_mib_extra", "io_stat_dedup"): assert not C.is_io_field(k), k -def test_mini_row_before_cutoff_is_halved_and_the_rest_is_untouched(): +def test_unstamped_mini_row_is_halved_and_the_rest_is_untouched(): before = copy.deepcopy(MINI) - out, raw = C.correct_row(MINI, ("mini",), CUT) + out, raw = C.correct_row(MINI, ("mini",)) assert MINI == before, "the input row must not change" assert out["io_write_mib"] == 8295.2 and out["io_write_mib_sum"] == 8295.2 and out["io_read_mib"] == 0.1 assert out["client_io_write_mib"] == 4.0 and out["server_container_io_write_mib"] == 3.0 and out["io_load_write_mib"] == 50.0 assert raw["io_write_mib"] == 16590.4 and "peak_mib_sum" not in raw and len(raw) == 9 - assert {k: v for k, v in out.items() if not C.is_io_field(k)} == {k: v for k, v in MINI.items() if not C.is_io_field(k)} + assert out["io_stat_dedup"] == "halved" + assert {k: v for k, v in out.items() if not C.is_io_field(k) and k != "io_stat_dedup"} == \ + {k: v for k, v in MINI.items() if not C.is_io_field(k)} + assert C.correct_row(dict(MINI, io_stat_dedup=None), ("mini",))[1] and C.correct_row(dict(MINI, io_stat_dedup=""), ("mini",))[1] + + +def test_stamped_row_is_never_halved(): + out, raw = C.correct_row(STAMPED, ("mini",)) + assert raw == {} and out is STAMPED and out["io_write_mib"] == 16590.4 and out["io_stat_dedup"] == 1 def test_rows_not_corrected(): - assert C.correct_row(dict(MINI, manifest="20261107T000000Z"), ("mini",), CUT)[1] == {} # at the cutoff: fixed - assert C.correct_row(dict(MINI, manifest="20261201T000000Z"), ("mini",), CUT)[1] == {} - assert C.correct_row(dict(MINI, bench_host="laptop"), ("mini",), CUT)[1] == {} - assert C.correct_row({k: v for k, v in MINI.items() if k != "bench_host"}, ("mini",), CUT)[1] == {} + assert C.correct_row(dict(MINI, bench_host="laptop"), ("mini",))[1] == {} + assert C.correct_row({k: v for k, v in MINI.items() if k != "bench_host"}, ("mini",))[1] == {} + assert C.correct_row(dict(MINI, bench_host="laptop", io_stat_dedup="junk"), ("mini",))[1] == {} # not our host: not read # None, text and bool pass through unhalved - odd = C.correct_row(dict(MINI, io_write_mib=None, io_read_mib="n/a", io_write_mib_sum=True), ("mini",), CUT)[1] + odd = C.correct_row(dict(MINI, io_write_mib=None, io_read_mib="n/a", io_write_mib_sum=True), ("mini",))[1] assert odd.keys() == {"io_read_mib_sum", "client_io_write_mib", "server_io_read_mib", "server_container_io_write_mib", "sidecar_x_io_write_mib", "io_load_write_mib"} + assert C.correct_row(dict(MINI, io_read_mib=None, io_write_mib=None, io_read_mib_sum=None, io_write_mib_sum=None, + client_io_write_mib=None, server_io_read_mib=None, server_container_io_write_mib=None, + sidecar_x_io_write_mib=None, io_load_write_mib=None), ("mini",))[0]["bench_host"] == "mini" -def test_cutoff_null_means_every_mini_row_predates_the_fix(): - assert C.correct_row(dict(MINI, manifest="20301231T000000Z"), ("mini",), None)[1]["io_write_mib"] == 16590.4 - - -def test_row_without_a_stamp_predates_the_fix_and_ts_utc_is_the_fallback(): - bare = {k: v for k, v in MINI.items() if k not in ("manifest", "ts_utc")} - assert C.correct_row(bare, ("mini",), CUT)[1] - late = dict(bare, ts_utc="2026-11-08T00:00:00+00:00") - assert not C.correct_row(late, ("mini",), CUT)[1] - assert C.correct_row(dict(late, ts_utc="2026-11-06T23:59:59+00:00"), ("mini",), CUT)[1] - assert C.correct_row(dict(bare, ts_utc="garbage"), ("mini",), CUT)[1] +def test_malformed_stamp_raises(): + for bad in (2, 0, "1", True, "halved", 1.5): + try: + C.correct_row(dict(MINI, io_stat_dedup=bad), ("mini",)) + except ValueError: + continue + raise AssertionError(f"stamp {bad!r} must raise") def test_correct_rows_report_names_each_corrected_row_with_its_raw_values(): - rows = [MINI, dict(MINI, run_id="r2", manifest="20261108T000000Z"), dict(MINI, run_id="r3", bench_host="laptop")] + rows = [MINI, STAMPED, dict(MINI, run_id="r3", bench_host="laptop")] out, report = C.correct_rows(rows, FIX) assert [r["run_id"] for r in out] == ["r1", "r2", "r3"] assert out[0]["io_write_mib"] == 8295.2 and out[1]["io_write_mib"] == 16590.4 and out[2]["io_write_mib"] == 16590.4 - assert report["rows_frozen"] == 3 and report["rows_corrected"] == 1 + assert report["rows_frozen"] == 3 and report["rows_corrected"] == 1 and report["rows_stamped_untouched"] == 1 assert report["corrected"][0]["run_id"] == "r1" and report["corrected"][0]["raw"]["io_write_mib"] == 16590.4 json.dumps(report) -def test_malformed_cutoff_fails_loudly(): - with tempfile.TemporaryDirectory() as d: - p = Path(d) / "fix.json" - p.write_text(json.dumps({"hosts": ["mini"], "fixed_from_manifest": "yesterday"})) - try: - C.load_fix(p) - except ValueError: - return - raise AssertionError("a malformed cutoff must raise") +def test_committed_fix_file_parses(): + assert C.load_fix()["hosts"] == ["mini"] -def test_committed_fix_file_parses(): - fix = C.load_fix() - assert fix["hosts"] == ["mini"] and "fixed_from_manifest" in fix +def test_the_stamp_is_declared_to_the_gates(): + import page_check + assert page_check._not_printed_reason("io_stat_dedup") + src = (HERE / "provenance_check.py").read_text() + assert '"io_stat_dedup")' in src.split("IGNORE_PREFIX = (")[1].split("def measured")[0] def test_freeze_writes_halved_rows_and_the_raw_values_to_the_sidecar(): import make_paper_tables as M - rows = [dict(MINI), dict(MINI, run_id="r2", rep=2, bench_host="laptop")] + rows = [dict(MINI), dict(STAMPED, rep=2), dict(MINI, run_id="r3", rep=3, bench_host="laptop")] before = copy.deepcopy(rows) with tempfile.TemporaryDirectory() as d: - saved = (M.RESULTS, M.GENERATED, C.load_fix) - M.RESULTS, M.GENERATED, C.load_fix = d, str(Path(d) / "generated"), lambda: FIX # not the committed cutoff, which moves + saved = (M.RESULTS, M.GENERATED) + M.RESULTS, M.GENERATED = d, str(Path(d) / "generated") try: M.freeze_paper_rows(rows) finally: - M.RESULTS, M.GENERATED, C.load_fix = saved + M.RESULTS, M.GENERATED = saved frozen = {r["run_id"]: r for r in csv.DictReader(open(Path(d) / "runs_paper.csv"))} report = json.load(open(Path(d) / "generated" / "io_stat_correction.json")) assert rows == before, "freeze_paper_rows must not change the rows the tables are built from" - assert float(frozen["r1"]["io_write_mib"]) == 8295.2 and float(frozen["r2"]["io_write_mib"]) == 16590.4 + assert float(frozen["r1"]["io_write_mib"]) == 8295.2 and frozen["r1"]["io_stat_dedup"] == "halved" + assert float(frozen["r2"]["io_write_mib"]) == 16590.4 and frozen["r2"]["io_stat_dedup"] == "1" + assert float(frozen["r3"]["io_write_mib"]) == 16590.4 and frozen["r3"]["io_stat_dedup"] == "" assert frozen["r1"]["peak_mib_sum"] == "123.0" and set(frozen["r1"]) == set(frozen["r2"]) assert report["rows_corrected"] == 1 and report["corrected"][0]["raw"]["io_write_mib"] == 16590.4