"""Validate all recorded runs before generating public tables and downloads.""" import argparse import csv import hashlib import io import json from pathlib import Path import shutil from experiment import PAYLOAD, digest, verify_case ROOT = Path(__file__).resolve().parent RUN_FILES = ("results-1.json", "results-2.json", "review-run.json") LABELS = { "row_count": "Row count", "byte_length": "Byte length", "prefix": "First 32 bytes", "whole_sha256": "Full-value SHA-256", "heap_without_toast": "Heap check without TOAST", "heap_with_toast": "Heap check with TOAST", "amcheck_default": "pg_amcheck: both routes", "amcheck_no_pointers": "pg_amcheck: dependent relation only", "amcheck_no_dependent_toast": "pg_amcheck: pointers only", "amcheck_neither": "pg_amcheck: neither route", } FILES = ("experiment.py", "test_experiment.py", "verify.py", "summary.json", "source-review.md", "README.md", "publication-README.md", "publication.json", "build-publication.py", "test_publication.py", *RUN_FILES) def derive(runs): assert len(runs) == 3 harness = digest((ROOT / "experiment.py").read_bytes()) rows = [] reference = runs[0] for repeat, result in enumerate(runs, 1): assert result["harness_sha256"] == harness assert result["fixture"] == reference["fixture"] assert result["fixture"]["expected_sha256"] == digest(PAYLOAD) assert [c["checksums"] for c in result["cases"]] == [True, False] for i, case in enumerate(result["cases"]): verify_case(case) assert case["settings"]["checksums"] == ("on" if case["checksums"] else "off") assert case["settings"]["ignore_checksum_failure"] == "off" assert case["settings"]["block_size"] == "8192" assert case["toast_chunks"] == 13 mutation = case["mutation"] assert mutation["before"] ^ mutation["after"] == 1 assert mutation["changed_bytes"] == 1 and mutation["xor_mask"] == 1 assert mutation["byte_offset"] == 32716 and mutation["block"] == 3 assert mutation["relation_bytes"] == 32768 assert mutation["checksum_recomputed"] is False for state in ("clean", "damaged"): probes = case[state]["probes"] assert set(probes) == set(LABELS) for name, probe in probes.items(): prior = reference["cases"][i][state]["probes"][name] assert probe["returncode"] == prior["returncode"] if probe["returncode"] == 0: assert probe["stdout"] == prior["stdout"] match = None if name == "whole_sha256" and probe["returncode"] == 0: match = probe["stdout"] == digest(PAYLOAD) rows.append(dict(repetition=repeat, checksums=case["checksums"], state=state, phase="online", probe=name, returncode=probe["returncode"], matches_expected=match)) offline = case[state]["offline_checksums"] if not case["checksums"]: assert offline["returncode"] != 0 and "not enabled" in offline["stderr"] elif state == "clean": assert offline["returncode"] == 0 else: assert offline["returncode"] != 0 and "Bad checksums: 1" in offline["stdout"] rows.append(dict(repetition=repeat, checksums=case["checksums"], state=state, phase="offline", probe="pg_checksums", returncode=offline["returncode"], matches_expected=None)) matrix = [] for name, label in LABELS.items(): cells = {} for case in reference["cases"]: probe = case["damaged"]["probes"][name] if probe["returncode"]: outcome = "Read/check error" elif name == "whole_sha256": outcome = "Different digest" elif name == "row_count": outcome = "1 row" elif name == "byte_length": outcome = "24,045 bytes" elif name == "prefix": outcome = "Unchanged prefix" elif name.startswith("heap_"): outcome = "No report rows" else: outcome = "Exit 0" cells["on" if case["checksums"] else "off"] = outcome matrix.append(dict(probe=name, label=label, **cells)) matrix.append(dict(probe="pg_checksums", label="Offline page checksums", on="One bad checksum", off="Refuses: not enabled")) return rows, matrix def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--public-root", type=Path, required=True) parser.add_argument("--surface", choices=("com", "dev"), required=True) args = parser.parse_args() if not __debug__: parser.error("Run without -O") runs = [json.loads((ROOT / name).read_text()) for name in RUN_FILES] rows, matrix = derive(runs) pub = json.loads((ROOT / "publication.json").read_text()) data = dict(schemaVersion=1, **pub, evidenceType="measured-api-behavior", generatedAt=max(r["generated_at"] for r in runs), executions=len(runs), freshClusters=sum(len(r["cases"]) for r in runs), rowCount=len(rows), onlineProbes=sum(r["phase"] == "online" for r in rows), offlineInvocations=sum(r["phase"] == "offline" for r in rows), matrix=matrix, environment=runs[-1]["environment"], license="CC0-1.0 for original Hesela code and synthetic observations; upstream sources retain their licenses.", licenseUrl="https://creativecommons.org/publicdomain/zero/1.0/", humanPage="https://hesela.com/analyses/postgresql-toast-read-coverage/", sourceUrl="https://hesela.com/analyses/postgresql-toast-read-coverage/", downloadPath="/data/datasets/postgresql-toast-read-coverage.csv", rawPath="/research/postgresql-toast-coverage/results-1.json", codePath="/research/postgresql-toast-coverage/experiment.py", generatorPath="/research/postgresql-toast-coverage/build-publication.py", columns=[dict(name=name, type=kind, description=description) for name, kind, description in ( ("repetition", "integer", "Execution 1 or 2; 3 is the review rerun using the same harness."), ("checksums", "boolean", "Checksum setting in a separately initialized cluster."), ("state", "string", "Clean fixture or the subsequent one-bit damaged fixture."), ("phase", "string", "Online probe or offline checksum utility."), ("probe", "string", "Stable probe identifier; raw JSON contains SQL/options."), ("returncode", "integer", "Process exit code, not an application-content verdict."), ("matches_expected", "boolean-or-null", "Only successful full-value hashes: true/false comparison with the trusted input; blank otherwise.") )], sha256={name: hashlib.sha256((ROOT / name).read_bytes()).hexdigest() for name in FILES}) public = args.public_root.resolve() if args.surface == "com": target = public / "research/postgresql-toast-publication.json" target.parent.mkdir(parents=True, exist_ok=True) target.write_text(json.dumps(data, indent=2) + "\n") else: research = public / "research/postgresql-toast-coverage" research.mkdir(parents=True, exist_ok=True) for name in FILES: shutil.copyfile(ROOT / name, research / name) target = public / "data/datasets" target.mkdir(parents=True, exist_ok=True) (target / (pub["slug"] + ".json")).write_text(json.dumps(data, indent=2) + "\n") buffer = io.StringIO() writer = csv.DictWriter(buffer, fieldnames=list(rows[0]), lineterminator="\n") writer.writeheader() writer.writerows({key: str(value).lower() if isinstance(value, bool) else value for key, value in row.items()} for row in rows) (target / (pub["slug"] + ".csv")).write_text(buffer.getvalue()) print(json.dumps({key: data[key] for key in ("rowCount", "onlineProbes", "offlineInvocations", "freshClusters")})) if __name__ == "__main__": main()