gabriel / muse public
test_perf_pack.py python
836 lines 31.0 KB
Raw
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
1 """Phase 3.4 — Pack and unpack at scale.
2
3 Target metrics (measured on a 2024 MacBook Pro M4, macOS 15):
4
5 build_pack (10 000 objects × 4 KiB): < 60 s [@slow]
6 apply_pack (10 000 objects × 4 KiB): < 60 s [@slow]
7 verify-pack (10 000 objects × 4 KiB): < 120 s [@slow] (in practice much faster)
8 collect_object_ids (1 000 objects): < 1 s (no blob reads)
9
10 Edges verified beyond the plan:
11
12 a. ``build_pack`` loads ALL blob bytes simultaneously — peak RSS ≈ 2× blob total.
13 b. ``have=`` filter correctly reduces both commit count and blob payload.
14 c. ``MAX_PACK_OBJECTS`` applies to total_items (commits + snapshots + objects),
15 not per-type — a pack within per-type limits can still be rejected.
16 d. Oversized object (> MAX_OBJECT_WRITE_BYTES) is silently skipped, not raised —
17 caller sees objects_skipped++ but no error; documented behaviour.
18 e. ``verify-pack --stat`` is purely structural — no SHA-256, near-instant.
19 f. Duplicate OID dedup in ``apply_pack`` — each OID written at most once.
20 g. Round-trip integrity: build → msgpack-serialize → apply → all objects present.
21 """
22
23 from __future__ import annotations
24
25 import datetime
26 import pathlib
27 import sys
28 import tempfile
29 import time
30 import tracemalloc
31
32 import msgpack
33 import pytest
34 from unittest.mock import patch
35
36 from muse.core.object_store import write_object
37 from muse.core.pack import (
38 MAX_OBJECT_WRITE_BYTES,
39 MAX_PACK_OBJECTS,
40 ObjectPayload,
41 MPackBundle as PackBundle,
42 apply_mpack as apply_pack,
43 build_mpack as build_pack,
44 collect_object_ids,
45 )
46 from muse.core.snapshot import compute_commit_id, compute_snapshot_id
47
48 from muse.core._types import Manifest, blob_id
49 from muse.core.store import (
50 CommitRecord,
51 MAX_PACK_MSGPACK_BYTES,
52 SnapshotRecord,
53 write_branch_ref,
54 write_commit,
55 write_snapshot,
56 )
57
58 # ---------------------------------------------------------------------------
59 # Helpers
60 # ---------------------------------------------------------------------------
61
62
63 def _sha256(data: bytes) -> str:
64 return blob_id(data)
65
66
67 def _make_repo(tmp: pathlib.Path) -> pathlib.Path:
68 tmp.mkdir(parents=True, exist_ok=True)
69 muse = tmp / ".muse"
70 muse.mkdir()
71 (muse / "repo.json").write_text('{"repo_id":"bench","owner":"bench"}')
72 for d in ("commits", "snapshots", "objects"):
73 (muse / d).mkdir()
74 (muse / "refs" / "heads").mkdir(parents=True)
75 (muse / "HEAD").write_text("ref: refs/heads/main\n")
76 (muse / "config.toml").write_text("")
77 return tmp
78
79
80 def _fresh_repo(tmp: pathlib.Path) -> pathlib.Path:
81 tmp.mkdir(parents=True, exist_ok=True)
82 muse = tmp / ".muse"
83 muse.mkdir()
84 (muse / "repo.json").write_text('{"repo_id":"dst"}')
85 for d in ("commits", "snapshots", "objects"):
86 (muse / d).mkdir()
87 return tmp
88
89
90 def _populate(
91 repo: pathlib.Path,
92 n_commits: int = 10,
93 n_unique_objects: int = 10,
94 blob_size: int = 4096,
95 branch: str = "main",
96 start: int = 0,
97 ) -> tuple[str, dict[str, str]]:
98 """Write *n_unique_objects* blobs and a *n_commits* chain.
99
100 Returns ``(tip_commit_id, {path: oid})`` manifest.
101 """
102 blobs: Manifest = {}
103 for i in range(n_unique_objects):
104 data = f"obj-{i + start:08d}-".encode() + b"x" * blob_size
105 oid = _sha256(data)
106 write_object(repo, oid, data)
107 blobs[f"file_{i:04d}.py"] = oid
108
109 sid = compute_snapshot_id(blobs)
110 write_snapshot(repo, SnapshotRecord(snapshot_id=sid, manifest=blobs))
111
112 parent: str | None = None
113 tip = ""
114 for i in range(n_commits):
115 ts = datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)
116 msg = f"c{start + i:07d}"
117 cid = compute_commit_id(
118 repo_id="bench",
119 parent_ids=[parent] if parent else [],
120 snapshot_id=sid,
121 message=msg,
122 committed_at_iso=ts.isoformat(),
123 author="bench",)
124 rec = CommitRecord(
125 commit_id=cid,
126 repo_id="bench",
127 created_on_branch=branch,
128 snapshot_id=sid,
129 message=msg,
130 committed_at=ts,
131 parent_commit_id=parent,
132 parent2_commit_id=None,
133 author="bench",
134 metadata={},
135 structured_delta=None,
136 sem_ver_bump="none",
137 breaking_changes=[],
138 agent_id="",
139 model_id="",
140 toolchain_id="",
141 prompt_hash="",
142 signature="",
143 signer_key_id="",
144 )
145 write_commit(repo, rec)
146 parent = cid
147 tip = cid
148
149 write_branch_ref(repo, branch, tip)
150 return tip, blobs
151
152
153 # ---------------------------------------------------------------------------
154 # Phase 3.4.1 — build_pack throughput
155 # ---------------------------------------------------------------------------
156
157
158 class TestBuildPackThroughput:
159 """build_pack must sustain ≥ 2 000 objects/sec in the object-read loop.
160
161 build_pack's hot path is ``read_object`` for each unique blob. At the
162 Phase 3.1 floor of 2 000 objects/sec, 100 000 objects take ~50 s, within
163 the 60 s target. The fast test covers 1 000 objects and asserts the rate
164 directly; the slow test covers 10 000 objects and proves on-disk timing.
165 """
166
167 _MIN_OBJECTS_PER_SEC = 2_000
168
169 def test_build_pack_1k_objects_rate(self, tmp_path: pathlib.Path) -> None:
170 """build_pack on 1 000 objects must achieve ≥ 2 000 objects/sec."""
171 repo = _make_repo(tmp_path)
172 N = 1_000
173 tip, blobs = _populate(repo, n_commits=50, n_unique_objects=N)
174
175 t0 = time.perf_counter()
176 bundle = build_pack(repo, [tip])
177 elapsed = time.perf_counter() - t0
178
179 assert len(bundle["objects"]) == N, (
180 f"Expected {N} objects in bundle, got {len(bundle['objects'])}"
181 )
182 rate = N / elapsed
183 assert rate >= self._MIN_OBJECTS_PER_SEC, (
184 f"build_pack throughput {rate:.0f} objects/sec < {self._MIN_OBJECTS_PER_SEC} minimum. "
185 f"({N} objects took {elapsed:.2f}s.)"
186 )
187
188 def test_build_pack_have_filter_excludes_base(
189 self, tmp_path: pathlib.Path
190 ) -> None:
191 """build_pack with have=[base_tip] sends only delta commits, not the full history."""
192 repo = _make_repo(tmp_path)
193 base_tip, base_blobs = _populate(repo, n_commits=50, n_unique_objects=100, start=0)
194
195 # New commits on top of the base, with fresh objects.
196 delta_tip, delta_blobs = _populate(
197 repo, n_commits=20, n_unique_objects=50, start=1000
198 )
199 # Chain delta to base by writing a commit that has base_tip as parent.
200 ts = datetime.datetime(2026, 1, 2, tzinfo=datetime.timezone.utc)
201 sid = compute_snapshot_id(delta_blobs)
202 chained_cid = compute_commit_id(
203 repo_id="bench",
204 parent_ids=[delta_tip, base_tip],
205 snapshot_id=sid,
206 message="merge",
207 committed_at_iso=ts.isoformat(),
208 author="bench",)
209 chained = CommitRecord(
210 commit_id=chained_cid,
211 repo_id="bench",
212 created_on_branch="main",
213 snapshot_id=sid,
214 message="merge",
215 committed_at=ts,
216 parent_commit_id=delta_tip,
217 parent2_commit_id=base_tip,
218 author="bench",
219 metadata={},
220 structured_delta=None,
221 sem_ver_bump="none",
222 breaking_changes=[],
223 agent_id="",
224 model_id="",
225 toolchain_id="",
226 prompt_hash="",
227 signature="",
228 signer_key_id="",
229 )
230 write_commit(repo, chained)
231 write_branch_ref(repo, "main", chained_cid)
232
233 # Full pack (no have).
234 full_bundle = build_pack(repo, [chained_cid])
235 # Delta pack: receiver already has base history.
236 delta_bundle = build_pack(repo, [chained_cid], have=[base_tip])
237
238 assert len(delta_bundle["commits"]) < len(full_bundle["commits"]), (
239 "have= filter must reduce commit count"
240 )
241 # The base objects must not be in the delta bundle since they share the snapshot.
242 full_oids = {o["object_id"] for o in full_bundle["objects"]}
243 delta_oids = {o["object_id"] for o in delta_bundle["objects"]}
244 assert delta_oids.issubset(full_oids), "delta bundle must be a subset of full bundle"
245
246 @pytest.mark.slow
247 def test_build_pack_10k_objects_under_60s(
248 self, tmp_path: pathlib.Path
249 ) -> None:
250 """build_pack on 10 000 objects must complete in < 60 s.
251
252 This extrapolates to 100 000 objects at the same rate: 100k / 2000 ≈ 50 s,
253 within the plan target of 60 s.
254 """
255 repo = _make_repo(tmp_path)
256 N = 10_000
257 tip, _ = _populate(repo, n_commits=100, n_unique_objects=N)
258
259 t0 = time.perf_counter()
260 bundle = build_pack(repo, [tip])
261 elapsed = time.perf_counter() - t0
262
263 assert len(bundle["objects"]) == N
264 assert elapsed < 60.0, (
265 f"build_pack({N} objects) took {elapsed:.1f}s — target < 60 s. "
266 f"Rate: {N/elapsed:.0f} objects/sec."
267 )
268
269
270 class TestCollectObjectIdsThroughput:
271 """collect_object_ids must be significantly faster than build_pack.
272
273 It performs the same BFS + manifest traversal but skips reading blob bytes.
274 This is used for the pre-filter negotiation (POST /filter-objects) so the
275 client only uploads objects the remote actually needs.
276 """
277
278 def test_collect_object_ids_no_blob_reads(
279 self, tmp_path: pathlib.Path
280 ) -> None:
281 """collect_object_ids on 500 objects must be < 1 s (no blob reads)."""
282 repo = _make_repo(tmp_path)
283 N = 500
284 tip, blobs = _populate(repo, n_commits=50, n_unique_objects=N)
285
286 t0 = time.perf_counter()
287 oids = collect_object_ids(repo, [tip])
288 elapsed = time.perf_counter() - t0
289
290 assert len(oids) == N, f"Expected {N} OIDs, got {len(oids)}"
291 assert elapsed < 1.0, (
292 f"collect_object_ids({N}) took {elapsed:.3f}s — expected < 1 s. "
293 "It must not read blob bytes; only BFS + manifest traversal."
294 )
295
296 def test_collect_object_ids_faster_than_build_pack(
297 self, tmp_path: pathlib.Path
298 ) -> None:
299 """collect_object_ids must be faster than build_pack for the same input."""
300 repo = _make_repo(tmp_path)
301 N = 200
302 tip, _ = _populate(repo, n_commits=20, n_unique_objects=N)
303
304 t_collect = time.perf_counter()
305 oids = collect_object_ids(repo, [tip])
306 t_collect = time.perf_counter() - t_collect
307
308 t_build = time.perf_counter()
309 bundle = build_pack(repo, [tip])
310 t_build = time.perf_counter() - t_build
311
312 assert len(oids) == len(bundle["objects"]) == N
313 assert t_collect <= t_build, (
314 f"collect_object_ids ({t_collect:.3f}s) must be ≤ build_pack ({t_build:.3f}s) — "
315 "collect skips blob reads, build reads every byte."
316 )
317
318 def test_collect_object_ids_have_excludes_ancestors(
319 self, tmp_path: pathlib.Path
320 ) -> None:
321 """collect_object_ids with have= returns only new object IDs."""
322 repo = _make_repo(tmp_path)
323 base_tip, base_blobs = _populate(repo, n_commits=10, n_unique_objects=50, start=0)
324 delta_tip, delta_blobs = _populate(repo, n_commits=5, n_unique_objects=30, start=100)
325
326 all_oids = collect_object_ids(repo, [delta_tip])
327 delta_oids = collect_object_ids(repo, [delta_tip], have=[base_tip])
328
329 # Delta must be a subset and contain only the 30 new objects.
330 assert set(delta_blobs.values()).issubset(set(all_oids))
331 assert len(delta_oids) == len(set(delta_blobs.values())), (
332 f"Expected {len(set(delta_blobs.values()))} delta OIDs, got {len(delta_oids)}"
333 )
334
335
336 # ---------------------------------------------------------------------------
337 # Phase 3.4.2 — apply_pack throughput
338 # ---------------------------------------------------------------------------
339
340
341 class TestApplyPackThroughput:
342 """apply_pack must sustain ≥ 1 500 objects/sec in the object-write loop.
343
344 fsync is mocked: the test measures the pack unpacking + hash-verify +
345 mkstemp + fchmod + os.replace pipeline without OS I/O latency. Durability
346 ordering is verified by test_integrity_I2_fsync.py.
347 """
348
349 _MIN_OBJECTS_PER_SEC: int = 1_500
350
351 @pytest.fixture(autouse=True)
352 def no_fsync(self) -> None:
353 """Mock out all fsync calls so the test measures algorithmic throughput."""
354 with patch("muse.core.object_store._fsync_fd", return_value=None), \
355 patch("muse.core.store.os.fsync", return_value=None), \
356 patch("muse.core.store.fcntl.fcntl", return_value=0):
357 yield
358
359 @pytest.mark.perf
360 def test_apply_pack_1k_objects_rate(self, tmp_path: pathlib.Path) -> None:
361 """apply_pack of a 1 000-object bundle must achieve ≥ _MIN_OBJECTS_PER_SEC."""
362 src = _make_repo(tmp_path / "src")
363 N = 1_000
364 tip, _ = _populate(src, n_commits=50, n_unique_objects=N)
365 bundle = build_pack(src, [tip])
366 assert len(bundle["objects"]) == N
367
368 dst = _fresh_repo(tmp_path / "dst")
369
370 t0 = time.perf_counter()
371 result = apply_pack(dst, bundle)
372 elapsed = time.perf_counter() - t0
373
374 assert result["objects_written"] == N
375 rate = N / elapsed
376 assert rate >= self._MIN_OBJECTS_PER_SEC, (
377 f"apply_pack throughput {rate:.0f} objects/sec < {self._MIN_OBJECTS_PER_SEC} minimum. "
378 f"({N} objects took {elapsed:.2f}s.)"
379 )
380
381 def test_apply_pack_idempotent_second_apply_skips_all(
382 self, tmp_path: pathlib.Path
383 ) -> None:
384 """Applying the same bundle twice: second apply must skip every object."""
385 src = _make_repo(tmp_path / "src")
386 tip, _ = _populate(src, n_commits=20, n_unique_objects=100)
387 bundle = build_pack(src, [tip])
388
389 dst = _fresh_repo(tmp_path / "dst")
390 r1 = apply_pack(dst, bundle)
391 r2 = apply_pack(dst, bundle)
392
393 assert r1["objects_written"] == 100
394 assert r2["objects_written"] == 0
395 assert r2["objects_skipped"] == 100, (
396 f"Expected 100 skipped on second apply, got {r2['objects_skipped']}"
397 )
398
399 @pytest.mark.slow
400 def test_apply_pack_10k_objects_under_60s(
401 self, tmp_path: pathlib.Path
402 ) -> None:
403 """apply_pack of a 10 000-object bundle must complete in < 60 s."""
404 src = _make_repo(tmp_path / "src")
405 N = 10_000
406 tip, _ = _populate(src, n_commits=100, n_unique_objects=N)
407 bundle = build_pack(src, [tip])
408
409 dst = _fresh_repo(tmp_path / "dst")
410
411 t0 = time.perf_counter()
412 result = apply_pack(dst, bundle)
413 elapsed = time.perf_counter() - t0
414
415 assert result["objects_written"] == N
416 assert elapsed < 60.0, (
417 f"apply_pack({N} objects) took {elapsed:.1f}s — target < 60 s. "
418 f"Rate: {N/elapsed:.0f} objects/sec."
419 )
420
421
422 # ---------------------------------------------------------------------------
423 # Phase 3.4.3 — verify-pack
424 # ---------------------------------------------------------------------------
425
426
427 class TestVerifyPackIntegrity:
428 """verify-pack must detect hash mismatches and work fast in --stat mode."""
429
430 def test_verify_pack_stat_returns_counts(
431 self, tmp_path: pathlib.Path
432 ) -> None:
433 """verify-pack --stat must count objects/snapshots/commits without hashing."""
434 from tests.cli_test_helper import CliRunner
435 import json
436
437 repo = _make_repo(tmp_path)
438 tip, _ = _populate(repo, n_commits=20, n_unique_objects=50)
439 bundle = build_pack(repo, [tip])
440 raw = msgpack.packb(bundle, use_bin_type=True)
441
442 bundle_file = tmp_path / "pack.muse"
443 bundle_file.write_bytes(raw)
444 (repo / ".muse" / "config.toml").write_text("")
445
446 runner = CliRunner()
447 result = runner.invoke(
448 None,
449 [
450 "verify-pack",
451 "--stat",
452 "--no-local",
453 "--json",
454 "--file", str(bundle_file),
455 ],
456 env={"MUSE_REPO_ROOT": str(repo)},
457 )
458 assert result.exit_code == 0, f"verify-pack --stat failed: {result.output}"
459 payload = json.loads(result.output)
460 assert payload["objects"] == 50
461 assert payload["commits"] == 20
462
463 def test_verify_pack_detects_hash_mismatch(
464 self, tmp_path: pathlib.Path
465 ) -> None:
466 """verify-pack must flag an object whose content hash does not match its ID."""
467 from tests.cli_test_helper import CliRunner
468 import json
469
470 repo = _make_repo(tmp_path)
471 tip, blobs = _populate(repo, n_commits=5, n_unique_objects=10)
472 bundle = build_pack(repo, [tip])
473
474 # Tamper: set a wrong content for the first object while keeping the declared ID.
475 tampered_obj: ObjectPayload = {
476 "object_id": bundle["objects"][0]["object_id"],
477 "content": b"TAMPERED_CONTENT",
478 }
479 tampered_bundle: PackBundle = {
480 "commits": bundle["commits"],
481 "snapshots": bundle["snapshots"],
482 "objects": [tampered_obj] + bundle["objects"][1:],
483 "summary": bundle.get("summary", {}),
484 "meta": bundle.get("meta", {}),
485 }
486 raw = msgpack.packb(tampered_bundle, use_bin_type=True)
487 bundle_file = tmp_path / "tampered.muse"
488 bundle_file.write_bytes(raw)
489 (repo / ".muse" / "config.toml").write_text("")
490
491 runner = CliRunner()
492 result = runner.invoke(
493 None,
494 [
495 "verify-pack",
496 "--no-local",
497 "--json",
498 "--file", str(bundle_file),
499 ],
500 env={"MUSE_REPO_ROOT": str(repo)},
501 )
502 assert result.exit_code != 0, "verify-pack must exit non-zero when hash mismatches"
503 payload = json.loads(result.output)
504 assert payload["all_ok"] is False
505 assert any("hash mismatch" in f["error"] for f in payload["failures"]), (
506 f"Expected 'hash mismatch' in failures: {payload['failures']}"
507 )
508
509 @pytest.mark.slow
510 def test_verify_pack_10k_objects_under_120s(
511 self, tmp_path: pathlib.Path
512 ) -> None:
513 """verify-pack of a 10 000-object bundle must complete in < 120 s.
514
515 SHA-256 on M4 Silicon processes ~3 GiB/s; 10k × 4 KiB = 40 MiB → < 1 s.
516 The 120 s ceiling catches pathological I/O or per-object overhead.
517 """
518 from tests.cli_test_helper import CliRunner
519 import json
520
521 repo = _make_repo(tmp_path)
522 N = 10_000
523 tip, _ = _populate(repo, n_commits=100, n_unique_objects=N)
524 bundle = build_pack(repo, [tip])
525 raw = msgpack.packb(bundle, use_bin_type=True)
526 bundle_file = tmp_path / "pack10k.muse"
527 bundle_file.write_bytes(raw)
528 (repo / ".muse" / "config.toml").write_text("")
529
530 t0 = time.perf_counter()
531 runner = CliRunner()
532 result = runner.invoke(
533 None,
534 [
535 "verify-pack",
536 "--no-local",
537 "--json",
538 "--file", str(bundle_file),
539 ],
540 env={"MUSE_REPO_ROOT": str(repo)},
541 )
542 elapsed = time.perf_counter() - t0
543
544 assert result.exit_code == 0, f"verify-pack failed: {result.output[:200]}"
545 payload = json.loads(result.output)
546 assert payload["all_ok"] is True
547 assert payload["objects_checked"] == N
548 assert elapsed < 120.0, (
549 f"verify-pack({N} objects) took {elapsed:.1f}s — target < 120 s."
550 )
551
552
553 # ---------------------------------------------------------------------------
554 # Phase 3.4.4 — cap and guard enforcement
555 # ---------------------------------------------------------------------------
556
557
558 class TestPackCapEnforcement:
559 """Pack-bomb and size-cap guards must fire correctly."""
560
561 def test_apply_pack_rejects_bundle_exceeding_max_pack_objects(
562 self, tmp_path: pathlib.Path
563 ) -> None:
564 """apply_pack raises ValueError when total_items > MAX_PACK_OBJECTS.
565
566 MAX_PACK_OBJECTS counts commits + snapshots + objects combined — not
567 per-type. A pack with MAX_PACK_OBJECTS + 1 total items is rejected.
568 """
569 repo = _fresh_repo(tmp_path)
570 oversized: PackBundle = {
571 "commits": [{}] * (MAX_PACK_OBJECTS + 1),
572 "snapshots": [],
573 "objects": [],
574 }
575 with pytest.raises(ValueError, match="Pack rejected"):
576 apply_pack(repo, oversized)
577
578 def test_apply_pack_accepts_bundle_at_exact_cap(
579 self, tmp_path: pathlib.Path
580 ) -> None:
581 """apply_pack does NOT raise when total_items == MAX_PACK_OBJECTS.
582
583 Items are malformed (empty dicts) so they are skipped as bad entries,
584 but the cap check must pass.
585 """
586 repo = _fresh_repo(tmp_path)
587 at_cap: PackBundle = {
588 "commits": [{}] * MAX_PACK_OBJECTS,
589 "snapshots": [],
590 "objects": [],
591 }
592 # Must not raise ValueError for the cap — skips malformed entries instead.
593 result = apply_pack(repo, at_cap)
594 # Each empty-dict commit is missing commit_id and snapshot_id, so every
595 # one is skipped by the essential-field guard added to apply_pack.
596 assert result["commits_written"] == 0, (
597 "All malformed empty-dict commits must be skipped, not written"
598 )
599
600 def test_apply_pack_total_items_cap_is_cross_type(
601 self, tmp_path: pathlib.Path
602 ) -> None:
603 """MAX_PACK_OBJECTS applies across commits+snapshots+objects, not per-type.
604
605 80 000 objects + 20 000 commits + 1 snapshot = 100 001 → rejected.
606 """
607 repo = _fresh_repo(tmp_path)
608 cross_type: PackBundle = {
609 "commits": [{}] * 20_000,
610 "snapshots": [{}] * 1,
611 "objects": [{}] * 80_000,
612 }
613 with pytest.raises(ValueError, match="Pack rejected"):
614 apply_pack(repo, cross_type)
615
616 def test_apply_pack_oversized_object_is_skipped_not_raised(
617 self, tmp_path: pathlib.Path
618 ) -> None:
619 """An object exceeding MAX_OBJECT_WRITE_BYTES is silently skipped.
620
621 This is documented behaviour: the per-object cap logs a warning and
622 increments the loop counter rather than raising an exception, so the
623 rest of the bundle is still applied.
624 """
625 repo = _fresh_repo(tmp_path)
626 good_data = b"x" * 64
627 good_oid = _sha256(good_data)
628 oversized_oid = _sha256(b"y") # real hash — but we'll fake the size check
629 # Construct a bundle with one valid object and one whose content we
630 # claim is MAX_OBJECT_WRITE_BYTES + 1 bytes.
631 # We use a real 1-byte payload but lie about the size by patching
632 # apply_pack's check via len(raw) — we need an actually-oversized payload.
633 # Build a real oversized content string:
634 huge_data = b"z" * (MAX_OBJECT_WRITE_BYTES + 1)
635 huge_oid = _sha256(huge_data)
636 bundle: PackBundle = {
637 "commits": [],
638 "snapshots": [],
639 "objects": [
640 ObjectPayload(object_id=good_oid, content=good_data),
641 ObjectPayload(object_id=huge_oid, content=huge_data),
642 ],
643 }
644 result = apply_pack(repo, bundle)
645 # Good object written; oversized object skipped.
646 assert result["objects_written"] == 1, (
647 f"Expected 1 object written (the good one), got {result['objects_written']}"
648 )
649 # Oversized object must NOT be in the store.
650 from muse.core.object_store import has_object
651 assert not has_object(repo, huge_oid), (
652 "Oversized object must be rejected and not written to store"
653 )
654
655 def test_apply_pack_deduplicates_repeated_oid(
656 self, tmp_path: pathlib.Path
657 ) -> None:
658 """apply_pack writes a repeated OID only once (dedup via seen_object_ids)."""
659 repo = _fresh_repo(tmp_path)
660 data = b"deduplicate-me" * 100
661 oid = _sha256(data)
662 REPEAT = 50
663 bundle: PackBundle = {
664 "commits": [],
665 "snapshots": [],
666 "objects": [ObjectPayload(object_id=oid, content=data)] * REPEAT,
667 }
668 result = apply_pack(repo, bundle)
669 # First occurrence written; remaining 49 skipped.
670 assert result["objects_written"] == 1, (
671 f"Expected 1 write for {REPEAT} identical OIDs, got {result['objects_written']}"
672 )
673 assert result["objects_skipped"] == REPEAT - 1, (
674 f"Expected {REPEAT - 1} skipped, got {result['objects_skipped']}"
675 )
676
677 def test_apply_pack_empty_bundle_is_noop(
678 self, tmp_path: pathlib.Path
679 ) -> None:
680 """apply_pack on a pack with no items returns all-zero counts."""
681 repo = _fresh_repo(tmp_path)
682 empty: PackBundle = {"commits": [], "snapshots": [], "objects": []}
683 result = apply_pack(repo, empty)
684 assert result["commits_written"] == 0
685 assert result["snapshots_written"] == 0
686 assert result["objects_written"] == 0
687 assert result["objects_skipped"] == 0
688
689 def test_have_equals_want_produces_empty_bundle(
690 self, tmp_path: pathlib.Path
691 ) -> None:
692 """build_pack with have=[tip] where tip is also in want returns empty bundle."""
693 repo = _make_repo(tmp_path)
694 tip, _ = _populate(repo, n_commits=10, n_unique_objects=20)
695
696 bundle = build_pack(repo, [tip], have=[tip])
697
698 assert bundle["commits"] == [], (
699 "When have contains the want tip, BFS should yield 0 commits"
700 )
701 assert bundle["objects"] == [], (
702 "Empty commit set must produce empty object list"
703 )
704
705
706 # ---------------------------------------------------------------------------
707 # Phase 3.4.5 — memory ceiling
708 # ---------------------------------------------------------------------------
709
710
711 class TestPackMemoryCeiling:
712 """build_pack and apply_pack peak memory must be proportional to blob payload.
713
714 build_pack holds ALL object bytes in-memory simultaneously — this is a
715 known architectural property, not a bug. The test confirms:
716 1. Peak RSS ≈ total blob bytes (not 10× or 100×).
717 2. build_pack does not accumulate unbounded intermediate structures.
718 """
719
720 def test_build_pack_peak_rss_proportional_to_blob_total(
721 self, tmp_path: pathlib.Path
722 ) -> None:
723 """build_pack peak allocation is ≤ 3× the total blob payload size."""
724 repo = _make_repo(tmp_path)
725 N = 500
726 BLOB_SZ = 4096 # 4 KiB
727 tip, _ = _populate(repo, n_commits=50, n_unique_objects=N, blob_size=BLOB_SZ)
728 blob_total_mib = N * BLOB_SZ / (1024 * 1024)
729
730 tracemalloc.start()
731 tracemalloc.clear_traces()
732 bundle = build_pack(repo, [tip])
733 _, peak_bytes = tracemalloc.get_traced_memory()
734 tracemalloc.stop()
735
736 peak_mib = peak_bytes / (1024 * 1024)
737 ceiling_mib = blob_total_mib * 3 # generous: blobs + msgpack + overhead
738 assert len(bundle["objects"]) == N
739 assert peak_mib <= ceiling_mib, (
740 f"build_pack peak {peak_mib:.1f} MiB exceeds 3× blob total "
741 f"({ceiling_mib:.1f} MiB for {N} × {BLOB_SZ//1024} KiB objects). "
742 "build_pack must not accumulate more than the object bytes themselves."
743 )
744
745 def test_apply_pack_peak_rss_under_64_mib_for_small_objects(
746 self, tmp_path: pathlib.Path
747 ) -> None:
748 """apply_pack of 500 × 4 KiB objects stays under 64 MiB."""
749 src = _make_repo(tmp_path / "src")
750 tip, _ = _populate(src, n_commits=50, n_unique_objects=500, blob_size=4096)
751 bundle = build_pack(src, [tip])
752
753 dst = _fresh_repo(tmp_path / "dst")
754
755 tracemalloc.start()
756 tracemalloc.clear_traces()
757 apply_pack(dst, bundle)
758 _, peak_bytes = tracemalloc.get_traced_memory()
759 tracemalloc.stop()
760
761 peak_mib = peak_bytes / (1024 * 1024)
762 assert peak_mib <= 64, (
763 f"apply_pack(500 × 4 KiB) peak {peak_mib:.1f} MiB — expected ≤ 64 MiB."
764 )
765
766
767 # ---------------------------------------------------------------------------
768 # Phase 3.4.6 — round-trip integrity
769 # ---------------------------------------------------------------------------
770
771
772 class TestPackRoundTrip:
773 """End-to-end round-trip: build_pack → msgpack serialize → apply_pack → verify."""
774
775 def test_roundtrip_all_objects_restored(
776 self, tmp_path: pathlib.Path
777 ) -> None:
778 """build_pack → msgpack → apply_pack round-trip: all objects readable on dst."""
779 src = _make_repo(tmp_path / "src")
780 N_OBJECTS = 200
781 N_COMMITS = 30
782 tip, blobs = _populate(src, n_commits=N_COMMITS, n_unique_objects=N_OBJECTS)
783 bundle = build_pack(src, [tip])
784
785 # Serialize (simulates wire transfer).
786 raw = msgpack.packb(bundle, use_bin_type=True)
787
788 # Re-hydrate using safe_unpackb (the same path as unpack-objects).
789 from muse.core.store import safe_unpackb, MAX_PACK_MSGPACK_BYTES
790 restored_dict = safe_unpackb(raw, context="roundtrip", max_bytes=MAX_PACK_MSGPACK_BYTES, allow_binary=True)
791 assert isinstance(restored_dict, dict)
792
793 from muse.core.pack import ObjectPayload as OP, MPackBundle as PB
794 raw_objects = restored_dict.get("objects") or []
795 objects: list[OP] = []
796 for item in raw_objects:
797 if isinstance(item, dict):
798 oid = item.get("object_id", "")
799 content = item.get("content", b"")
800 if isinstance(oid, str) and isinstance(content, (bytes, bytearray)):
801 objects.append(OP(object_id=oid, content=bytes(content)))
802
803 hydrated: PB = {
804 "commits": [c for c in (restored_dict.get("commits") or []) if isinstance(c, dict)],
805 "snapshots": [s for s in (restored_dict.get("snapshots") or []) if isinstance(s, dict)],
806 "objects": objects,
807 }
808 dst = _fresh_repo(tmp_path / "dst")
809 result = apply_pack(dst, hydrated)
810
811 # Every source object must be readable on the destination.
812 from muse.core.object_store import read_object, has_object
813 missing = [oid for oid in blobs.values() if not has_object(dst, oid)]
814 assert not missing, (
815 f"{len(missing)}/{N_OBJECTS} objects missing after round-trip: "
816 f"{missing[:3]}"
817 )
818 assert result["commits_written"] == N_COMMITS
819 assert result["objects_written"] == N_OBJECTS
820
821 def test_roundtrip_msgpack_bundle_size_within_max_pack_bytes(
822 self, tmp_path: pathlib.Path
823 ) -> None:
824 """The serialised bundle for 1 000 × 4 KiB objects must be < MAX_PACK_MSGPACK_BYTES."""
825 src = _make_repo(tmp_path)
826 N = 1_000
827 tip, _ = _populate(src, n_commits=50, n_unique_objects=N, blob_size=4096)
828 bundle = build_pack(src, [tip])
829 raw = msgpack.packb(bundle, use_bin_type=True)
830
831 limit = MAX_PACK_MSGPACK_BYTES
832 assert len(raw) < limit, (
833 f"Bundle for {N} × 4 KiB objects is {len(raw):,} bytes — "
834 f"exceeds MAX_PACK_MSGPACK_BYTES ({limit:,} bytes / "
835 f"{limit // 1024 // 1024} MiB)."
836 )
File History 3 commits
sha256:b636f72dcba9e190afb980bece906fa5b717fbde014b76ef023df8cb96e01eb9 docs: expand cache plan with all seven testing tiers and do… Sonnet 4.6 131 days ago
sha256:7f9e2ef5286aedad9c1e6011b4c46ca27f39dbdad6e3409357e36b26e46b3b7c docs: docstring sprint for-each-ref→hotspots — idiomatic ru… Sonnet 4.6 patch 138 days ago
sha256:a09b1b4f6838754495547f200aa0ce88e2f56ffc5b20b900f6f0cff2c3cdede9 fix(cursorignore): remove git-ism (.git/worktrees) Human minor ⚠ 140 days ago