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