gabriel / muse public
test_perf_pack.py python
833 lines 30.9 KB
Raw
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ breaking 126 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 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_pack throughput
153 # ---------------------------------------------------------------------------
154
155
156 class TestBuildPackThroughput:
157 """build_pack must sustain ≥ 2 000 objects/sec in the object-read loop.
158
159 build_pack'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_pack 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_pack(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_pack 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_pack 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_pack(repo, [chained_cid])
233 # Delta pack: receiver already has base history.
234 delta_bundle = build_pack(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_pack 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_pack(repo, [tip])
259 elapsed = time.perf_counter() - t0
260
261 assert len(bundle["objects"]) == N
262 assert elapsed < 60.0, (
263 f"build_pack({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_pack.
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_pack 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_pack(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_pack ({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_pack throughput
335 # ---------------------------------------------------------------------------
336
337
338 class TestApplyPackThroughput:
339 """apply_pack must sustain ≥ 1 500 objects/sec in the object-write loop.
340
341 fsync is mocked: the test measures the pack unpacking + hash-verify +
342 mkstemp + fchmod + os.replace pipeline without OS I/O latency. Durability
343 ordering is verified by test_integrity_I2_fsync.py.
344 """
345
346 _MIN_OBJECTS_PER_SEC: int = 1_500
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_pack 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_pack(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_pack(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_pack 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_pack(src, [tip])
385
386 dst = _fresh_repo(tmp_path / "dst")
387 r1 = apply_pack(dst, bundle)
388 r2 = apply_pack(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_pack 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_pack(src, [tip])
405
406 dst = _fresh_repo(tmp_path / "dst")
407
408 t0 = time.perf_counter()
409 result = apply_pack(dst, bundle)
410 elapsed = time.perf_counter() - t0
411
412 assert result["objects_written"] == N
413 assert elapsed < 60.0, (
414 f"apply_pack({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_pack(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_pack(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: PackBundle = {
477 "commits": bundle["commits"],
478 "snapshots": bundle["snapshots"],
479 "objects": [tampered_obj] + bundle["objects"][1:],
480 "summary": bundle.get("summary", {}),
481 "meta": bundle.get("meta", {}),
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 (config_toml_path(repo)).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_pack(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 (config_toml_path(repo)).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_pack 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: PackBundle = {
568 "commits": [{}] * (MAX_PACK_OBJECTS + 1),
569 "snapshots": [],
570 "objects": [],
571 }
572 with pytest.raises(ValueError, match="Pack rejected"):
573 apply_pack(repo, oversized)
574
575 def test_apply_pack_accepts_bundle_at_exact_cap(
576 self, tmp_path: pathlib.Path
577 ) -> None:
578 """apply_pack 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: PackBundle = {
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_pack(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_pack.
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: PackBundle = {
606 "commits": [{}] * 20_000,
607 "snapshots": [{}] * 1,
608 "objects": [{}] * 80_000,
609 }
610 with pytest.raises(ValueError, match="Pack rejected"):
611 apply_pack(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 = blob_id(good_data)
625 oversized_oid = blob_id(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_pack'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 = blob_id(huge_data)
633 bundle: PackBundle = {
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_pack(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_pack 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 = blob_id(data)
659 REPEAT = 50
660 bundle: PackBundle = {
661 "commits": [],
662 "snapshots": [],
663 "objects": [ObjectPayload(object_id=oid, content=data)] * REPEAT,
664 }
665 result = apply_pack(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_pack on a pack with no items returns all-zero counts."""
678 repo = _fresh_repo(tmp_path)
679 empty: PackBundle = {"commits": [], "snapshots": [], "objects": []}
680 result = apply_pack(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_pack 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_pack(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_pack and apply_pack peak memory must be proportional to blob payload.
710
711 build_pack 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_pack 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_pack 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_pack(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_pack peak {peak_mib:.1f} MiB exceeds 3× blob total "
738 f"({ceiling_mib:.1f} MiB for {N} × {BLOB_SZ//1024} KiB objects). "
739 "build_pack 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_pack 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_pack(src, [tip])
749
750 dst = _fresh_repo(tmp_path / "dst")
751
752 tracemalloc.start()
753 tracemalloc.clear_traces()
754 apply_pack(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_pack(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_pack → msgpack serialize → apply_pack → verify."""
771
772 def test_roundtrip_all_objects_restored(
773 self, tmp_path: pathlib.Path
774 ) -> None:
775 """build_pack → msgpack → apply_pack 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_pack(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_pack(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_pack(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 1 commit
sha256:51ce277f663e01a43eaffbe77509b1de7ac2d4251b55d23306304bcdeb92c90d feat(pack): delta-encode snapshots in MPackBundle wire format Sonnet 4.6 minor ⚠ 126 days ago