gabriel / muse public
test_mpack_perf_ops.py python
833 lines 30.9 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_mpack (10 000 objects × 4 KiB): < 60 s [@slow]
6 apply_mpack (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_mpack`` 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_mpack`` — 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,
42 apply_mpack,
43 build_mpack,
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_mpack throughput
155 # ---------------------------------------------------------------------------
156
157
158 class TestBuildPackThroughput:
159 """build_mpack must sustain ≥ 2 000 objects/sec in the object-read loop.
160
161 build_mpack'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_mpack 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_mpack(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_mpack 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_mpack 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_mpack(repo, [chained_cid])
235 # Delta pack: receiver already has base history.
236 delta_bundle = build_mpack(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_mpack 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_mpack(repo, [tip])
261 elapsed = time.perf_counter() - t0
262
263 assert len(bundle["objects"]) == N
264 assert elapsed < 60.0, (
265 f"build_mpack({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_mpack.
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_mpack 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_mpack(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_mpack ({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_mpack throughput
338 # ---------------------------------------------------------------------------
339
340
341 class TestApplyPackThroughput:
342 """apply_mpack must sustain ≥ 100 objects/sec in the object-write loop.
343
344 fsync is mocked but real disk writes still occur, so the floor is set
345 conservatively to catch algorithmic regressions without flapping under
346 system load. Durability ordering is verified by test_integrity_I2_fsync.py.
347 """
348
349 _MIN_OBJECTS_PER_SEC: int = 100
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_mpack 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_mpack(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_mpack(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_mpack 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_mpack(src, [tip])
388
389 dst = _fresh_repo(tmp_path / "dst")
390 r1 = apply_mpack(dst, bundle)
391 r2 = apply_mpack(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_mpack 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_mpack(src, [tip])
408
409 dst = _fresh_repo(tmp_path / "dst")
410
411 t0 = time.perf_counter()
412 result = apply_mpack(dst, bundle)
413 elapsed = time.perf_counter() - t0
414
415 assert result["objects_written"] == N
416 assert elapsed < 60.0, (
417 f"apply_mpack({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_mpack(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_mpack(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: MPackBundle = {
480 **bundle,
481 "objects": [tampered_obj] + bundle["objects"][1:],
482 }
483 raw = msgpack.packb(tampered_bundle, use_bin_type=True)
484 bundle_file = tmp_path / "tampered.muse"
485 bundle_file.write_bytes(raw)
486 (repo / ".muse" / "config.toml").write_text("")
487
488 runner = CliRunner()
489 result = runner.invoke(
490 None,
491 [
492 "verify-pack",
493 "--no-local",
494 "--json",
495 "--file", str(bundle_file),
496 ],
497 env={"MUSE_REPO_ROOT": str(repo)},
498 )
499 assert result.exit_code != 0, "verify-pack must exit non-zero when hash mismatches"
500 payload = json.loads(result.output)
501 assert payload["all_ok"] is False
502 assert any("hash mismatch" in f["error"] for f in payload["failures"]), (
503 f"Expected 'hash mismatch' in failures: {payload['failures']}"
504 )
505
506 @pytest.mark.slow
507 def test_verify_pack_10k_objects_under_120s(
508 self, tmp_path: pathlib.Path
509 ) -> None:
510 """verify-pack of a 10 000-object bundle must complete in < 120 s.
511
512 SHA-256 on M4 Silicon processes ~3 GiB/s; 10k × 4 KiB = 40 MiB → < 1 s.
513 The 120 s ceiling catches pathological I/O or per-object overhead.
514 """
515 from tests.cli_test_helper import CliRunner
516 import json
517
518 repo = _make_repo(tmp_path)
519 N = 10_000
520 tip, _ = _populate(repo, n_commits=100, n_unique_objects=N)
521 bundle = build_mpack(repo, [tip])
522 raw = msgpack.packb(bundle, use_bin_type=True)
523 bundle_file = tmp_path / "pack10k.muse"
524 bundle_file.write_bytes(raw)
525 (repo / ".muse" / "config.toml").write_text("")
526
527 t0 = time.perf_counter()
528 runner = CliRunner()
529 result = runner.invoke(
530 None,
531 [
532 "verify-pack",
533 "--no-local",
534 "--json",
535 "--file", str(bundle_file),
536 ],
537 env={"MUSE_REPO_ROOT": str(repo)},
538 )
539 elapsed = time.perf_counter() - t0
540
541 assert result.exit_code == 0, f"verify-pack failed: {result.output[:200]}"
542 payload = json.loads(result.output)
543 assert payload["all_ok"] is True
544 assert payload["objects_checked"] == N
545 assert elapsed < 120.0, (
546 f"verify-pack({N} objects) took {elapsed:.1f}s — target < 120 s."
547 )
548
549
550 # ---------------------------------------------------------------------------
551 # Phase 3.4.4 — cap and guard enforcement
552 # ---------------------------------------------------------------------------
553
554
555 class TestPackCapEnforcement:
556 """Pack-bomb and size-cap guards must fire correctly."""
557
558 def test_apply_pack_rejects_bundle_exceeding_max_pack_objects(
559 self, tmp_path: pathlib.Path
560 ) -> None:
561 """apply_mpack raises ValueError when total_items > MAX_PACK_OBJECTS.
562
563 MAX_PACK_OBJECTS counts commits + snapshots + objects combined — not
564 per-type. A pack with MAX_PACK_OBJECTS + 1 total items is rejected.
565 """
566 repo = _fresh_repo(tmp_path)
567 oversized: MPackBundle = {
568 "commits": [{}] * (MAX_PACK_OBJECTS + 1),
569 "snapshots": [],
570 "objects": [],
571 }
572 with pytest.raises(ValueError, match="Pack rejected"):
573 apply_mpack(repo, oversized)
574
575 def test_apply_pack_accepts_bundle_at_exact_cap(
576 self, tmp_path: pathlib.Path
577 ) -> None:
578 """apply_mpack does NOT raise when total_items == MAX_PACK_OBJECTS.
579
580 Items are malformed (empty dicts) so they are skipped as bad entries,
581 but the cap check must pass.
582 """
583 repo = _fresh_repo(tmp_path)
584 at_cap: MPackBundle = {
585 "commits": [{}] * MAX_PACK_OBJECTS,
586 "snapshots": [],
587 "objects": [],
588 }
589 # Must not raise ValueError for the cap — skips malformed entries instead.
590 result = apply_mpack(repo, at_cap)
591 # Each empty-dict commit is missing commit_id and snapshot_id, so every
592 # one is skipped by the essential-field guard added to apply_mpack.
593 assert result["commits_written"] == 0, (
594 "All malformed empty-dict commits must be skipped, not written"
595 )
596
597 def test_apply_pack_total_items_cap_is_cross_type(
598 self, tmp_path: pathlib.Path
599 ) -> None:
600 """MAX_PACK_OBJECTS applies across commits+snapshots+objects, not per-type.
601
602 80 000 objects + 20 000 commits + 1 snapshot = 100 001 → rejected.
603 """
604 repo = _fresh_repo(tmp_path)
605 cross_type: MPackBundle = {
606 "commits": [{}] * 20_000,
607 "snapshots": [{}] * 1,
608 "objects": [{}] * 80_000,
609 }
610 with pytest.raises(ValueError, match="Pack rejected"):
611 apply_mpack(repo, cross_type)
612
613 def test_apply_pack_oversized_object_is_skipped_not_raised(
614 self, tmp_path: pathlib.Path
615 ) -> None:
616 """An object exceeding MAX_OBJECT_WRITE_BYTES is silently skipped.
617
618 This is documented behaviour: the per-object cap logs a warning and
619 increments the loop counter rather than raising an exception, so the
620 rest of the bundle is still applied.
621 """
622 repo = _fresh_repo(tmp_path)
623 good_data = b"x" * 64
624 good_oid = _sha256(good_data)
625 oversized_oid = _sha256(b"y") # real hash — but we'll fake the size check
626 # Construct a bundle with one valid object and one whose content we
627 # claim is MAX_OBJECT_WRITE_BYTES + 1 bytes.
628 # We use a real 1-byte payload but lie about the size by patching
629 # apply_mpack's check via len(raw) — we need an actually-oversized payload.
630 # Build a real oversized content string:
631 huge_data = b"z" * (MAX_OBJECT_WRITE_BYTES + 1)
632 huge_oid = _sha256(huge_data)
633 bundle: MPackBundle = {
634 "commits": [],
635 "snapshots": [],
636 "objects": [
637 ObjectPayload(object_id=good_oid, content=good_data),
638 ObjectPayload(object_id=huge_oid, content=huge_data),
639 ],
640 }
641 result = apply_mpack(repo, bundle)
642 # Good object written; oversized object skipped.
643 assert result["objects_written"] == 1, (
644 f"Expected 1 object written (the good one), got {result['objects_written']}"
645 )
646 # Oversized object must NOT be in the store.
647 from muse.core.object_store import has_object
648 assert not has_object(repo, huge_oid), (
649 "Oversized object must be rejected and not written to store"
650 )
651
652 def test_apply_pack_deduplicates_repeated_oid(
653 self, tmp_path: pathlib.Path
654 ) -> None:
655 """apply_mpack writes a repeated OID only once (dedup via seen_object_ids)."""
656 repo = _fresh_repo(tmp_path)
657 data = b"deduplicate-me" * 100
658 oid = _sha256(data)
659 REPEAT = 50
660 bundle: MPackBundle = {
661 "commits": [],
662 "snapshots": [],
663 "objects": [ObjectPayload(object_id=oid, content=data)] * REPEAT,
664 }
665 result = apply_mpack(repo, bundle)
666 # First occurrence written; remaining 49 skipped.
667 assert result["objects_written"] == 1, (
668 f"Expected 1 write for {REPEAT} identical OIDs, got {result['objects_written']}"
669 )
670 assert result["objects_skipped"] == REPEAT - 1, (
671 f"Expected {REPEAT - 1} skipped, got {result['objects_skipped']}"
672 )
673
674 def test_apply_pack_empty_bundle_is_noop(
675 self, tmp_path: pathlib.Path
676 ) -> None:
677 """apply_mpack on a pack with no items returns all-zero counts."""
678 repo = _fresh_repo(tmp_path)
679 empty: MPackBundle = {"commits": [], "snapshots": [], "objects": []}
680 result = apply_mpack(repo, empty)
681 assert result["commits_written"] == 0
682 assert result["snapshots_written"] == 0
683 assert result["objects_written"] == 0
684 assert result["objects_skipped"] == 0
685
686 def test_have_equals_want_produces_empty_bundle(
687 self, tmp_path: pathlib.Path
688 ) -> None:
689 """build_mpack with have=[tip] where tip is also in want returns empty bundle."""
690 repo = _make_repo(tmp_path)
691 tip, _ = _populate(repo, n_commits=10, n_unique_objects=20)
692
693 bundle = build_mpack(repo, [tip], have=[tip])
694
695 assert bundle["commits"] == [], (
696 "When have contains the want tip, BFS should yield 0 commits"
697 )
698 assert bundle["objects"] == [], (
699 "Empty commit set must produce empty object list"
700 )
701
702
703 # ---------------------------------------------------------------------------
704 # Phase 3.4.5 — memory ceiling
705 # ---------------------------------------------------------------------------
706
707
708 class TestPackMemoryCeiling:
709 """build_mpack and apply_mpack peak memory must be proportional to blob payload.
710
711 build_mpack holds ALL object bytes in-memory simultaneously — this is a
712 known architectural property, not a bug. The test confirms:
713 1. Peak RSS ≈ total blob bytes (not 10× or 100×).
714 2. build_mpack does not accumulate unbounded intermediate structures.
715 """
716
717 def test_build_pack_peak_rss_proportional_to_blob_total(
718 self, tmp_path: pathlib.Path
719 ) -> None:
720 """build_mpack peak allocation is ≤ 3× the total blob payload size."""
721 repo = _make_repo(tmp_path)
722 N = 500
723 BLOB_SZ = 4096 # 4 KiB
724 tip, _ = _populate(repo, n_commits=50, n_unique_objects=N, blob_size=BLOB_SZ)
725 blob_total_mib = N * BLOB_SZ / (1024 * 1024)
726
727 tracemalloc.start()
728 tracemalloc.clear_traces()
729 bundle = build_mpack(repo, [tip])
730 _, peak_bytes = tracemalloc.get_traced_memory()
731 tracemalloc.stop()
732
733 peak_mib = peak_bytes / (1024 * 1024)
734 ceiling_mib = blob_total_mib * 3 # generous: blobs + msgpack + overhead
735 assert len(bundle["objects"]) == N
736 assert peak_mib <= ceiling_mib, (
737 f"build_mpack peak {peak_mib:.1f} MiB exceeds 3× blob total "
738 f"({ceiling_mib:.1f} MiB for {N} × {BLOB_SZ//1024} KiB objects). "
739 "build_mpack must not accumulate more than the object bytes themselves."
740 )
741
742 def test_apply_pack_peak_rss_under_64_mib_for_small_objects(
743 self, tmp_path: pathlib.Path
744 ) -> None:
745 """apply_mpack of 500 × 4 KiB objects stays under 64 MiB."""
746 src = _make_repo(tmp_path / "src")
747 tip, _ = _populate(src, n_commits=50, n_unique_objects=500, blob_size=4096)
748 bundle = build_mpack(src, [tip])
749
750 dst = _fresh_repo(tmp_path / "dst")
751
752 tracemalloc.start()
753 tracemalloc.clear_traces()
754 apply_mpack(dst, bundle)
755 _, peak_bytes = tracemalloc.get_traced_memory()
756 tracemalloc.stop()
757
758 peak_mib = peak_bytes / (1024 * 1024)
759 assert peak_mib <= 64, (
760 f"apply_mpack(500 × 4 KiB) peak {peak_mib:.1f} MiB — expected ≤ 64 MiB."
761 )
762
763
764 # ---------------------------------------------------------------------------
765 # Phase 3.4.6 — round-trip integrity
766 # ---------------------------------------------------------------------------
767
768
769 class TestPackRoundTrip:
770 """End-to-end round-trip: build_mpack → msgpack serialize → apply_mpack → verify."""
771
772 def test_roundtrip_all_objects_restored(
773 self, tmp_path: pathlib.Path
774 ) -> None:
775 """build_mpack → msgpack → apply_mpack round-trip: all objects readable on dst."""
776 src = _make_repo(tmp_path / "src")
777 N_OBJECTS = 200
778 N_COMMITS = 30
779 tip, blobs = _populate(src, n_commits=N_COMMITS, n_unique_objects=N_OBJECTS)
780 bundle = build_mpack(src, [tip])
781
782 # Serialize (simulates wire transfer).
783 raw = msgpack.packb(bundle, use_bin_type=True)
784
785 # Re-hydrate using safe_unpackb (the same path as unpack-objects).
786 from muse.core.store import safe_unpackb, MAX_PACK_MSGPACK_BYTES
787 restored_dict = safe_unpackb(raw, context="roundtrip", max_bytes=MAX_PACK_MSGPACK_BYTES, allow_binary=True)
788 assert isinstance(restored_dict, dict)
789
790 from muse.core.pack import ObjectPayload as OP, MPackBundle as PB
791 raw_objects = restored_dict.get("objects") or []
792 objects: list[OP] = []
793 for item in raw_objects:
794 if isinstance(item, dict):
795 oid = item.get("object_id", "")
796 content = item.get("content", b"")
797 if isinstance(oid, str) and isinstance(content, (bytes, bytearray)):
798 objects.append(OP(object_id=oid, content=bytes(content)))
799
800 hydrated: PB = {
801 "commits": [c for c in (restored_dict.get("commits") or []) if isinstance(c, dict)],
802 "snapshots": [s for s in (restored_dict.get("snapshots") or []) if isinstance(s, dict)],
803 "objects": objects,
804 }
805 dst = _fresh_repo(tmp_path / "dst")
806 result = apply_mpack(dst, hydrated)
807
808 # Every source object must be readable on the destination.
809 from muse.core.object_store import read_object, has_object
810 missing = [oid for oid in blobs.values() if not has_object(dst, oid)]
811 assert not missing, (
812 f"{len(missing)}/{N_OBJECTS} objects missing after round-trip: "
813 f"{missing[:3]}"
814 )
815 assert result["commits_written"] == N_COMMITS
816 assert result["objects_written"] == N_OBJECTS
817
818 def test_roundtrip_msgpack_bundle_size_within_max_pack_bytes(
819 self, tmp_path: pathlib.Path
820 ) -> None:
821 """The serialised bundle for 1 000 × 4 KiB objects must be < MAX_PACK_MSGPACK_BYTES."""
822 src = _make_repo(tmp_path)
823 N = 1_000
824 tip, _ = _populate(src, n_commits=50, n_unique_objects=N, blob_size=4096)
825 bundle = build_mpack(src, [tip])
826 raw = msgpack.packb(bundle, use_bin_type=True)
827
828 limit = MAX_PACK_MSGPACK_BYTES
829 assert len(raw) < limit, (
830 f"Bundle for {N} × 4 KiB objects is {len(raw):,} bytes — "
831 f"exceeds MAX_PACK_MSGPACK_BYTES ({limit:,} bytes / "
832 f"{limit // 1024 // 1024} MiB)."
833 )
File History 2 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 137 days ago