test_mpack_gc_phase4.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 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" |