gabriel / musehub public
test_wire_step4c_scale.py python
280 lines 9.5 KB
Raw
sha256:ad616c6113d6c00f4efed6b2993734ca46d3e9b5bee25addd4ce8ae6b57136e5 chore: bump version to 0.2.0rc11; typing audit clean + all … Sonnet 4.6 minor ⚠ breaking 55 days ago
1 """Wire protocol — scale gate matching real-world musehub repo dimensions.
2
3 Step 4 (test_wire_step4_e2e.py) uses 100 commits sharing ONE snapshot.
4 Real musehub push: 1028 commits × 1022 DISTINCT snapshots × 5151 objects.
5 That's the case that hangs — this test reproduces it.
6
7 Each commit gets its own snapshot (one file changed per commit), so the
8 server must INSERT ~1000 snapshot rows, ~1000 commit rows, and ~5000
9 object rows. Gate: full server pipeline < 10s.
10 """
11 from __future__ import annotations
12
13 import datetime
14 import hashlib
15 import pathlib
16 import time
17
18 import httpx
19 import msgpack
20 import pytest
21 import pytest_asyncio
22 from httpx import AsyncClient, ASGITransport
23 from sqlalchemy.ext.asyncio import AsyncSession
24
25 from musehub.auth.request_signing import MSignContext, require_signed_request, optional_signed_request
26 from musehub.db.database import get_db
27 from musehub.main import app
28
29 from muse.core.object_store import write_object
30 from muse.core.mpack import build_mpack, build_wire_mpack
31 from muse.core.paths import muse_dir
32 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
33 from muse.core.commits import CommitRecord, write_commit
34 from muse.core.refs import write_branch_ref
35 from muse.core.snapshots import SnapshotRecord, write_snapshot
36 from muse.core.types import blob_id
37
38
39 # ---------------------------------------------------------------------------
40 # Scale parameters — match real musehub repo dimensions
41 # ---------------------------------------------------------------------------
42
43 _N_COMMITS = 1_000
44 _N_OBJECTS = 5_000 # total distinct blobs
45 _BLOB_SIZE = 4_096
46 _E2E_GATE_S = 10.0 # generous gate; goal is not to hang
47
48
49 # ---------------------------------------------------------------------------
50 # Auth + fixtures
51 # ---------------------------------------------------------------------------
52
53 _AUTH_CTX = MSignContext(
54 handle="gabriel",
55 identity_id="sha256:" + "0" * 64,
56 is_agent=False,
57 is_admin=True,
58 )
59
60
61 @pytest_asyncio.fixture()
62 async def client(db_session: AsyncSession):
63 async def _override_get_db():
64 yield db_session
65
66 app.dependency_overrides[get_db] = _override_get_db
67 app.dependency_overrides[require_signed_request] = lambda: _AUTH_CTX
68 app.dependency_overrides[optional_signed_request] = lambda: _AUTH_CTX
69
70 async with AsyncClient(
71 transport=ASGITransport(app=app),
72 base_url="https://localhost:1337",
73 ) as c:
74 yield c
75
76 app.dependency_overrides.clear()
77
78
79 @pytest_asyncio.fixture()
80 async def repo(client: AsyncClient):
81 resp = await client.post(
82 "/api/repos",
83 json={"owner": "gabriel", "name": "wire-scale", "visibility": "public", "initialize": False},
84 )
85 assert resp.status_code in (200, 201), f"repo create failed: {resp.text}"
86 data = resp.json()
87 slug = data["slug"]
88 repo_id = data["repoId"]
89 yield slug
90 await client.delete(f"/api/repos/{repo_id}")
91
92
93 # ---------------------------------------------------------------------------
94 # Local repo builder — one distinct snapshot per commit
95 # ---------------------------------------------------------------------------
96
97 def _make_repo(tmp: pathlib.Path) -> pathlib.Path:
98 tmp.mkdir(parents=True, exist_ok=True)
99 dot = muse_dir(tmp)
100 dot.mkdir()
101 (dot / "repo.json").write_text('{"repo_id":"scale","owner":"gabriel"}')
102 for d in ("commits", "snapshots", "objects"):
103 (dot / d).mkdir()
104 (dot / "refs" / "heads").mkdir(parents=True)
105 (dot / "HEAD").write_text("ref: refs/heads/main\n")
106 (dot / "config.toml").write_text("")
107 return tmp
108
109
110 def _populate(repo: pathlib.Path) -> str:
111 """Create _N_OBJECTS blobs, then _N_COMMITS commits each with a distinct snapshot.
112
113 Each commit swaps out one file so every snapshot is unique — this matches
114 the real-world case where every commit touches at least one file.
115 """
116 # Create all blobs up front.
117 blob_ids: list[str] = []
118 for i in range(_N_OBJECTS):
119 data = f"scale-{i:08d}-".encode() + b"x" * _BLOB_SIZE
120 oid = blob_id(data)
121 write_object(repo, oid, data)
122 blob_ids.append(oid)
123
124 # Base manifest: files 0…N_OBJECTS-1
125 base_manifest: dict[str, str] = {
126 f"file_{i:04d}.py": blob_ids[i] for i in range(_N_OBJECTS)
127 }
128
129 parent: str | None = None
130 tip = ""
131 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
132
133 for i in range(_N_COMMITS):
134 # Each commit replaces one file with a re-hashed variant, making the
135 # snapshot unique (different manifest → different snapshot_id).
136 manifest = dict(base_manifest)
137 variant = f"commit-{i:05d}-variant".encode() + b"y" * _BLOB_SIZE
138 variant_oid = blob_id(variant)
139 write_object(repo, variant_oid, variant)
140 manifest[f"file_{i % _N_OBJECTS:04d}.py"] = variant_oid
141
142 sid = compute_snapshot_id(manifest)
143 write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=manifest))
144
145 msg = f"commit-{i:05d}"
146 cid = compute_commit_id(
147 parent_ids=[parent] if parent else [],
148 snapshot_id=sid,
149 message=msg,
150 committed_at_iso=ts.isoformat(),
151 author="gabriel",
152 )
153 rec = CommitRecord(
154 commit_id=cid,
155 branch="main",
156 snapshot_id=sid,
157 message=msg,
158 committed_at=ts,
159 parent_commit_id=parent,
160 parent2_commit_id=None,
161 author="gabriel",
162 metadata={},
163 structured_delta=None,
164 sem_ver_bump="none",
165 breaking_changes=[],
166 agent_id="",
167 model_id="",
168 toolchain_id="",
169 prompt_hash="",
170 signature="",
171 signer_key_id="",
172 )
173 write_commit(repo, rec)
174 parent = cid
175 tip = cid
176 ts += datetime.timedelta(seconds=60)
177
178 write_branch_ref(repo, "main", tip)
179 return tip
180
181
182 # ---------------------------------------------------------------------------
183 # THE test
184 # ---------------------------------------------------------------------------
185
186 @pytest.mark.asyncio
187 async def test_scale_unpack_mpack(
188 client: AsyncClient, repo: str, tmp_path: pathlib.Path, db_session: AsyncSession
189 ) -> None:
190 """Scale gate: 1000 commits × 1000 distinct snapshots × 5000 objects < 10s."""
191 owner = "gabriel"
192 branch = "main"
193
194 local_repo = _make_repo(tmp_path / "repo")
195
196 t_build = time.perf_counter()
197 head = _populate(local_repo)
198 mpack = build_mpack(local_repo, [head], have=[])
199 wire_bytes = build_wire_mpack(mpack)
200 mpack_key = blob_id(wire_bytes)
201 print(f"\n build+pack: {time.perf_counter()-t_build:.2f}s "
202 f"({len(wire_bytes)//1024} KiB)", flush=True)
203
204 t_server = time.perf_counter()
205
206 # 1. mpack-presign
207 presign_resp = await client.post(
208 f"/{owner}/{repo}/push/mpack-presign",
209 content=msgpack.packb({"mpack_key": mpack_key, "size_bytes": len(wire_bytes)}, use_bin_type=True),
210 headers={"Content-Type": "application/x-msgpack"},
211 )
212 assert presign_resp.status_code == 200, presign_resp.text
213 upload_url = presign_resp.json().get("upload_url") or presign_resp.json().get("uploadUrl")
214 assert upload_url
215
216 t_presign = time.perf_counter()
217
218 # 2. PUT → MinIO
219 async with httpx.AsyncClient() as raw:
220 put_resp = await raw.put(upload_url, content=wire_bytes)
221 assert put_resp.status_code in (200, 204), f"PUT failed {put_resp.status_code}"
222
223 t_put = time.perf_counter()
224
225 # 3. unpack-mpack
226 unpack_resp = await client.post(
227 f"/{owner}/{repo}/push/unpack-mpack",
228 content=msgpack.packb(
229 {"mpack_key": mpack_key, "branch": branch, "head": head},
230 use_bin_type=True,
231 ),
232 headers={"Content-Type": "application/x-msgpack"},
233 )
234 assert unpack_resp.status_code == 200, unpack_resp.text
235 result = unpack_resp.json()
236
237 t_unpack = time.perf_counter()
238
239 # 4. verify branch head
240 refs_resp = await client.get(f"/{owner}/{repo}/refs")
241 assert refs_resp.status_code == 200
242 branch_heads = refs_resp.json().get("branch_heads", {})
243
244 t_verify = time.perf_counter()
245
246 server_ms = (
247 (t_presign - t_server) +
248 (t_unpack - t_put) +
249 (t_verify - t_unpack)
250 ) * 1000
251 put_ms = (t_put - t_presign) * 1000
252
253 if "job_id" in result and result.get("commits_in_mpack", 0) == 0:
254 from musehub.services.musehub_wire import process_mpack_index_job
255 job_result = await process_mpack_index_job(db_session, result["job_id"])
256 await db_session.commit()
257 result = {**result, **job_result}
258
259 assert result.get("commits_written") == _N_COMMITS, result
260 assert branch_heads.get(branch) == head, branch_heads
261
262 gate_ms = _E2E_GATE_S * 1000
263 assert server_ms < gate_ms, (
264 f"Scale gate FAIL: {server_ms:.0f}ms > {gate_ms:.0f}ms\n"
265 f" presign={( t_presign-t_server)*1000:.0f}ms "
266 f"unpack={(t_unpack-t_put)*1000:.0f}ms "
267 f"verify={(t_verify-t_unpack)*1000:.0f}ms"
268 )
269
270 print(
271 f"\n Scale gate — 1000 commits × 1000 snapshots × 5000 objects\n"
272 f" mpack: {len(wire_bytes)//1024} KiB\n"
273 f" presign: {(t_presign-t_server)*1000:.0f}ms\n"
274 f" PUT → MinIO: {put_ms:.0f}ms (not counted)\n"
275 f" unpack-mpack: {(t_unpack-t_put)*1000:.0f}ms\n"
276 f" verify refs: {(t_verify-t_unpack)*1000:.0f}ms\n"
277 f" server total: {server_ms:.0f}ms (gate {gate_ms:.0f}ms)\n"
278 f" commits written: {result.get('commits_written')}\n"
279 f" objects written: {result.get('objects_written')}"
280 )
File History 2 commits
sha256:ad616c6113d6c00f4efed6b2993734ca46d3e9b5bee25addd4ce8ae6b57136e5 chore: bump version to 0.2.0rc11; typing audit clean + all … Sonnet 4.6 minor ⚠ 55 days ago
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595 fix: typing audit — 0 violations, 0 untyped defs across all… Sonnet 4.6 minor ⚠ 58 days ago