"""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.types import blob_id, fake_id from musehub.db.musehub_repo_models import ( MusehubBranch, MusehubCommit, MusehubCommitRef, MusehubFetchMPackCache, MusehubObject, MusehubObjectRef, ) 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_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 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 # ══════════════════════════════════════════════════════════════════════════════ # AU_01 — repair-object leaves cache row intact → stale HIT (RED-AS-BUG) # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_AU_01_repair_object_leaves_stale_cache_hit( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """AU_01 — RED-AS-BUG: wire_repair_object does not bust the cache. After the repair, the (repo_id, tip_T) cache row still exists and wire_fetch_mpack(want=[T], have=[]) returns the pre-repair mpack_id. EXPECTED POST-FIX BEHAVIOR (Phase 2, RP_01 + RP_04): wire_fetch_mpack raises MPackNotReadyError — cache row was deleted in the same transaction as the repair. """ content = b"au01-test-object-mwp8" object_id = blob_id(content) tip_T = fake_id("tip-au01") mpack_M = fake_id("mpack-au01") 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="au01") _stub_push_backend(monkeypatch) _stub_fetch_backend(monkeypatch) # Repair the object — cache is NOT busted (the bug) result = await wire_repair_object( db_session, repo.repo_id, object_id, content, caller_id="gabriel" ) assert result["repaired"] is True # BUG: the cache row survived the repair → fetch HITs the stale mpack_M fetch_result = await wire_fetch_mpack( db_session, repo.repo_id, want=[tip_T], have=[] ) assert fetch_result["mpack_id"] == mpack_M, ( "BUG (AU_01): cache row survived wire_repair_object — " "fetch returns pre-repair mpack_M. " "After Phase 2 fix (RP_01+RP_04), must raise MPackNotReadyError." ) # Confirm the stale row is still in the table remaining = await _cache_rows_for_repo(db_session, repo.repo_id) assert len(remaining) == 1, "cache row count must be 1 (bug: row was not busted)" # ══════════════════════════════════════════════════════════════════════════════ # AU_02 — repair-snapshot leaves cache row intact → stale HIT (RED-AS-BUG) # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_AU_02_repair_snapshot_leaves_stale_cache_hit( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """AU_02 — RED-AS-BUG: wire_repair_snapshot does not bust the cache. After the snapshot repair, the (repo_id, tip_T) cache row still exists and wire_fetch_mpack(want=[T], have=[]) returns the pre-repair mpack_id. EXPECTED POST-FIX BEHAVIOR (Phase 2, RP_02 + RP_05): wire_fetch_mpack raises MPackNotReadyError. """ manifest: dict[str, str] = {"file.txt": fake_id("obj-au02")} directories: list[str] = [] snapshot_id = compute_snapshot_id(manifest, directories) tip_T = fake_id("tip-au02") mpack_M = fake_id("mpack-au02") 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="au02") _stub_fetch_backend(monkeypatch) # Repair the snapshot — cache is NOT busted (the bug) result = await wire_repair_snapshot( db_session, repo.repo_id, snapshot_id, manifest, directories, caller_id="gabriel", ) assert result["repaired"] is True # BUG: cache row survived → fetch HITs the stale mpack_M fetch_result = await wire_fetch_mpack( db_session, repo.repo_id, want=[tip_T], have=[] ) assert fetch_result["mpack_id"] == mpack_M, ( "BUG (AU_02): cache row survived wire_repair_snapshot — " "fetch returns pre-repair mpack_M. " "After Phase 2 fix (RP_02+RP_05), must raise MPackNotReadyError." ) remaining = await _cache_rows_for_repo(db_session, repo.repo_id) assert len(remaining) == 1, "cache row must still exist (bug: row was not busted)" # ══════════════════════════════════════════════════════════════════════════════ # AU_03 — repair-commit leaves cache row intact → stale HIT (RED-AS-BUG) # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_AU_03_repair_commit_leaves_stale_cache_hit( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """AU_03 — RED-AS-BUG: wire_repair_commit does not bust the cache. This is the exact failure from the gabriel/muse rc10 incident: - repair-commit fixed the row's signer_public_key - the cache kept serving the corrupted mpack for 7 more days - the manual fix was: UPDATE musehub_fetch_mpack_cache SET expires_at = NOW() - INTERVAL '1 second' EXPECTED POST-FIX BEHAVIOR (Phase 2, RP_03 + RP_06): wire_fetch_mpack raises MPackNotReadyError. """ committed_at = "2026-01-01T00:00:00+00:00" snapshot_id = fake_id("snap-au03") message = "au03 audit test commit" commit_id, wire_dict = _build_valid_commit_dict( snapshot_id=snapshot_id, message=message, committed_at=committed_at, ) tip_T = commit_id # use the commit_id as the branch tip — matches real usage mpack_M = fake_id("mpack-au03") 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="au03") _stub_fetch_backend(monkeypatch) # Repair the commit — cache is NOT busted (the bug) result = await wire_repair_commit( db_session, repo.repo_id, wire_dict, caller_id="gabriel" ) assert result["repaired"] is True # BUG: cache row survived → fetch HITs the stale mpack_M fetch_result = await wire_fetch_mpack( db_session, repo.repo_id, want=[tip_T], have=[] ) assert fetch_result["mpack_id"] == mpack_M, ( "BUG (AU_03): cache row survived wire_repair_commit — " "fetch returns pre-repair mpack_M. " "This is the gabriel/muse rc10 incident. " "After Phase 2 fix (RP_03+RP_06), must raise MPackNotReadyError." ) remaining = await _cache_rows_for_repo(db_session, repo.repo_id) assert len(remaining) == 1, "cache row must still exist (bug: row was not busted)" # ══════════════════════════════════════════════════════════════════════════════ # AU_04 — force-push leaves old-tip cache row intact → stale HIT (RED-AS-BUG) # ══════════════════════════════════════════════════════════════════════════════ @pytest.mark.asyncio @pytest.mark.tier2 async def test_AU_04_force_push_leaves_old_tip_cache_row( db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch, ) -> None: """AU_04 — RED-AS-BUG: wire_push_unpack_mpack(force=True) does not bust the old-tip cache row. After force-push branch main from T_old → T_new, the (repo_id, T_old) cache row still exists. A client that still holds T_old as its remote tip and requests a fresh clone (want=[T_old], have=[]) gets a HIT on content that is no longer on any branch. EXPECTED POST-FIX BEHAVIOR (Phase 3, FP_01 + FP_02): After force-push, the (repo_id, T_old) row is gone; the fetch raises MPackNotReadyError. """ from musehub.core.genesis import compute_branch_id from musehub.services.musehub_wire_push import wire_push_unpack_mpack T_old = fake_id("tip-old-au04") T_new = fake_id("tip-new-au04") mpack_old = fake_id("mpack-old-au04") mpack_key = fake_id("mpack-key-au04") repo = await create_repo(db_session, owner="gabriel", visibility="public") # Create branch main pointing to T_old db_session.add(MusehubBranch( branch_id=compute_branch_id(repo.repo_id, "main"), repo_id=repo.repo_id, name="main", head_commit_id=T_old, )) await db_session.commit() # Seed a fresh cache row for the old tip await _insert_fresh_cache_row(db_session, repo.repo_id, T_old, mpack_old, seed="au04-old") # Mock all backends: the push backend needs to handle get_backend().get(mpack_key) # returning a minimal valid mpack (empty), plus the fetch backend for the clone. push_backend = AsyncMock() push_backend.get = AsyncMock(return_value=_empty_mpack_bytes()) push_backend.delete = AsyncMock() monkeypatch.setattr("musehub.services.musehub_wire_push.get_backend", lambda: push_backend) fetch_backend = AsyncMock() fetch_backend.presign_mpack_get = AsyncMock(return_value="https://r2.example/old-mpack") monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: fetch_backend) # Force-push main from T_old to T_new # wire_push_unpack_mpack will parse the mpack — provide a minimal valid one. # The push result is secondary; what matters is the cache row audit. try: await wire_push_unpack_mpack( db_session, repo.repo_id, mpack_key=mpack_key, pusher_id="gabriel", branch="main", head_commit_id=T_new, force=True, ) except Exception: # wire_push_unpack_mpack may raise if the mpack is empty (no commits) — # that's fine for this audit test. We only care whether the cache row # was busted, regardless of whether the push succeeded. pass # BUG: the old-tip cache row still exists after force-push remaining = await _cache_rows_for_repo(db_session, repo.repo_id) old_tip_rows = [r for r in remaining if r.tip_commit_id == T_old] assert len(old_tip_rows) == 1, ( "BUG (AU_04): (repo_id, T_old) cache row survived force-push. " "A clone of T_old will HIT this stale row. " "After Phase 3 fix (FP_01+FP_02), this row must be gone." ) # Confirm the stale HIT: a clone of T_old still gets the pre-force-push mpack fetch_result = await wire_fetch_mpack( db_session, repo.repo_id, want=[T_old], have=[] ) assert fetch_result["mpack_id"] == mpack_old, ( "BUG (AU_04): wire_fetch_mpack HITs stale mpack_old for the superseded T_old tip." ) def _empty_mpack_bytes() -> bytes: """Minimal valid msgpack bytes that wire_push_unpack_mpack will accept as an mpack. The push handler calls msgpack.unpackb on the raw bytes from backend.get(). An empty dict `{}` is the smallest payload that won't raise a parse error, though the push will fail on missing fields (commits/objects) — that's fine for this audit, which only needs the cache row check, not a successful push. """ import msgpack return msgpack.packb({}, use_bin_type=True)