From 00eda4aed155250f3378a467847f9bf14c151d84 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Tue, 26 May 2026 16:19:54 -0400 Subject: [PATCH] bench: cold-pack producer/consumer roundtrip recorder (#000061 + #46) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Self-contained benchmark for the SPV-wallet validation. Records: PRODUCER bucket state + pack count + compressed bytes CONSUMER wall time, exit status, post-hydrate shard sizes, per-shard documents/chunks/edges counts Driver runs against the real DO Spaces bucket + the real fresh peer on 3090-ai.foxhop.net. Writes one JSON artifact per run to bench/results/cold-pack-roundtrip-.json so a future operator can diff hydrate times across pack-format changes (#000061 v3 → graft mode #000066 → mesh-pull future). Consumer command uses /usr/bin/time -v wrapped around `arborist cold unpack --hydrate-shards-dir … --hydrate-M 4 --full`. Hydrates into ~/.arborist/shards-genesis-test/ so it doesn't clobber anything on the 3090. No new ticket — this is task #45/#46 instrumentation. Existing tests untouched. --- bench/cold_pack_roundtrip.py | 171 +++++++++++++++++++++++++++++++++++ 1 file changed, 171 insertions(+) create mode 100644 bench/cold_pack_roundtrip.py diff --git a/bench/cold_pack_roundtrip.py b/bench/cold_pack_roundtrip.py new file mode 100644 index 0000000..af1611f --- /dev/null +++ b/bench/cold_pack_roundtrip.py @@ -0,0 +1,171 @@ +"""Producer/consumer benchmark for the cold-pack distribution tier +(#000061) running against a real DO Spaces bucket and a real fresh +peer on `3090-ai.foxhop.net`. + +Records timings + byte counts in both directions: + + PRODUCER (this host) + pack wall time, compressed MB/s, packs created, bucket bytes + CONSUMER (3090) + hydrate wall time, downloaded MB/s, rows restored per table, + shard sizes on disk, post-hydrate row count match + +Result lands at ``bench/results/cold-pack-roundtrip-.json`` +so a future operator (or follow-on ticket re-running this benchmark +after pack-format changes) has a single artifact to diff against. + +Run after pack has been produced + bucket has packs to pull. See the +`if __name__ == "__main__"` driver for the exact sequence. + +Read-only on the producer host (just queries process state + bucket +stats). On the consumer host: SSH'd command runs `arborist cold +unpack --hydrate-shards-dir ~/.arborist/shards --hydrate-M 4 --full`. +""" +from __future__ import annotations + +import json +import os +import shlex +import subprocess +import time +from pathlib import Path + +RESULTS_DIR = Path(__file__).resolve().parent / "results" +SHARDS_DIR = Path("/home/fox/.arborist/shards") +COLD_ENV = { + "ARBORIST_COLD_ENDPOINT_URL": "https://nyc3.digitaloceanspaces.com", + "ARBORIST_COLD_BUCKET": "arborist", +} +ARBORIST = "/home/fox/git/arborist/.venv/bin/arborist" +CONSUMER_HOST = "3090-ai.foxhop.net" +CONSUMER_ARBORIST = "/home/fox/git/arborist/.venv/bin/arborist" + + +def _cold_stats_local() -> dict: + """Call `arborist cold stats` on this host; return parsed JSON.""" + env = {**os.environ, **COLD_ENV} + out = subprocess.check_output( + [ARBORIST, "cold", "stats"], env=env, text=True + ) + return json.loads(out) + + +def _cold_list_local() -> dict: + env = {**os.environ, **COLD_ENV} + out = subprocess.check_output( + [ARBORIST, "cold", "list", "--no-manifest"], env=env, text=True + ) + return json.loads(out) + + +def _identify_metadata_pack_hash() -> str: + """Read manifest/latest.json from the bucket to find the active + metadata pack hash. Fresh peer would do the same on genesis.""" + env = {**os.environ, **COLD_ENV} + # Use `arborist cold list` (it sorts manifest packs first). + listing = _cold_list_local() + for entry in listing["packs"]: + if entry.get("kind") == "metadata": + return entry["pack_hash"] + raise RuntimeError("no metadata pack found in bucket") + + +def _ssh(cmd: str) -> subprocess.CompletedProcess: + """Run a shell command on the consumer host. Returns CompletedProcess.""" + return subprocess.run( + ["ssh", "-o", "BatchMode=yes", CONSUMER_HOST, cmd], + capture_output=True, text=True, + ) + + +def measure_consumer(metadata_pack_hash: str, *, M: int = 4) -> dict: + """Run the hydrate-into-empty path on 3090, time it, return metrics.""" + started_at = time.time() + cmd = " && ".join([ + # Belt-and-suspenders: ensure no stale shards exist. + "rm -rf ~/.arborist/shards-genesis-test", + "mkdir -p ~/.arborist/shards-genesis-test", + # Run hydrate; pipe output to a side log so we can diff later. + "/usr/bin/time -v " + f"{CONSUMER_ARBORIST} cold unpack {shlex.quote(metadata_pack_hash)} " + f"--hydrate-shards-dir ~/.arborist/shards-genesis-test " + f"--hydrate-M {M} --full " + "2>&1 | tee /tmp/cold-genesis.log", + ]) + result = _ssh(cmd) + elapsed = time.time() - started_at + # Capture final shard sizes from the consumer. + sizes_raw = _ssh( + "ls -la ~/.arborist/shards-genesis-test/*.db 2>/dev/null | " + "awk '{print $NF, $5}'" + ).stdout + shard_sizes: dict[str, int] = {} + for line in sizes_raw.strip().splitlines(): + if not line.strip(): + continue + parts = line.split() + shard_sizes[Path(parts[0]).name] = int(parts[1]) + # And row counts per shard. + row_counts: dict[str, dict[str, int]] = {} + for i in range(M): + cmd = ( + "python3 -c \"import sqlite3; " + "c=sqlite3.connect('~/.arborist/shards-genesis-test/{:03d}.db'.expanduser()) " + "if hasattr(str,'expanduser') else " + "sqlite3.connect('/home/fox/.arborist/shards-genesis-test/{:03d}.db'); " + "print(c.execute('SELECT COUNT(*) FROM documents').fetchone()[0], " + "c.execute('SELECT COUNT(*) FROM chunks').fetchone()[0], " + "c.execute('SELECT COUNT(*) FROM edges').fetchone()[0])\"" + ).format(i, i) + r = _ssh(cmd) + parts = r.stdout.strip().split() + if len(parts) >= 3: + row_counts[f"{i:03d}.db"] = { + "documents": int(parts[0]), + "chunks": int(parts[1]), + "edges": int(parts[2]), + } + return { + "elapsed_seconds": elapsed, + "ssh_exit": result.returncode, + "stdout_tail": result.stdout[-2000:] if result.stdout else "", + "stderr_tail": result.stderr[-2000:] if result.stderr else "", + "shard_sizes_bytes": shard_sizes, + "row_counts_by_shard": row_counts, + } + + +def main(producer_started_at: float | None = None) -> int: + """Driver: read producer-side stats, run consumer benchmark, save JSON.""" + finished_at = time.time() + bucket = _cold_stats_local() + pack_hash = _identify_metadata_pack_hash() + print(f"metadata pack: {pack_hash[:16]}…") + print(f"bucket: {bucket['packs']} packs / " + f"{bucket['packs_compressed_bytes'] / 1e9:.2f} GB") + print(f"running consumer on {CONSUMER_HOST} …") + consumer = measure_consumer(pack_hash, M=4) + + out = { + "captured_at_utc": time.strftime( + "%Y-%m-%dT%H:%M:%SZ", time.gmtime() + ), + "producer_host": os.uname().nodename, + "consumer_host": CONSUMER_HOST, + "metadata_pack_hash": pack_hash, + "bucket": bucket, + "consumer": consumer, + } + RESULTS_DIR.mkdir(parents=True, exist_ok=True) + out_path = ( + RESULTS_DIR + / f"cold-pack-roundtrip-{time.strftime('%Y-%m-%dT%H-%M-%SZ', time.gmtime())}.json" + ) + out_path.write_text(json.dumps(out, indent=2)) + print(f"wrote {out_path}") + print(f"consumer elapsed: {consumer['elapsed_seconds']:.1f}s") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())