gabriel / muse public
test_perf_phase3.py python
1,178 lines 41.0 KB
Raw
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 147 days ago
1 """Phase 3 — Performance regression tests.
2
3 Target metrics (measured on a 2024 MacBook Pro M4, macOS 15):
4
5 Phase 3.1 — Linux-kernel commit throughput
6 write_commit: ≥ 1 000 commits/sec
7 write_object: ≥ 2 000 objects/sec
8 build_snapshot_manifest: ≥ 10 000 files/sec
9 muse commit (e2e, 1 000-file workdir): < 5 000 ms
10
11 Phase 3.2 — Concurrent agent write storm
12 200 threads × write_object: all objects readable, no corruption
13 200 threads × write_commit: all commits readable, no corruption
14 100 threads × write_head_commit: last-write-wins, valid ID written
15
16 Phase 3.3 — Memory ceiling
17 write_commit (10 000 commits): peak RSS < 512 MiB
18 build_snapshot_manifest (5 000 files): peak RSS < 128 MiB
19
20 Phase 3.3 extended — Linux-scale memory ceiling (100k / 75k)
21 get_all_commits (100 000 commits): peak RSS < 2 GiB [@slow]
22 get_commits_for_branch walk-cap: max_walk_commits bounds RSS
23 muse log --json pseudo-streaming: no double-buffer
24 build_snapshot_manifest (75 000 files): peak RSS < 512 MiB [@slow]
25 find_merge_base (deep chain): cap fires, no OOM
26 walk_commits_between: silent truncation at cap, not OOM
27
28 All slow tests are decorated with ``@pytest.mark.slow`` and are skipped by
29 default. Run the full suite with ``pytest tests/test_perf_phase3.py -v``.
30 """
31
32 from __future__ import annotations
33
34 type _FileStore = dict[str, bytes]
35
36 import datetime
37 import hashlib
38 import os
39 import pathlib
40 import resource
41 import sys
42 import threading
43 import time
44 import tracemalloc
45
46 import pytest
47 from unittest.mock import patch
48
49 from muse.core.object_store import (
50 _created_object_shards,
51 has_object,
52 object_path,
53 read_object,
54 write_object,
55 )
56 from muse.core.merge_engine import find_merge_base
57 from muse.core.snapshot import build_snapshot_manifest, compute_commit_id
58 from muse.core.store import (
59 CommitRecord,
60 get_all_commits,
61 get_commits_for_branch,
62 read_commit,
63 walk_commits_between_result,
64 write_branch_ref,
65 write_commit,
66 write_head_commit,
67 )
68
69 # ---------------------------------------------------------------------------
70 # Helpers
71 # ---------------------------------------------------------------------------
72
73
74 from muse.core._types import blob_id
75
76
77 def _sha256(data: bytes) -> str:
78 return blob_id(data)
79
80
81 def _repo(tmp_path: pathlib.Path) -> pathlib.Path:
82 muse_dir = tmp_path / ".muse"
83 muse_dir.mkdir()
84 (muse_dir / "repo.json").write_text('{"repo_id": "bench", "owner": "bench"}')
85 (muse_dir / "commits").mkdir()
86 (muse_dir / "snapshots").mkdir()
87 (muse_dir / "refs" / "heads").mkdir(parents=True)
88 (muse_dir / "HEAD").write_text("ref: refs/heads/main\n")
89 return tmp_path
90
91
92 def _write_chain(
93 repo: pathlib.Path,
94 branch: str,
95 n: int,
96 snap_id: str = "f" * 64,
97 start: int = 0,
98 ) -> str:
99 """Write a linear commit chain of length *n* and return the tip commit ID.
100
101 Sets the branch ref to the tip so ``get_commits_for_branch`` can walk it.
102 """
103 parent: str | None = None
104 tip = ""
105 for i in range(start, start + n):
106 msg = f"chain-{i:07d}"
107 ts = datetime.datetime(2026, 1, 1, i % 3600 // 3600, i % 3600 % 60, tzinfo=datetime.timezone.utc)
108 cid = compute_commit_id([parent] if parent else [], snap_id, msg, ts.isoformat())
109 rec = CommitRecord(
110 commit_id=cid,
111 repo_id="bench",
112 branch=branch,
113 snapshot_id=snap_id,
114 message=msg,
115 committed_at=ts,
116 parent_commit_id=parent,
117 parent2_commit_id=None,
118 author="chain-agent",
119 metadata={},
120 structured_delta=None,
121 sem_ver_bump="none",
122 breaking_changes=[],
123 agent_id="",
124 model_id="",
125 toolchain_id="",
126 prompt_hash="",
127 signature="",
128 signer_key_id="",
129 )
130 write_commit(repo, rec)
131 parent = cid
132 tip = cid
133 write_branch_ref(repo, branch, tip)
134 return tip
135
136
137 def _make_commit(index: int, snap_id: str, parent: str | None = None) -> CommitRecord:
138 """Build a CommitRecord whose ``commit_id`` passes content-hash verification."""
139 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
140 msg = f"commit-{index:07d}"
141 cid = compute_commit_id([parent] if parent else [], snap_id, msg, ts.isoformat())
142 return CommitRecord(
143 commit_id=cid,
144 repo_id="bench",
145 branch="main",
146 snapshot_id=snap_id,
147 message=msg,
148 committed_at=ts,
149 parent_commit_id=parent,
150 parent2_commit_id=None,
151 author="perf-agent",
152 metadata={},
153 structured_delta=None,
154 sem_ver_bump="none",
155 breaking_changes=[],
156 agent_id="",
157 model_id="",
158 toolchain_id="",
159 prompt_hash="",
160 signature="",
161 signer_key_id="",
162 )
163
164
165 # ---------------------------------------------------------------------------
166 # Phase 3.1 — Linux-kernel commit throughput
167 # ---------------------------------------------------------------------------
168
169
170 class TestWriteCommitThroughput:
171 """write_commit must sustain ≥ 1 000 commits/sec in isolation.
172
173 The Linux-kernel migration target is 472 commits/sec over 850 000 commits
174 (< 30 min). A 1 000 commits/sec floor gives comfortable headroom for
175 real workload overhead (snapshot building, object writes, disk pressure).
176
177 fsync is mocked: these tests measure msgpack serialisation + filesystem
178 metadata throughput, not OS I/O durability. Durability ordering is
179 verified by test_integrity_I2_fsync.py.
180 """
181
182 _MIN_COMMITS_PER_SEC = 1_000
183
184 @pytest.fixture(autouse=True)
185 def no_fsync(self) -> None:
186 """Mock out all fsync calls so the test measures algorithmic throughput."""
187 with patch("muse.core.store.os.fsync", return_value=None), \
188 patch("muse.core.store.fcntl.fcntl", return_value=0):
189 yield
190
191 @pytest.mark.slow
192 def test_write_commit_throughput_10k(self, tmp_path: pathlib.Path) -> None:
193 """Write 10 000 commits and assert throughput ≥ 1 000 commits/sec."""
194 repo = _repo(tmp_path)
195 snap_id = "a" * 64
196 N = 10_000
197 commits = [_make_commit(i, snap_id) for i in range(N)]
198
199 t0 = time.perf_counter()
200 for rec in commits:
201 write_commit(repo, rec)
202 elapsed = time.perf_counter() - t0
203 rate = N / elapsed
204
205 assert rate >= self._MIN_COMMITS_PER_SEC, (
206 f"write_commit throughput {rate:.0f} commits/sec is below the "
207 f"minimum {self._MIN_COMMITS_PER_SEC} commits/sec. "
208 f"(10k commits took {elapsed:.2f}s. "
209 f"Linux-kernel migration target: 472 commits/sec.)"
210 )
211
212 def test_write_commit_throughput_1k_fast(self, tmp_path: pathlib.Path) -> None:
213 """Smoke-speed: 1 000 commits must complete within 5 seconds."""
214 repo = _repo(tmp_path)
215 snap_id = "b" * 64
216 N = 1_000
217 commits = [_make_commit(i, snap_id) for i in range(N)]
218
219 t0 = time.perf_counter()
220 for rec in commits:
221 write_commit(repo, rec)
222 elapsed = time.perf_counter() - t0
223
224 assert elapsed <= 5.0, (
225 f"1 000 write_commit calls took {elapsed:.2f}s — expected ≤ 5.0s."
226 )
227
228
229 class TestWriteObjectThroughput:
230 """write_object must sustain ≥ 1 500 objects/sec in isolation.
231
232 fsync is mocked: the test measures mkstemp + hash-verify + fchmod +
233 os.replace throughput without OS I/O latency. Durability ordering is
234 verified by test_integrity_I2_fsync.py.
235 """
236
237 _MIN_OBJECTS_PER_SEC = 1_500
238
239 @pytest.fixture(autouse=True)
240 def no_fsync(self) -> None:
241 """Mock out all fsync calls so the test measures algorithmic throughput."""
242 with patch("muse.core.object_store._fsync_fd", return_value=None):
243 yield
244
245 @pytest.mark.slow
246 def test_write_object_throughput_10k(self, tmp_path: pathlib.Path) -> None:
247 """Write 10 000 4-KiB objects and assert throughput ≥ 2 000 objects/sec."""
248 repo = _repo(tmp_path)
249 N = 10_000
250 items = [
251 (
252 _sha256(f"perf-obj-{i:08d}".encode() * 16),
253 f"perf-obj-{i:08d}".encode() * 16,
254 )
255 for i in range(N)
256 ]
257
258 t0 = time.perf_counter()
259 for oid, content in items:
260 write_object(repo, oid, content)
261 elapsed = time.perf_counter() - t0
262 rate = N / elapsed
263
264 assert rate >= self._MIN_OBJECTS_PER_SEC, (
265 f"write_object throughput {rate:.0f} objects/sec is below the "
266 f"minimum {self._MIN_OBJECTS_PER_SEC} objects/sec. "
267 f"(10k objects took {elapsed:.2f}s.)"
268 )
269
270 @pytest.mark.perf
271 def test_write_object_throughput_1k_fast(self, tmp_path: pathlib.Path) -> None:
272 """Smoke-speed: 1 000 objects must complete within 2 seconds."""
273 repo = _repo(tmp_path)
274 N = 1_000
275 items = [
276 (
277 _sha256(f"fast-obj-{i:08d}".encode() * 8),
278 f"fast-obj-{i:08d}".encode() * 8,
279 )
280 for i in range(N)
281 ]
282
283 t0 = time.perf_counter()
284 for oid, content in items:
285 write_object(repo, oid, content)
286 elapsed = time.perf_counter() - t0
287
288 assert elapsed <= 2.0, (
289 f"1 000 write_object calls took {elapsed:.2f}s — expected ≤ 2.0s."
290 )
291
292
293 class TestSnapshotManifestThroughput:
294 """build_snapshot_manifest must sustain ≥ 10 000 files/sec."""
295
296 _MIN_FILES_PER_SEC = 10_000
297
298 @pytest.mark.slow
299 def test_snapshot_5k_files(self, tmp_path: pathlib.Path) -> None:
300 """Build manifest of 5 000 files; assert ≥ 10 000 files/sec."""
301 root = tmp_path / "workdir"
302 root.mkdir()
303 (root / ".muse").mkdir()
304 (root / ".muse" / "repo.json").write_text('{"repo_id": "bench"}')
305
306 # 50 dirs × 100 files = 5 000 files
307 for d in range(50):
308 dp = root / f"pkg_{d:03d}"
309 dp.mkdir()
310 for f in range(100):
311 (dp / f"file_{f:03d}.py").write_bytes(
312 f"# content-{d}-{f}\n".encode() * 50
313 )
314
315 t0 = time.perf_counter()
316 manifest = build_snapshot_manifest(root)
317 elapsed = time.perf_counter() - t0
318 rate = len(manifest) / elapsed
319
320 assert len(manifest) == 5_000
321 assert rate >= self._MIN_FILES_PER_SEC, (
322 f"build_snapshot_manifest throughput {rate:.0f} files/sec is below "
323 f"the minimum {self._MIN_FILES_PER_SEC} files/sec. "
324 f"(5k files took {elapsed:.3f}s.)"
325 )
326
327 def test_snapshot_500_files_fast(self, tmp_path: pathlib.Path) -> None:
328 """Smoke-speed: 500-file manifest must complete within 500 ms."""
329 root = tmp_path / "workdir"
330 root.mkdir()
331 (root / ".muse").mkdir()
332 (root / ".muse" / "repo.json").write_text('{"repo_id": "bench"}')
333
334 for d in range(25):
335 dp = root / f"pkg_{d:02d}"
336 dp.mkdir()
337 for f in range(20):
338 (dp / f"file_{f:02d}.py").write_bytes(b"x" * 200)
339
340 t0 = time.perf_counter()
341 manifest = build_snapshot_manifest(root)
342 elapsed = time.perf_counter() - t0
343
344 assert len(manifest) == 500
345 assert elapsed <= 0.5, (
346 f"500-file manifest took {elapsed*1000:.1f} ms — expected ≤ 500 ms."
347 )
348
349
350 # ---------------------------------------------------------------------------
351 # Phase 3.2 — Concurrent agent write storm
352 # ---------------------------------------------------------------------------
353
354
355 class TestConcurrentWriteObjectStorm:
356 """200 threads writing distinct objects — no corruption, no data loss."""
357
358 def test_200_threads_write_distinct_objects(self, tmp_path: pathlib.Path) -> None:
359 """200 threads each write 50 distinct objects; all must be readable after join."""
360 repo = _repo(tmp_path)
361 N_THREADS = 200
362 N_PER_THREAD = 50
363 written: _FileStore = {}
364 lock = threading.Lock()
365 errors: list[str] = []
366
367 # Pre-compute all objects to avoid per-thread hashing noise.
368 all_items: list[list[tuple[str, bytes]]] = []
369 for t in range(N_THREADS):
370 thread_items: list[tuple[str, bytes]] = []
371 for i in range(N_PER_THREAD):
372 content = f"thread-{t:03d}-obj-{i:03d}".encode() * 4
373 oid = _sha256(content)
374 thread_items.append((oid, content))
375 with lock:
376 written[oid] = content
377 all_items.append(thread_items)
378
379 def writer(items: list[tuple[str, bytes]]) -> None:
380 try:
381 for oid, content in items:
382 write_object(repo, oid, content)
383 except Exception as exc:
384 with lock:
385 errors.append(str(exc))
386
387 threads = [
388 threading.Thread(target=writer, args=(all_items[t],))
389 for t in range(N_THREADS)
390 ]
391 for th in threads:
392 th.start()
393 for th in threads:
394 th.join(timeout=30.0)
395
396 assert not errors, f"Write errors during concurrent storm: {errors[:3]}"
397
398 # Verify every object is readable and byte-identical.
399 missing: list[str] = []
400 corrupt: list[str] = []
401 for oid, expected in written.items():
402 if not has_object(repo, oid):
403 missing.append(oid[:8])
404 else:
405 actual = read_object(repo, oid)
406 if actual != expected:
407 corrupt.append(oid[:8])
408
409 assert not missing, f"{len(missing)} objects missing after concurrent write storm"
410 assert not corrupt, f"{len(corrupt)} objects corrupted after concurrent write storm"
411
412
413 class TestConcurrentWriteCommitStorm:
414 """200 threads writing distinct commits — all must be readable after join."""
415
416 def test_200_threads_write_distinct_commits(self, tmp_path: pathlib.Path) -> None:
417 """200 threads each write 25 distinct commits; all must be readable."""
418 repo = _repo(tmp_path)
419 N_THREADS = 200
420 N_PER_THREAD = 25
421 snap_id = "c" * 64
422 all_records: list[list[CommitRecord]] = []
423 errors: list[str] = []
424 lock = threading.Lock()
425
426 for t in range(N_THREADS):
427 thread_recs: list[CommitRecord] = []
428 for i in range(N_PER_THREAD):
429 rec = _make_commit(t * N_PER_THREAD + i, snap_id)
430 thread_recs.append(rec)
431 all_records.append(thread_recs)
432
433 def writer(recs: list[CommitRecord]) -> None:
434 try:
435 for rec in recs:
436 write_commit(repo, rec)
437 except Exception as exc:
438 with lock:
439 errors.append(str(exc))
440
441 threads = [
442 threading.Thread(target=writer, args=(all_records[t],))
443 for t in range(N_THREADS)
444 ]
445 for th in threads:
446 th.start()
447 for th in threads:
448 th.join(timeout=30.0)
449
450 assert not errors, f"Write errors during concurrent commit storm: {errors[:3]}"
451
452 # All commits must be readable.
453 total = N_THREADS * N_PER_THREAD
454 missing: list[str] = []
455 for recs in all_records:
456 for rec in recs:
457 result = read_commit(repo, rec.commit_id)
458 if result is None:
459 missing.append(rec.commit_id[:8])
460
461 assert not missing, (
462 f"{len(missing)}/{total} commits missing after concurrent write storm"
463 )
464
465
466 class TestConcurrentWriteBranchRef:
467 """100 threads racing on write_branch_ref — last write wins, no corruption."""
468
469 def test_100_threads_write_branch_ref_last_write_wins(
470 self, tmp_path: pathlib.Path
471 ) -> None:
472 """100 threads each writing write_branch_ref; result must be a valid 64-char hex ID."""
473 repo = _repo(tmp_path)
474 refs_dir = repo / ".muse" / "refs" / "heads" # already created by _repo()
475
476 N = 100
477 snap_id = "d" * 64
478 commit_ids: list[str] = []
479 errors: list[str] = []
480 lock = threading.Lock()
481
482 for i in range(N):
483 cid = compute_commit_id([], snap_id, f"head-{i}", "2026-01-01T00:00:00+00:00")
484 commit_ids.append(cid)
485
486 def writer(cid: str) -> None:
487 try:
488 write_branch_ref(repo, "main", cid)
489 except Exception as exc:
490 with lock:
491 errors.append(str(exc))
492
493 threads = [threading.Thread(target=writer, args=(commit_ids[i],)) for i in range(N)]
494 for th in threads:
495 th.start()
496 for th in threads:
497 th.join(timeout=10.0)
498
499 assert not errors, f"Errors during concurrent write_branch_ref: {errors[:3]}"
500
501 # The branch ref must contain exactly one of the written IDs.
502 ref_file = refs_dir / "main"
503 assert ref_file.exists(), "Branch ref file was lost after concurrent writes"
504 final_cid = ref_file.read_text(encoding="utf-8").strip()
505 # Commit IDs are now "sha256:<64hex>" = 71 chars
506 assert final_cid.startswith("sha256:"), f"Branch ref has unexpected format: {final_cid!r}"
507 hex_part = final_cid[len("sha256:"):]
508 assert len(hex_part) == 64, f"Branch ref hex part has unexpected length: {final_cid!r}"
509 assert all(c in "0123456789abcdef" for c in hex_part), (
510 f"Branch ref is not a valid hex ID: {final_cid!r}"
511 )
512 assert final_cid in commit_ids, (
513 f"Branch ref {final_cid[:19]} is not one of the written IDs"
514 )
515
516
517 # ---------------------------------------------------------------------------
518 # Phase 3.3 — Memory ceiling under load
519 # ---------------------------------------------------------------------------
520
521
522 class TestMemoryCeiling:
523 """Peak memory must not grow proportionally to the number of objects or commits."""
524
525 _MAX_WRITE_COMMIT_MIB = 512
526 _MAX_SNAPSHOT_MIB = 128
527
528 @pytest.mark.slow
529 def test_write_10k_commits_peak_rss_under_512_mib(
530 self, tmp_path: pathlib.Path
531 ) -> None:
532 """Writing 10 000 commits must not exceed 512 MiB peak RSS.
533
534 write_commit buffers one commit at a time; it must not accumulate
535 a list of all commits in memory.
536 """
537 repo = _repo(tmp_path)
538 snap_id = "e" * 64
539 N = 10_000
540
541 tracemalloc.start()
542 tracemalloc.clear_traces()
543
544 for i in range(N):
545 rec = _make_commit(i, snap_id)
546 write_commit(repo, rec)
547
548 _, peak_bytes = tracemalloc.get_traced_memory()
549 tracemalloc.stop()
550
551 peak_mib = peak_bytes / (1024 * 1024)
552 assert peak_mib <= self._MAX_WRITE_COMMIT_MIB, (
553 f"write_commit peak allocation {peak_mib:.1f} MiB exceeds "
554 f"{self._MAX_WRITE_COMMIT_MIB} MiB for 10k commits. "
555 "write_commit must stream one record at a time."
556 )
557
558 def test_snapshot_5k_files_peak_rss_under_128_mib(
559 self, tmp_path: pathlib.Path
560 ) -> None:
561 """build_snapshot_manifest on 5 000 small files stays under 128 MiB."""
562 root = tmp_path / "workdir"
563 root.mkdir()
564 (root / ".muse").mkdir()
565 (root / ".muse" / "repo.json").write_text('{"repo_id": "bench"}')
566
567 for d in range(50):
568 dp = root / f"d_{d:02d}"
569 dp.mkdir()
570 for f in range(100):
571 (dp / f"f_{f:02d}.txt").write_bytes(b"x" * 512)
572
573 tracemalloc.start()
574 tracemalloc.clear_traces()
575 build_snapshot_manifest(root)
576 _, peak_bytes = tracemalloc.get_traced_memory()
577 tracemalloc.stop()
578
579 peak_mib = peak_bytes / (1024 * 1024)
580 assert peak_mib <= self._MAX_SNAPSHOT_MIB, (
581 f"build_snapshot_manifest peak allocation {peak_mib:.1f} MiB "
582 f"exceeds {self._MAX_SNAPSHOT_MIB} MiB for 5k files. "
583 "The manifest dict must not accumulate large intermediate buffers."
584 )
585
586
587 # ---------------------------------------------------------------------------
588 # Phase 3.1 — Shard-cache amortisation (regression guard)
589 # ---------------------------------------------------------------------------
590
591
592 class TestShardCacheAmortisation:
593 """The shard-validation cache eliminates O(objects) resolve() calls.
594
595 This test is not a timing test — it verifies the structural invariant:
596 after N writes to the same shard, _created_object_shards contains exactly
597 that shard entry and subsequent writes to the same shard do not re-trigger
598 path-resolution (proved by exercising the idempotent write path).
599 """
600
601 def test_shard_cache_populated_after_first_write(
602 self, tmp_path: pathlib.Path
603 ) -> None:
604 """After the first write to a shard, _created_object_shards contains its path."""
605 repo = _repo(tmp_path)
606 content = b"shard-cache-test"
607 oid = _sha256(content)
608 shard_str = str(object_path(repo, oid).parent)
609
610 # Ensure cache starts without this shard.
611 _created_object_shards.discard(shard_str)
612
613 write_object(repo, oid, content)
614
615 assert shard_str in _created_object_shards, (
616 f"Shard {oid[:2]} not recorded in _created_object_shards after first write. "
617 "The amortisation optimisation is not active."
618 )
619
620 def test_repeated_writes_to_same_shard_succeed(
621 self, tmp_path: pathlib.Path
622 ) -> None:
623 """50 distinct writes to the same shard all land correctly."""
624 repo = _repo(tmp_path)
625 # Force a fixed shard prefix by constructing objects with the same prefix.
626 # We use a known prefix and craft content that happens to hash to it.
627 # Easier: just write 50 objects and verify they all land.
628 N = 50
629 items = [
630 (
631 _sha256(f"repeat-shard-{i:04d}".encode()),
632 f"repeat-shard-{i:04d}".encode(),
633 )
634 for i in range(N)
635 ]
636
637 for oid, content in items:
638 write_object(repo, oid, content)
639
640 missing = [oid[:8] for oid, _ in items if not has_object(repo, oid)]
641 assert not missing, (
642 f"{len(missing)} objects missing after repeated writes: {missing[:5]}"
643 )
644
645 def test_idempotent_write_bypasses_shard_write(
646 self, tmp_path: pathlib.Path
647 ) -> None:
648 """write_object returns False (idempotent) on the second call for the same OID."""
649 repo = _repo(tmp_path)
650 content = b"idempotent-check"
651 oid = _sha256(content)
652
653 first = write_object(repo, oid, content)
654 second = write_object(repo, oid, content)
655
656 assert first is True, "First write should return True"
657 assert second is False, "Second write should return False (already exists)"
658
659
660 # ---------------------------------------------------------------------------
661 # Phase 3.3 extended — Linux-scale memory ceiling
662 # ---------------------------------------------------------------------------
663
664
665 class TestGetAllCommitsMemory:
666 """get_all_commits must not OOM under Linux-scale commit counts.
667
668 This is the highest-risk accumulator: it loads *every* CommitRecord in the
669 store into a list simultaneously with no cap. At 100k commits × ~2 KB per
670 serialised record the baseline is ~200 MiB. Commits with large
671 ``structured_delta`` payloads can be 100 KB each (100k × 100 KB = 10 GiB).
672 The 64 MiB per-record msgpack cap (MAX_MSGPACK_BYTES) provides the guard.
673
674 The @slow variants build real on-disk commit chains; the fast variant uses
675 a small chain to confirm the structural property with tracemalloc.
676 """
677
678 def test_get_all_commits_1k_peak_rss_under_128_mib(
679 self, tmp_path: pathlib.Path
680 ) -> None:
681 """get_all_commits on 1 000 commits stays under 128 MiB (fast smoke)."""
682 repo = _repo(tmp_path)
683 N = 1_000
684 snap_id = "aa" * 32
685 for i in range(N):
686 write_commit(repo, _make_commit(i, snap_id))
687
688 tracemalloc.start()
689 tracemalloc.clear_traces()
690 results = get_all_commits(repo)
691 _, peak_bytes = tracemalloc.get_traced_memory()
692 tracemalloc.stop()
693
694 assert len(results) == N, f"Expected {N} commits, got {len(results)}"
695 peak_mib = peak_bytes / (1024 * 1024)
696 assert peak_mib <= 128, (
697 f"get_all_commits({N}) peak {peak_mib:.1f} MiB — expected ≤ 128 MiB. "
698 "CommitRecord size has grown; re-audit the dataclass fields."
699 )
700
701 @pytest.mark.slow
702 def test_get_all_commits_100k_under_2_gib(
703 self, tmp_path: pathlib.Path
704 ) -> None:
705 """get_all_commits on 100 000 commits stays under 2 GiB.
706
707 100k × minimal CommitRecord ≈ 200 MiB. The 2 GiB ceiling allows a 10×
708 margin for realistic payloads (metadata, structured_delta) while still
709 catching runaway accumulation.
710 """
711 repo = _repo(tmp_path)
712 N = 100_000
713 snap_id = "bb" * 32
714
715 for i in range(N):
716 write_commit(repo, _make_commit(i, snap_id))
717
718 _MAX_MIB = 2_048 # 2 GiB
719
720 tracemalloc.start()
721 tracemalloc.clear_traces()
722 results = get_all_commits(repo)
723 _, peak_bytes = tracemalloc.get_traced_memory()
724 tracemalloc.stop()
725
726 assert len(results) == N
727 peak_mib = peak_bytes / (1024 * 1024)
728 assert peak_mib <= _MAX_MIB, (
729 f"get_all_commits(100k) peak {peak_mib:.1f} MiB exceeds {_MAX_MIB} MiB. "
730 "The function loads all CommitRecords into a list simultaneously — "
731 "consider streaming or paginating for very large repos."
732 )
733
734
735 class TestGetCommitsForBranchWalkCap:
736 """get_commits_for_branch must honour its walk cap.
737
738 The cap is the primary memory guard for ``muse log --json``. A branch with
739 N commits deeper than the cap must return exactly cap records, not N — even
740 when filters are active and the caller passes max_count=0.
741 """
742
743 def test_walk_cap_bounds_returned_records(
744 self, tmp_path: pathlib.Path
745 ) -> None:
746 """Chain of 500 commits with cap=100 returns exactly 100 records."""
747 repo = _repo(tmp_path)
748 N = 500
749 CAP = 100
750 _write_chain(repo, "main", N)
751
752 results = get_commits_for_branch(repo, "bench", "main", max_count=CAP)
753
754 assert len(results) == CAP, (
755 f"Expected cap={CAP} records, got {len(results)} — "
756 "walk cap is not being respected."
757 )
758
759 def test_walk_cap_memory_bounded_deep_chain(
760 self, tmp_path: pathlib.Path
761 ) -> None:
762 """2 000-commit chain with default cap stays under 64 MiB."""
763 repo = _repo(tmp_path)
764 N = 2_000
765 CAP = 500
766 _write_chain(repo, "main", N)
767
768 tracemalloc.start()
769 tracemalloc.clear_traces()
770 results = get_commits_for_branch(repo, "bench", "main", max_count=CAP)
771 _, peak_bytes = tracemalloc.get_traced_memory()
772 tracemalloc.stop()
773
774 assert len(results) == CAP
775 peak_mib = peak_bytes / (1024 * 1024)
776 assert peak_mib <= 64, (
777 f"get_commits_for_branch(cap={CAP}) peak {peak_mib:.1f} MiB — "
778 f"expected ≤ 64 MiB for {CAP} records."
779 )
780
781
782 class TestLogJsonStreaming:
783 """``muse log --json`` must not double-buffer commit JSON strings.
784
785 The previous implementation accumulated ``commit_jsons: list[str]`` before
786 writing to stdout — doubling peak memory vs. the CommitRecord list alone.
787 The fix uses a first-item flag to write each record inline immediately.
788 This test verifies the fix: tracemalloc peak for a 500-commit log should
789 not contain the signature of a large string buffer.
790 """
791
792 def test_log_json_output_is_valid_json(
793 self, tmp_path: pathlib.Path
794 ) -> None:
795 """muse log --json on a 100-commit branch emits valid JSON with all commits."""
796 import json
797
798 from tests.cli_test_helper import CliRunner
799
800 repo = _repo(tmp_path)
801 N = 100
802 _write_chain(repo, "main", N)
803 (repo / ".muse" / "config.toml").write_text("")
804
805 runner = CliRunner()
806 result = runner.invoke(
807 None,
808 ["log", "--format", "json", "--max-count", str(N)],
809 env={"MUSE_REPO_ROOT": str(repo)},
810 )
811 assert result.exit_code == 0, f"muse log --json failed:\n{result.output}"
812
813 payload = json.loads(result.output)
814 assert "commits" in payload, f"Missing 'commits' key: {payload}"
815 assert len(payload["commits"]) == N, (
816 f"Expected {N} commits in JSON output, got {len(payload['commits'])}"
817 )
818
819 def test_log_json_empty_repo_returns_empty_array(
820 self, tmp_path: pathlib.Path
821 ) -> None:
822 """muse log --json on a branch with no commits returns empty commits array."""
823 import json
824
825 from tests.cli_test_helper import CliRunner
826
827 repo = _repo(tmp_path)
828 (repo / ".muse" / "config.toml").write_text("")
829
830 runner = CliRunner()
831 result = runner.invoke(
832 None,
833 ["log", "--format", "json"],
834 env={"MUSE_REPO_ROOT": str(repo)},
835 )
836 # No commits → either exit 0 with empty commits array or a graceful
837 # "(no commits)" message; both are acceptable.
838 if result.exit_code == 0 and result.output.startswith("{"):
839 payload = json.loads(result.output)
840 assert payload.get("commits") == []
841
842
843 class TestFindMergeBaseMemory:
844 """find_merge_base must fire its cap cleanly, not OOM.
845
846 The default ``max_ancestors`` cap is 50 000. A chain deeper than the cap
847 must raise ``MuseCLIError`` with an actionable message rather than
848 exhausting all available memory. This test confirms the error path, not
849 the OOM path.
850 """
851
852 def test_merge_base_cap_raises_gracefully_on_deep_chain(
853 self, tmp_path: pathlib.Path
854 ) -> None:
855 """Chain deeper than max_ancestors raises MuseCLIError, not MemoryError."""
856 from muse.core.errors import MuseCLIError
857
858 repo = _repo(tmp_path)
859 (repo / ".muse" / "config.toml").write_text(
860 "[limits]\nmax_ancestors = 50\n"
861 )
862
863 tip_a = _write_chain(repo, "branchA", 60, snap_id="cc" * 32)
864 # branchB is entirely separate — no common ancestor with branchA.
865 tip_b = _write_chain(repo, "branchB", 60, snap_id="dd" * 32, start=1000)
866
867 with pytest.raises(MuseCLIError, match="max_ancestors"):
868 find_merge_base(repo, tip_a, tip_b)
869
870 def test_merge_base_found_within_cap(
871 self, tmp_path: pathlib.Path
872 ) -> None:
873 """find_merge_base finds the base when both branches are within cap."""
874 repo = _repo(tmp_path)
875 (repo / ".muse" / "config.toml").write_text(
876 "[limits]\nmax_ancestors = 500\n"
877 )
878
879 # Build a common root commit.
880 snap_id = "ee" * 32
881 root_cid = compute_commit_id([], snap_id, "root", "2026-01-01T00:00:00+00:00")
882 root_rec = CommitRecord(
883 commit_id=root_cid,
884 repo_id="bench",
885 branch="main",
886 snapshot_id=snap_id,
887 message="root",
888 committed_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
889 parent_commit_id=None,
890 parent2_commit_id=None,
891 author="test",
892 metadata={},
893 structured_delta=None,
894 sem_ver_bump="none",
895 breaking_changes=[],
896 agent_id="",
897 model_id="",
898 toolchain_id="",
899 prompt_hash="",
900 signature="",
901 signer_key_id="",
902 )
903 write_commit(repo, root_rec)
904
905 # Extend branchA and branchB from the same root.
906 parent_a: str | None = root_cid
907 tip_a = root_cid
908 for i in range(20):
909 msg = f"a-{i:04d}"
910 ts = datetime.datetime(2026, 1, 2, tzinfo=datetime.timezone.utc)
911 cid = compute_commit_id([parent_a] if parent_a else [], snap_id, msg, ts.isoformat())
912 rec = CommitRecord(
913 commit_id=cid,
914 repo_id="bench",
915 branch="branchA",
916 snapshot_id=snap_id,
917 message=msg,
918 committed_at=ts,
919 parent_commit_id=parent_a,
920 parent2_commit_id=None,
921 author="test",
922 metadata={},
923 structured_delta=None,
924 sem_ver_bump="none",
925 breaking_changes=[],
926 agent_id="",
927 model_id="",
928 toolchain_id="",
929 prompt_hash="",
930 signature="",
931 signer_key_id="",
932 )
933 write_commit(repo, rec)
934 parent_a = cid
935 tip_a = cid
936
937 parent_b: str | None = root_cid
938 tip_b = root_cid
939 for i in range(15):
940 msg = f"b-{i:04d}"
941 ts = datetime.datetime(2026, 1, 3, tzinfo=datetime.timezone.utc)
942 cid = compute_commit_id([parent_b] if parent_b else [], snap_id, msg, ts.isoformat())
943 rec = CommitRecord(
944 commit_id=cid,
945 repo_id="bench",
946 branch="branchB",
947 snapshot_id=snap_id,
948 message=msg,
949 committed_at=ts,
950 parent_commit_id=parent_b,
951 parent2_commit_id=None,
952 author="test",
953 metadata={},
954 structured_delta=None,
955 sem_ver_bump="none",
956 breaking_changes=[],
957 agent_id="",
958 model_id="",
959 toolchain_id="",
960 prompt_hash="",
961 signature="",
962 signer_key_id="",
963 )
964 write_commit(repo, rec)
965 parent_b = cid
966 tip_b = cid
967
968 base = find_merge_base(repo, tip_a, tip_b)
969 assert base == root_cid, (
970 f"Expected merge base {root_cid[:8]}, got {base[:8] if base else None}"
971 )
972
973 def test_merge_base_memory_bounded_within_cap(
974 self, tmp_path: pathlib.Path
975 ) -> None:
976 """find_merge_base BFS uses bounded memory proportional to max_ancestors."""
977 from muse.core.errors import MuseCLIError
978
979 repo = _repo(tmp_path)
980 CAP = 200
981 (repo / ".muse" / "config.toml").write_text(
982 f"[limits]\nmax_ancestors = {CAP}\n"
983 )
984
985 # Build a shallow divergence: 50 commits each, well inside the cap.
986 snap_id = "ff" * 32
987 root_cid = compute_commit_id([], snap_id, "base", "2026-01-01T00:00:00+00:00")
988 root_rec = CommitRecord(
989 commit_id=root_cid,
990 repo_id="bench",
991 branch="main",
992 snapshot_id=snap_id,
993 message="base",
994 committed_at=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
995 parent_commit_id=None,
996 parent2_commit_id=None,
997 author="t",
998 metadata={},
999 structured_delta=None,
1000 sem_ver_bump="none",
1001 breaking_changes=[],
1002 agent_id="",
1003 model_id="",
1004 toolchain_id="",
1005 prompt_hash="",
1006 signature="",
1007 signer_key_id="",
1008 )
1009 write_commit(repo, root_rec)
1010
1011 def _extend(parent: str, prefix: str, n: int) -> str:
1012 tip = parent
1013 for i in range(n):
1014 msg = f"{prefix}-{i:04d}"
1015 ts = datetime.datetime(2026, 2, 1, tzinfo=datetime.timezone.utc)
1016 cid = compute_commit_id([tip], snap_id, msg, ts.isoformat())
1017 rec = CommitRecord(
1018 commit_id=cid,
1019 repo_id="bench",
1020 branch=prefix,
1021 snapshot_id=snap_id,
1022 message=msg,
1023 committed_at=ts,
1024 parent_commit_id=tip,
1025 parent2_commit_id=None,
1026 author="t",
1027 metadata={},
1028 structured_delta=None,
1029 sem_ver_bump="none",
1030 breaking_changes=[],
1031 agent_id="",
1032 model_id="",
1033 toolchain_id="",
1034 prompt_hash="",
1035 signature="",
1036 signer_key_id="",
1037 )
1038 write_commit(repo, rec)
1039 tip = cid
1040 return tip
1041
1042 tip_a = _extend(root_cid, "xa", 50)
1043 tip_b = _extend(root_cid, "xb", 50)
1044
1045 tracemalloc.start()
1046 tracemalloc.clear_traces()
1047 try:
1048 base = find_merge_base(repo, tip_a, tip_b)
1049 except MuseCLIError:
1050 base = None # Cap triggered — that's also a valid outcome.
1051 _, peak_bytes = tracemalloc.get_traced_memory()
1052 tracemalloc.stop()
1053
1054 peak_mib = peak_bytes / (1024 * 1024)
1055 assert peak_mib <= 64, (
1056 f"find_merge_base peak {peak_mib:.1f} MiB — expected ≤ 64 MiB "
1057 f"for two 50-commit branches (cap={CAP})."
1058 )
1059
1060
1061 class TestSnapshotManifest75kFiles:
1062 """build_snapshot_manifest on 75 000 files must stay under 512 MiB.
1063
1064 The manifest dict holds only ``{rel_path: sha256_hex}`` — strings only.
1065 75k × (60-char path + 64-char hash) ≈ 9.3 MiB for the dict alone.
1066 The peak should be dominated by the stat-cache msgpack load, not by
1067 the manifest itself. 512 MiB is a generous ceiling to catch any
1068 accidental full-file-content buffering.
1069 """
1070
1071 @pytest.mark.slow
1072 def test_75k_files_peak_rss_under_512_mib(
1073 self, tmp_path: pathlib.Path
1074 ) -> None:
1075 """build_snapshot_manifest on 75 000 small files stays under 512 MiB."""
1076 root = tmp_path / "workdir"
1077 root.mkdir()
1078 (root / ".muse").mkdir()
1079 (root / ".muse" / "repo.json").write_text('{"repo_id": "bench"}')
1080
1081 # 750 dirs × 100 files = 75 000 files, each 128 bytes.
1082 for d in range(750):
1083 dp = root / f"pkg_{d:04d}"
1084 dp.mkdir()
1085 for f in range(100):
1086 (dp / f"f_{f:03d}.py").write_bytes(b"x" * 128)
1087
1088 tracemalloc.start()
1089 tracemalloc.clear_traces()
1090 manifest = build_snapshot_manifest(root)
1091 _, peak_bytes = tracemalloc.get_traced_memory()
1092 tracemalloc.stop()
1093
1094 assert len(manifest) == 75_000, (
1095 f"Expected 75 000 files in manifest, got {len(manifest)}"
1096 )
1097 peak_mib = peak_bytes / (1024 * 1024)
1098 assert peak_mib <= 512, (
1099 f"build_snapshot_manifest(75k) peak {peak_mib:.1f} MiB exceeds 512 MiB. "
1100 "File content must never be loaded into memory — only stat + SHA-256."
1101 )
1102
1103 def test_10k_files_peak_rss_under_64_mib(
1104 self, tmp_path: pathlib.Path
1105 ) -> None:
1106 """10k-file manifest stays under 64 MiB (fast smoke for the ceiling property)."""
1107 root = tmp_path / "workdir"
1108 root.mkdir()
1109 (root / ".muse").mkdir()
1110 (root / ".muse" / "repo.json").write_text('{"repo_id": "bench"}')
1111
1112 for d in range(100):
1113 dp = root / f"pkg_{d:03d}"
1114 dp.mkdir()
1115 for f in range(100):
1116 (dp / f"f_{f:03d}.py").write_bytes(b"y" * 64)
1117
1118 tracemalloc.start()
1119 tracemalloc.clear_traces()
1120 manifest = build_snapshot_manifest(root)
1121 _, peak_bytes = tracemalloc.get_traced_memory()
1122 tracemalloc.stop()
1123
1124 assert len(manifest) == 10_000
1125 peak_mib = peak_bytes / (1024 * 1024)
1126 assert peak_mib <= 64, (
1127 f"build_snapshot_manifest(10k) peak {peak_mib:.1f} MiB — expected ≤ 64 MiB."
1128 )
1129
1130
1131 class TestWalkCommitsBetweenCap:
1132 """walk_commits_between must truncate at its cap, never OOM.
1133
1134 This function is used by ``muse status --json`` for ahead/behind counts.
1135 A branch 100k commits ahead of remote must return at most ``max_commits``
1136 records and set ``truncated=True`` — it must never allocate memory
1137 proportional to the full chain depth.
1138 """
1139
1140 def test_truncates_at_cap_not_oom(
1141 self, tmp_path: pathlib.Path
1142 ) -> None:
1143 """Chain of 1 000 commits with cap=100 truncates, doesn't exhaust memory."""
1144 repo = _repo(tmp_path)
1145 N = 1_000
1146 CAP = 100
1147 tip = _write_chain(repo, "main", N)
1148
1149 result = walk_commits_between_result(repo, tip, max_commits=CAP)
1150
1151 assert result["truncated"] is True, (
1152 "Expected truncated=True for chain longer than cap"
1153 )
1154 assert len(result["commits"]) == CAP, (
1155 f"Expected exactly {CAP} commits, got {len(result['commits'])}"
1156 )
1157
1158 def test_truncation_memory_bounded(
1159 self, tmp_path: pathlib.Path
1160 ) -> None:
1161 """walk_commits_between_result peak memory is bounded by cap, not chain depth."""
1162 repo = _repo(tmp_path)
1163 N = 2_000
1164 CAP = 200
1165 tip = _write_chain(repo, "main", N)
1166
1167 tracemalloc.start()
1168 tracemalloc.clear_traces()
1169 result = walk_commits_between_result(repo, tip, max_commits=CAP)
1170 _, peak_bytes = tracemalloc.get_traced_memory()
1171 tracemalloc.stop()
1172
1173 assert result["count"] == CAP
1174 peak_mib = peak_bytes / (1024 * 1024)
1175 assert peak_mib <= 32, (
1176 f"walk_commits_between_result(cap={CAP}) peak {peak_mib:.1f} MiB — "
1177 "memory must be proportional to cap, not chain depth."
1178 )
File History 2 commits
sha256:88ac91129873e6a496e9189515aa690eb893ae25d69c8f72af141a2be5068eb3 docs: docstring sprint contract→find-symbol — idiomatic run… Sonnet 4.6 patch 147 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 150 days ago