"""MWP-8 — fetch-mpack cache invalidation (musehub#111). AUDIT VERDICT (Phase 0 — AU_01..AU_04): ======================================== All three repair endpoints (wire_repair_object, wire_repair_snapshot, wire_repair_commit) and the force-push path (wire_push_unpack_mpack with force=True) share the same gap: 1. repair-object (AU_01) — wire_repair_object performs upsert + session.commit() but NEVER touches musehub_fetch_mpack_cache. A fresh cache row for the tip that reaches the repaired object survives unchanged. A subsequent fresh-clone request (want=[tip], have=[]) returns a HIT on the pre-repair mpack_id. 2. repair-snapshot (AU_02) — wire_repair_snapshot resets manifest_blob + session.commit() but NEVER touches musehub_fetch_mpack_cache. Same HIT on stale bytes. 3. repair-commit (AU_03) — wire_repair_commit force-overwrites identity fields + session.commit() but NEVER touches musehub_fetch_mpack_cache. Same HIT. This is the exact mechanism behind the gabriel/muse rc10 incident: the repair fixed the DB row but the cache kept serving the corrupted mpack for 7 more days. 4. force-push (AU_04) — wire_push_unpack_mpack(force=True) advances the branch pointer and enqueues intel jobs but NEVER deletes the old-tip cache row. A client that still knows the old tip gets a HIT on content that is no longer on any branch. Expected post-fix behavior (implemented in Phases 2–3): • AU_01..AU_03 → after repair, wire_fetch_mpack(want=[tip], have=[]) raises MPackNotReadyError (cache row deleted in same transaction as repair). • AU_04 → after force-push, (repo_id, old_tip) row is gone; the same fetch raises MPackNotReadyError. Phase 2 (RP_04..RP_06) and Phase 3 (FP_02) replace these assertions with the corrected contracts. Run: python3 -m pytest tests/test_mwp8_cache_invalidation.py -q --tb=short """ from __future__ import annotations import hashlib from datetime import datetime, timedelta, timezone from typing import Any from unittest.mock import AsyncMock, patch import pytest from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from muse.core.ids import hash_commit, long_id from muse.core.mpack import build_wire_mpack from muse.core.types import blob_id, fake_id from musehub.db.musehub_repo_models import ( MusehubBranch, MusehubCommit, MusehubCommitRef, MusehubFetchMPackCache, MusehubObject, MusehubObjectRef, ) from musehub.db.musehub_jobs_models import MusehubBackgroundJob from musehub.muse_cli.snapshot import compute_snapshot_id from musehub.services.musehub_wire_fetch import wire_fetch_mpack from musehub.services.musehub_wire_push import ( wire_push_unpack_mpack, wire_repair_commit, wire_repair_object, wire_repair_snapshot, ) from musehub.services.musehub_wire_shared import MPackNotReadyError from tests.factories import create_branch, create_repo # ── helpers ─────────────────────────────────────────────────────────────────── def _now() -> datetime: return datetime.now(tz=timezone.utc) async def _insert_fresh_cache_row( session: AsyncSession, repo_id: str, tip_commit_id: str, mpack_id: str, seed: str, ) -> MusehubFetchMPackCache: """Seed a non-expired cache row (tip → mpack_id) into musehub_fetch_mpack_cache.""" now = _now() row = MusehubFetchMPackCache( cache_id=fake_id(f"cache-{seed}"), repo_id=repo_id, tip_commit_id=tip_commit_id, mpack_id=mpack_id, created_at=now, expires_at=now + timedelta(days=7), ) session.add(row) await session.commit() return row async def _cache_rows_for_repo( session: AsyncSession, repo_id: str, ) -> list[MusehubFetchMPackCache]: """Return all cache rows for a given repo_id.""" rows = await session.execute( select(MusehubFetchMPackCache).where( MusehubFetchMPackCache.repo_id == repo_id ) ) return list(rows.scalars().all()) def _stub_push_backend(monkeypatch: pytest.MonkeyPatch) -> AsyncMock: """Patch the storage backend used by musehub_wire_push (repair calls backend.put).""" backend = AsyncMock() backend.put = AsyncMock(return_value="s3://test-bucket/object") monkeypatch.setattr("musehub.services.musehub_wire_push.get_backend", lambda: backend) return backend def _stub_fetch_backend(monkeypatch: pytest.MonkeyPatch, presign_url: str = "https://r2.example/mpack") -> AsyncMock: """Patch the storage backend used by musehub_wire_fetch (HIT calls backend.presign_mpack_get).""" backend = AsyncMock() backend.presign_mpack_get = AsyncMock(return_value=presign_url) monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend) return backend def _stub_gc_backend(monkeypatch: pytest.MonkeyPatch) -> AsyncMock: """Patch the storage backend used by musehub_gc (invalidate calls backend.delete).""" backend = AsyncMock() backend.delete = AsyncMock() monkeypatch.setattr("musehub.services.musehub_gc.get_backend", lambda: backend) return backend def _build_force_push_backend( monkeypatch: pytest.MonkeyPatch, mpack_bytes: bytes, ) -> AsyncMock: """Patch the storage backend used inside wire_push_unpack_mpack. wire_push_unpack_mpack imports get_backend locally from musehub.storage.backends, so we patch that site (not the module-level musehub.services.musehub_wire_push one). The returned mock has get_mpack pre-configured to return mpack_bytes. """ backend = AsyncMock() backend.get_mpack = AsyncMock(return_value=mpack_bytes) backend.quarantine_mpack = AsyncMock() backend.delete = AsyncMock() monkeypatch.setattr("musehub.storage.backends.get_backend", lambda: backend) return backend async def _insert_object_row( session: AsyncSession, repo_id: str, object_id: str, ) -> None: """Insert a minimal MusehubObject + ref so wire_repair_object can upsert it.""" from sqlalchemy.dialects.postgresql import insert as _pg_insert await session.execute( _pg_insert(MusehubObject) .values(object_id=object_id, path="test.bin", size_bytes=0, storage_uri="s3://bucket/x", content_cache=None) .on_conflict_do_nothing(index_elements=["object_id"]) ) await session.execute( _pg_insert(MusehubObjectRef) .values(repo_id=repo_id, object_id=object_id) .on_conflict_do_nothing(index_elements=["repo_id", "object_id"]) ) await session.commit() def _build_valid_commit_dict( *, snapshot_id: str, message: str, committed_at: str, author: str = "gabriel", ) -> tuple[str, dict[str, Any]]: """Compute a canonical (commit_id, wire_dict) pair that passes wire_repair_commit's hash check.""" commit_id = hash_commit( parent_ids=[], snapshot_id=snapshot_id, message=message, committed_at_iso=committed_at, author=author, signer_public_key="", ) wire_dict = { "commit_id": commit_id, "branch": "main", "snapshot_id": snapshot_id, "message": message, "committed_at": committed_at, "parent_commit_id": None, "parent2_commit_id": None, "author": author, "signer_public_key": "", "signature": "", "signer_key_id": "", } return commit_id, wire_dict async def _insert_commit_row( session: AsyncSession, repo_id: str, commit_id: str, *, snapshot_id: str, message: str, committed_at: str, author: str = "gabriel", ) -> MusehubCommit: """Insert a MusehubCommit + MusehubCommitRef row that wire_repair_commit can look up.""" ts = datetime.fromisoformat(committed_at.replace("Z", "+00:00")) row = MusehubCommit( commit_id=commit_id, message=message, author=author, branch="main", parent_ids=[], snapshot_id=snapshot_id, timestamp=ts, signer_public_key="", signature="", signer_key_id="", ) session.add(row) session.add(MusehubCommitRef(repo_id=repo_id, commit_id=commit_id)) await session.commit() return row # ══════════════════════════════════════════════════════════════════════════════ # RP_04 — repair-object busts cache (post-fix contract, converts AU_01) # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_RP_04_repair_object_busts_cache( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """RP_04 — wire_repair_object invalidates the cache in the same transaction. After repair, wire_fetch_mpack(want=[T], have=[]) raises MPackNotReadyError (the cache row is gone — the next clone will MISS-rebuild from repaired rows). Row count for repo_id is 0. Converts AU_01 (red-as-bug) to the post-fix correctness contract. """ content = b"rp04-test-object-mwp8" object_id = blob_id(content) tip_T = fake_id("tip-rp04") mpack_M = fake_id("mpack-rp04") repo = await create_repo(db_session, owner="gabriel", visibility="public") await _insert_object_row(db_session, repo.repo_id, object_id) await _insert_fresh_cache_row(db_session, repo.repo_id, tip_T, mpack_M, seed="rp04") _stub_push_backend(monkeypatch) _stub_gc_backend(monkeypatch) result = await wire_repair_object( db_session, repo.repo_id, object_id, content, caller_id="gabriel" ) assert result["repaired"] is True # Post-fix: cache row is gone — fetch raises MPackNotReadyError (MISS) with pytest.raises(MPackNotReadyError): await wire_fetch_mpack( db_session, repo.repo_id, want=[tip_T], have=[], force_build=False ) remaining = await _cache_rows_for_repo(db_session, repo.repo_id) assert len(remaining) == 0, "cache must be fully busted after repair-object" # ══════════════════════════════════════════════════════════════════════════════ # RP_05 — repair-snapshot busts cache (post-fix contract, converts AU_02) # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_RP_05_repair_snapshot_busts_cache( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """RP_05 — wire_repair_snapshot invalidates the cache in the same transaction. Converts AU_02 (red-as-bug) to the post-fix correctness contract. """ manifest: dict[str, str] = {"file.txt": fake_id("obj-rp05")} directories: list[str] = [] snapshot_id = compute_snapshot_id(manifest, directories) tip_T = fake_id("tip-rp05") mpack_M = fake_id("mpack-rp05") repo = await create_repo(db_session, owner="gabriel", visibility="public") await _insert_fresh_cache_row(db_session, repo.repo_id, tip_T, mpack_M, seed="rp05") _stub_gc_backend(monkeypatch) result = await wire_repair_snapshot( db_session, repo.repo_id, snapshot_id, manifest, directories, caller_id="gabriel", ) assert result["repaired"] is True with pytest.raises(MPackNotReadyError): await wire_fetch_mpack( db_session, repo.repo_id, want=[tip_T], have=[], force_build=False ) remaining = await _cache_rows_for_repo(db_session, repo.repo_id) assert len(remaining) == 0, "cache must be fully busted after repair-snapshot" # ══════════════════════════════════════════════════════════════════════════════ # RP_06 — repair-commit busts cache (post-fix contract, converts AU_03) # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_RP_06_repair_commit_busts_cache( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """RP_06 — wire_repair_commit invalidates the cache in the same transaction. This is the direct fix for the gabriel/muse rc10 incident: repair-commit now busts the cache so the next clone rebuilds from the corrected commit row. Converts AU_03 (red-as-bug) to the post-fix correctness contract. """ committed_at = "2026-01-01T00:00:00+00:00" snapshot_id = fake_id("snap-rp06") message = "rp06 post-fix repair-commit test" commit_id, wire_dict = _build_valid_commit_dict( snapshot_id=snapshot_id, message=message, committed_at=committed_at, ) tip_T = commit_id mpack_M = fake_id("mpack-rp06") repo = await create_repo(db_session, owner="gabriel", visibility="public") await _insert_commit_row( db_session, repo.repo_id, commit_id, snapshot_id=snapshot_id, message=message, committed_at=committed_at, ) await _insert_fresh_cache_row(db_session, repo.repo_id, tip_T, mpack_M, seed="rp06") _stub_gc_backend(monkeypatch) result = await wire_repair_commit( db_session, repo.repo_id, wire_dict, caller_id="gabriel" ) assert result["repaired"] is True with pytest.raises(MPackNotReadyError): await wire_fetch_mpack( db_session, repo.repo_id, want=[tip_T], have=[], force_build=False ) remaining = await _cache_rows_for_repo(db_session, repo.repo_id) assert len(remaining) == 0, "cache must be fully busted after repair-commit" # ══════════════════════════════════════════════════════════════════════════════ # RP_07 — atomicity guard: failed repair commit rolls back the invalidation # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_RP_07_repair_rollback_keeps_cache( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """RP_07 — if the repair's session.commit() fails, the invalidation rolls back too. No half-state: the cache row must survive a repair that fails to commit, because both the repair write and the cache deletion are in the same transaction (Design Decision D2). This prevents a scenario where: - invalidate_fetch_mpack_cache deletes the row (no commit) - session.commit() fails (e.g. constraint violation, DB error) - the repair is rejected — but the cache is nuked → next clone is a MISS on an un-repaired repo, hiding the failure from the operator. Uses wire_repair_object as the representative trigger. """ content = b"rp07-atomicity-guard-mwp8" object_id = blob_id(content) tip_T = fake_id("tip-rp07") mpack_M = fake_id("mpack-rp07") repo = await create_repo(db_session, owner="gabriel", visibility="public") repo_id = repo.repo_id # capture before any potential rollback await _insert_object_row(db_session, repo_id, object_id) await _insert_fresh_cache_row(db_session, repo_id, tip_T, mpack_M, seed="rp07") _stub_push_backend(monkeypatch) _stub_gc_backend(monkeypatch) # Patch session.commit to raise after setup is complete — simulates a DB # commit failure (e.g. a concurrent constraint violation). original_commit = db_session.commit commit_calls = 0 async def failing_commit() -> None: nonlocal commit_calls commit_calls += 1 raise RuntimeError("RP_07: simulated commit failure") monkeypatch.setattr(db_session, "commit", failing_commit) with pytest.raises(RuntimeError, match="RP_07"): await wire_repair_object( db_session, repo_id, object_id, content, caller_id="gabriel" ) # Restore commit so we can rollback and then query monkeypatch.setattr(db_session, "commit", original_commit) await db_session.rollback() # The cache row must still exist — invalidation was not committed remaining = await _cache_rows_for_repo(db_session, repo_id) assert len(remaining) == 1, ( "cache row must survive a repair whose commit failed — " "invalidation must be in the same transaction (D2)" ) # ══════════════════════════════════════════════════════════════════════════════ # FP_02 — force-push busts old + new tip rows (post-fix, converts AU_04) # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_FP_02_force_push_busts_old_and_new_tip( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """FP_02 — force-push deletes cache rows for both the old and new tip (FP_01 wiring). Scenario: - Branch main currently points to T_old; a fresh cache row exists for T_old. - A cache row for T_new also exists (e.g. T_new was once a prior branch head — the hazard described in Background 'Trigger 2, hazard 2'). - Force-push main → T_new. Post-fix contract: - Both (repo_id, T_old) and (repo_id, T_new) rows are deleted in the same transaction as the branch advance (Design Decision D1 tip-scoped, D2 same-tx). - A subsequent clone of T_old raises MPackNotReadyError (MISS). Converts AU_04 (red-as-bug) to the post-fix correctness contract. """ from musehub.core.genesis import compute_branch_id mpack_bytes = build_wire_mpack({"objects": [], "commits": [], "snapshots": []}) mpack_key = blob_id(mpack_bytes) T_old = fake_id("tip-old-fp02") T_new = fake_id("tip-new-fp02") mpack_old = fake_id("mpack-fp02-old") mpack_new_cached = fake_id("mpack-fp02-new-cached") repo = await create_repo(db_session, owner="gabriel", visibility="public") repo_id = repo.repo_id db_session.add(MusehubBranch( branch_id=compute_branch_id(repo_id, "main"), repo_id=repo_id, name="main", head_commit_id=T_old, )) await db_session.commit() # Seed cache rows for both tips (old + new) await _insert_fresh_cache_row(db_session, repo_id, T_old, mpack_old, seed="fp02-old") await _insert_fresh_cache_row(db_session, repo_id, T_new, mpack_new_cached, seed="fp02-new") _build_force_push_backend(monkeypatch, mpack_bytes) _stub_gc_backend(monkeypatch) await wire_push_unpack_mpack( db_session, repo_id, mpack_key=mpack_key, pusher_id="gabriel", branch="main", head_commit_id=T_new, force=True, ) remaining = await _cache_rows_for_repo(db_session, repo_id) remaining_tips = {r.tip_commit_id for r in remaining} assert T_old not in remaining_tips, "(repo_id, T_old) must be gone after force-push" assert T_new not in remaining_tips, "(repo_id, T_new) must be gone after force-push" # A clone of T_old is now a MISS with pytest.raises(MPackNotReadyError): await wire_fetch_mpack(db_session, repo_id, want=[T_old], have=[], force_build=False) # ══════════════════════════════════════════════════════════════════════════════ # FP_03 — normal (non-force) push does NOT invalidate sibling cache rows # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_FP_03_normal_push_does_not_invalidate( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """FP_03 — a non-force push must not invalidate sibling branch cache rows (Goal #3 / D3). Scenario: - Repo has a 'dev' branch tip (T_dev) with a fresh cache row — the sibling. - 'main' branch does not yet exist (first push to it). - Push main → T_main (force=False, new branch creation). With current_head = None (new branch), the invalidation guard `force and current_head and current_head != tip` is False on two counts (force=False, current_head=None). The sibling T_dev row must survive intact. This asserts Design Decision D3: only force-push triggers invalidation; the common fast-forward path pays zero invalidation cost. """ mpack_bytes = build_wire_mpack({"objects": [], "commits": [], "snapshots": []}) mpack_key = blob_id(mpack_bytes) T_dev = fake_id("tip-dev-fp03") T_main = fake_id("tip-main-fp03") mpack_dev = fake_id("mpack-dev-fp03") repo = await create_repo(db_session, owner="gabriel", visibility="public") repo_id = repo.repo_id # Seed the sibling dev branch cache row (no branch row needed — the cache # row tests whether the sibling survives the push to main) await _insert_fresh_cache_row(db_session, repo_id, T_dev, mpack_dev, seed="fp03-dev") _build_force_push_backend(monkeypatch, mpack_bytes) _stub_gc_backend(monkeypatch) # Non-force push creating a new 'main' branch — current_head is None await wire_push_unpack_mpack( db_session, repo_id, mpack_key=mpack_key, pusher_id="gabriel", branch="main", head_commit_id=T_main, force=False, ) remaining = await _cache_rows_for_repo(db_session, repo_id) remaining_tips = {r.tip_commit_id for r in remaining} assert T_dev in remaining_tips, "sibling dev cache row must survive a non-force push" # ══════════════════════════════════════════════════════════════════════════════ # FP_04 — force-push still enqueues a prebuild for T_new (loop closed) # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_FP_04_force_push_enqueues_prebuild_for_new_tip( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """FP_04 — after force-push, a fetch.mpack.prebuild job is enqueued for T_new. The invalidation (FP_01) deletes old cache rows; the existing enqueue_push_intel path (unchanged by this ticket) then repopulates T_new with a fresh mpack. This test verifies the loop is still closed — force-push = delete stale rows + enqueue rebuild — so the next clone is a fast HIT on correct content, not a slow synchronous MISS-build or a stale HIT. Asserts Design Decision D4: correctness via deletion; speed via prebuild. """ from musehub.db.musehub_jobs_models import MusehubBackgroundJob from musehub.core.genesis import compute_branch_id mpack_bytes = build_wire_mpack({"objects": [], "commits": [], "snapshots": []}) mpack_key = blob_id(mpack_bytes) T_old = fake_id("tip-old-fp04") T_new = fake_id("tip-new-fp04") mpack_old = fake_id("mpack-fp04-old") repo = await create_repo(db_session, owner="gabriel", visibility="public") repo_id = repo.repo_id db_session.add(MusehubBranch( branch_id=compute_branch_id(repo_id, "main"), repo_id=repo_id, name="main", head_commit_id=T_old, )) await db_session.commit() await _insert_fresh_cache_row(db_session, repo_id, T_old, mpack_old, seed="fp04-old") _build_force_push_backend(monkeypatch, mpack_bytes) _stub_gc_backend(monkeypatch) await wire_push_unpack_mpack( db_session, repo_id, mpack_key=mpack_key, pusher_id="gabriel", branch="main", head_commit_id=T_new, force=True, ) # A fetch.mpack.prebuild job must exist for the repo (enqueued by enqueue_push_intel) job_rows = (await db_session.execute( select(MusehubBackgroundJob).where( MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", MusehubBackgroundJob.repo_id == repo_id, ) )).scalars().all() assert len(job_rows) >= 1, ( "force-push must enqueue a fetch.mpack.prebuild job so T_new gets a fresh " "cache row after invalidation (D4 — correctness via deletion, speed via prebuild)" ) # ══════════════════════════════════════════════════════════════════════════════ # Phase 1 — The invalidation primitive (IV_01..IV_06) # ══════════════════════════════════════════════════════════════════════════════ # ── IV_01 ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.tier2 async def test_IV_01_invalidate_all_deletes_every_row_for_repo( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """IV_01 — repo-wide invalidation deletes all rows for target repo; sibling repo untouched. Seed 3 rows for repo A (mix of fresh + expired) and 2 rows for repo B. invalidate_fetch_mpack_cache(session, A) must return 3, leave A with 0 rows, and leave B's 2 rows intact (cross-repo isolation). """ from musehub.services.musehub_gc import invalidate_fetch_mpack_cache _stub_push_backend(monkeypatch) repo_a = await create_repo(db_session, owner="gabriel", visibility="public") repo_b = await create_repo(db_session, owner="gabriel", visibility="public") # 3 rows for A — mix of fresh and expired for seed in ("iv01-a1", "iv01-a2", "iv01-a3"): expired = seed.endswith("a3") now = _now() db_session.add(MusehubFetchMPackCache( cache_id=fake_id(f"cache-{seed}"), repo_id=repo_a.repo_id, tip_commit_id=fake_id(f"tip-{seed}"), mpack_id=fake_id(f"mpack-{seed}"), created_at=now, expires_at=now - timedelta(minutes=5) if expired else now + timedelta(days=7), )) await db_session.commit() # 2 rows for B — all fresh for seed in ("iv01-b1", "iv01-b2"): await _insert_fresh_cache_row(db_session, repo_b.repo_id, fake_id(f"tip-{seed}"), fake_id(f"mpack-{seed}"), seed=seed) deleted = await invalidate_fetch_mpack_cache(db_session, repo_a.repo_id) await db_session.commit() assert deleted == 3, f"expected 3 deleted, got {deleted}" assert len(await _cache_rows_for_repo(db_session, repo_a.repo_id)) == 0, "repo A must have 0 rows" assert len(await _cache_rows_for_repo(db_session, repo_b.repo_id)) == 2, "repo B must be untouched" # ── IV_02 ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.tier2 async def test_IV_02_invalidate_deletes_regardless_of_expiry( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """IV_02 — invalidate deletes fresh (non-expired) rows; gc_fetch_mpack_cache does not. This is the key behavioral difference: gc only removes expired rows, invalidate removes regardless of expires_at. """ from musehub.services.musehub_gc import gc_fetch_mpack_cache, invalidate_fetch_mpack_cache _stub_push_backend(monkeypatch) repo = await create_repo(db_session, owner="gabriel", visibility="public") tip = fake_id("tip-iv02") mpack_id = fake_id("mpack-iv02") # Insert a FRESH (non-expired) row now = _now() db_session.add(MusehubFetchMPackCache( cache_id=fake_id("cache-iv02"), repo_id=repo.repo_id, tip_commit_id=tip, mpack_id=mpack_id, created_at=now, expires_at=now + timedelta(days=7), )) await db_session.commit() # gc_fetch_mpack_cache must NOT delete the fresh row gc_deleted = await gc_fetch_mpack_cache(db_session, repo.repo_id) assert gc_deleted == 0, "gc must not delete fresh rows" assert len(await _cache_rows_for_repo(db_session, repo.repo_id)) == 1, "row must survive gc" # invalidate_fetch_mpack_cache MUST delete the fresh row inv_deleted = await invalidate_fetch_mpack_cache(db_session, repo.repo_id) await db_session.commit() assert inv_deleted == 1, "invalidate must delete regardless of expiry" assert len(await _cache_rows_for_repo(db_session, repo.repo_id)) == 0, "row must be gone after invalidate" # ── IV_03 ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.tier2 async def test_IV_03_invalidate_tip_scoped_leaves_siblings( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """IV_03 — tip-scoped invalidation deletes only the named tips; siblings survive. Seed rows for tips [X, Y, Z]. invalidate(..., tips=[X]) must delete only X; Y and Z must be untouched. """ from musehub.services.musehub_gc import invalidate_fetch_mpack_cache _stub_push_backend(monkeypatch) repo = await create_repo(db_session, owner="gabriel", visibility="public") tip_x = fake_id("tip-x-iv03") tip_y = fake_id("tip-y-iv03") tip_z = fake_id("tip-z-iv03") for tip, seed in ((tip_x, "iv03-x"), (tip_y, "iv03-y"), (tip_z, "iv03-z")): await _insert_fresh_cache_row(db_session, repo.repo_id, tip, fake_id(f"mpack-{seed}"), seed=seed) deleted = await invalidate_fetch_mpack_cache(db_session, repo.repo_id, tips=[tip_x]) await db_session.commit() assert deleted == 1, "only 1 row (X) must be deleted" remaining = await _cache_rows_for_repo(db_session, repo.repo_id) remaining_tips = {r.tip_commit_id for r in remaining} assert tip_x not in remaining_tips, "X must be gone" assert tip_y in remaining_tips, "Y must survive" assert tip_z in remaining_tips, "Z must survive" # ── IV_04 ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.tier2 async def test_IV_04_invalidate_best_effort_r2_delete_swallows_failure( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """IV_04 — backend.delete raising must not propagate; DB row still deleted. Mirrors the best-effort pattern from test_bc2_gc_delete_failure_is_swallowed. R2 unavailability must not block cache invalidation — the DB row is the source of truth for the HIT gate, not the R2 object. """ from musehub.services.musehub_gc import invalidate_fetch_mpack_cache backend = AsyncMock() backend.delete = AsyncMock(side_effect=RuntimeError("R2 unavailable")) monkeypatch.setattr("musehub.services.musehub_gc.get_backend", lambda: backend) repo = await create_repo(db_session, owner="gabriel", visibility="public") await _insert_fresh_cache_row(db_session, repo.repo_id, fake_id("tip-iv04"), fake_id("mpack-iv04"), seed="iv04") # Must not raise even though backend.delete raises deleted = await invalidate_fetch_mpack_cache(db_session, repo.repo_id) await db_session.commit() assert deleted == 1, "row must be deleted from DB even when R2 delete fails" backend.delete.assert_awaited_once() assert len(await _cache_rows_for_repo(db_session, repo.repo_id)) == 0, "DB row must be gone" # ── IV_05 ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.tier2 async def test_IV_05_invalidate_does_not_commit( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """IV_05 — invalidate_fetch_mpack_cache issues no COMMIT (Design Decision D2). The caller owns the transaction. Rows deleted within the open transaction must be invisible after a session.rollback(), proving the deletion was not committed by the primitive itself. """ from musehub.services.musehub_gc import invalidate_fetch_mpack_cache _stub_push_backend(monkeypatch) repo = await create_repo(db_session, owner="gabriel", visibility="public") repo_id = repo.repo_id # capture before rollback; ORM objects expire on rollback await _insert_fresh_cache_row(db_session, repo_id, fake_id("tip-iv05"), fake_id("mpack-iv05"), seed="iv05") # Rows are visible before invalidation assert len(await _cache_rows_for_repo(db_session, repo_id)) == 1 deleted = await invalidate_fetch_mpack_cache(db_session, repo_id) assert deleted == 1 # Within the open transaction, the row appears deleted assert len(await _cache_rows_for_repo(db_session, repo_id)) == 0, "row deleted within transaction" # Roll back — if the primitive had committed, this would be a no-op and the row would stay gone. # If the primitive did NOT commit (correct), the row is restored. await db_session.rollback() assert len(await _cache_rows_for_repo(db_session, repo_id)) == 1, ( "row must be restored after rollback — invalidate must not commit (D2)" ) # ── IV_06 ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.tier2 async def test_IV_06_invalidate_empty_repo_is_noop_returns_zero( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """IV_06 — invalidating a repo with no cache rows is a no-op returning 0.""" from musehub.services.musehub_gc import invalidate_fetch_mpack_cache _stub_push_backend(monkeypatch) repo = await create_repo(db_session, owner="gabriel", visibility="public") deleted = await invalidate_fetch_mpack_cache(db_session, repo.repo_id) await db_session.commit() assert deleted == 0 assert len(await _cache_rows_for_repo(db_session, repo.repo_id)) == 0 # ══════════════════════════════════════════════════════════════════════════════ # Phase 4 — Optional prebuild-on-repair + E2E clone-after-repair (EX_01..EX_03) # ══════════════════════════════════════════════════════════════════════════════ # ── EX_01 ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.tier2 async def test_EX_01_repair_enqueues_prebuild( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """EX_01 — each repair function enqueues a fetch.mpack.prebuild job after invalidation. Closing the loop: after repair busts the cache, the worker rebuilds a fresh mpack from the corrected rows so the NEXT clone is a fast HIT, not a slow synchronous MISS-build (Design Decision D4 — correctness via deletion, speed via prebuild). Uses wire_repair_commit as the representative repair trigger. The branch must exist so _enqueue_repair_prebuild has branch heads to cover. """ from musehub.core.genesis import compute_branch_id committed_at = "2026-01-01T00:00:00+00:00" snapshot_id = fake_id("snap-ex01") message = "ex01 prebuild-on-repair test" commit_id, wire_dict = _build_valid_commit_dict( snapshot_id=snapshot_id, message=message, committed_at=committed_at, ) repo = await create_repo(db_session, owner="gabriel", visibility="public") repo_id = repo.repo_id # Branch must exist so _enqueue_repair_prebuild finds a tip to cover db_session.add(MusehubBranch( branch_id=compute_branch_id(repo_id, "main"), repo_id=repo_id, name="main", head_commit_id=commit_id, )) await _insert_commit_row( db_session, repo_id, commit_id, snapshot_id=snapshot_id, message=message, committed_at=committed_at, ) _stub_gc_backend(monkeypatch) result = await wire_repair_commit(db_session, repo_id, wire_dict, caller_id="gabriel") assert result["repaired"] is True # A fetch.mpack.prebuild job must be enqueued for the repo job_rows = (await db_session.execute( select(MusehubBackgroundJob).where( MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", MusehubBackgroundJob.repo_id == repo_id, ) )).scalars().all() assert len(job_rows) >= 1, ( "repair must enqueue a fetch.mpack.prebuild job so the next clone " "is a fast HIT on corrected content (EX_01 — D4)" ) # The job payload must cover the branch tip job = job_rows[0] assert commit_id in job.payload.get("tip_commit_ids", []), ( "prebuild job payload must include the branch tip so the worker " "rebuilds the mpack for the corrected content" ) # ── EX_02 ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.tier2 @pytest.mark.wire async def test_EX_02_clone_after_repair_reflects_repair( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """EX_02 — load-bearing E2E: repair busts cache; second clone sees corrected rows. This is the regression test for the gabriel/muse rc10 incident: - A push populates musehub_fetch_mpack_cache (simulated by seeding a fresh row). - A first clone is a fast HIT returning mpack_M_stale. - An operator calls repair-commit to fix a signer identity field. - A second clone attempt raises MPackNotReadyError (MISS) — the stale mpack_M is gone and the cache row was deleted in the same transaction as the repair. - The corrected commit row is in the DB and will be served by the next rebuild. - A fetch.mpack.prebuild job is enqueued so the rebuild happens before the next real client request. Combines RP_06 (repair-commit invalidates) + EX_01 (prebuild enqueued) into a single narrative to prove the full chain is wired correctly. """ from musehub.core.genesis import compute_branch_id committed_at = "2026-01-15T12:00:00+00:00" snapshot_id = fake_id("snap-ex02") message = "ex02 e2e repair test" commit_id, wire_dict = _build_valid_commit_dict( snapshot_id=snapshot_id, message=message, committed_at=committed_at, ) repo = await create_repo(db_session, owner="gabriel", visibility="public") repo_id = repo.repo_id # Branch pointing to this commit (main tip = commit_id) db_session.add(MusehubBranch( branch_id=compute_branch_id(repo_id, "main"), repo_id=repo_id, name="main", head_commit_id=commit_id, )) await _insert_commit_row( db_session, repo_id, commit_id, snapshot_id=snapshot_id, message=message, committed_at=committed_at, ) # Seed the pre-repair mpack cache row (simulates state after a push + prebuild) mpack_stale = fake_id("mpack-stale-ex02") await _insert_fresh_cache_row(db_session, repo_id, commit_id, mpack_stale, seed="ex02") _stub_gc_backend(monkeypatch) _stub_fetch_backend(monkeypatch, presign_url="https://r2.example/stale-mpack") # ── First clone: cache HIT on pre-repair mpack ──────────────────────────── first_clone = await wire_fetch_mpack(db_session, repo_id, want=[commit_id], have=[]) assert first_clone["mpack_id"] == mpack_stale, "first clone must HIT the pre-repair mpack" # ── Repair: fix the commit's signer field ───────────────────────────────── result = await wire_repair_commit(db_session, repo_id, wire_dict, caller_id="gabriel") assert result["repaired"] is True # ── Second clone: MISS — cache was busted by the repair in the same tx ──── with pytest.raises(MPackNotReadyError): await wire_fetch_mpack(db_session, repo_id, want=[commit_id], have=[], force_build=False) # The stale cache row is gone remaining = await _cache_rows_for_repo(db_session, repo_id) assert len(remaining) == 0, "cache must be empty after repair" # The corrected commit row is in the DB — next rebuild will serve correct bytes from sqlalchemy import select as _select from musehub.db.musehub_repo_models import MusehubCommit as _MC commit_row = (await db_session.execute( _select(_MC).where(_MC.commit_id == commit_id) )).scalar_one() assert commit_row.signer_public_key == wire_dict["signer_public_key"], ( "the commit row must have the repaired signer_public_key — " "the next rebuild will serve this corrected content" ) # A prebuild job was enqueued to close the loop job_rows = (await db_session.execute( select(MusehubBackgroundJob).where( MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", MusehubBackgroundJob.repo_id == repo_id, ) )).scalars().all() assert len(job_rows) >= 1, "repair must enqueue a prebuild so the next clone is a fast HIT" # ── EX_03 ───────────────────────────────────────────────────────────────────── @pytest.mark.asyncio @pytest.mark.tier2 async def test_EX_03_invalidation_is_logged( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, ) -> None: """EX_03 — invalidation emits a log line naming repo, scope, and rows deleted. Operators need visibility into when the cache was busted — the gabriel/muse rc10 incident was invisible because there was no invalidation AND no log. This test asserts that after a repair trigger, a log record exists that an operator can grep for to confirm the bust happened. Checks the logger.info in invalidate_fetch_mpack_cache: "invalidate_fetch_mpack_cache: repo=... tips=all deleted=N rows" """ import logging committed_at = "2026-02-01T00:00:00+00:00" snapshot_id = fake_id("snap-ex03") message = "ex03 logging test" commit_id, wire_dict = _build_valid_commit_dict( snapshot_id=snapshot_id, message=message, committed_at=committed_at, ) repo = await create_repo(db_session, owner="gabriel", visibility="public") repo_id = repo.repo_id await _insert_commit_row( db_session, repo_id, commit_id, snapshot_id=snapshot_id, message=message, committed_at=committed_at, ) tip_T = commit_id await _insert_fresh_cache_row(db_session, repo_id, tip_T, fake_id("mpack-ex03"), seed="ex03") _stub_gc_backend(monkeypatch) with caplog.at_level(logging.INFO, logger="musehub.services.musehub_gc"): result = await wire_repair_commit(db_session, repo_id, wire_dict, caller_id="gabriel") assert result["repaired"] is True # The invalidation log line must appear, naming repo + deletion count gc_logs = [r.message for r in caplog.records if "invalidate_fetch_mpack_cache" in r.message] assert len(gc_logs) >= 1, ( "invalidate_fetch_mpack_cache must emit a log line so operators can confirm " "the bust happened (the rc10 incident was invisible — this log line is the fix)" ) log_line = gc_logs[0] assert "deleted=1" in log_line or "deleted" in log_line, ( f"log line must name the rows deleted; got: {log_line!r}" ) assert repo_id[:8] in log_line or "repo=" in log_line, ( f"log line must name the repo; got: {log_line!r}" )