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