push_timing_test.py
python
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef
debug(push/stream): instrument O-frame decode path with INF…
Sonnet 4.6
patch
121 days ago
| 1 | """Push timing diagnostic — scale ladder against staging. |
| 2 | |
| 3 | Sends synthetic pushes at increasing scale and reports timing per phase. |
| 4 | Use this to find exactly where the wall is. |
| 5 | |
| 6 | Usage: |
| 7 | python3 scripts/push_timing_test.py # default scale ladder |
| 8 | python3 scripts/push_timing_test.py 500 # single: 500 objects |
| 9 | python3 scripts/push_timing_test.py 500 827 # 500 objects + 827 commits |
| 10 | python3 scripts/push_timing_test.py --target local # vs localhost |
| 11 | |
| 12 | Each run reports: |
| 13 | T_connect — TCP + TLS handshake |
| 14 | T_first — time from request sent to first response byte |
| 15 | T_total — time from connection to last response byte |
| 16 | ok — did the push succeed? |
| 17 | """ |
| 18 | from __future__ import annotations |
| 19 | |
| 20 | import asyncio |
| 21 | import hashlib |
| 22 | import os |
| 23 | import sys |
| 24 | import time |
| 25 | from pathlib import Path |
| 26 | |
| 27 | import httpx |
| 28 | import msgpack |
| 29 | |
| 30 | sys.path.insert(0, str(Path.home() / "ecosystem" / "muse")) |
| 31 | sys.path.insert(0, str(Path.home() / "ecosystem" / "musehub")) |
| 32 | |
| 33 | MKCERT_CA = str(Path.home() / "Library/Application Support/mkcert/rootCA.pem") |
| 34 | |
| 35 | TARGETS = { |
| 36 | "staging": ("https://staging.musehub.ai", True), |
| 37 | "local": ("https://localhost:1337", MKCERT_CA), |
| 38 | } |
| 39 | |
| 40 | target_name = "staging" |
| 41 | for arg in sys.argv[1:]: |
| 42 | if arg.startswith("--target="): |
| 43 | target_name = arg.split("=", 1)[1] |
| 44 | elif arg == "--target" and sys.argv.index(arg) + 1 < len(sys.argv): |
| 45 | target_name = sys.argv[sys.argv.index(arg) + 1] |
| 46 | |
| 47 | BASE_URL, SSL_VERIFY = TARGETS.get(target_name, TARGETS["staging"]) |
| 48 | OWNER = "gabriel" |
| 49 | REPO = "timing-test" |
| 50 | ROUTE = f"/{OWNER}/{REPO}/push/stream" |
| 51 | WIRE_CONTENT_TYPE = "application/x-muse-wire" |
| 52 | |
| 53 | |
| 54 | def _oid(data: bytes) -> str: |
| 55 | return "sha256:" + hashlib.sha256(data).hexdigest() |
| 56 | |
| 57 | |
| 58 | def _pack_frame(frame_type: str, payload: dict) -> bytes: |
| 59 | from muse.core.mpack import MuseWireFrameWriter |
| 60 | fw = MuseWireFrameWriter() |
| 61 | return fw.wrap(frame_type=frame_type, payload=msgpack.packb(payload, use_bin_type=True)) |
| 62 | |
| 63 | |
| 64 | def _header_frame(n_objects: int, n_commits: int) -> bytes: |
| 65 | return _pack_frame("H", { |
| 66 | "t": "H", "branch": "main", "force": True, |
| 67 | "have": [], "head": "", "n_objects": n_objects, "n_commits": n_commits, |
| 68 | }) |
| 69 | |
| 70 | |
| 71 | def _object_frame(size_bytes: int, index: int) -> tuple[str, bytes]: |
| 72 | raw = os.urandom(max(64, size_bytes)) |
| 73 | if raw[:2] == b"#!": |
| 74 | raw = b"\x00" + raw[1:] |
| 75 | oid = _oid(raw) |
| 76 | frame = _pack_frame("O", { |
| 77 | "t": "O", "id": oid, "path": f"file_{index:04d}.dat", |
| 78 | "enc": "raw", "content": raw, |
| 79 | }) |
| 80 | return oid, frame |
| 81 | |
| 82 | |
| 83 | def _content_snapshot_id(manifest: dict) -> str: |
| 84 | _SEP = "\x00" |
| 85 | parts = sorted( |
| 86 | f"{path}{_SEP}{oid.split(':', 1)[1]}" for path, oid in manifest.items() |
| 87 | ) |
| 88 | return _oid(_SEP.join(parts).encode()) |
| 89 | |
| 90 | |
| 91 | def _commit_frame(n_commits: int, object_ids: list[str]) -> bytes: |
| 92 | commits = [] |
| 93 | prev_id = "" |
| 94 | commit_ids = [_oid(os.urandom(32)) for _ in range(n_commits)] |
| 95 | |
| 96 | manifests = [] |
| 97 | for i in range(n_commits): |
| 98 | manifest: dict[str, str] = {} |
| 99 | if object_ids: |
| 100 | obj_idx = i % len(object_ids) |
| 101 | manifest[f"file_{obj_idx:04d}.dat"] = object_ids[obj_idx] |
| 102 | manifests.append(manifest) |
| 103 | |
| 104 | snap_ids = [_content_snapshot_id(m) for m in manifests] |
| 105 | |
| 106 | for i in range(n_commits): |
| 107 | cid = commit_ids[i] |
| 108 | commits.append({ |
| 109 | "commit_id": cid, |
| 110 | "parent_commit_id": prev_id if prev_id else None, |
| 111 | "snapshot_id": snap_ids[i], |
| 112 | "branch": "main", |
| 113 | "message": f"timing test commit {i}", |
| 114 | "author": OWNER, |
| 115 | "committed_at": "2026-04-28T00:00:00+00:00", |
| 116 | "signature": "", "signer_key_id": "", |
| 117 | "agent_id": "timing-test", "model_id": "n/a", "metadata": {}, |
| 118 | }) |
| 119 | prev_id = cid |
| 120 | |
| 121 | snapshots = [ |
| 122 | {"snapshot_id": snap_ids[i], "manifest": manifests[i]} |
| 123 | for i in range(n_commits) |
| 124 | ] |
| 125 | return _pack_frame("C", {"t": "C", "commits": commits, "snapshots": snapshots}) |
| 126 | |
| 127 | |
| 128 | def _end_frame(n_objects: int, n_commits: int) -> bytes: |
| 129 | return _pack_frame("E", {"t": "E", "n_objects": n_objects, "n_commits": n_commits}) |
| 130 | |
| 131 | |
| 132 | def _auth_header(route: str) -> str: |
| 133 | from muse.cli.config import get_signing_identity |
| 134 | from muse.core.msign import build_msign_header |
| 135 | s = get_signing_identity(remote_url=BASE_URL) |
| 136 | if s is None: |
| 137 | raise RuntimeError(f"No signing identity for {BASE_URL}") |
| 138 | return build_msign_header(s, "POST", f"{BASE_URL}{route}", b"") |
| 139 | |
| 140 | |
| 141 | async def run_push(n_objects: int, n_commits: int, obj_size_bytes: int = 512, *, label: str = "") -> dict: |
| 142 | object_ids: list[str] = [] |
| 143 | obj_frames: list[bytes] = [] |
| 144 | for i in range(n_objects): |
| 145 | oid, frame = _object_frame(obj_size_bytes, i) |
| 146 | object_ids.append(oid) |
| 147 | obj_frames.append(frame) |
| 148 | |
| 149 | body = ( |
| 150 | _header_frame(n_objects, n_commits) |
| 151 | + b"".join(obj_frames) |
| 152 | + (_commit_frame(n_commits, object_ids) if n_commits else b"") |
| 153 | + _end_frame(n_objects, n_commits) |
| 154 | ) |
| 155 | |
| 156 | headers = { |
| 157 | "Content-Type": WIRE_CONTENT_TYPE, |
| 158 | "Accept": WIRE_CONTENT_TYPE, |
| 159 | "Authorization": _auth_header(ROUTE), |
| 160 | } |
| 161 | |
| 162 | result: dict = { |
| 163 | "label": label or f"{n_objects}obj+{n_commits}commit", |
| 164 | "n_objects": n_objects, "n_commits": n_commits, |
| 165 | "body_kb": round(len(body) / 1024, 1), |
| 166 | "t_connect": None, "t_first_byte": None, "t_total": None, |
| 167 | "status": None, "ok": None, "cf524": False, |
| 168 | "error": None, "progress_frames": [], |
| 169 | } |
| 170 | |
| 171 | try: |
| 172 | t0 = time.perf_counter() |
| 173 | t_first = None |
| 174 | chunks = [] |
| 175 | |
| 176 | async with httpx.AsyncClient( |
| 177 | http2=False, timeout=300.0, verify=SSL_VERIFY, |
| 178 | limits=httpx.Limits(max_keepalive_connections=0), |
| 179 | ) as client: |
| 180 | async with client.stream("POST", f"{BASE_URL}{ROUTE}", content=body, headers=headers) as resp: |
| 181 | result["t_connect"] = round((time.perf_counter() - t0) * 1000, 1) |
| 182 | result["status"] = resp.status_code |
| 183 | |
| 184 | if resp.status_code == 524: |
| 185 | result["cf524"] = True |
| 186 | result["error"] = "CF 524 timeout" |
| 187 | return result |
| 188 | |
| 189 | async for chunk in resp.aiter_bytes(): |
| 190 | if t_first is None: |
| 191 | t_first = time.perf_counter() - t0 |
| 192 | result["t_first_byte"] = round(t_first * 1000, 1) |
| 193 | chunks.append(chunk) |
| 194 | |
| 195 | result["t_total"] = round((time.perf_counter() - t0) * 1000, 1) |
| 196 | |
| 197 | unpacker = msgpack.Unpacker(raw=False) |
| 198 | unpacker.feed(b"".join(chunks)) |
| 199 | for frame in unpacker: |
| 200 | t = frame.get("t") |
| 201 | if t == "P": |
| 202 | result["progress_frames"].append(frame.get("msg", "")) |
| 203 | elif t == "R": |
| 204 | result["ok"] = frame.get("ok") |
| 205 | elif t == "X": |
| 206 | result["ok"] = False |
| 207 | result["error"] = frame.get("msg", "error frame") |
| 208 | |
| 209 | except httpx.ReadTimeout: |
| 210 | result["error"] = "client timeout (>300s)" |
| 211 | result["t_total"] = 300_000 |
| 212 | except Exception as exc: |
| 213 | result["error"] = str(exc)[:120] |
| 214 | |
| 215 | return result |
| 216 | |
| 217 | |
| 218 | def _fmt(r: dict) -> str: |
| 219 | ok_str = "✅" if r["ok"] else ("❌" if r["ok"] is False else "?") |
| 220 | cf_str = " ⚠️ CF524" if r["cf524"] else "" |
| 221 | err_str = f" ERROR: {r['error']}" if r["error"] else "" |
| 222 | p_str = f" [{len(r['progress_frames'])} P-frames]" if r["progress_frames"] else "" |
| 223 | return ( |
| 224 | f" {r['label']:30s} {r['body_kb']:8.1f} KB " |
| 225 | f"connect={str(r['t_connect'] or '?'):>6}ms " |
| 226 | f"first={str(r['t_first_byte'] or '?'):>7}ms " |
| 227 | f"total={str(r['t_total'] or '?'):>8}ms " |
| 228 | f"{ok_str}{cf_str}{p_str}{err_str}" |
| 229 | ) |
| 230 | |
| 231 | |
| 232 | async def main() -> None: |
| 233 | args = [a for a in sys.argv[1:] if not a.startswith("--target")] |
| 234 | |
| 235 | if len(args) == 1: |
| 236 | scales = [(int(args[0]), 1)] |
| 237 | elif len(args) == 2: |
| 238 | scales = [(int(args[0]), int(args[1]))] |
| 239 | else: |
| 240 | scales = [ |
| 241 | (1, 1), |
| 242 | (10, 1), |
| 243 | (50, 1), |
| 244 | (100, 1), |
| 245 | (500, 1), |
| 246 | (500, 100), |
| 247 | (500, 500), |
| 248 | (500, 827), |
| 249 | (1000, 827), |
| 250 | ] |
| 251 | |
| 252 | print(f"\nPush timing test → {BASE_URL}{ROUTE}") |
| 253 | print(f" {'label':30s} {'body':>8} {'connect':>10} {'first':>10} {'total':>12} status") |
| 254 | print(" " + "-" * 100) |
| 255 | |
| 256 | for n_obj, n_com in scales: |
| 257 | r = await run_push(n_obj, n_com, label=f"{n_obj}obj+{n_com}commit") |
| 258 | print(_fmt(r)) |
| 259 | sys.stdout.flush() |
| 260 | |
| 261 | if r["cf524"]: |
| 262 | print(f"\n⚠️ CF 524 fired at {n_obj} objects. This is the wall.") |
| 263 | break |
| 264 | if r["t_total"] and r["t_total"] > 90_000: |
| 265 | print(f"\n⚠️ Batch took {r['t_total']}ms (>90s). Next scale will likely 524.") |
| 266 | break |
| 267 | |
| 268 | print() |
| 269 | |
| 270 | |
| 271 | if __name__ == "__main__": |
| 272 | asyncio.run(main()) |
File History
1 commit
sha256:a34090cc4a394a78bd72cbbe34b08cc59525141e19135b6c0ab154f10611b9ef
debug(push/stream): instrument O-frame decode path with INF…
Sonnet 4.6
patch
121 days ago