gabriel / musehub public
test_fetch_mpack_prebuild.py python
592 lines 26.0 KB Cold
Raw
sha256:f58d788df3ccdda8f8987b428418db655a38582309239b99d7b9715ea6dff618 feat(#92): phase 5 — GC expired fetch mpack cache entries (… Sonnet 4.6 patch 103 days ago
1 """TDD — fetch.mpack.prebuild job handler and wire_fetch_mpack cache (issue #92 Phases 2–5).
2
3 Test IDs:
4 FMC_07 Unit: mock wire_fetch_mpack, confirm cache rows written for each tip,
5 confirm existing fresh entries are skipped
6 FMC_08 Integration: insert a job row, run the handler, verify
7 MusehubFetchMPackCache row exists with correct repo_id/tip/mpack_id
8 FMC_13 Unit: cache hit returns presigned URL without entering blob-load path
9 FMC_14 Unit: cache miss builds and writes cache row
10 FMC_18 Integration: enqueue_push_intel inserts fetch.mpack.prebuild with branch tips
11 FMC_20 Unit: gc_fetch_mpack_cache deletes expired rows + R2 objects; fresh rows untouched
12 """
13 from __future__ import annotations
14
15 import hashlib
16 from datetime import datetime, timedelta, timezone
17 from unittest.mock import AsyncMock, patch
18
19 import pytest
20 from sqlalchemy import select
21 from sqlalchemy.ext.asyncio import AsyncSession
22
23 from musehub.core.genesis import compute_job_id
24 from musehub.db.musehub_jobs_models import MusehubBackgroundJob
25 from musehub.db.musehub_repo_models import MusehubFetchMPackCache
26 from musehub.services.musehub_gc import gc_fetch_mpack_cache
27 from musehub.services.musehub_jobs import enqueue_push_intel
28 from musehub.services.musehub_wire_fetch import process_fetch_mpack_prebuild_job, wire_fetch_mpack
29 from musehub.services.musehub_wire_shared import MPackNotReadyError
30 from tests.factories import create_branch, create_repo
31
32
33 def _now() -> datetime:
34 return datetime.now(tz=timezone.utc)
35
36
37 def _fake_commit_id(seed: str) -> str:
38 return "sha256:" + hashlib.sha256(seed.encode()).hexdigest()
39
40
41 def _fake_mpack_id(seed: str) -> str:
42 return "sha256:" + hashlib.sha256(f"mpack-{seed}".encode()).hexdigest()
43
44
45 async def _insert_job(
46 session: AsyncSession,
47 repo_id: str,
48 tip_commit_ids: list[str],
49 ) -> str:
50 now = _now()
51 job_id = compute_job_id(repo_id, "fetch.mpack.prebuild", now.isoformat())
52 session.add(MusehubBackgroundJob(
53 job_id=job_id,
54 repo_id=repo_id,
55 job_type="fetch.mpack.prebuild",
56 payload={"tip_commit_ids": tip_commit_ids},
57 status="pending",
58 created_at=now,
59 attempt=0,
60 ))
61 await session.flush()
62 return job_id
63
64
65 # ── FMC_07 ────────────────────────────────────────────────────────────────────
66
67 @pytest.mark.tier2
68 async def test_fmc_07_builds_uncached_tips_in_one_combined_mpack(db_session: AsyncSession) -> None:
69 """FMC_07a: the handler builds all uncached tips in a SINGLE combined wire_fetch_mpack
70 call (want=[all uncached tips]) — not one call per tip. The per-tip cache rows are
71 written inside wire_fetch_mpack (covered by FMC_14), so it is mocked here and we assert
72 the handler's orchestration + counts only."""
73 repo = await create_repo(db_session, owner="gabriel", visibility="public")
74 tip_a = _fake_commit_id("tip-a")
75 tip_b = _fake_commit_id("tip-b")
76 combined_mpack = _fake_mpack_id("combined")
77
78 # The handler reads live branch tips from MusehubBranch (Phase 1 fix) —
79 # create branches so the handler sees tip_a and tip_b as the live tip set.
80 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip_a)
81 await create_branch(db_session, repo.repo_id, name="dev", head_commit_id=tip_b)
82 job_id = await _insert_job(db_session, repo.repo_id, [tip_a, tip_b])
83 await db_session.commit()
84
85 with patch(
86 "musehub.services.musehub_wire_fetch.wire_fetch_mpack",
87 new_callable=AsyncMock,
88 return_value={"mpack_id": combined_mpack, "mpack_url": "https://r2.example/c", "commit_count": 2, "blob_count": 5},
89 ) as mock_build:
90 result = await process_fetch_mpack_prebuild_job(db_session, job_id)
91 await db_session.commit()
92
93 # One combined build covering both uncached tips — not one call per tip.
94 assert mock_build.call_count == 1
95 assert set(mock_build.call_args.kwargs["want"]) == {tip_a, tip_b}
96 assert result["tips_requested"] == 2
97 assert result["tips_built"] == 2
98 assert result["tips_skipped"] == 0
99
100
101 @pytest.mark.tier2
102 async def test_fmc_07b_skips_tips_with_fresh_cache(db_session: AsyncSession) -> None:
103 """FMC_07b: prebuild skips ALL tips only when every tip shares the same cached mpack_id.
104
105 When one tip is already cached with a DIFFERENT mpack_id than would be built for a
106 new tip, the prebuild must rebuild ALL tips together so that every tip ends up
107 pointing to the same combined mpack_id — required for the clone cache-hit check
108 (len(mpack_ids)==1 across all want tips).
109 """
110 repo = await create_repo(db_session, owner="gabriel", visibility="public")
111 tip_cached = _fake_commit_id("tip-cached")
112 tip_new = _fake_commit_id("tip-new")
113 existing_mpack = _fake_mpack_id("existing")
114 new_mpack = _fake_mpack_id("new")
115
116 # Pre-populate a fresh cache entry for tip_cached (different mpack_id from what
117 # the combined build will produce — diverged state).
118 cache_id = hashlib.sha256((repo.repo_id + tip_cached).encode()).hexdigest()
119 db_session.add(MusehubFetchMPackCache(
120 cache_id=cache_id,
121 repo_id=repo.repo_id,
122 tip_commit_id=tip_cached,
123 mpack_id=existing_mpack,
124 created_at=_now(),
125 expires_at=_now() + timedelta(days=7),
126 ))
127 # The handler reads live branch tips from MusehubBranch (Phase 1 fix).
128 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip_cached)
129 await create_branch(db_session, repo.repo_id, name="dev", head_commit_id=tip_new)
130 job_id = await _insert_job(db_session, repo.repo_id, [tip_cached, tip_new])
131 await db_session.commit()
132
133 with patch(
134 "musehub.services.musehub_wire_fetch.wire_fetch_mpack",
135 new_callable=AsyncMock,
136 return_value={"mpack_id": new_mpack, "mpack_url": "https://r2.example/new", "commit_count": 2, "blob_count": 2},
137 ) as mock_build:
138 result = await process_fetch_mpack_prebuild_job(db_session, job_id)
139 await db_session.commit()
140
141 # Both tips must be rebuilt together so they share one mpack_id.
142 assert mock_build.call_count == 1
143 actual_want = mock_build.call_args[1].get("want") or mock_build.call_args[0][2]
144 assert set(actual_want) == {tip_cached, tip_new}, (
145 f"want must include ALL tips; got {actual_want}"
146 )
147 assert result["tips_built"] == 2
148 assert result["tips_skipped"] == 0
149
150 # The previously-cached entry must be updated to the new combined mpack_id.
151 cached_row = (await db_session.execute(
152 select(MusehubFetchMPackCache)
153 .where(MusehubFetchMPackCache.repo_id == repo.repo_id)
154 .where(MusehubFetchMPackCache.tip_commit_id == tip_cached)
155 )).scalar_one()
156 assert cached_row.mpack_id == new_mpack, (
157 "existing cache entry must be updated to the combined mpack_id"
158 )
159
160
161 @pytest.mark.tier2
162 async def test_fmc_07c_empty_payload_is_a_noop(db_session: AsyncSession) -> None:
163 """FMC_07c: job with no tip_commit_ids returns zeros without calling wire_fetch_mpack."""
164 repo = await create_repo(db_session, owner="gabriel", visibility="public")
165 job_id = await _insert_job(db_session, repo.repo_id, [])
166 await db_session.commit()
167
168 with patch(
169 "musehub.services.musehub_wire_fetch.wire_fetch_mpack",
170 new_callable=AsyncMock,
171 ) as mock_build:
172 result = await process_fetch_mpack_prebuild_job(db_session, job_id)
173
174 assert mock_build.call_count == 0
175 assert result["tips_requested"] == 0
176 assert result["tips_built"] == 0
177
178
179 # ── FMC_08 ────────────────────────────────────────────────────────────────────
180
181 @pytest.mark.tier2
182 async def test_fmc_08_cache_row_has_correct_fields(db_session: AsyncSession) -> None:
183 """FMC_08: running the handler end to end (through a REAL wire_fetch_mpack) writes a
184 cache row with matching repo_id, tip, and mpack_id. Cache rows are written inside
185 wire_fetch_mpack, so this exercises the full handler -> build -> cache-write chain
186 rather than mocking the build away. Only the expensive externals (DAG walk, mpack
187 bytes, storage) are stubbed — same pattern as FMC_14."""
188 from types import SimpleNamespace
189
190 repo = await create_repo(db_session, owner="gabriel", visibility="public")
191 tip = _fake_commit_id("integration-tip")
192 built_mpack = _fake_mpack_id("integration")
193
194 # The handler reads live branch tips from MusehubBranch (Phase 1 fix).
195 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip)
196 job_id = await _insert_job(db_session, repo.repo_id, [tip])
197 await db_session.commit()
198
199 mock_backend = AsyncMock()
200 mock_backend.put_mpack.return_value = None
201 mock_backend.presign_mpack_get.return_value = "https://r2.example/int"
202 mock_backend.delete.return_value = None
203 fake_proxy = SimpleNamespace(commit_id=tip, snapshot_id=None, parent_ids=[])
204
205 with patch("musehub.services.musehub_wire_fetch.get_backend", return_value=mock_backend), \
206 patch("musehub.services.musehub_wire_fetch._walk_commit_delta", new_callable=AsyncMock,
207 return_value={tip: fake_proxy}), \
208 patch("muse.core.mpack.build_wire_mpack", return_value=b"MUSE\x00fake-mpack"), \
209 patch("musehub.services.musehub_wire_fetch.blob_id", return_value=built_mpack):
210 result = await process_fetch_mpack_prebuild_job(db_session, job_id)
211 await db_session.commit()
212
213 assert result["tips_built"] == 1
214
215 row = (await db_session.execute(
216 select(MusehubFetchMPackCache)
217 .where(MusehubFetchMPackCache.repo_id == repo.repo_id)
218 .where(MusehubFetchMPackCache.tip_commit_id == tip)
219 )).scalar_one()
220
221 assert row.repo_id == repo.repo_id
222 assert row.tip_commit_id == tip
223 assert row.mpack_id == built_mpack
224 assert row.expires_at > _now()
225
226
227 # ── FMC_13 ────────────────────────────────────────────────────────────────────
228
229 @pytest.mark.tier2
230 async def test_fmc_13_cache_hit_returns_presigned_url_without_blob_load(db_session: AsyncSession) -> None:
231 """FMC_13: cache hit returns the presigned URL immediately; blob-load path is never entered."""
232 repo = await create_repo(db_session, owner="gabriel", visibility="public")
233 tip = _fake_commit_id("fmc13-tip")
234 cached_mpack = _fake_mpack_id("fmc13-tip")
235 expected_url = "https://r2.example/cached-fmc13"
236
237 # Pre-populate a fresh cache entry.
238 cache_id = hashlib.sha256((repo.repo_id + tip).encode()).hexdigest()
239 db_session.add(MusehubFetchMPackCache(
240 cache_id=cache_id,
241 repo_id=repo.repo_id,
242 tip_commit_id=tip,
243 mpack_id=cached_mpack,
244 created_at=_now(),
245 expires_at=_now() + timedelta(days=7),
246 ))
247 await db_session.commit()
248
249 mock_backend = AsyncMock()
250 mock_backend.presign_mpack_get.return_value = expected_url
251
252 with patch("musehub.services.musehub_wire_fetch.get_backend", return_value=mock_backend), \
253 patch("musehub.services.musehub_wire_fetch._walk_commit_delta", new_callable=AsyncMock) as mock_walk:
254 result = await wire_fetch_mpack(db_session, repo.repo_id, want=[tip], have=[])
255
256 # The expensive DAG walk must never have been called.
257 mock_walk.assert_not_called()
258 # The returned URL must match the presigned URL for the cached mpack.
259 assert result["mpack_url"] == expected_url
260 assert result["mpack_id"] == cached_mpack
261
262
263 @pytest.mark.tier2
264 async def test_fmc_13b_duplicate_branch_heads_share_one_cache_tip(db_session: AsyncSession) -> None:
265 """FMC_13b: main/dev at one commit is one cache requirement, not two.
266
267 ``muse clone`` sends one want entry per branch, so two branches at the same
268 commit produce duplicate commit IDs. The cache is keyed by unique commit
269 ID and must still hit instead of rebuilding forever.
270 """
271 repo = await create_repo(db_session, owner="gabriel", visibility="public")
272 shared_tip = _fake_commit_id("fmc13b-main-dev-tip")
273 cached_mpack = _fake_mpack_id("fmc13b-main-dev-tip")
274 expected_url = "https://r2.example/cached-fmc13b"
275
276 cache_id = hashlib.sha256((repo.repo_id + shared_tip).encode()).hexdigest()
277 db_session.add(MusehubFetchMPackCache(
278 cache_id=cache_id,
279 repo_id=repo.repo_id,
280 tip_commit_id=shared_tip,
281 mpack_id=cached_mpack,
282 created_at=_now(),
283 expires_at=_now() + timedelta(days=7),
284 ))
285 await db_session.commit()
286
287 mock_backend = AsyncMock()
288 mock_backend.presign_mpack_get.return_value = expected_url
289
290 with patch("musehub.services.musehub_wire_fetch.get_backend", return_value=mock_backend), \
291 patch("musehub.services.musehub_wire_fetch._walk_commit_delta", new_callable=AsyncMock) as mock_walk:
292 result = await wire_fetch_mpack(
293 db_session,
294 repo.repo_id,
295 want=[shared_tip, shared_tip],
296 have=[],
297 )
298
299 mock_walk.assert_not_called()
300 assert result["mpack_url"] == expected_url
301 assert result["mpack_id"] == cached_mpack
302
303
304 @pytest.mark.tier2
305 async def test_fmc_13c_prebuild_deduplicates_live_branch_heads(db_session: AsyncSession) -> None:
306 """FMC_13c: a prebuild skips one cached commit shared by two branches."""
307 repo = await create_repo(db_session, owner="gabriel", visibility="public")
308 shared_tip = _fake_commit_id("fmc13c-main-dev-tip")
309 cached_mpack = _fake_mpack_id("fmc13c-main-dev-tip")
310
311 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=shared_tip)
312 await create_branch(db_session, repo.repo_id, name="dev", head_commit_id=shared_tip)
313 job_id = await _insert_job(db_session, repo.repo_id, [shared_tip, shared_tip])
314 cache_id = hashlib.sha256((repo.repo_id + shared_tip).encode()).hexdigest()
315 db_session.add(MusehubFetchMPackCache(
316 cache_id=cache_id,
317 repo_id=repo.repo_id,
318 tip_commit_id=shared_tip,
319 mpack_id=cached_mpack,
320 created_at=_now(),
321 expires_at=_now() + timedelta(days=7),
322 ))
323 await db_session.commit()
324
325 with patch(
326 "musehub.services.musehub_wire_fetch.wire_fetch_mpack",
327 new_callable=AsyncMock,
328 ) as mock_build:
329 result = await process_fetch_mpack_prebuild_job(db_session, job_id)
330
331 mock_build.assert_not_called()
332 assert result["tips_requested"] == 1
333 assert result["tips_built"] == 0
334 assert result["tips_skipped"] == 1
335
336
337 # ── FMC_14 ────────────────────────────────────────────────────────────────────
338
339 @pytest.mark.tier2
340 async def test_fmc_14_cache_miss_builds_and_writes_cache_row(db_session: AsyncSession) -> None:
341 """FMC_14: on a cache miss, wire_fetch_mpack builds the mpack and writes a cache row."""
342 from types import SimpleNamespace
343
344 repo = await create_repo(db_session, owner="gabriel", visibility="public")
345 tip = _fake_commit_id("fmc14-tip")
346 built_mpack = _fake_mpack_id("fmc14-tip")
347 built_url = "https://r2.example/built-fmc14"
348
349 # No cache row exists — this is a cold miss.
350 mock_backend = AsyncMock()
351 mock_backend.put_mpack.return_value = None
352 mock_backend.presign_mpack_get.return_value = built_url
353 mock_backend.delete.return_value = None
354
355 # _walk_commit_delta returns one proxy commit with no snapshot so that
356 # wire_fetch_mpack proceeds to build an empty-but-valid mpack without
357 # needing real commit/snapshot/object rows in the DB.
358 fake_proxy = SimpleNamespace(commit_id=tip, snapshot_id=None, parent_ids=[])
359
360 with patch("musehub.services.musehub_wire_fetch.get_backend", return_value=mock_backend), \
361 patch("musehub.services.musehub_wire_fetch._walk_commit_delta", new_callable=AsyncMock,
362 return_value={tip: fake_proxy}), \
363 patch("muse.core.mpack.build_wire_mpack", return_value=b"MUSE\x00fake-mpack"), \
364 patch("musehub.services.musehub_wire_fetch.blob_id", return_value=built_mpack):
365 result = await wire_fetch_mpack(db_session, repo.repo_id, want=[tip], have=[], force_build=True)
366 await db_session.commit()
367
368 assert result["mpack_url"] == built_url
369 assert result["mpack_id"] == built_mpack
370
371 # The cache row must have been written.
372 row = (await db_session.execute(
373 select(MusehubFetchMPackCache)
374 .where(MusehubFetchMPackCache.repo_id == repo.repo_id)
375 .where(MusehubFetchMPackCache.tip_commit_id == tip)
376 )).scalar_one()
377 assert row.mpack_id == built_mpack
378 assert row.expires_at > _now()
379
380
381 # ── FMC_18 ────────────────────────────────────────────────────────────────────
382
383 @pytest.mark.tier2
384 async def test_fmc_18_enqueue_push_intel_creates_prebuild_job_with_branch_tips(
385 db_session: AsyncSession,
386 ) -> None:
387 """FMC_18: enqueue_push_intel enqueues fetch.mpack.prebuild with all branch tip commit IDs."""
388 repo = await create_repo(db_session, owner="gabriel", visibility="public")
389
390 tip_a = _fake_commit_id("branch-main")
391 tip_b = _fake_commit_id("branch-dev")
392 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip_a)
393 await create_branch(db_session, repo.repo_id, name="dev", head_commit_id=tip_b)
394
395 await enqueue_push_intel(
396 db_session,
397 repo.repo_id,
398 head=tip_a,
399 domain_id=None,
400 branch="main",
401 )
402 await db_session.commit()
403
404 job_row = (await db_session.execute(
405 select(MusehubBackgroundJob)
406 .where(MusehubBackgroundJob.repo_id == repo.repo_id)
407 .where(MusehubBackgroundJob.job_type == "fetch.mpack.prebuild")
408 .where(MusehubBackgroundJob.status == "pending")
409 )).scalar_one()
410
411 tip_ids = set(job_row.payload.get("tip_commit_ids", []))
412 assert tip_a in tip_ids, f"main tip {tip_a[:20]} missing from payload"
413 assert tip_b in tip_ids, f"dev tip {tip_b[:20]} missing from payload"
414
415
416 # ── FMC_20 ────────────────────────────────────────────────────────────────────
417
418 @pytest.mark.tier2
419 async def test_fmc_20_gc_deletes_expired_rows_and_r2_objects_leaves_fresh_untouched(
420 db_session: AsyncSession,
421 ) -> None:
422 """FMC_20: gc_fetch_mpack_cache deletes expired rows + R2 objects; fresh rows survive."""
423 repo = await create_repo(db_session, owner="gabriel", visibility="public")
424
425 tip_expired_a = _fake_commit_id("expired-a")
426 tip_expired_b = _fake_commit_id("expired-b")
427 tip_fresh = _fake_commit_id("fresh")
428
429 mpack_expired_a = _fake_mpack_id("expired-a")
430 mpack_expired_b = _fake_mpack_id("expired-b")
431 mpack_fresh = _fake_mpack_id("fresh")
432
433 past = _now() - timedelta(days=1)
434 future = _now() + timedelta(days=6)
435
436 for tip, mpack, exp in [
437 (tip_expired_a, mpack_expired_a, past),
438 (tip_expired_b, mpack_expired_b, past),
439 (tip_fresh, mpack_fresh, future),
440 ]:
441 cache_id = hashlib.sha256((repo.repo_id + tip).encode()).hexdigest()
442 db_session.add(MusehubFetchMPackCache(
443 cache_id=cache_id,
444 repo_id=repo.repo_id,
445 tip_commit_id=tip,
446 mpack_id=mpack,
447 created_at=_now(),
448 expires_at=exp,
449 ))
450 await db_session.commit()
451
452 mock_backend = AsyncMock()
453 mock_backend.delete.return_value = None
454
455 with patch("musehub.services.musehub_gc.get_backend", return_value=mock_backend):
456 n_deleted = await gc_fetch_mpack_cache(db_session, repo.repo_id)
457 await db_session.commit()
458
459 assert n_deleted == 2
460
461 # R2 delete called exactly once for each expired mpack — in any order.
462 deleted_mpack_ids = {call.args[0] for call in mock_backend.delete.call_args_list}
463 assert mpack_expired_a in deleted_mpack_ids
464 assert mpack_expired_b in deleted_mpack_ids
465 assert mpack_fresh not in deleted_mpack_ids
466
467 # Expired rows gone from DB.
468 remaining = (await db_session.execute(
469 select(MusehubFetchMPackCache)
470 .where(MusehubFetchMPackCache.repo_id == repo.repo_id)
471 )).scalars().all()
472 remaining_tips = {r.tip_commit_id for r in remaining}
473 assert tip_expired_a not in remaining_tips
474 assert tip_expired_b not in remaining_tips
475
476 # Fresh row survives.
477 assert tip_fresh in remaining_tips
478
479
480 # ── FMC_21 ────────────────────────────────────────────────────────────────────
481 # Regression for musehub issue #39 (muse clone fails / hangs): a repo that goes
482 # stale (cache TTL expires with no further pushes) previously had no self-heal
483 # path — wire_fetch_mpack would raise MPackNotReadyError() forever on every
484 # retry, since nothing ever enqueued a new fetch.mpack.prebuild job for the
485 # current tips. Reproduced live against gabriel/muse on 2026-09-09: cache rows
486 # for both branch heads existed but had expired 5 days earlier, and the client
487 # retried the full 120s budget every single attempt with zero chance of success.
488
489 @pytest.mark.tier2
490 async def test_fmc_21_cache_miss_enqueues_prebuild_job(db_session: AsyncSession) -> None:
491 """FMC_21: on a real (non-force_build) cache miss, a fetch.mpack.prebuild job is enqueued."""
492 repo = await create_repo(db_session, owner="gabriel", visibility="public")
493 tip = _fake_commit_id("fmc21-tip")
494 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip)
495
496 with pytest.raises(MPackNotReadyError):
497 await wire_fetch_mpack(db_session, repo.repo_id, want=[tip], have=[])
498 await db_session.commit()
499
500 job_row = (await db_session.execute(
501 select(MusehubBackgroundJob)
502 .where(MusehubBackgroundJob.repo_id == repo.repo_id)
503 .where(MusehubBackgroundJob.job_type == "fetch.mpack.prebuild")
504 .where(MusehubBackgroundJob.status == "pending")
505 )).scalar_one()
506 assert tip in set(job_row.payload.get("tip_commit_ids", []))
507
508
509 @pytest.mark.tier2
510 async def test_fmc_21b_repeated_cache_miss_does_not_duplicate_pending_job(db_session: AsyncSession) -> None:
511 """FMC_21b: a second MISS (e.g. the client's own retry loop) does not enqueue a duplicate job."""
512 repo = await create_repo(db_session, owner="gabriel", visibility="public")
513 tip = _fake_commit_id("fmc21b-tip")
514 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip)
515
516 for _ in range(3):
517 with pytest.raises(MPackNotReadyError):
518 await wire_fetch_mpack(db_session, repo.repo_id, want=[tip], have=[])
519 await db_session.commit()
520
521 rows = (await db_session.execute(
522 select(MusehubBackgroundJob)
523 .where(MusehubBackgroundJob.repo_id == repo.repo_id)
524 .where(MusehubBackgroundJob.job_type == "fetch.mpack.prebuild")
525 .where(MusehubBackgroundJob.status == "pending")
526 )).scalars().all()
527 assert len(rows) == 1
528
529
530 @pytest.mark.tier2
531 async def test_fmc_21c_expired_cache_self_heals_end_to_end(db_session: AsyncSession) -> None:
532 """FMC_21c: expired cache -> MISS enqueues job -> running the job -> next fetch is a HIT.
533
534 This is the exact sequence that was broken for gabriel/muse: a dormant
535 repo's cache entry ages past its TTL and, before this fix, could never
536 recover without a fresh push.
537 """
538 from types import SimpleNamespace
539
540 repo = await create_repo(db_session, owner="gabriel", visibility="public")
541 tip = _fake_commit_id("fmc21c-tip")
542 stale_mpack = _fake_mpack_id("fmc21c-stale")
543 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip)
544
545 # Seed an *expired* cache row -- exactly what we found live on staging.
546 cache_id = hashlib.sha256((repo.repo_id + tip).encode()).hexdigest()
547 db_session.add(MusehubFetchMPackCache(
548 cache_id=cache_id,
549 repo_id=repo.repo_id,
550 tip_commit_id=tip,
551 mpack_id=stale_mpack,
552 created_at=_now() - timedelta(days=8),
553 expires_at=_now() - timedelta(days=1),
554 ))
555 await db_session.commit()
556
557 # Step 1 -- MISS (expired row doesn't count as a hit) must self-heal.
558 with pytest.raises(MPackNotReadyError):
559 await wire_fetch_mpack(db_session, repo.repo_id, want=[tip], have=[])
560 await db_session.commit()
561
562 job_row = (await db_session.execute(
563 select(MusehubBackgroundJob)
564 .where(MusehubBackgroundJob.repo_id == repo.repo_id)
565 .where(MusehubBackgroundJob.job_type == "fetch.mpack.prebuild")
566 .where(MusehubBackgroundJob.status == "pending")
567 )).scalar_one()
568
569 # Step 2 -- run the job the worker would have picked up.
570 fresh_mpack = _fake_mpack_id("fmc21c-fresh")
571 fake_proxy = SimpleNamespace(commit_id=tip, snapshot_id=None, parent_ids=[])
572 mock_backend = AsyncMock()
573 mock_backend.put_mpack.return_value = None
574 mock_backend.presign_mpack_get.return_value = "https://r2.example/fresh-fmc21c"
575 mock_backend.delete.return_value = None
576
577 with patch("musehub.services.musehub_wire_fetch.get_backend", return_value=mock_backend), \
578 patch("musehub.services.musehub_wire_fetch._walk_commit_delta", new_callable=AsyncMock,
579 return_value={tip: fake_proxy}), \
580 patch("muse.core.mpack.build_wire_mpack", return_value=b"MUSE\x00fake-mpack"), \
581 patch("musehub.services.musehub_wire_fetch.blob_id", return_value=fresh_mpack):
582 await process_fetch_mpack_prebuild_job(db_session, job_row.job_id)
583 await db_session.commit()
584
585 # Step 3 -- the next fetch is now a real cache HIT on the freshly built mpack.
586 with patch("musehub.services.musehub_wire_fetch.get_backend", return_value=mock_backend), \
587 patch("musehub.services.musehub_wire_fetch._walk_commit_delta", new_callable=AsyncMock) as mock_walk:
588 result = await wire_fetch_mpack(db_session, repo.repo_id, want=[tip], have=[])
589
590 mock_walk.assert_not_called()
591 assert result["mpack_id"] == fresh_mpack
592 assert result["mpack_id"] != stale_mpack
File History 4 commits
sha256:f58d788df3ccdda8f8987b428418db655a38582309239b99d7b9715ea6dff618 feat(#92): phase 5 — GC expired fetch mpack cache entries (… Sonnet 4.6 patch 103 days ago
sha256:d50f9cf9829dfbe35721a23b81ad256c729ddf9dd565a0a9e56d27847e255632 feat(#92): phase 4 — enqueue fetch.mpack.prebuild on push (… Sonnet 4.6 patch 103 days ago
sha256:1c5b7a0aba79472f4b10e52326dc010bdab1a498c9e195593d0707860478a034 feat(#92): phase 3 — cache lookup in wire_fetch_mpack (FMC_… Sonnet 4.6 patch 103 days ago
sha256:0e447fc3f6b7887d5d9e86b557c659ef7d0b05e2e09ddb0cb551ada240e48a51 feat(phase2): fetch.mpack.prebuild job handler + worker dis… Sonnet 4.6 patch 103 days ago