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