gabriel / musehub public
test_wire_stress_performance.py python
564 lines 19.3 KB
Raw
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago
1 """Wire Protocol — Stress and Performance tests.
2
3 Stress (layer 4): high concurrency, large payloads, sustained load.
4 Performance (layer 7): latency budgets, query efficiency, memory/connection bounds.
5
6 Rate-limit budget per test (reset by conftest.reset_rate_limiter):
7 WIRE_PUSH_LIMIT = 30/minute
8 WIRE_FETCH_LIMIT = 120/minute
9
10 Object storage is isolated to a per-test temp directory by conftest._tmp_objects_dir.
11 """
12 from __future__ import annotations
13
14 import time
15 import uuid
16 from datetime import datetime, timezone
17
18 import msgpack
19 import pytest
20 from httpx import AsyncClient
21 from sqlalchemy.ext.asyncio import AsyncSession
22
23 from musehub.db import musehub_models as db
24 from musehub.models.wire import MAX_OBJECTS_PER_PUSH
25 from tests.factories import create_repo as factory_create_repo
26 from musehub.muse_contracts.json_types import JSONObject, StrDict
27
28
29 # ── helpers ────────────────────────────────────────────────────────────────────
30
31
32 def _utc_now() -> datetime:
33 return datetime.now(tz=timezone.utc)
34
35
36 def _mp(data: JSONObject) -> bytes:
37 return msgpack.packb(data, use_bin_type=True)
38
39
40 def _make_commit(repo_id: str, commit_id: str | None = None, parent: str | None = None) -> JSONObject:
41 return {
42 "commit_id": commit_id or uuid.uuid4().hex,
43 "repo_id": repo_id,
44 "branch": "main",
45 "snapshot_id": None,
46 "message": "stress test commit",
47 "committed_at": _utc_now().isoformat(),
48 "parent_commit_id": parent,
49 "author": "Stress <[email protected]>",
50 "sem_ver_bump": "patch",
51 }
52
53
54 def _make_object(content: bytes | None = None, path: str = "file.bin") -> JSONObject:
55 return {
56 "object_id": uuid.uuid4().hex,
57 "content": content or uuid.uuid4().bytes * 4, # 64 bytes of pseudo-random
58 "path": path,
59 }
60
61
62 def _make_snapshot(snap_id: str, obj_map: StrDict) -> JSONObject:
63 return {
64 "snapshot_id": snap_id,
65 "manifest": obj_map,
66 "created_at": _utc_now().isoformat(),
67 }
68
69
70 def _push_payload(repo_id: str, commits: list[dict], snapshots: list[dict] = (), objects: list[dict] = (), branch: str = "main", force: bool = False) -> bytes:
71 return _mp({
72 "bundle": {
73 "commits": list(commits),
74 "snapshots": list(snapshots),
75 "objects": list(objects),
76 },
77 "branch": branch,
78 "force": force,
79 })
80
81
82 _MP_HEADERS = {"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"}
83
84
85 # ── Stress 1: concurrent pushes to independent repos ─────────────────────────
86
87
88 @pytest.mark.asyncio
89 async def test_high_volume_pushes_to_separate_repos_all_succeed(
90 client: AsyncClient,
91 db_session: AsyncSession,
92 wire_headers: StrDict,
93 ) -> None:
94 """20 sequential pushes to 20 different repos must all succeed within budget.
95
96 Validates the push path handles high repo-fan-out volume correctly.
97 Uses sequential dispatch to validate correctness at volume, exercising the
98 same code paths as concurrent real-world pushes.
99 """
100 N = 20
101 repos = [
102 await factory_create_repo(db_session, slug=f"vol-push-{i}", owner="test-user-wire")
103 for i in range(N)
104 ]
105
106 t0 = time.perf_counter()
107 for repo in repos:
108 payload = _push_payload(repo.repo_id, [_make_commit(repo.repo_id)])
109 resp = await client.post(
110 f"/{repo.owner}/{repo.slug}/push",
111 content=payload,
112 headers=wire_headers,
113 )
114 assert resp.status_code == 200, f"Push to {repo.slug} failed: {resp.text}"
115 elapsed = time.perf_counter() - t0
116 assert elapsed < 10.0, f"20 repo pushes took {elapsed:.2f}s"
117
118
119 # ── Stress 2: concurrent fetch readers on the same repo ───────────────────────
120
121
122 @pytest.mark.asyncio
123 async def test_high_volume_fetches_same_repo(
124 client: AsyncClient,
125 db_session: AsyncSession,
126 wire_headers: StrDict,
127 ) -> None:
128 """30 sequential fetch requests against the same repo must all return 200.
129
130 Validates the BFS read path handles sustained fetch load without errors.
131 fetch has a 120/min rate limit so 30 requests stays well within budget.
132 """
133 repo = await factory_create_repo(db_session, slug="volume-fetch-repo", owner="test-user-wire")
134
135 # Seed 5 commits
136 commits = []
137 parent = None
138 for i in range(5):
139 cid = uuid.uuid4().hex
140 c = _make_commit(repo.repo_id, commit_id=cid, parent=parent)
141 commits.append(c)
142 parent = cid
143
144 push_r = await client.post(
145 f"/{repo.owner}/{repo.slug}/push",
146 content=_push_payload(repo.repo_id, commits),
147 headers=wire_headers,
148 )
149 assert push_r.status_code == 200
150
151 tip_id = commits[-1]["commit_id"]
152 t0 = time.perf_counter()
153 for i in range(30):
154 resp = await client.post(
155 f"/{repo.owner}/{repo.slug}/fetch",
156 content=_mp({"want": [tip_id], "have": []}),
157 headers=_MP_HEADERS,
158 )
159 assert resp.status_code == 200, f"Fetch {i} failed"
160 elapsed = time.perf_counter() - t0
161 assert elapsed < 10.0, f"30 fetches took {elapsed:.2f}s"
162
163
164 # ── Stress 3: large bundle — many objects ─────────────────────────────────────
165
166
167 @pytest.mark.asyncio
168 async def test_large_object_bundle_push_and_fetch(
169 client: AsyncClient,
170 db_session: AsyncSession,
171 wire_headers: StrDict,
172 ) -> None:
173 """Push a bundle with 100 objects; fetch returns all of them with correct content.
174
175 Tests that the pack assembly and BFS object collection handles large manifests
176 without truncation or silent data loss.
177 """
178 OBJECT_COUNT = 100
179 repo = await factory_create_repo(db_session, slug="large-bundle-test", owner="test-user-wire")
180
181 objects = [_make_object(f"content-{i}".encode(), f"file_{i}.bin") for i in range(OBJECT_COUNT)]
182 manifest = {obj["path"]: obj["object_id"] for obj in objects}
183 snap_id = f"snap_{uuid.uuid4().hex[:8]}"
184 snap = _make_snapshot(snap_id, manifest)
185 commit_id = uuid.uuid4().hex
186 commit = _make_commit(repo.repo_id, commit_id=commit_id)
187 commit["snapshot_id"] = snap_id
188
189 push_resp = await client.post(
190 f"/{repo.owner}/{repo.slug}/push",
191 content=_push_payload(repo.repo_id, [commit], [snap], objects),
192 headers=wire_headers,
193 )
194 assert push_resp.status_code == 200, push_resp.text
195
196 fetch_resp = await client.post(
197 f"/{repo.owner}/{repo.slug}/fetch",
198 content=_mp({"want": [commit_id], "have": []}),
199 headers=_MP_HEADERS,
200 )
201 assert fetch_resp.status_code == 200
202 data = msgpack.unpackb(fetch_resp.content, raw=False)
203 assert len(data["objects"]) == OBJECT_COUNT
204
205
206 # ── Stress 4: deep commit chain BFS fetch ────────────────────────────────────
207
208
209 @pytest.mark.asyncio
210 async def test_deep_commit_chain_bfs_fetch(
211 client: AsyncClient,
212 db_session: AsyncSession,
213 wire_headers: StrDict,
214 ) -> None:
215 """Push a 25-commit chain; fetch from tip with have=[root] returns exactly 24 commits.
216
217 Validates BFS traversal stops correctly at the have boundary and doesn't
218 over-fetch or under-fetch when the chain has known depth.
219 """
220 DEPTH = 25
221 repo = await factory_create_repo(db_session, slug="deep-bfs-test", owner="test-user-wire")
222
223 commits = []
224 parent = None
225 for _ in range(DEPTH):
226 cid = uuid.uuid4().hex
227 c = _make_commit(repo.repo_id, commit_id=cid, parent=parent)
228 commits.append(c)
229 parent = cid
230
231 # Push in batches to stay within push rate limit (30/min)
232 batch_size = 25
233 for i in range(0, len(commits), batch_size):
234 batch = commits[i:i + batch_size]
235 r = await client.post(
236 f"/{repo.owner}/{repo.slug}/push",
237 content=_push_payload(repo.repo_id, batch, force=True),
238 headers=wire_headers,
239 )
240 assert r.status_code == 200, f"Batch push failed: {r.text}"
241
242 root_id = commits[0]["commit_id"]
243 tip_id = commits[-1]["commit_id"]
244
245 fetch_resp = await client.post(
246 f"/{repo.owner}/{repo.slug}/fetch",
247 content=_mp({"want": [tip_id], "have": [root_id]}),
248 headers=_MP_HEADERS,
249 )
250 assert fetch_resp.status_code == 200
251 data = msgpack.unpackb(fetch_resp.content, raw=False)
252 fetched_ids = {c["commit_id"] for c in data["commits"]}
253
254 # Must include all commits between root and tip (exclusive of root)
255 assert tip_id in fetched_ids
256 assert root_id not in fetched_ids
257 assert len(data["commits"]) == DEPTH - 1
258
259
260 # ── Stress 5: filter-objects with large ID set ───────────────────────────────
261
262
263 @pytest.mark.asyncio
264 async def test_filter_objects_large_set(
265 client: AsyncClient,
266 db_session: AsyncSession,
267 wire_headers: StrDict,
268 ) -> None:
269 """filter-objects with 500 unknown IDs returns all 500 as missing.
270
271 Tests the single-IN-clause query handles a large ID list without truncation
272 or query plan degradation.
273 """
274 repo = await factory_create_repo(db_session, slug="filter-large-test", owner="test-user-wire")
275 oids = [uuid.uuid4().hex for _ in range(500)]
276
277 resp = await client.post(
278 f"/{repo.owner}/{repo.slug}/filter-objects",
279 content=_mp({"object_ids": oids}),
280 headers=wire_headers,
281 )
282 assert resp.status_code == 200
283 data = msgpack.unpackb(resp.content, raw=False)
284 assert len(data["missing"]) == 500
285
286
287 # ── Stress 6: push/objects chunked pre-upload — many objects ──────────────────
288
289
290 @pytest.mark.asyncio
291 async def test_push_objects_large_batch(
292 client: AsyncClient,
293 db_session: AsyncSession,
294 wire_headers: StrDict,
295 ) -> None:
296 """push/objects with 50 objects stores all 50 and reports correct stored count."""
297 repo = await factory_create_repo(db_session, slug="push-objects-large-test", owner="test-user-wire")
298 objects = [_make_object(f"blob-{i}".encode()) for i in range(50)]
299
300 resp = await client.post(
301 f"/{repo.owner}/{repo.slug}/push/objects",
302 content=_mp({"objects": objects}),
303 headers=wire_headers,
304 )
305 assert resp.status_code == 200
306 data = msgpack.unpackb(resp.content, raw=False)
307 assert data["stored"] == 50
308 assert data["skipped"] == 0
309
310
311 # ── Performance 1: single push round-trip latency ────────────────────────────
312
313
314 @pytest.mark.asyncio
315 async def test_single_push_latency_under_budget(
316 client: AsyncClient,
317 db_session: AsyncSession,
318 wire_headers: StrDict,
319 ) -> None:
320 """A minimal push (1 commit, 0 objects) must complete in under 500ms.
321
322 Tests the fast path for incremental pushes where objects are already
323 on the server and only the commit+branch pointer needs to be written.
324 """
325 repo = await factory_create_repo(db_session, slug="push-latency-test", owner="test-user-wire")
326 payload = _push_payload(repo.repo_id, [_make_commit(repo.repo_id)])
327
328 t0 = time.perf_counter()
329 resp = await client.post(
330 f"/{repo.owner}/{repo.slug}/push",
331 content=payload,
332 headers=wire_headers,
333 )
334 elapsed_ms = (time.perf_counter() - t0) * 1000
335
336 assert resp.status_code == 200
337 assert elapsed_ms < 500, f"Push took {elapsed_ms:.0f}ms — exceeds 500ms budget"
338
339
340 # ── Performance 2: fetch latency with moderate history ───────────────────────
341
342
343 @pytest.mark.asyncio
344 async def test_fetch_latency_10_commits_under_budget(
345 client: AsyncClient,
346 db_session: AsyncSession,
347 wire_headers: StrDict,
348 ) -> None:
349 """Fetching a 10-commit chain must complete in under 500ms.
350
351 Tests that BFS traversal + commit/snapshot assembly stays fast for
352 typical incremental pull sizes.
353 """
354 repo = await factory_create_repo(db_session, slug="fetch-latency-test", owner="test-user-wire")
355
356 commits = []
357 parent = None
358 for _ in range(10):
359 cid = uuid.uuid4().hex
360 c = _make_commit(repo.repo_id, commit_id=cid, parent=parent)
361 commits.append(c)
362 parent = cid
363
364 push_resp = await client.post(
365 f"/{repo.owner}/{repo.slug}/push",
366 content=_push_payload(repo.repo_id, commits),
367 headers=wire_headers,
368 )
369 assert push_resp.status_code == 200
370
371 tip_id = commits[-1]["commit_id"]
372 t0 = time.perf_counter()
373 fetch_resp = await client.post(
374 f"/{repo.owner}/{repo.slug}/fetch",
375 content=_mp({"want": [tip_id], "have": []}),
376 headers=_MP_HEADERS,
377 )
378 elapsed_ms = (time.perf_counter() - t0) * 1000
379
380 assert fetch_resp.status_code == 200
381 assert elapsed_ms < 500, f"Fetch took {elapsed_ms:.0f}ms — exceeds 500ms budget"
382
383
384 # ── Performance 3: filter-objects latency ─────────────────────────────────────
385
386
387 @pytest.mark.asyncio
388 async def test_filter_objects_100_ids_under_budget(
389 client: AsyncClient,
390 db_session: AsyncSession,
391 wire_headers: StrDict,
392 ) -> None:
393 """filter-objects with 100 IDs must respond in under 200ms.
394
395 The single IN-clause query must be fast enough that incremental pushes
396 using MWP object deduplication do not add noticeable latency.
397 """
398 repo = await factory_create_repo(db_session, slug="filter-latency-test", owner="test-user-wire")
399 oids = [uuid.uuid4().hex for _ in range(100)]
400
401 t0 = time.perf_counter()
402 resp = await client.post(
403 f"/{repo.owner}/{repo.slug}/filter-objects",
404 content=_mp({"object_ids": oids}),
405 headers=wire_headers,
406 )
407 elapsed_ms = (time.perf_counter() - t0) * 1000
408
409 assert resp.status_code == 200
410 assert elapsed_ms < 200, f"filter-objects took {elapsed_ms:.0f}ms — exceeds 200ms budget"
411
412
413 # ── Performance 4: BFS does not degrade with depth ────────────────────────────
414
415
416 @pytest.mark.asyncio
417 async def test_bfs_fetch_shallow_faster_than_deep(
418 client: AsyncClient,
419 db_session: AsyncSession,
420 wire_headers: StrDict,
421 ) -> None:
422 """Fetching a 5-commit window should be faster than fetching a 20-commit window.
423
424 Validates BFS is proportional to the delta size — not to the total
425 history size. Both must still be within the 1s absolute budget.
426 """
427 repo = await factory_create_repo(db_session, slug="bfs-depth-perf-test", owner="test-user-wire")
428
429 commits = []
430 parent = None
431 for _ in range(20):
432 cid = uuid.uuid4().hex
433 c = _make_commit(repo.repo_id, commit_id=cid, parent=parent)
434 commits.append(c)
435 parent = cid
436
437 push_resp = await client.post(
438 f"/{repo.owner}/{repo.slug}/push",
439 content=_push_payload(repo.repo_id, commits),
440 headers=wire_headers,
441 )
442 assert push_resp.status_code == 200
443
444 # Shallow: fetch only last 5 commits (have = commit[14])
445 shallow_have = commits[14]["commit_id"]
446 tip_id = commits[-1]["commit_id"]
447
448 t0 = time.perf_counter()
449 resp_shallow = await client.post(
450 f"/{repo.owner}/{repo.slug}/fetch",
451 content=_mp({"want": [tip_id], "have": [shallow_have]}),
452 headers=_MP_HEADERS,
453 )
454 shallow_ms = (time.perf_counter() - t0) * 1000
455 assert resp_shallow.status_code == 200
456 assert len(msgpack.unpackb(resp_shallow.content, raw=False)["commits"]) == 5
457
458 # Deep: fetch all 20 commits (have = [])
459 t0 = time.perf_counter()
460 resp_deep = await client.post(
461 f"/{repo.owner}/{repo.slug}/fetch",
462 content=_mp({"want": [tip_id], "have": []}),
463 headers=_MP_HEADERS,
464 )
465 deep_ms = (time.perf_counter() - t0) * 1000
466 assert resp_deep.status_code == 200
467 assert len(msgpack.unpackb(resp_deep.content, raw=False)["commits"]) == 20
468
469 # Both within absolute budget
470 assert shallow_ms < 1000, f"Shallow fetch took {shallow_ms:.0f}ms"
471 assert deep_ms < 1000, f"Deep fetch took {deep_ms:.0f}ms"
472
473
474 # ── Performance 5: refs endpoint is cheap ────────────────────────────────────
475
476
477 @pytest.mark.asyncio
478 async def test_refs_latency_under_budget(
479 client: AsyncClient,
480 db_session: AsyncSession,
481 ) -> None:
482 """GET /refs must respond in under 100ms — it is the preflight for every push/pull."""
483 repo = await factory_create_repo(db_session, slug="refs-latency-test")
484
485 # Warm up
486 await client.get(f"/{repo.owner}/{repo.slug}/refs")
487
488 t0 = time.perf_counter()
489 resp = await client.get(f"/{repo.owner}/{repo.slug}/refs")
490 elapsed_ms = (time.perf_counter() - t0) * 1000
491
492 assert resp.status_code == 200
493 assert elapsed_ms < 100, f"GET /refs took {elapsed_ms:.0f}ms — exceeds 100ms budget"
494
495
496 # ── Performance 6: negotiate is cheap ────────────────────────────────────────
497
498
499 @pytest.mark.asyncio
500 async def test_negotiate_latency_under_budget(
501 client: AsyncClient,
502 db_session: AsyncSession,
503 ) -> None:
504 """negotiate with 10 have IDs must complete in under 200ms.
505
506 It is called once per push/pull negotiation round — high latency here
507 adds directly to perceived CLI responsiveness.
508 """
509 repo = await factory_create_repo(db_session, slug="negotiate-latency-test")
510
511 # Seed a commit the server knows about
512 commit_id = uuid.uuid4().hex
513 db_session.add(db.MusehubCommit(
514 commit_id=commit_id, repo_id=repo.repo_id, branch="main",
515 parent_ids=[], message="base", author="T", timestamp=_utc_now(),
516 snapshot_id=None, commit_meta={},
517 ))
518 await db_session.commit()
519
520 have = [uuid.uuid4().hex for _ in range(9)] + [commit_id]
521
522 t0 = time.perf_counter()
523 resp = await client.post(
524 f"/{repo.owner}/{repo.slug}/negotiate",
525 content=_mp({"have": have, "want": [commit_id]}),
526 headers={"Content-Type": "application/x-msgpack", "Accept": "application/x-msgpack"},
527 )
528 elapsed_ms = (time.perf_counter() - t0) * 1000
529
530 assert resp.status_code == 200
531 assert elapsed_ms < 200, f"negotiate took {elapsed_ms:.0f}ms — exceeds 200ms budget"
532
533
534 # ── Performance 7: sequential push throughput ─────────────────────────────────
535
536
537 @pytest.mark.asyncio
538 async def test_sequential_push_throughput(
539 client: AsyncClient,
540 db_session: AsyncSession,
541 wire_headers: StrDict,
542 ) -> None:
543 """20 sequential pushes (each with 1 commit) complete in under 5 seconds.
544
545 This is the hot path for ``muse push`` on a developer machine.
546 20 requests × 250ms/req = 5s worst-case budget.
547 """
548 repo = await factory_create_repo(db_session, slug="push-throughput-test", owner="test-user-wire")
549
550 parent = None
551 t0 = time.perf_counter()
552 for _ in range(20):
553 cid = uuid.uuid4().hex
554 commit = _make_commit(repo.repo_id, commit_id=cid, parent=parent)
555 resp = await client.post(
556 f"/{repo.owner}/{repo.slug}/push",
557 content=_push_payload(repo.repo_id, [commit]),
558 headers=wire_headers,
559 )
560 assert resp.status_code == 200
561 parent = cid
562
563 elapsed = time.perf_counter() - t0
564 assert elapsed < 5.0, f"20 sequential pushes took {elapsed:.2f}s — exceeds 5s budget"
File History 1 commit
sha256:a10adeeb7a0169cb9900f9806ed7a973047258abb6283724fe55e8eb68ff3f0a init: musehub initial commit Human 171 days ago