gabriel / musehub public
test_mwp8_cache_invalidation.py python
482 lines 19.1 KB
Raw
sha256:0e5174da777684df43b2db71f282079030ef28bacffb93e19c29ee712b2a8bc4 test(mwp8/phase0): audit — repair+force-push leave stale ca… Sonnet 4.6 21 days ago
1 """MWP-8 — fetch-mpack cache invalidation (musehub#111).
2
3 AUDIT VERDICT (Phase 0 — AU_01..AU_04):
4 ========================================
5 All three repair endpoints (wire_repair_object, wire_repair_snapshot,
6 wire_repair_commit) and the force-push path (wire_push_unpack_mpack with
7 force=True) share the same gap:
8
9 1. repair-object (AU_01) — wire_repair_object performs upsert + session.commit()
10 but NEVER touches musehub_fetch_mpack_cache. A fresh cache row for the tip
11 that reaches the repaired object survives unchanged. A subsequent fresh-clone
12 request (want=[tip], have=[]) returns a HIT on the pre-repair mpack_id.
13
14 2. repair-snapshot (AU_02) — wire_repair_snapshot resets manifest_blob +
15 session.commit() but NEVER touches musehub_fetch_mpack_cache. Same HIT
16 on stale bytes.
17
18 3. repair-commit (AU_03) — wire_repair_commit force-overwrites identity fields +
19 session.commit() but NEVER touches musehub_fetch_mpack_cache. Same HIT.
20 This is the exact mechanism behind the gabriel/muse rc10 incident: the
21 repair fixed the DB row but the cache kept serving the corrupted mpack for
22 7 more days.
23
24 4. force-push (AU_04) — wire_push_unpack_mpack(force=True) advances the
25 branch pointer and enqueues intel jobs but NEVER deletes the old-tip cache
26 row. A client that still knows the old tip gets a HIT on content that is
27 no longer on any branch.
28
29 Expected post-fix behavior (implemented in Phases 2–3):
30 • AU_01..AU_03 → after repair, wire_fetch_mpack(want=[tip], have=[]) raises
31 MPackNotReadyError (cache row deleted in same transaction as repair).
32 • AU_04 → after force-push, (repo_id, old_tip) row is gone; the same
33 fetch raises MPackNotReadyError.
34
35 Phase 2 (RP_04..RP_06) and Phase 3 (FP_02) replace these assertions with the
36 corrected contracts.
37
38 Run:
39 python3 -m pytest tests/test_mwp8_cache_invalidation.py -q --tb=short
40 """
41 from __future__ import annotations
42
43 import hashlib
44 from datetime import datetime, timedelta, timezone
45 from typing import Any
46 from unittest.mock import AsyncMock, patch
47
48 import pytest
49 from sqlalchemy import select
50 from sqlalchemy.ext.asyncio import AsyncSession
51
52 from muse.core.ids import hash_commit, long_id
53 from muse.core.types import blob_id, fake_id
54 from musehub.db.musehub_repo_models import (
55 MusehubBranch,
56 MusehubCommit,
57 MusehubCommitRef,
58 MusehubFetchMPackCache,
59 MusehubObject,
60 MusehubObjectRef,
61 )
62 from musehub.muse_cli.snapshot import compute_snapshot_id
63 from musehub.services.musehub_wire_fetch import wire_fetch_mpack
64 from musehub.services.musehub_wire_push import (
65 wire_repair_commit,
66 wire_repair_object,
67 wire_repair_snapshot,
68 )
69 from musehub.services.musehub_wire_shared import MPackNotReadyError
70 from tests.factories import create_branch, create_repo
71
72
73 # ── helpers ───────────────────────────────────────────────────────────────────
74
75
76 def _now() -> datetime:
77 return datetime.now(tz=timezone.utc)
78
79
80 async def _insert_fresh_cache_row(
81 session: AsyncSession,
82 repo_id: str,
83 tip_commit_id: str,
84 mpack_id: str,
85 seed: str,
86 ) -> MusehubFetchMPackCache:
87 """Seed a non-expired cache row (tip → mpack_id) into musehub_fetch_mpack_cache."""
88 now = _now()
89 row = MusehubFetchMPackCache(
90 cache_id=fake_id(f"cache-{seed}"),
91 repo_id=repo_id,
92 tip_commit_id=tip_commit_id,
93 mpack_id=mpack_id,
94 created_at=now,
95 expires_at=now + timedelta(days=7),
96 )
97 session.add(row)
98 await session.commit()
99 return row
100
101
102 async def _cache_rows_for_repo(
103 session: AsyncSession,
104 repo_id: str,
105 ) -> list[MusehubFetchMPackCache]:
106 """Return all cache rows for a given repo_id."""
107 rows = await session.execute(
108 select(MusehubFetchMPackCache).where(
109 MusehubFetchMPackCache.repo_id == repo_id
110 )
111 )
112 return list(rows.scalars().all())
113
114
115 def _stub_push_backend(monkeypatch: pytest.MonkeyPatch) -> AsyncMock:
116 """Patch the storage backend used by musehub_wire_push (repair calls backend.put)."""
117 backend = AsyncMock()
118 backend.put = AsyncMock(return_value="s3://test-bucket/object")
119 monkeypatch.setattr("musehub.services.musehub_wire_push.get_backend", lambda: backend)
120 return backend
121
122
123 def _stub_fetch_backend(monkeypatch: pytest.MonkeyPatch, presign_url: str = "https://r2.example/mpack") -> AsyncMock:
124 """Patch the storage backend used by musehub_wire_fetch (HIT calls backend.presign_mpack_get)."""
125 backend = AsyncMock()
126 backend.presign_mpack_get = AsyncMock(return_value=presign_url)
127 monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend)
128 return backend
129
130
131 async def _insert_object_row(
132 session: AsyncSession,
133 repo_id: str,
134 object_id: str,
135 ) -> None:
136 """Insert a minimal MusehubObject + ref so wire_repair_object can upsert it."""
137 from sqlalchemy.dialects.postgresql import insert as _pg_insert
138 await session.execute(
139 _pg_insert(MusehubObject)
140 .values(object_id=object_id, path="test.bin", size_bytes=0, storage_uri="s3://bucket/x", content_cache=None)
141 .on_conflict_do_nothing(index_elements=["object_id"])
142 )
143 await session.execute(
144 _pg_insert(MusehubObjectRef)
145 .values(repo_id=repo_id, object_id=object_id)
146 .on_conflict_do_nothing(index_elements=["repo_id", "object_id"])
147 )
148 await session.commit()
149
150
151 def _build_valid_commit_dict(
152 *,
153 snapshot_id: str,
154 message: str,
155 committed_at: str,
156 author: str = "gabriel",
157 ) -> tuple[str, dict[str, Any]]:
158 """Compute a canonical (commit_id, wire_dict) pair that passes wire_repair_commit's hash check."""
159 commit_id = hash_commit(
160 parent_ids=[],
161 snapshot_id=snapshot_id,
162 message=message,
163 committed_at_iso=committed_at,
164 author=author,
165 signer_public_key="",
166 )
167 wire_dict = {
168 "commit_id": commit_id,
169 "branch": "main",
170 "snapshot_id": snapshot_id,
171 "message": message,
172 "committed_at": committed_at,
173 "parent_commit_id": None,
174 "parent2_commit_id": None,
175 "author": author,
176 "signer_public_key": "",
177 "signature": "",
178 "signer_key_id": "",
179 }
180 return commit_id, wire_dict
181
182
183 async def _insert_commit_row(
184 session: AsyncSession,
185 repo_id: str,
186 commit_id: str,
187 *,
188 snapshot_id: str,
189 message: str,
190 committed_at: str,
191 author: str = "gabriel",
192 ) -> MusehubCommit:
193 """Insert a MusehubCommit + MusehubCommitRef row that wire_repair_commit can look up."""
194 ts = datetime.fromisoformat(committed_at.replace("Z", "+00:00"))
195 row = MusehubCommit(
196 commit_id=commit_id,
197 message=message,
198 author=author,
199 branch="main",
200 parent_ids=[],
201 snapshot_id=snapshot_id,
202 timestamp=ts,
203 signer_public_key="",
204 signature="",
205 signer_key_id="",
206 )
207 session.add(row)
208 session.add(MusehubCommitRef(repo_id=repo_id, commit_id=commit_id))
209 await session.commit()
210 return row
211
212
213 # ══════════════════════════════════════════════════════════════════════════════
214 # AU_01 — repair-object leaves cache row intact → stale HIT (RED-AS-BUG)
215 # ══════════════════════════════════════════════════════════════════════════════
216
217
218 @pytest.mark.asyncio
219 @pytest.mark.tier2
220 async def test_AU_01_repair_object_leaves_stale_cache_hit(
221 db_session: AsyncSession,
222 monkeypatch: pytest.MonkeyPatch,
223 ) -> None:
224 """AU_01 — RED-AS-BUG: wire_repair_object does not bust the cache.
225
226 After the repair, the (repo_id, tip_T) cache row still exists and
227 wire_fetch_mpack(want=[T], have=[]) returns the pre-repair mpack_id.
228
229 EXPECTED POST-FIX BEHAVIOR (Phase 2, RP_01 + RP_04):
230 wire_fetch_mpack raises MPackNotReadyError — cache row was deleted in
231 the same transaction as the repair.
232 """
233 content = b"au01-test-object-mwp8"
234 object_id = blob_id(content)
235 tip_T = fake_id("tip-au01")
236 mpack_M = fake_id("mpack-au01")
237
238 repo = await create_repo(db_session, owner="gabriel", visibility="public")
239 await _insert_object_row(db_session, repo.repo_id, object_id)
240 await _insert_fresh_cache_row(db_session, repo.repo_id, tip_T, mpack_M, seed="au01")
241
242 _stub_push_backend(monkeypatch)
243 _stub_fetch_backend(monkeypatch)
244
245 # Repair the object — cache is NOT busted (the bug)
246 result = await wire_repair_object(
247 db_session, repo.repo_id, object_id, content, caller_id="gabriel"
248 )
249 assert result["repaired"] is True
250
251 # BUG: the cache row survived the repair → fetch HITs the stale mpack_M
252 fetch_result = await wire_fetch_mpack(
253 db_session, repo.repo_id, want=[tip_T], have=[]
254 )
255 assert fetch_result["mpack_id"] == mpack_M, (
256 "BUG (AU_01): cache row survived wire_repair_object — "
257 "fetch returns pre-repair mpack_M. "
258 "After Phase 2 fix (RP_01+RP_04), must raise MPackNotReadyError."
259 )
260
261 # Confirm the stale row is still in the table
262 remaining = await _cache_rows_for_repo(db_session, repo.repo_id)
263 assert len(remaining) == 1, "cache row count must be 1 (bug: row was not busted)"
264
265
266 # ══════════════════════════════════════════════════════════════════════════════
267 # AU_02 — repair-snapshot leaves cache row intact → stale HIT (RED-AS-BUG)
268 # ══════════════════════════════════════════════════════════════════════════════
269
270
271 @pytest.mark.asyncio
272 @pytest.mark.tier2
273 async def test_AU_02_repair_snapshot_leaves_stale_cache_hit(
274 db_session: AsyncSession,
275 monkeypatch: pytest.MonkeyPatch,
276 ) -> None:
277 """AU_02 — RED-AS-BUG: wire_repair_snapshot does not bust the cache.
278
279 After the snapshot repair, the (repo_id, tip_T) cache row still exists and
280 wire_fetch_mpack(want=[T], have=[]) returns the pre-repair mpack_id.
281
282 EXPECTED POST-FIX BEHAVIOR (Phase 2, RP_02 + RP_05):
283 wire_fetch_mpack raises MPackNotReadyError.
284 """
285 manifest: dict[str, str] = {"file.txt": fake_id("obj-au02")}
286 directories: list[str] = []
287 snapshot_id = compute_snapshot_id(manifest, directories)
288 tip_T = fake_id("tip-au02")
289 mpack_M = fake_id("mpack-au02")
290
291 repo = await create_repo(db_session, owner="gabriel", visibility="public")
292 await _insert_fresh_cache_row(db_session, repo.repo_id, tip_T, mpack_M, seed="au02")
293
294 _stub_fetch_backend(monkeypatch)
295
296 # Repair the snapshot — cache is NOT busted (the bug)
297 result = await wire_repair_snapshot(
298 db_session, repo.repo_id, snapshot_id, manifest, directories,
299 caller_id="gabriel",
300 )
301 assert result["repaired"] is True
302
303 # BUG: cache row survived → fetch HITs the stale mpack_M
304 fetch_result = await wire_fetch_mpack(
305 db_session, repo.repo_id, want=[tip_T], have=[]
306 )
307 assert fetch_result["mpack_id"] == mpack_M, (
308 "BUG (AU_02): cache row survived wire_repair_snapshot — "
309 "fetch returns pre-repair mpack_M. "
310 "After Phase 2 fix (RP_02+RP_05), must raise MPackNotReadyError."
311 )
312
313 remaining = await _cache_rows_for_repo(db_session, repo.repo_id)
314 assert len(remaining) == 1, "cache row must still exist (bug: row was not busted)"
315
316
317 # ══════════════════════════════════════════════════════════════════════════════
318 # AU_03 — repair-commit leaves cache row intact → stale HIT (RED-AS-BUG)
319 # ══════════════════════════════════════════════════════════════════════════════
320
321
322 @pytest.mark.asyncio
323 @pytest.mark.tier2
324 async def test_AU_03_repair_commit_leaves_stale_cache_hit(
325 db_session: AsyncSession,
326 monkeypatch: pytest.MonkeyPatch,
327 ) -> None:
328 """AU_03 — RED-AS-BUG: wire_repair_commit does not bust the cache.
329
330 This is the exact failure from the gabriel/muse rc10 incident:
331 - repair-commit fixed the row's signer_public_key
332 - the cache kept serving the corrupted mpack for 7 more days
333 - the manual fix was: UPDATE musehub_fetch_mpack_cache SET expires_at = NOW() - INTERVAL '1 second'
334
335 EXPECTED POST-FIX BEHAVIOR (Phase 2, RP_03 + RP_06):
336 wire_fetch_mpack raises MPackNotReadyError.
337 """
338 committed_at = "2026-01-01T00:00:00+00:00"
339 snapshot_id = fake_id("snap-au03")
340 message = "au03 audit test commit"
341 commit_id, wire_dict = _build_valid_commit_dict(
342 snapshot_id=snapshot_id,
343 message=message,
344 committed_at=committed_at,
345 )
346
347 tip_T = commit_id # use the commit_id as the branch tip — matches real usage
348 mpack_M = fake_id("mpack-au03")
349
350 repo = await create_repo(db_session, owner="gabriel", visibility="public")
351 await _insert_commit_row(
352 db_session, repo.repo_id, commit_id,
353 snapshot_id=snapshot_id,
354 message=message,
355 committed_at=committed_at,
356 )
357 await _insert_fresh_cache_row(db_session, repo.repo_id, tip_T, mpack_M, seed="au03")
358
359 _stub_fetch_backend(monkeypatch)
360
361 # Repair the commit — cache is NOT busted (the bug)
362 result = await wire_repair_commit(
363 db_session, repo.repo_id, wire_dict, caller_id="gabriel"
364 )
365 assert result["repaired"] is True
366
367 # BUG: cache row survived → fetch HITs the stale mpack_M
368 fetch_result = await wire_fetch_mpack(
369 db_session, repo.repo_id, want=[tip_T], have=[]
370 )
371 assert fetch_result["mpack_id"] == mpack_M, (
372 "BUG (AU_03): cache row survived wire_repair_commit — "
373 "fetch returns pre-repair mpack_M. "
374 "This is the gabriel/muse rc10 incident. "
375 "After Phase 2 fix (RP_03+RP_06), must raise MPackNotReadyError."
376 )
377
378 remaining = await _cache_rows_for_repo(db_session, repo.repo_id)
379 assert len(remaining) == 1, "cache row must still exist (bug: row was not busted)"
380
381
382 # ══════════════════════════════════════════════════════════════════════════════
383 # AU_04 — force-push leaves old-tip cache row intact → stale HIT (RED-AS-BUG)
384 # ══════════════════════════════════════════════════════════════════════════════
385
386
387 @pytest.mark.asyncio
388 @pytest.mark.tier2
389 async def test_AU_04_force_push_leaves_old_tip_cache_row(
390 db_session: AsyncSession,
391 monkeypatch: pytest.MonkeyPatch,
392 ) -> None:
393 """AU_04 — RED-AS-BUG: wire_push_unpack_mpack(force=True) does not bust the old-tip cache row.
394
395 After force-push branch main from T_old → T_new, the (repo_id, T_old) cache
396 row still exists. A client that still holds T_old as its remote tip and
397 requests a fresh clone (want=[T_old], have=[]) gets a HIT on content that is
398 no longer on any branch.
399
400 EXPECTED POST-FIX BEHAVIOR (Phase 3, FP_01 + FP_02):
401 After force-push, the (repo_id, T_old) row is gone; the fetch raises
402 MPackNotReadyError.
403 """
404 from musehub.core.genesis import compute_branch_id
405 from musehub.services.musehub_wire_push import wire_push_unpack_mpack
406
407 T_old = fake_id("tip-old-au04")
408 T_new = fake_id("tip-new-au04")
409 mpack_old = fake_id("mpack-old-au04")
410 mpack_key = fake_id("mpack-key-au04")
411
412 repo = await create_repo(db_session, owner="gabriel", visibility="public")
413 # Create branch main pointing to T_old
414 db_session.add(MusehubBranch(
415 branch_id=compute_branch_id(repo.repo_id, "main"),
416 repo_id=repo.repo_id,
417 name="main",
418 head_commit_id=T_old,
419 ))
420 await db_session.commit()
421
422 # Seed a fresh cache row for the old tip
423 await _insert_fresh_cache_row(db_session, repo.repo_id, T_old, mpack_old, seed="au04-old")
424
425 # Mock all backends: the push backend needs to handle get_backend().get(mpack_key)
426 # returning a minimal valid mpack (empty), plus the fetch backend for the clone.
427 push_backend = AsyncMock()
428 push_backend.get = AsyncMock(return_value=_empty_mpack_bytes())
429 push_backend.delete = AsyncMock()
430 monkeypatch.setattr("musehub.services.musehub_wire_push.get_backend", lambda: push_backend)
431
432 fetch_backend = AsyncMock()
433 fetch_backend.presign_mpack_get = AsyncMock(return_value="https://r2.example/old-mpack")
434 monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: fetch_backend)
435
436 # Force-push main from T_old to T_new
437 # wire_push_unpack_mpack will parse the mpack — provide a minimal valid one.
438 # The push result is secondary; what matters is the cache row audit.
439 try:
440 await wire_push_unpack_mpack(
441 db_session,
442 repo.repo_id,
443 mpack_key=mpack_key,
444 pusher_id="gabriel",
445 branch="main",
446 head_commit_id=T_new,
447 force=True,
448 )
449 except Exception:
450 # wire_push_unpack_mpack may raise if the mpack is empty (no commits) —
451 # that's fine for this audit test. We only care whether the cache row
452 # was busted, regardless of whether the push succeeded.
453 pass
454
455 # BUG: the old-tip cache row still exists after force-push
456 remaining = await _cache_rows_for_repo(db_session, repo.repo_id)
457 old_tip_rows = [r for r in remaining if r.tip_commit_id == T_old]
458 assert len(old_tip_rows) == 1, (
459 "BUG (AU_04): (repo_id, T_old) cache row survived force-push. "
460 "A clone of T_old will HIT this stale row. "
461 "After Phase 3 fix (FP_01+FP_02), this row must be gone."
462 )
463
464 # Confirm the stale HIT: a clone of T_old still gets the pre-force-push mpack
465 fetch_result = await wire_fetch_mpack(
466 db_session, repo.repo_id, want=[T_old], have=[]
467 )
468 assert fetch_result["mpack_id"] == mpack_old, (
469 "BUG (AU_04): wire_fetch_mpack HITs stale mpack_old for the superseded T_old tip."
470 )
471
472
473 def _empty_mpack_bytes() -> bytes:
474 """Minimal valid msgpack bytes that wire_push_unpack_mpack will accept as an mpack.
475
476 The push handler calls msgpack.unpackb on the raw bytes from backend.get().
477 An empty dict `{}` is the smallest payload that won't raise a parse error,
478 though the push will fail on missing fields (commits/objects) — that's fine
479 for this audit, which only needs the cache row check, not a successful push.
480 """
481 import msgpack
482 return msgpack.packb({}, use_bin_type=True)
File History 1 commit
sha256:0e5174da777684df43b2db71f282079030ef28bacffb93e19c29ee712b2a8bc4 test(mwp8/phase0): audit — repair+force-push leave stale ca… Sonnet 4.6 21 days ago