feat: split probe records into line and sidecar

A record is now one line of records/<emulator>.jsonl in the measurements repository, under 20 KB, and a gzip sidecar holding every repetition, the core logs and the raw timings, its sha256 in the line. record_version 2, core-probe-data.schema.json for the sidecar. core_probe --record appends to a measurements clone and refuses a directory inside this repository's tracked tree.

The seven records of issue 85 moved to Abdess/retrobios-measurements: each line plus its sidecar rebuilds the first file, and keeps its first id.

scripts/core_probe.py

@@ -10,8 +10,10 @@ Three modes share one runner:
 
 Each repetition runs in its own process, under strace when it is installed,
 so the files a core opened are read from the kernel rather than guessed.
-`--record DIR` writes one JSON record per invocation, carrying what is needed
-to reproduce it (core build, frontend commit, host, inputs by hash).
+`--record DIR` appends one record per invocation to a clone of the
+measurements repository, carrying what is needed to reproduce it (core build,
+frontend commit, host, inputs by hash). measurement_records.py splits it into
+a slim line and a gzip sidecar.
 """
 
 from __future__ import annotations
@@ -37,7 +39,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 
 ROOT = Path(__file__).resolve().parent.parent
-RECORD_VERSION = 1
+MEASUREMENTS_CLONE = ROOT / ".cache" / "measurements"
 MIN_BENCH_REPETITIONS = 5
 DEFAULT_WORK_DIR = ROOT / "tmp" / "core_probe"
 BASE_OBSERVABLES = ("frame_sha1", "video_sha1", "audio_sha1", "state_sha1")
@@ -769,7 +771,18 @@ def build_parser() -> argparse.ArgumentParser:
     common.add_argument("--core-repo", help="source repository URL of the core build")
     common.add_argument("--core-commit", help="commit the core was built from")
     common.add_argument("--build-command", help="command and flags that built the core")
-    common.add_argument("--record", metavar="DIR", help="write one JSON record under DIR/<core>/")
+    common.add_argument("--record", metavar="DIR",
+                        help="append the record to the measurements clone DIR "
+                             f"(records/<emulator>.jsonl and data/<emulator>/), "
+                             f"for example {MEASUREMENTS_CLONE.relative_to(ROOT)}")
+    common.add_argument("--emulator", help="profile the record is filed under (default: the "
+                                           "profile whose cores: names the .so, else the "
+                                           "core's library name)")
+    common.add_argument("--subject", help="slug naming what the record tests, shared by its "
+                                          "reruns (default: derived from the claim and method, "
+                                          "or from the inputs of a run or bench)")
+    common.add_argument("--supersedes", metavar="ID",
+                        help="id of an earlier record of the same emulator this one corrects")
     common.add_argument("--max-load", type=float, default=None,
                         help="refuse to record when the 1-minute load average exceeds this "
                              "(default: a quarter of the CPU count)")
@@ -809,6 +822,18 @@ def validate_args(args: argparse.Namespace) -> list[str]:
         problems.append("--frames must be >= 1 and --warmup >= 0")
     if args.png_dir and not args.frame_hash_every:
         problems.append("--png-dir needs --frame-hash-every")
+    if args.record:
+        problems.extend(record_target_problems(Path(args.record)))
+        if not args.emulator and Path(args.core).is_file():
+            try:
+                default_emulator(Path(args.core), None)
+            except ProbeError as exc:
+                problems.append(str(exc))
+    if (args.subject or args.supersedes or args.emulator) and not args.record:
+        problems.append("--emulator, --subject and --supersedes need --record")
+    for flag, value in (("--subject", args.subject), ("--emulator", args.emulator)):
+        if value and not re.fullmatch(r"[a-z0-9]+(?:[-_][a-z0-9]+)*", value):
+            problems.append(f"{flag} {value!r}: lowercase letters, digits, - and _ only")
     if args.mode == "bench" and args.record and args.repeat < MIN_BENCH_REPETITIONS:
         problems.append(f"a recorded benchmark needs --repeat >= {MIN_BENCH_REPETITIONS}")
     if args.mode == "check":
@@ -882,12 +907,83 @@ def check_result(variants: list[dict], observable: str) -> dict:
     }
 
 
-def record_path(directory: Path, core_name: str, record: dict) -> Path:
-    canonical = json.dumps(record, sort_keys=True, separators=(",", ":")).encode()
-    digest = hashlib.sha256(canonical).hexdigest()[:12]
-    date = record["recorded_at"][:10]
-    slug = re.sub(r"[^a-z0-9]+", "_", core_name.lower()).strip("_") or "core"
-    return directory / slug / f"{date}-{digest}.json"
+def record_target_problems(directory: Path) -> list[str]:
+    """Why records may not be written under ``directory``.
+
+    Records live in their own repository. A directory inside this one's
+    tracked tree would put them in its history, megabytes per campaign;
+    an ignored directory (tmp/, the clone under .cache/) is fine.
+    """
+    target = directory.resolve()
+    if target != ROOT and ROOT not in target.parents:
+        return []
+    try:
+        ignored = subprocess.run(["git", "check-ignore", "-q", str(target)], cwd=ROOT,
+                                 capture_output=True, check=False).returncode
+    except OSError as exc:
+        return [f"--record {directory}: cannot ask git whether it is tracked ({exc})"]
+    if ignored == 0:
+        return []
+    if ignored != 1:
+        # 128: not a git checkout, nothing to keep records out of.
+        return []
+    return [f"--record {directory}: inside the tracked tree of this repository. Records go "
+            f"to the measurements repository: clone it with "
+            f"`python scripts/sync_measurements.py --fetch` and pass "
+            f"--record {MEASUREMENTS_CLONE.relative_to(ROOT)}"]
+
+
+def default_emulator(core_file: Path, library_name: str | None) -> str:
+    """The profile whose cores: names the .so, else the library name as a slug."""
+    stem = core_file.name.split(".")[0].removesuffix("_libretro")
+    sys.path.insert(0, str(ROOT / "scripts"))
+    from common import load_emulator_profiles  # noqa: PLC0415
+
+    owners = sorted(key for key, profile in load_emulator_profiles(str(ROOT / "emulators")).items()
+                    if stem == key or stem in (profile.get("cores") or []))
+    if len(owners) == 1:
+        return owners[0]
+    if len(owners) > 1:
+        raise ProbeError(f"{core_file.name}: profiles {', '.join(owners)} all name core "
+                         f"{stem}, pass --emulator")
+    slug = re.sub(r"[^a-z0-9]+", "_", (library_name or stem).lower()).strip("_")
+    return slug or "core"
+
+
+def default_subject(record: dict) -> str:
+    """A slug shared by the reruns of one test: a check by its claim and
+    method, a run or bench by its inputs."""
+    from measurement_records import canonical, sha256_hex, slugify  # noqa: PLC0415
+
+    check = record.get("check")
+    if check:
+        digest = sha256_hex(canonical([check["claim"], check["method"], check["observable"],
+                                       check["variable"]]))[:8]
+        return f"{slugify(check['claim'])}-{digest}"
+    inputs = {k: v for k, v in record["inputs"].items() if k != "repetitions"}
+    content = (record["inputs"].get("content") or {}).get("label") or "no content"
+    digest = sha256_hex(canonical([record["core"]["sha256"], inputs]))[:8]
+    return f"{record['kind']}-{slugify(content, 30)}-{digest}"
+
+
+def write_record(directory: Path, full: dict) -> tuple[Path, Path]:
+    """Append the record to its emulator's .jsonl and write its sidecar."""
+    from measurement_records import (  # noqa: PLC0415
+        line_text, parse_lines, records_path, seal, sidecar_path, split)
+
+    slim, sidecar = split(full)
+    sealed, blob = seal(slim, sidecar)
+    lines = directory / records_path(sealed["emulator"])
+    existing = parse_lines(lines.read_text(encoding="utf-8"), str(lines)) if lines.is_file() else []
+    if sealed["supersedes"] and sealed["supersedes"] not in {r["id"] for r in existing}:
+        raise ProbeError(f"--supersedes {sealed['supersedes']}: no such record in {lines}")
+    data = directory / sidecar_path(sealed["emulator"], sealed["id"])
+    data.parent.mkdir(parents=True, exist_ok=True)
+    data.write_bytes(blob)
+    lines.parent.mkdir(parents=True, exist_ok=True)
+    with lines.open("a", encoding="utf-8") as handle:
+        handle.write(line_text(sealed) + "\n")
+    return lines, data
 
 
 def probe(args: argparse.Namespace) -> int:
@@ -915,8 +1011,13 @@ def probe(args: argparse.Namespace) -> int:
         shutil.rmtree(work, ignore_errors=True)
     load_after = load_average()
     first = variants[0]["repetitions"][0]
+    from measurement_records import RECORD_VERSION  # noqa: PLC0415
+
     record: dict = {
         "record_version": RECORD_VERSION,
+        "emulator": None,
+        "subject": None,
+        "supersedes": args.supersedes,
         "kind": args.mode,
         "recorded_at": dt.datetime.now(dt.timezone.utc).replace(microsecond=0).isoformat(),
         "core": {**core_identity(args), "library_name": first.get("library_name"),
@@ -951,13 +1052,21 @@ def probe(args: argparse.Namespace) -> int:
         record["check"] = {"claim": args.claim, "method": args.method, "observable": args.observable,
                            "variable": args.vary[0][0],
                            "result": check_result(variants, args.observable)}
-    text = json.dumps(record, indent=2, sort_keys=False)
+    try:
+        record["emulator"] = args.emulator or default_emulator(Path(args.core), first.get("library_name"))
+    except ProbeError as exc:
+        print(f"ERROR: {exc}", file=sys.stderr)
+        return 2
+    record["subject"] = args.subject or default_subject(record)
     if args.record:
-        path = record_path(Path(args.record), first.get("library_name") or Path(args.core).stem, record)
-        path.parent.mkdir(parents=True, exist_ok=True)
-        path.write_text(text + "\n", encoding="utf-8")
-        print(f"recorded {path}", file=sys.stderr)
-    print(text)
+        try:
+            lines, data = write_record(Path(args.record), record)
+        except ProbeError as exc:
+            print(f"ERROR: {exc}", file=sys.stderr)
+            return 2
+        print(f"recorded {lines} (line {sum(1 for _ in lines.open(encoding='utf-8'))}) and {data}",
+              file=sys.stderr)
+    print(json.dumps(record, indent=2, sort_keys=False))
     return 0
 
 

scripts/measurement_records.py

@@ -0,0 +1,246 @@
+"""Core probe records as the measurements repository stores them.
+
+A run of core_probe.py produces one full record. It is stored as two parts:
+
+  records/<emulator>.jsonl            one slim record per line, append-only
+  data/<emulator>/<id>.json.gz        the bulky rest, named by the record id
+
+The slim record keeps what a reader needs to judge and reproduce the result:
+claim, method, observable, result, statistics, host, inputs by hash, core
+build and frontend commit, and per variant the values of its first
+repetition. The sidecar holds every repetition in full (per-frame hashes,
+core logs, declared options), the raw timing values the statistics were
+computed from, and any compared value too large to sit in a line. Its
+sha256 is in the record, so the line vouches for it. ``join`` puts the two
+back into the full record, ``split`` is its inverse.
+
+A record id is the date and the first 12 hex digits of the sha256 of the
+record without its id. A record carried over from the first layout keeps
+the id it had, which hashes that full record.
+
+Standard library only: core_probe.py, sync_measurements.py and the site
+read it.
+"""
+
+from __future__ import annotations
+
+import copy
+import gzip
+import hashlib
+import json
+import re
+from collections.abc import Iterable
+
+RECORD_VERSION = 2
+MEASUREMENTS_REPO = "https://github.com/Abdess/retrobios-measurements"
+RECORDS_DIR = "records"
+DATA_DIR = "data"
+# Fields a repetition keeps in the slim record, per variant, from the first one.
+VARIANT_SUMMARY = ("observables", "files_opened", "files_probed_absent", "tracing")
+# Observables too large for a line: their values stay in the sidecar.
+BULKY_OBSERVABLES = ("frame_hashes",)
+ID_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}-[0-9a-f]{12}$")
+SLUG_PATTERN = re.compile(r"^[a-z0-9]+(?:[-_][a-z0-9]+)*$")
+
+
+class RecordError(ValueError):
+    """A record or sidecar that does not match its counterpart."""
+
+
+def canonical(value: object) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
+
+
+def sha256_hex(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def slugify(text: str, limit: int = 40) -> str:
+    """Lowercase words joined by hyphens, cut at a word boundary."""
+    words = re.findall(r"[a-z0-9]+", text.lower())
+    out = ""
+    for word in words:
+        candidate = f"{out}-{word}" if out else word
+        if len(candidate) > limit:
+            break
+        out = candidate
+    return out or "record"
+
+
+def record_id(slim: dict) -> str:
+    body = {key: value for key, value in slim.items() if key != "id"}
+    return f"{slim['recorded_at'][:10]}-{sha256_hex(canonical(body))[:12]}"
+
+
+def line_text(slim: dict) -> str:
+    """The record as one line of its .jsonl file, keys in a fixed order."""
+    return json.dumps(slim, sort_keys=True, separators=(",", ":"), ensure_ascii=False)
+
+
+def sidecar_path(emulator: str, rid: str) -> str:
+    return f"{DATA_DIR}/{emulator}/{rid}.json.gz"
+
+
+def records_path(emulator: str) -> str:
+    return f"{RECORDS_DIR}/{emulator}.jsonl"
+
+
+def sidecar_bytes(sidecar: dict) -> bytes:
+    """gzip of the sidecar JSON, with no name and no time in the header."""
+    return gzip.compress(canonical(sidecar), compresslevel=9, mtime=0)
+
+
+def read_sidecar(blob: bytes) -> dict:
+    return json.loads(gzip.decompress(blob))
+
+
+def _value_digest(value: object) -> str:
+    return hashlib.sha1(canonical(value)).hexdigest()
+
+
+def split(full: dict) -> tuple[dict, dict]:
+    """The full record as (slim record without id and data, sidecar)."""
+    slim = copy.deepcopy(full)
+    sidecar: dict = {"variants": [], "timing_raw": {}, "group_values": {}}
+    variants = []
+    for variant in slim["variants"]:
+        reps = variant["repetitions"]
+        sidecar["variants"].append({"label": variant["label"], "repetitions": reps})
+        first = reps[0]
+        summary = {key: first.get(key) for key in VARIANT_SUMMARY}
+        summary["observables"] = {
+            key: value for key, value in (first.get("observables") or {}).items()
+            if key not in BULKY_OBSERVABLES
+        }
+        variants.append({
+            "label": variant["label"],
+            "system_files": variant["system_files"],
+            "flags": variant["flags"],
+            "repetitions": len(reps),
+            "loaded": all(rep.get("loaded") for rep in reps),
+            **summary,
+        })
+    slim["variants"] = variants
+    for name, stats in (slim.get("timing") or {}).items():
+        sidecar["timing_raw"][name] = stats.pop("raw")
+    for group in ((slim.get("check") or {}).get("result") or {}).get("groups", []):
+        if isinstance(group.get("value"), (list, dict)):
+            digest = _value_digest(group["value"])
+            sidecar["group_values"][digest] = group.pop("value")
+            group["value_sha1"] = digest
+    return slim, sidecar
+
+
+def join(slim: dict, sidecar: dict) -> dict:
+    """The full record a slim record and its sidecar were split from."""
+    full = copy.deepcopy(slim)
+    full.pop("id", None)
+    full.pop("data", None)
+    labels = [v["label"] for v in full["variants"]]
+    if labels != [v["label"] for v in sidecar["variants"]]:
+        raise RecordError(f"variants differ: record {labels}, sidecar "
+                          f"{[v['label'] for v in sidecar['variants']]}")
+    full["variants"] = [
+        {"label": variant["label"], "system_files": variant["system_files"],
+         "repetitions": copy.deepcopy(stored["repetitions"]), "flags": variant["flags"]}
+        for variant, stored in zip(full["variants"], sidecar["variants"])
+    ]
+    for name, raw in sidecar.get("timing_raw", {}).items():
+        full["timing"][name]["raw"] = list(raw)
+    for group in ((full.get("check") or {}).get("result") or {}).get("groups", []):
+        digest = group.pop("value_sha1", None)
+        if digest is not None:
+            group["value"] = copy.deepcopy(sidecar["group_values"][digest])
+    return full
+
+
+def seal(slim: dict, sidecar: dict, rid: str | None = None) -> tuple[dict, bytes]:
+    """The slim record with its data reference and id, and the sidecar bytes.
+
+    ``rid`` keeps an id given before (a record carried over from the first
+    layout); otherwise the id is computed.
+    """
+    blob = sidecar_bytes(sidecar)
+    sealed = {**slim, "data": {"sha256": sha256_hex(blob), "bytes": len(blob)}}
+    sealed["id"] = rid or record_id(sealed)
+    return sealed, blob
+
+
+def check_pair(slim: dict, blob: bytes) -> list[str]:
+    """What is wrong between a stored line and its stored sidecar."""
+    problems = []
+    data = slim.get("data") or {}
+    if sha256_hex(blob) != data.get("sha256"):
+        problems.append(f"{slim.get('id')}: sidecar sha256 {sha256_hex(blob)} != {data.get('sha256')}")
+    if len(blob) != data.get("bytes"):
+        problems.append(f"{slim.get('id')}: sidecar is {len(blob)} bytes, record says {data.get('bytes')}")
+    if not problems:
+        try:
+            full = join(slim, read_sidecar(blob))
+        except (RecordError, KeyError, OSError, ValueError) as exc:
+            return [f"{slim.get('id')}: record and sidecar do not join: {exc}"]
+        if slim.get("migrated_from_version") == 1:
+            digest = sha256_hex(canonical(first_layout(full)))[:12]
+            if str(slim.get("id", "")).rsplit("-", 1)[-1] != digest:
+                problems.append(f"{slim.get('id')}: id does not hash the first-layout record ({digest})")
+    return problems
+
+
+V2_IDENTITY = ("emulator", "subject", "supersedes", "migrated_from_version")
+
+
+def first_layout(full: dict) -> dict:
+    """A full record as the first layout wrote it: no emulator, subject or
+    supersedes, record_version 1."""
+    old = {key: value for key, value in full.items() if key not in V2_IDENTITY}
+    old["record_version"] = 1
+    return old
+
+
+def check_line(slim: dict) -> list[str]:
+    """The structural rules a line holds, without a JSON Schema validator."""
+    rid = slim.get("id", "")
+    problems = []
+    if slim.get("record_version") != RECORD_VERSION:
+        problems.append(f"{rid}: record_version {slim.get('record_version')} != {RECORD_VERSION}")
+    if not ID_PATTERN.match(str(rid)):
+        problems.append(f"{rid!r}: not an id of the form YYYY-MM-DD-<12 hex>")
+    elif str(slim.get("recorded_at", ""))[:10] != rid[:10]:
+        problems.append(f"{rid}: id date differs from recorded_at {slim.get('recorded_at')}")
+    elif slim.get("migrated_from_version") is None and record_id(slim) != rid:
+        problems.append(f"{rid}: id does not hash the record ({record_id(slim)})")
+    for key in ("emulator", "subject"):
+        if not SLUG_PATTERN.match(str(slim.get(key, ""))):
+            problems.append(f"{rid}: {key} {slim.get(key)!r} is not a lowercase slug")
+    return problems
+
+
+def parse_lines(text: str, label: str) -> list[dict]:
+    out = []
+    for number, line in enumerate(text.splitlines(), 1):
+        if not line.strip():
+            raise RecordError(f"{label}:{number}: empty line")
+        try:
+            out.append(json.loads(line))
+        except json.JSONDecodeError as exc:
+            raise RecordError(f"{label}:{number}: {exc}") from exc
+    return out
+
+
+def appended_only(before: str, after: str) -> str | None:
+    """None when ``after`` keeps every line of ``before``, in place, and
+    only adds lines at its end; otherwise the first line that moved."""
+    old, new = before.splitlines(), after.splitlines()
+    for number, line in enumerate(old, 1):
+        if number > len(new):
+            return f"line {number} removed"
+        if new[number - 1] != line:
+            return f"line {number} changed"
+    return None
+
+
+def live(records: Iterable[dict]) -> list[dict]:
+    """The records no later record supersedes."""
+    records = list(records)
+    superseded = {r["supersedes"] for r in records if r.get("supersedes")}
+    return [r for r in records if r["id"] not in superseded]