gabriel / musehub public
test_mpack_gc_phase4.py python
438 lines 16.2 KB
Raw
sha256:34035d72cef530c1ab9d6a6f53be18d803dde705fc3157617d70352a96d0747b Merge branch 'fix/wire-push-external-parent-manifest' into dev Human 8 days ago
1 """TDD — Phase 4: mpack GC / consolidation (issue #63).
2
3 PG-1 process_mpack_gc_job merges all mpack index rows for a repo into a single
4 consolidated mpack and updates the mpack index so every object_id maps
5 to the new consolidated mpack_key.
6
7 PG-2 After GC, the number of distinct mpack_ids for any repo is ≤ 1
8 (a freshly-consolidated repo has exactly one mpack).
9
10 PG-3 The consolidated mpack is a valid msgpack mpack containing all objects
11 from the source mpacks.
12
13 PG-4 process_mpack_gc_job is idempotent: running it twice produces the same
14 consolidated mpack_key (same content → same sha256 key) and does not
15 create duplicate mpack index rows.
16
17 PG-5 A repo with ≤ 1 distinct mpack is skipped (no-op) — GC only acts when
18 consolidation would reduce mpack count.
19
20 PG-6 After GC, wire_fetch_mpack returns exactly 1 mpack_url (the consolidated
21 mpack) for a repo that was previously spread across N mpacks.
22 """
23 from __future__ import annotations
24
25 import datetime
26
27 import msgpack
28 import pytest
29 from sqlalchemy import select, func
30 from sqlalchemy.ext.asyncio import AsyncSession
31
32 from muse.core.mpack import build_wire_mpack
33 from muse.core.types import blob_id
34 from musehub.db import musehub_repo_models as db
35 from musehub.core.genesis import compute_identity_id
36 from musehub.services.musehub_repository import create_repo
37
38 pytestmark = pytest.mark.wire
39
40
41 # ---------------------------------------------------------------------------
42 # Helpers
43 # ---------------------------------------------------------------------------
44
45 def _make_mpack(
46 objects: dict[str, bytes],
47 commits: list[dict] | None = None,
48 snapshots: list[dict] | None = None,
49 branch_heads: dict[str, str] | None = None,
50 ) -> tuple[bytes, str]:
51 wire_bytes = build_wire_mpack({
52 "commits": commits or [],
53 "snapshots": snapshots or [],
54 "blobs": [
55 {"object_id": oid, "content": data}
56 for oid, data in objects.items()
57 ],
58 "tags": [],
59 })
60 mpack_key = blob_id(wire_bytes)
61 return wire_bytes, mpack_key
62
63
64 async def _store_mpack(wire_bytes: bytes, mpack_key: str) -> None:
65 import musehub.storage.backends as _backends_mod
66 backend = _backends_mod.get_backend()
67 await backend.put_mpack(mpack_key, wire_bytes)
68
69
70 async def _push_and_index(
71 session: AsyncSession,
72 repo_id: str,
73 objects: dict[str, bytes],
74 commits: list[dict] | None = None,
75 snapshots: list[dict] | None = None,
76 branch_heads: dict[str, str] | None = None,
77 ) -> str:
78 """Push an mpack and run process_mpack_index_job. Returns mpack_key."""
79 from musehub.core.genesis import compute_job_id
80 from musehub.db.musehub_jobs_models import MusehubBackgroundJob
81 from musehub.services.musehub_wire import process_mpack_index_job
82
83 wire_bytes, mpack_key = _make_mpack(objects, commits, snapshots, branch_heads)
84 await _store_mpack(wire_bytes, mpack_key)
85
86 now = datetime.datetime.now(datetime.timezone.utc)
87 job_id = compute_job_id(repo_id, "mpack.index", now.isoformat())
88 session.add(MusehubBackgroundJob(
89 job_id=job_id,
90 repo_id=repo_id,
91 job_type="mpack.index",
92 payload={
93 "mpack_key": mpack_key,
94 "branch": "main",
95 "head": (commits or [{}])[-1].get("commit_id", ""),
96 "pusher_id": "",
97 "declared_objects_count": len(objects),
98 "declared_commits_count": len(commits or []),
99 },
100 status="pending",
101 created_at=now,
102 attempt=0,
103 ))
104 await session.commit()
105 await process_mpack_index_job(session, job_id)
106 await session.commit()
107 return mpack_key
108
109
110 def _make_commit_chain(n: int, seed: str, parent_tip: str | None = None) -> tuple[list[dict], list[dict], str, dict[str, str]]:
111 """Return (commits, snapshots, tip_commit_id, objects_dict)."""
112 objects: dict[str, bytes] = {}
113 commits = []
114 snapshots = []
115 parent_id = parent_tip
116
117 for i in range(n):
118 oid = blob_id(f"{seed}-obj-{i}".encode())
119 objects[oid] = f"{seed}-obj-{i}".encode()
120 snap_id = blob_id(f"{seed}-snap-{i}".encode())
121 snapshots.append({
122 "snapshot_id": snap_id,
123 "parent_snapshot_id": None,
124 "delta_upsert": {f"file_{i}.txt": oid},
125 "delta_remove": [],
126 })
127 cid = blob_id(f"{seed}-commit-{i}-p={parent_id}".encode())
128 commits.append({
129 "commit_id": cid,
130 "branch": "main",
131 "message": f"commit {i}",
132 "author": "gabriel",
133 "committed_at": datetime.datetime(2025, 1, 1, tzinfo=datetime.timezone.utc).isoformat(),
134 "parent_commit_id": parent_id,
135 "parent2_commit_id": None,
136 "snapshot_id": snap_id,
137 "agent_id": "",
138 "model_id": "",
139 "toolchain_id": "",
140 "sem_ver_bump": "none",
141 "breaking_changes": [],
142 "signature": "",
143 "signer_key_id": "",
144 "signer_public_key": "",
145 "prompt_hash": "",
146 })
147 parent_id = cid
148
149 return commits, snapshots, parent_id, objects # type: ignore[return-value]
150
151
152 async def _repo_pack_oids(session: AsyncSession, repo_id: str) -> list[str]:
153 """Return object_ids that belong to this repo (via MusehubObjectRef)."""
154 q = await session.execute(
155 select(db.MusehubObjectRef.object_id).where(db.MusehubObjectRef.repo_id == repo_id)
156 )
157 return [row[0] for row in q]
158
159
160 async def _distinct_mpack_count(session: AsyncSession, repo_id: str) -> int:
161 """Count distinct mpacks that cover this repo's objects."""
162 oid_set = await _repo_pack_oids(session, repo_id)
163 if not oid_set:
164 return 0
165 result = await session.execute(
166 select(db.MusehubMPackIndex.mpack_id)
167 .where(db.MusehubMPackIndex.entity_id.in_(oid_set))
168 .where(db.MusehubMPackIndex.entity_type == "object")
169 .distinct()
170 )
171 return len(result.all())
172
173
174 # ---------------------------------------------------------------------------
175 # PG-1
176 # ---------------------------------------------------------------------------
177
178 @pytest.mark.asyncio
179 async def test_pg1_gc_consolidates_multiple_mpacks(db_session: AsyncSession) -> None:
180 """process_mpack_gc_job merges N mpacks into one and updates the mpack index."""
181 from musehub.services.musehub_wire import process_mpack_gc_job
182
183 repo = await create_repo(
184 db_session,
185 name="pg-test-1",
186 owner="gabriel",
187 owner_user_id=compute_identity_id(b"gabriel"),
188 visibility="public",
189 initialize=False,
190 )
191
192 # Three separate pushes → three mpacks
193 commits_a, snaps_a, tip_a, objs_a = _make_commit_chain(2, "pg1a")
194 await _push_and_index(db_session, repo.repo_id, objs_a, commits_a, snaps_a,
195 branch_heads={"main": tip_a})
196
197 commits_b, snaps_b, tip_b, objs_b = _make_commit_chain(2, "pg1b", parent_tip=tip_a)
198 await _push_and_index(db_session, repo.repo_id, objs_b, commits_b, snaps_b,
199 branch_heads={"main": tip_b})
200
201 commits_c, snaps_c, tip_c, objs_c = _make_commit_chain(2, "pg1c", parent_tip=tip_b)
202 await _push_and_index(db_session, repo.repo_id, objs_c, commits_c, snaps_c,
203 branch_heads={"main": tip_c})
204
205 before = await _distinct_mpack_count(db_session, repo.repo_id)
206 assert before == 3, f"expected 3 mpacks before GC, got {before}"
207
208 result = await process_mpack_gc_job(db_session, repo.repo_id)
209 await db_session.commit()
210
211 assert result["packs_before"] == 3
212 assert result["packs_after"] == 1
213 assert result["consolidated_key"], "no consolidated mpack_key returned"
214
215
216 # ---------------------------------------------------------------------------
217 # PG-2
218 # ---------------------------------------------------------------------------
219
220 @pytest.mark.asyncio
221 async def test_pg2_gc_leaves_exactly_one_mpack(db_session: AsyncSession) -> None:
222 """After GC, distinct mpack_id count for the repo is exactly 1."""
223 from musehub.services.musehub_wire import process_mpack_gc_job
224
225 repo = await create_repo(
226 db_session,
227 name="pg-test-2",
228 owner="gabriel",
229 owner_user_id=compute_identity_id(b"gabriel"),
230 visibility="public",
231 initialize=False,
232 )
233
234 for seed in ("pg2a", "pg2b", "pg2c", "pg2d"):
235 commits, snaps, tip, objs = _make_commit_chain(2, seed)
236 await _push_and_index(db_session, repo.repo_id, objs, commits, snaps,
237 branch_heads={"main": tip})
238
239 await process_mpack_gc_job(db_session, repo.repo_id)
240 await db_session.commit()
241
242 after = await _distinct_mpack_count(db_session, repo.repo_id)
243 assert after == 1, f"expected 1 mpack after GC, got {after}"
244
245
246 # ---------------------------------------------------------------------------
247 # PG-3
248 # ---------------------------------------------------------------------------
249
250 @pytest.mark.asyncio
251 async def test_pg3_consolidated_mpack_contains_all_objects(db_session: AsyncSession) -> None:
252 """The consolidated mpack stored in MinIO is a valid msgpack mpack with all objects."""
253 import musehub.storage.backends as _backends_mod
254 from musehub.services.musehub_wire import process_mpack_gc_job
255
256 repo = await create_repo(
257 db_session,
258 name="pg-test-3",
259 owner="gabriel",
260 owner_user_id=compute_identity_id(b"gabriel"),
261 visibility="public",
262 initialize=False,
263 )
264
265 all_oids: set[str] = set()
266 commits_a, snaps_a, tip_a, objs_a = _make_commit_chain(3, "pg3a")
267 await _push_and_index(db_session, repo.repo_id, objs_a, commits_a, snaps_a,
268 branch_heads={"main": tip_a})
269 all_oids.update(objs_a)
270
271 commits_b, snaps_b, tip_b, objs_b = _make_commit_chain(3, "pg3b", parent_tip=tip_a)
272 await _push_and_index(db_session, repo.repo_id, objs_b, commits_b, snaps_b,
273 branch_heads={"main": tip_b})
274 all_oids.update(objs_b)
275
276 result = await process_mpack_gc_job(db_session, repo.repo_id)
277 await db_session.commit()
278
279 consolidated_key = result["consolidated_key"]
280 backend = _backends_mod.get_backend()
281 mpack_bytes = await backend.get_mpack(consolidated_key)
282 assert mpack_bytes, "consolidated mpack not found in storage"
283
284 if mpack_bytes[:4] == b"MUSE":
285 from muse.core.mpack import parse_wire_mpack as _parse_wm
286 mpack = _parse_wm(mpack_bytes)
287 else:
288 mpack = msgpack.unpackb(mpack_bytes, raw=False)
289 stored_oids = {obj["object_id"] for obj in mpack.get("blobs", [])}
290 assert all_oids == stored_oids, (
291 f"missing objects in consolidated mpack: {all_oids - stored_oids}"
292 )
293
294
295 # ---------------------------------------------------------------------------
296 # PG-4
297 # ---------------------------------------------------------------------------
298
299 @pytest.mark.asyncio
300 async def test_pg4_gc_is_idempotent(db_session: AsyncSession) -> None:
301 """Running GC twice produces the same consolidated_key and no duplicate rows."""
302 from musehub.services.musehub_wire import process_mpack_gc_job
303
304 repo = await create_repo(
305 db_session,
306 name="pg-test-4",
307 owner="gabriel",
308 owner_user_id=compute_identity_id(b"gabriel"),
309 visibility="public",
310 initialize=False,
311 )
312
313 commits_a, snaps_a, tip_a, objs_a = _make_commit_chain(2, "pg4a")
314 await _push_and_index(db_session, repo.repo_id, objs_a, commits_a, snaps_a,
315 branch_heads={"main": tip_a})
316
317 commits_b, snaps_b, tip_b, objs_b = _make_commit_chain(2, "pg4b", parent_tip=tip_a)
318 await _push_and_index(db_session, repo.repo_id, objs_b, commits_b, snaps_b,
319 branch_heads={"main": tip_b})
320
321 result1 = await process_mpack_gc_job(db_session, repo.repo_id)
322 await db_session.commit()
323
324 result2 = await process_mpack_gc_job(db_session, repo.repo_id)
325 await db_session.commit()
326
327 assert result1["consolidated_key"] == result2["consolidated_key"], (
328 "second GC produced a different key — consolidation is not content-addressed"
329 )
330 # No duplicate rows — each (object_id, mpack_id) pair must be unique.
331 # Query via object refs since mpack index is global.
332 oid_set = await _repo_pack_oids(db_session, repo.repo_id)
333 total_q = await db_session.execute(
334 select(func.count()).select_from(db.MusehubMPackIndex)
335 .where(db.MusehubMPackIndex.entity_id.in_(oid_set))
336 .where(db.MusehubMPackIndex.entity_type == "object")
337 )
338 total = total_q.scalar_one()
339 all_oids = len(objs_a) + len(objs_b)
340 assert total == all_oids, f"expected {all_oids} rows (no dups), got {total}"
341
342
343 # ---------------------------------------------------------------------------
344 # PG-5
345 # ---------------------------------------------------------------------------
346
347 @pytest.mark.asyncio
348 async def test_pg5_gc_skips_single_mpack_repo(db_session: AsyncSession) -> None:
349 """GC is a no-op when the repo already has ≤ 1 mpack."""
350 from musehub.services.musehub_wire import process_mpack_gc_job
351
352 repo = await create_repo(
353 db_session,
354 name="pg-test-5",
355 owner="gabriel",
356 owner_user_id=compute_identity_id(b"gabriel"),
357 visibility="public",
358 initialize=False,
359 )
360
361 commits, snaps, tip, objs = _make_commit_chain(3, "pg5")
362 await _push_and_index(db_session, repo.repo_id, objs, commits, snaps,
363 branch_heads={"main": tip})
364
365 result = await process_mpack_gc_job(db_session, repo.repo_id)
366 await db_session.commit()
367
368 assert result["skipped"] is True, "expected GC to be skipped for single-mpack repo"
369 assert result.get("packs_before", 1) <= 1
370 after = await _distinct_mpack_count(db_session, repo.repo_id)
371 assert after == 1
372
373
374 # ---------------------------------------------------------------------------
375 # PG-6
376 # ---------------------------------------------------------------------------
377
378 @pytest.mark.asyncio
379 async def test_pg6_fetch_works_after_gc_and_index_converges(db_session: AsyncSession) -> None:
380 """After GC, wire_fetch_mpack still works and MPackIndex collapses to 1 covering mpack.
381
382 Pre-GC: two pushes → two source mpacks in MPackIndex.
383 Post-GC: consolidated mpack → all objects point to the same mpack_id.
384 wire_fetch_mpack must not raise in either state.
385 """
386 from musehub.services.musehub_wire import process_mpack_gc_job, wire_fetch_mpack
387
388 repo = await create_repo(
389 db_session,
390 name="pg-test-6",
391 owner="gabriel",
392 owner_user_id=compute_identity_id(b"gabriel"),
393 visibility="public",
394 initialize=False,
395 )
396
397 commits_a, snaps_a, tip_a, objs_a = _make_commit_chain(3, "pg6a")
398 await _push_and_index(db_session, repo.repo_id, objs_a, commits_a, snaps_a,
399 branch_heads={"main": tip_a})
400
401 commits_b, snaps_b, tip_b, objs_b = _make_commit_chain(3, "pg6b", parent_tip=tip_a)
402 await _push_and_index(db_session, repo.repo_id, objs_b, commits_b, snaps_b,
403 branch_heads={"main": tip_b})
404
405 # Before GC: 2 distinct covering mpacks in MPackIndex for this repo's objects
406 assert await _distinct_mpack_count(db_session, repo.repo_id) == 2, (
407 "expected 2 distinct mpacks in MPackIndex before GC"
408 )
409
410 # Before GC: fetch must not raise (force_build=True bypasses prebuild cache requirement)
411 before_result = await wire_fetch_mpack(
412 db_session, repo.repo_id,
413 want=[tip_b], have=[],
414 ttl_seconds=3600,
415 force_build=True,
416 )
417 assert "mpack_url" in before_result, "fetch must return mpack_url field before GC"
418
419 # Run GC
420 gc_result = await process_mpack_gc_job(db_session, repo.repo_id)
421 await db_session.commit()
422
423 assert not gc_result["skipped"], "GC should not skip when 2 mpacks exist"
424 assert gc_result["packs_after"] == 1
425
426 # After GC: exactly 1 distinct covering mpack in MPackIndex
427 assert await _distinct_mpack_count(db_session, repo.repo_id) == 1, (
428 "expected 1 distinct mpack in MPackIndex after GC"
429 )
430
431 # After GC: fetch must still not raise
432 after_result = await wire_fetch_mpack(
433 db_session, repo.repo_id,
434 want=[tip_b], have=[],
435 ttl_seconds=3600,
436 force_build=True,
437 )
438 assert "mpack_url" in after_result, "fetch must return mpack_url field after GC"
File History 16 commits
sha256:34035d72cef530c1ab9d6a6f53be18d803dde705fc3157617d70352a96d0747b Merge branch 'fix/wire-push-external-parent-manifest' into dev Human 8 days ago
sha256:fc04e4cae9e1774d6a21b65c45daeed0e6787eb581d13aa1b03bfe9384a34226 Merge branch 'fix/two-column-scroll-layout' into dev Human 8 days ago
sha256:408916fc5973ba59c6e4eebaa80ebdcc801c0a63205651e25009d11548f79454 chore: bump version to 0.2.0.dev2 — nightly.2, matching muse Sonnet 4.6 patch 11 days ago
sha256:d035733f21ccff27735fddebfbbe0ed24565a32a22db8de5885402262671ecd2 chore: bump version to 0.2.0rc15 for musehub#113 fix release Sonnet 4.6 patch 14 days ago
sha256:0032d6cfa33bc3c8367436ad768e7dd0e339b4332153160247da8266cb5fa352 Merge branch 'task/version-tags-phase3-server' into dev Human 17 days ago
sha256:4669620efda9ff41c55bdefd1f7bfe1c239d468428744c84ead9957e5a003a53 merge: rescue snapshot-recovery hardening (c00aa21d) into d… Opus 4.8 minor 29 days ago
sha256:a59da49c4611b970fc4b6ae48678ce4943261c213a07ddbd73ce9201df869b4a fix: remove false-positive proposal_comments index drop fro… Sonnet 4.6 patch 33 days ago
sha256:0a240d6dbff234f07d98a28a4a9a68db702f3f9ff9260196f24219bdb1c0b6f3 feat: render markdown mists as HTML with heading anchor links Sonnet 4.6 patch 34 days ago
sha256:24a7d47486ebc4ebd1832830580e177ec6f877b48dced8c000e198cdec4ce9d6 Merge 'task/bump-version-rc12' into 'dev' — proposal: Bump … Human 35 days ago
sha256:b9ff931d147e0114a1f17060f415b89ed551c170a91ff226c70437aa5c85f9ee Merge 'task/bump-version-rc12' into 'dev' — proposal: Bump … Human 35 days ago
sha256:d1122d21e73471879b460037b22c0b50fded7c423444a176f248428f75dac39c Merge 'task/fix-issue-pagination-cursor' into 'dev' — propo… Human 35 days ago
sha256:01e18975e73d2b3cd5b6db7929c895bef9aa6e0d4391dc5b2adfc548b41318dd Merge 'feat/adding-debug-logs-to-staging' into 'dev' — prop… Human 35 days ago
sha256:6b1949fc2797ca4c1936a637a4cbfec828ef56cf52398a2e74ca3c4f494e728f fix: use wire_bytes not mpack_bytes_raw in compute_object_b… Sonnet 4.6 patch 48 days ago
sha256:b99f2455dc346966d040133f5203297e6e3ef5803a93728a2c30568d0a0f7583 rename: delta_add → delta_upsert across wire format, models… Sonnet 4.6 patch 50 days ago
sha256:1bb2b2331b696382042e1249385aafd0c7bdd17167b7b972be72b90ae625bd80 fix: repair 400+ failing tests, fix mpack integrity check, … Sonnet 4.6 minor 50 days ago
sha256:4aed3d8601c8dd3ed37074de35f11f4a9699a0a4b99d43727048fd3f8e6fd13d chore: doc sweep, ignore wrangler build state, misc fixes Sonnet 4.6 minor 50 days ago