"""Push timing diagnostic — scale ladder against staging. Sends synthetic pushes at increasing scale and reports timing per phase. Use this to find exactly where the wall is. Usage: python3 scripts/push_timing_test.py # default scale ladder python3 scripts/push_timing_test.py 500 # single: 500 objects python3 scripts/push_timing_test.py 500 827 # 500 objects + 827 commits python3 scripts/push_timing_test.py --target local # vs localhost Each run reports: T_connect — TCP + TLS handshake T_first — time from request sent to first response byte T_total — time from connection to last response byte ok — did the push succeed? """ from __future__ import annotations import asyncio import hashlib import os import sys import time from pathlib import Path import httpx import msgpack sys.path.insert(0, str(Path.home() / "ecosystem" / "muse")) sys.path.insert(0, str(Path.home() / "ecosystem" / "musehub")) MKCERT_CA = str(Path.home() / "Library/Application Support/mkcert/rootCA.pem") TARGETS = { "staging": ("https://staging.musehub.ai", True), "local": ("https://localhost:1337", MKCERT_CA), } target_name = "staging" for arg in sys.argv[1:]: if arg.startswith("--target="): target_name = arg.split("=", 1)[1] elif arg == "--target" and sys.argv.index(arg) + 1 < len(sys.argv): target_name = sys.argv[sys.argv.index(arg) + 1] BASE_URL, SSL_VERIFY = TARGETS.get(target_name, TARGETS["staging"]) OWNER = "gabriel" REPO = "timing-test" ROUTE = f"/{OWNER}/{REPO}/push/stream" WIRE_CONTENT_TYPE = "application/x-muse-wire" def _oid(data: bytes) -> str: return "sha256:" + hashlib.sha256(data).hexdigest() def _pack_frame(frame_type: str, payload: dict) -> bytes: from muse.core.mpack import MuseWireFrameWriter fw = MuseWireFrameWriter() return fw.wrap(frame_type=frame_type, payload=msgpack.packb(payload, use_bin_type=True)) def _header_frame(n_objects: int, n_commits: int) -> bytes: return _pack_frame("H", { "t": "H", "branch": "main", "force": True, "have": [], "head": "", "n_objects": n_objects, "n_commits": n_commits, }) def _object_frame(size_bytes: int, index: int) -> tuple[str, bytes]: raw = os.urandom(max(64, size_bytes)) if raw[:2] == b"#!": raw = b"\x00" + raw[1:] oid = _oid(raw) frame = _pack_frame("O", { "t": "O", "id": oid, "path": f"file_{index:04d}.dat", "enc": "raw", "content": raw, }) return oid, frame def _content_snapshot_id(manifest: dict) -> str: _SEP = "\x00" parts = sorted( f"{path}{_SEP}{oid.split(':', 1)[1]}" for path, oid in manifest.items() ) return _oid(_SEP.join(parts).encode()) def _commit_frame(n_commits: int, object_ids: list[str]) -> bytes: commits = [] prev_id = "" commit_ids = [_oid(os.urandom(32)) for _ in range(n_commits)] manifests = [] for i in range(n_commits): manifest: dict[str, str] = {} if object_ids: obj_idx = i % len(object_ids) manifest[f"file_{obj_idx:04d}.dat"] = object_ids[obj_idx] manifests.append(manifest) snap_ids = [_content_snapshot_id(m) for m in manifests] for i in range(n_commits): cid = commit_ids[i] commits.append({ "commit_id": cid, "parent_commit_id": prev_id if prev_id else None, "snapshot_id": snap_ids[i], "branch": "main", "message": f"timing test commit {i}", "author": OWNER, "committed_at": "2026-04-28T00:00:00+00:00", "signature": "", "signer_key_id": "", "agent_id": "timing-test", "model_id": "n/a", "metadata": {}, }) prev_id = cid snapshots = [ {"snapshot_id": snap_ids[i], "manifest": manifests[i]} for i in range(n_commits) ] return _pack_frame("C", {"t": "C", "commits": commits, "snapshots": snapshots}) def _end_frame(n_objects: int, n_commits: int) -> bytes: return _pack_frame("E", {"t": "E", "n_objects": n_objects, "n_commits": n_commits}) def _auth_header(route: str) -> str: from muse.cli.config import get_signing_identity from muse.core.msign import build_msign_header s = get_signing_identity(remote_url=BASE_URL) if s is None: raise RuntimeError(f"No signing identity for {BASE_URL}") return build_msign_header(s, "POST", f"{BASE_URL}{route}", b"") async def run_push(n_objects: int, n_commits: int, obj_size_bytes: int = 512, *, label: str = "") -> dict: object_ids: list[str] = [] obj_frames: list[bytes] = [] for i in range(n_objects): oid, frame = _object_frame(obj_size_bytes, i) object_ids.append(oid) obj_frames.append(frame) body = ( _header_frame(n_objects, n_commits) + b"".join(obj_frames) + (_commit_frame(n_commits, object_ids) if n_commits else b"") + _end_frame(n_objects, n_commits) ) headers = { "Content-Type": WIRE_CONTENT_TYPE, "Accept": WIRE_CONTENT_TYPE, "Authorization": _auth_header(ROUTE), } result: dict = { "label": label or f"{n_objects}obj+{n_commits}commit", "n_objects": n_objects, "n_commits": n_commits, "body_kb": round(len(body) / 1024, 1), "t_connect": None, "t_first_byte": None, "t_total": None, "status": None, "ok": None, "cf524": False, "error": None, "progress_frames": [], } try: t0 = time.perf_counter() t_first = None chunks = [] async with httpx.AsyncClient( http2=False, timeout=300.0, verify=SSL_VERIFY, limits=httpx.Limits(max_keepalive_connections=0), ) as client: async with client.stream("POST", f"{BASE_URL}{ROUTE}", content=body, headers=headers) as resp: result["t_connect"] = round((time.perf_counter() - t0) * 1000, 1) result["status"] = resp.status_code if resp.status_code == 524: result["cf524"] = True result["error"] = "CF 524 timeout" return result async for chunk in resp.aiter_bytes(): if t_first is None: t_first = time.perf_counter() - t0 result["t_first_byte"] = round(t_first * 1000, 1) chunks.append(chunk) result["t_total"] = round((time.perf_counter() - t0) * 1000, 1) unpacker = msgpack.Unpacker(raw=False) unpacker.feed(b"".join(chunks)) for frame in unpacker: t = frame.get("t") if t == "P": result["progress_frames"].append(frame.get("msg", "")) elif t == "R": result["ok"] = frame.get("ok") elif t == "X": result["ok"] = False result["error"] = frame.get("msg", "error frame") except httpx.ReadTimeout: result["error"] = "client timeout (>300s)" result["t_total"] = 300_000 except Exception as exc: result["error"] = str(exc)[:120] return result def _fmt(r: dict) -> str: ok_str = "✅" if r["ok"] else ("❌" if r["ok"] is False else "?") cf_str = " ⚠️ CF524" if r["cf524"] else "" err_str = f" ERROR: {r['error']}" if r["error"] else "" p_str = f" [{len(r['progress_frames'])} P-frames]" if r["progress_frames"] else "" return ( f" {r['label']:30s} {r['body_kb']:8.1f} KB " f"connect={str(r['t_connect'] or '?'):>6}ms " f"first={str(r['t_first_byte'] or '?'):>7}ms " f"total={str(r['t_total'] or '?'):>8}ms " f"{ok_str}{cf_str}{p_str}{err_str}" ) async def main() -> None: args = [a for a in sys.argv[1:] if not a.startswith("--target")] if len(args) == 1: scales = [(int(args[0]), 1)] elif len(args) == 2: scales = [(int(args[0]), int(args[1]))] else: scales = [ (1, 1), (10, 1), (50, 1), (100, 1), (500, 1), (500, 100), (500, 500), (500, 827), (1000, 827), ] print(f"\nPush timing test → {BASE_URL}{ROUTE}") print(f" {'label':30s} {'body':>8} {'connect':>10} {'first':>10} {'total':>12} status") print(" " + "-" * 100) for n_obj, n_com in scales: r = await run_push(n_obj, n_com, label=f"{n_obj}obj+{n_com}commit") print(_fmt(r)) sys.stdout.flush() if r["cf524"]: print(f"\n⚠️ CF 524 fired at {n_obj} objects. This is the wall.") break if r["t_total"] and r["t_total"] > 90_000: print(f"\n⚠️ Batch took {r['t_total']}ms (>90s). Next scale will likely 524.") break print() if __name__ == "__main__": asyncio.run(main())