gabriel / musehub public
test_fetch_mpack_prebuild.py python
402 lines 17.5 KB
Raw
sha256:f58d788df3ccdda8f8987b428418db655a38582309239b99d7b9715ea6dff618 feat(#92): phase 5 — GC expired fetch mpack cache entries (… Sonnet 4.6 patch 34 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 tests.factories import create_branch, create_repo
30
31
32 def _now() -> datetime:
33 return datetime.now(tz=timezone.utc)
34
35
36 def _fake_commit_id(seed: str) -> str:
37 return "sha256:" + hashlib.sha256(seed.encode()).hexdigest()
38
39
40 def _fake_mpack_id(seed: str) -> str:
41 return "sha256:" + hashlib.sha256(f"mpack-{seed}".encode()).hexdigest()
42
43
44 async def _insert_job(
45 session: AsyncSession,
46 repo_id: str,
47 tip_commit_ids: list[str],
48 ) -> str:
49 now = _now()
50 job_id = compute_job_id(repo_id, "fetch.mpack.prebuild", now.isoformat())
51 session.add(MusehubBackgroundJob(
52 job_id=job_id,
53 repo_id=repo_id,
54 job_type="fetch.mpack.prebuild",
55 payload={"tip_commit_ids": tip_commit_ids},
56 status="pending",
57 created_at=now,
58 attempt=0,
59 ))
60 await session.flush()
61 return job_id
62
63
64 # ── FMC_07 ────────────────────────────────────────────────────────────────────
65
66 @pytest.mark.tier2
67 async def test_fmc_07_builds_uncached_tips_in_one_combined_mpack(db_session: AsyncSession) -> None:
68 """FMC_07a: the handler builds all uncached tips in a SINGLE combined wire_fetch_mpack
69 call (want=[all uncached tips]) — not one call per tip. The per-tip cache rows are
70 written inside wire_fetch_mpack (covered by FMC_14), so it is mocked here and we assert
71 the handler's orchestration + counts only."""
72 repo = await create_repo(db_session, owner="gabriel", visibility="public")
73 tip_a = _fake_commit_id("tip-a")
74 tip_b = _fake_commit_id("tip-b")
75 combined_mpack = _fake_mpack_id("combined")
76
77 # The handler reads live branch tips from MusehubBranch (Phase 1 fix) —
78 # create branches so the handler sees tip_a and tip_b as the live tip set.
79 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip_a)
80 await create_branch(db_session, repo.repo_id, name="dev", head_commit_id=tip_b)
81 job_id = await _insert_job(db_session, repo.repo_id, [tip_a, tip_b])
82 await db_session.commit()
83
84 with patch(
85 "musehub.services.musehub_wire_fetch.wire_fetch_mpack",
86 new_callable=AsyncMock,
87 return_value={"mpack_id": combined_mpack, "mpack_url": "https://r2.example/c", "commit_count": 2, "blob_count": 5},
88 ) as mock_build:
89 result = await process_fetch_mpack_prebuild_job(db_session, job_id)
90 await db_session.commit()
91
92 # One combined build covering both uncached tips — not one call per tip.
93 assert mock_build.call_count == 1
94 assert set(mock_build.call_args.kwargs["want"]) == {tip_a, tip_b}
95 assert result["tips_requested"] == 2
96 assert result["tips_built"] == 2
97 assert result["tips_skipped"] == 0
98
99
100 @pytest.mark.tier2
101 async def test_fmc_07b_skips_tips_with_fresh_cache(db_session: AsyncSession) -> None:
102 """FMC_07b: prebuild skips ALL tips only when every tip shares the same cached mpack_id.
103
104 When one tip is already cached with a DIFFERENT mpack_id than would be built for a
105 new tip, the prebuild must rebuild ALL tips together so that every tip ends up
106 pointing to the same combined mpack_id — required for the clone cache-hit check
107 (len(mpack_ids)==1 across all want tips).
108 """
109 repo = await create_repo(db_session, owner="gabriel", visibility="public")
110 tip_cached = _fake_commit_id("tip-cached")
111 tip_new = _fake_commit_id("tip-new")
112 existing_mpack = _fake_mpack_id("existing")
113 new_mpack = _fake_mpack_id("new")
114
115 # Pre-populate a fresh cache entry for tip_cached (different mpack_id from what
116 # the combined build will produce — diverged state).
117 cache_id = hashlib.sha256((repo.repo_id + tip_cached).encode()).hexdigest()
118 db_session.add(MusehubFetchMPackCache(
119 cache_id=cache_id,
120 repo_id=repo.repo_id,
121 tip_commit_id=tip_cached,
122 mpack_id=existing_mpack,
123 created_at=_now(),
124 expires_at=_now() + timedelta(days=7),
125 ))
126 # The handler reads live branch tips from MusehubBranch (Phase 1 fix).
127 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip_cached)
128 await create_branch(db_session, repo.repo_id, name="dev", head_commit_id=tip_new)
129 job_id = await _insert_job(db_session, repo.repo_id, [tip_cached, tip_new])
130 await db_session.commit()
131
132 with patch(
133 "musehub.services.musehub_wire_fetch.wire_fetch_mpack",
134 new_callable=AsyncMock,
135 return_value={"mpack_id": new_mpack, "mpack_url": "https://r2.example/new", "commit_count": 2, "blob_count": 2},
136 ) as mock_build:
137 result = await process_fetch_mpack_prebuild_job(db_session, job_id)
138 await db_session.commit()
139
140 # Both tips must be rebuilt together so they share one mpack_id.
141 assert mock_build.call_count == 1
142 actual_want = mock_build.call_args[1].get("want") or mock_build.call_args[0][2]
143 assert set(actual_want) == {tip_cached, tip_new}, (
144 f"want must include ALL tips; got {actual_want}"
145 )
146 assert result["tips_built"] == 2
147 assert result["tips_skipped"] == 0
148
149 # The previously-cached entry must be updated to the new combined mpack_id.
150 cached_row = (await db_session.execute(
151 select(MusehubFetchMPackCache)
152 .where(MusehubFetchMPackCache.repo_id == repo.repo_id)
153 .where(MusehubFetchMPackCache.tip_commit_id == tip_cached)
154 )).scalar_one()
155 assert cached_row.mpack_id == new_mpack, (
156 "existing cache entry must be updated to the combined mpack_id"
157 )
158
159
160 @pytest.mark.tier2
161 async def test_fmc_07c_empty_payload_is_a_noop(db_session: AsyncSession) -> None:
162 """FMC_07c: job with no tip_commit_ids returns zeros without calling wire_fetch_mpack."""
163 repo = await create_repo(db_session, owner="gabriel", visibility="public")
164 job_id = await _insert_job(db_session, repo.repo_id, [])
165 await db_session.commit()
166
167 with patch(
168 "musehub.services.musehub_wire_fetch.wire_fetch_mpack",
169 new_callable=AsyncMock,
170 ) as mock_build:
171 result = await process_fetch_mpack_prebuild_job(db_session, job_id)
172
173 assert mock_build.call_count == 0
174 assert result["tips_requested"] == 0
175 assert result["tips_built"] == 0
176
177
178 # ── FMC_08 ────────────────────────────────────────────────────────────────────
179
180 @pytest.mark.tier2
181 async def test_fmc_08_cache_row_has_correct_fields(db_session: AsyncSession) -> None:
182 """FMC_08: running the handler end to end (through a REAL wire_fetch_mpack) writes a
183 cache row with matching repo_id, tip, and mpack_id. Cache rows are written inside
184 wire_fetch_mpack, so this exercises the full handler -> build -> cache-write chain
185 rather than mocking the build away. Only the expensive externals (DAG walk, mpack
186 bytes, storage) are stubbed — same pattern as FMC_14."""
187 from types import SimpleNamespace
188
189 repo = await create_repo(db_session, owner="gabriel", visibility="public")
190 tip = _fake_commit_id("integration-tip")
191 built_mpack = _fake_mpack_id("integration")
192
193 # The handler reads live branch tips from MusehubBranch (Phase 1 fix).
194 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip)
195 job_id = await _insert_job(db_session, repo.repo_id, [tip])
196 await db_session.commit()
197
198 mock_backend = AsyncMock()
199 mock_backend.put_mpack.return_value = None
200 mock_backend.presign_mpack_get.return_value = "https://r2.example/int"
201 mock_backend.delete.return_value = None
202 fake_proxy = SimpleNamespace(commit_id=tip, snapshot_id=None, parent_ids=[])
203
204 with patch("musehub.services.musehub_wire_fetch.get_backend", return_value=mock_backend), \
205 patch("musehub.services.musehub_wire_fetch._walk_commit_delta", new_callable=AsyncMock,
206 return_value={tip: fake_proxy}), \
207 patch("muse.core.mpack.build_wire_mpack", return_value=b"MUSE\x00fake-mpack"), \
208 patch("musehub.services.musehub_wire_fetch.blob_id", return_value=built_mpack):
209 result = await process_fetch_mpack_prebuild_job(db_session, job_id)
210 await db_session.commit()
211
212 assert result["tips_built"] == 1
213
214 row = (await db_session.execute(
215 select(MusehubFetchMPackCache)
216 .where(MusehubFetchMPackCache.repo_id == repo.repo_id)
217 .where(MusehubFetchMPackCache.tip_commit_id == tip)
218 )).scalar_one()
219
220 assert row.repo_id == repo.repo_id
221 assert row.tip_commit_id == tip
222 assert row.mpack_id == built_mpack
223 assert row.expires_at > _now()
224
225
226 # ── FMC_13 ────────────────────────────────────────────────────────────────────
227
228 @pytest.mark.tier2
229 async def test_fmc_13_cache_hit_returns_presigned_url_without_blob_load(db_session: AsyncSession) -> None:
230 """FMC_13: cache hit returns the presigned URL immediately; blob-load path is never entered."""
231 repo = await create_repo(db_session, owner="gabriel", visibility="public")
232 tip = _fake_commit_id("fmc13-tip")
233 cached_mpack = _fake_mpack_id("fmc13-tip")
234 expected_url = "https://r2.example/cached-fmc13"
235
236 # Pre-populate a fresh cache entry.
237 cache_id = hashlib.sha256((repo.repo_id + tip).encode()).hexdigest()
238 db_session.add(MusehubFetchMPackCache(
239 cache_id=cache_id,
240 repo_id=repo.repo_id,
241 tip_commit_id=tip,
242 mpack_id=cached_mpack,
243 created_at=_now(),
244 expires_at=_now() + timedelta(days=7),
245 ))
246 await db_session.commit()
247
248 mock_backend = AsyncMock()
249 mock_backend.presign_mpack_get.return_value = expected_url
250
251 with patch("musehub.services.musehub_wire_fetch.get_backend", return_value=mock_backend), \
252 patch("musehub.services.musehub_wire_fetch._walk_commit_delta", new_callable=AsyncMock) as mock_walk:
253 result = await wire_fetch_mpack(db_session, repo.repo_id, want=[tip], have=[])
254
255 # The expensive DAG walk must never have been called.
256 mock_walk.assert_not_called()
257 # The returned URL must match the presigned URL for the cached mpack.
258 assert result["mpack_url"] == expected_url
259 assert result["mpack_id"] == cached_mpack
260
261
262 # ── FMC_14 ────────────────────────────────────────────────────────────────────
263
264 @pytest.mark.tier2
265 async def test_fmc_14_cache_miss_builds_and_writes_cache_row(db_session: AsyncSession) -> None:
266 """FMC_14: on a cache miss, wire_fetch_mpack builds the mpack and writes a cache row."""
267 from types import SimpleNamespace
268
269 repo = await create_repo(db_session, owner="gabriel", visibility="public")
270 tip = _fake_commit_id("fmc14-tip")
271 built_mpack = _fake_mpack_id("fmc14-tip")
272 built_url = "https://r2.example/built-fmc14"
273
274 # No cache row exists — this is a cold miss.
275 mock_backend = AsyncMock()
276 mock_backend.put_mpack.return_value = None
277 mock_backend.presign_mpack_get.return_value = built_url
278 mock_backend.delete.return_value = None
279
280 # _walk_commit_delta returns one proxy commit with no snapshot so that
281 # wire_fetch_mpack proceeds to build an empty-but-valid mpack without
282 # needing real commit/snapshot/object rows in the DB.
283 fake_proxy = SimpleNamespace(commit_id=tip, snapshot_id=None, parent_ids=[])
284
285 with patch("musehub.services.musehub_wire_fetch.get_backend", return_value=mock_backend), \
286 patch("musehub.services.musehub_wire_fetch._walk_commit_delta", new_callable=AsyncMock,
287 return_value={tip: fake_proxy}), \
288 patch("muse.core.mpack.build_wire_mpack", return_value=b"MUSE\x00fake-mpack"), \
289 patch("musehub.services.musehub_wire_fetch.blob_id", return_value=built_mpack):
290 result = await wire_fetch_mpack(db_session, repo.repo_id, want=[tip], have=[], force_build=True)
291 await db_session.commit()
292
293 assert result["mpack_url"] == built_url
294 assert result["mpack_id"] == built_mpack
295
296 # The cache row must have been written.
297 row = (await db_session.execute(
298 select(MusehubFetchMPackCache)
299 .where(MusehubFetchMPackCache.repo_id == repo.repo_id)
300 .where(MusehubFetchMPackCache.tip_commit_id == tip)
301 )).scalar_one()
302 assert row.mpack_id == built_mpack
303 assert row.expires_at > _now()
304
305
306 # ── FMC_18 ────────────────────────────────────────────────────────────────────
307
308 @pytest.mark.tier2
309 async def test_fmc_18_enqueue_push_intel_creates_prebuild_job_with_branch_tips(
310 db_session: AsyncSession,
311 ) -> None:
312 """FMC_18: enqueue_push_intel enqueues fetch.mpack.prebuild with all branch tip commit IDs."""
313 repo = await create_repo(db_session, owner="gabriel", visibility="public")
314
315 tip_a = _fake_commit_id("branch-main")
316 tip_b = _fake_commit_id("branch-dev")
317 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=tip_a)
318 await create_branch(db_session, repo.repo_id, name="dev", head_commit_id=tip_b)
319
320 await enqueue_push_intel(
321 db_session,
322 repo.repo_id,
323 head=tip_a,
324 domain_id=None,
325 branch="main",
326 )
327 await db_session.commit()
328
329 job_row = (await db_session.execute(
330 select(MusehubBackgroundJob)
331 .where(MusehubBackgroundJob.repo_id == repo.repo_id)
332 .where(MusehubBackgroundJob.job_type == "fetch.mpack.prebuild")
333 .where(MusehubBackgroundJob.status == "pending")
334 )).scalar_one()
335
336 tip_ids = set(job_row.payload.get("tip_commit_ids", []))
337 assert tip_a in tip_ids, f"main tip {tip_a[:20]} missing from payload"
338 assert tip_b in tip_ids, f"dev tip {tip_b[:20]} missing from payload"
339
340
341 # ── FMC_20 ────────────────────────────────────────────────────────────────────
342
343 @pytest.mark.tier2
344 async def test_fmc_20_gc_deletes_expired_rows_and_r2_objects_leaves_fresh_untouched(
345 db_session: AsyncSession,
346 ) -> None:
347 """FMC_20: gc_fetch_mpack_cache deletes expired rows + R2 objects; fresh rows survive."""
348 repo = await create_repo(db_session, owner="gabriel", visibility="public")
349
350 tip_expired_a = _fake_commit_id("expired-a")
351 tip_expired_b = _fake_commit_id("expired-b")
352 tip_fresh = _fake_commit_id("fresh")
353
354 mpack_expired_a = _fake_mpack_id("expired-a")
355 mpack_expired_b = _fake_mpack_id("expired-b")
356 mpack_fresh = _fake_mpack_id("fresh")
357
358 past = _now() - timedelta(days=1)
359 future = _now() + timedelta(days=6)
360
361 for tip, mpack, exp in [
362 (tip_expired_a, mpack_expired_a, past),
363 (tip_expired_b, mpack_expired_b, past),
364 (tip_fresh, mpack_fresh, future),
365 ]:
366 cache_id = hashlib.sha256((repo.repo_id + tip).encode()).hexdigest()
367 db_session.add(MusehubFetchMPackCache(
368 cache_id=cache_id,
369 repo_id=repo.repo_id,
370 tip_commit_id=tip,
371 mpack_id=mpack,
372 created_at=_now(),
373 expires_at=exp,
374 ))
375 await db_session.commit()
376
377 mock_backend = AsyncMock()
378 mock_backend.delete.return_value = None
379
380 with patch("musehub.services.musehub_gc.get_backend", return_value=mock_backend):
381 n_deleted = await gc_fetch_mpack_cache(db_session, repo.repo_id)
382 await db_session.commit()
383
384 assert n_deleted == 2
385
386 # R2 delete called exactly once for each expired mpack — in any order.
387 deleted_mpack_ids = {call.args[0] for call in mock_backend.delete.call_args_list}
388 assert mpack_expired_a in deleted_mpack_ids
389 assert mpack_expired_b in deleted_mpack_ids
390 assert mpack_fresh not in deleted_mpack_ids
391
392 # Expired rows gone from DB.
393 remaining = (await db_session.execute(
394 select(MusehubFetchMPackCache)
395 .where(MusehubFetchMPackCache.repo_id == repo.repo_id)
396 )).scalars().all()
397 remaining_tips = {r.tip_commit_id for r in remaining}
398 assert tip_expired_a not in remaining_tips
399 assert tip_expired_b not in remaining_tips
400
401 # Fresh row survives.
402 assert tip_fresh in remaining_tips
File History 4 commits
sha256:f58d788df3ccdda8f8987b428418db655a38582309239b99d7b9715ea6dff618 feat(#92): phase 5 — GC expired fetch mpack cache entries (… Sonnet 4.6 patch 34 days ago
sha256:d50f9cf9829dfbe35721a23b81ad256c729ddf9dd565a0a9e56d27847e255632 feat(#92): phase 4 — enqueue fetch.mpack.prebuild on push (… Sonnet 4.6 patch 34 days ago
sha256:1c5b7a0aba79472f4b10e52326dc010bdab1a498c9e195593d0707860478a034 feat(#92): phase 3 — cache lookup in wire_fetch_mpack (FMC_… Sonnet 4.6 patch 34 days ago
sha256:0e447fc3f6b7887d5d9e86b557c659ef7d0b05e2e09ddb0cb551ada240e48a51 feat(phase2): fetch.mpack.prebuild job handler + worker dis… Sonnet 4.6 patch 34 days ago