gabriel / musehub public
test_mwp4_prebuild_ordering.py python
340 lines 13.3 KB
Raw
sha256:2c523da45351334b5c4dbefed4dc3dd553b3faa8737a4e6caf301e5dc82141be test(mwp4): Phase 0 RED reproduction tests for RC-4 ordering race Sonnet 4.6 22 days ago
1 """MWP-4 — Gate fetch.mpack.prebuild behind mpack.index (fixes RC-4).
2
3 Sub-ticket: musehub#109 — https://staging.musehub.ai/gabriel/musehub/issues/109
4 Master: muse#58 — https://staging.musehub.ai/gabriel/muse/issues/58
5 Predecessors:
6 musehub#106 (MWP-1, generation authority) — closed.
7 musehub#107 (MWP-2, walk fallback) — closed.
8 musehub#108 (MWP-3, job-enqueue idempotency) — closed.
9
10 Root cause (RC-4): ``claim_next_job`` orders only by ``created_at`` with no
11 tiebreaker. When ``mpack.index`` and ``fetch.mpack.prebuild`` share the same
12 ``created_at`` (the normal case from ``enqueue_push_intel``), or the prebuild
13 lands with an earlier timestamp due to cross-push interleaving, PostgreSQL may
14 return the prebuild first. The prebuild then calls ``wire_fetch_mpack`` with
15 ``force_build=True``, hits the index-coverage check at
16 ``musehub_wire_fetch.py:681–694``, and raises ``FetchNotIndexedError`` because
17 ``MusehubMPackIndex`` is still empty. The broad ``except Exception`` at
18 ``musehub_wire_fetch.py:1164–1167`` swallows the error silently: ``tips_built``
19 stays ``0``, no ``MusehubFetchMPackCache`` row is written, and the next clone
20 returns 503.
21
22 Phase 0 — RED reproduction (no production code changes):
23 MWP4_01 RED — ``claim_next_job`` returns the prebuild before the index
24 when the prebuild has an earlier ``created_at``. Asserts
25 the desired correct behavior → FAILS before Phase 2 fix.
26 MWP4_02 — handler-level proof: prebuild run before index swallows
27 FetchNotIndexedError and writes no cache row. Asserts the
28 broken outcome → PASSES before fix (proves the bug).
29 Must be updated in Phase 3 when the swallow is narrowed.
30 """
31
32 from __future__ import annotations
33
34 import hashlib
35 from datetime import datetime, timedelta, timezone
36
37 import pytest
38 from sqlalchemy import delete, func, select
39 from sqlalchemy.ext.asyncio import AsyncSession
40
41 from muse.core.ids import hash_snapshot
42 from muse.core.mpack import build_wire_mpack
43 from muse.core.types import blob_id
44 from musehub.core.genesis import compute_job_id
45 from musehub.db.musehub_jobs_models import MusehubBackgroundJob
46 from musehub.db.musehub_repo_models import MusehubFetchMPackCache, MusehubMPackIndex
47 from musehub.services.musehub_jobs import claim_next_job
48 from musehub.services.musehub_wire_fetch import process_fetch_mpack_prebuild_job
49 from musehub.services.musehub_wire_push import wire_push_unpack_mpack
50 from tests.factories import create_repo
51
52
53 # ---------------------------------------------------------------------------
54 # Helpers
55 # ---------------------------------------------------------------------------
56
57
58 def _fake_commit_id(seed: str) -> str:
59 return "sha256:" + hashlib.sha256(seed.encode()).hexdigest()
60
61
62 def _now() -> datetime:
63 return datetime.now(tz=timezone.utc)
64
65
66 def _job_row(
67 repo_id: str,
68 job_type: str,
69 payload: dict,
70 created_at: datetime,
71 status: str = "pending",
72 ) -> MusehubBackgroundJob:
73 """Build a MusehubBackgroundJob row ready to be added to a session."""
74 job_id = compute_job_id(repo_id, job_type, created_at.isoformat())
75 return MusehubBackgroundJob(
76 job_id=job_id,
77 repo_id=repo_id,
78 job_type=job_type,
79 payload=payload,
80 status=status,
81 created_at=created_at,
82 attempt=0,
83 )
84
85
86 _E2E_OWNER = "gabriel"
87
88
89 def _e2e_raw_commit(
90 cid: str,
91 parent: str | None,
92 snap_id: str,
93 branch: str = "main",
94 ) -> dict:
95 return {
96 "commit_id": cid,
97 "branch": branch,
98 "message": f"commit {cid[:12]}",
99 "author": _E2E_OWNER,
100 "committed_at": _now().isoformat(),
101 "parent_commit_id": parent,
102 "parent2_commit_id": None,
103 "snapshot_id": snap_id,
104 "agent_id": "",
105 "model_id": "",
106 "toolchain_id": "",
107 "sem_ver_bump": "none",
108 "breaking_changes": [],
109 "signature": "",
110 "signer_key_id": "",
111 "signer_public_key": "",
112 "prompt_hash": "",
113 }
114
115
116 async def _e2e_push(
117 session: AsyncSession,
118 repo_id: str,
119 cid: str,
120 parent: str | None,
121 snap_id: str,
122 fpath: str,
123 oid: str,
124 data: bytes,
125 branch: str = "main",
126 ) -> str:
127 """Build a single-commit mpack, store it in the conftest MemoryBackend, push it.
128
129 The conftest autouse fixture patches ``musehub.storage.backends.get_backend``
130 to a ``MemoryBackend`` instance shared across all modules. Pre-seeding
131 ``backend._mpacks[key]`` is equivalent to a MinIO PUT — both
132 ``wire_push_unpack_mpack`` and ``process_mpack_index_job`` call
133 ``get_backend()`` and see the same bytes.
134
135 Returns the ``mpack_key`` so callers can locate the enqueued index job.
136 """
137 import musehub.storage.backends as _b_mod
138 backend = _b_mod.get_backend()
139
140 mpack_bytes = build_wire_mpack({
141 "commits": [_e2e_raw_commit(cid, parent, snap_id, branch=branch)],
142 "snapshots": [{
143 "snapshot_id": snap_id,
144 "parent_snapshot_id": None,
145 "delta_upsert": {fpath: oid},
146 "delta_remove": [],
147 "directories": [],
148 }],
149 "blobs": [{"object_id": oid, "content": data}],
150 "tags": [],
151 })
152 mpack_key = blob_id(mpack_bytes)
153 backend._mpacks[mpack_key] = mpack_bytes
154
155 await wire_push_unpack_mpack(
156 session,
157 repo_id,
158 mpack_key,
159 pusher_id=_E2E_OWNER,
160 branch=branch,
161 head_commit_id=cid,
162 commits_count=1,
163 blobs_count=1,
164 force=True,
165 )
166 await session.commit()
167 return mpack_key
168
169
170 # ---------------------------------------------------------------------------
171 # MWP4_01 — claim_next_job returns prebuild before a pending index (RED)
172 # ---------------------------------------------------------------------------
173
174
175 @pytest.mark.asyncio
176 async def test_mwp4_01_claim_returns_prebuild_before_index(
177 db_session: AsyncSession,
178 ) -> None:
179 """MWP4_01 RED: claim_next_job must NOT return fetch.mpack.prebuild while
180 an mpack.index for the same repo is pending.
181
182 Setup: insert a fetch.mpack.prebuild with created_at = now-5s (earlier
183 than the mpack.index at now). This simulates the cross-push race where
184 push A's prebuild is enqueued before push B's index job arrives.
185
186 Pre-fix (expected to FAIL):
187 claim_next_job orders solely by created_at; the prebuild (earlier
188 timestamp) is returned ahead of the index. The assertion
189 ``claimed.job_type == "mpack.index"`` therefore fails.
190
191 Post-fix (Phase 2 — correlated NOT EXISTS barrier):
192 The barrier prevents fetch.mpack.prebuild from being claimed while any
193 mpack.index for the same repo_id is pending or running. The index is
194 returned instead, and the prebuild waits until the index drains.
195 """
196 repo = await create_repo(db_session, owner="gabriel")
197 repo_id = str(repo.repo_id)
198
199 now = _now()
200 earlier = now - timedelta(seconds=5)
201
202 # Insert prebuild FIRST with an earlier timestamp to force it ahead in the
203 # ORDER BY created_at ordering — this is the race we're fixing.
204 prebuild = _job_row(
205 repo_id,
206 "fetch.mpack.prebuild",
207 {"tip_commit_ids": [_fake_commit_id("mwp4-01-tip")]},
208 created_at=earlier,
209 )
210 index = _job_row(
211 repo_id,
212 "mpack.index",
213 {
214 "mpack_key": "sha256:" + hashlib.sha256(b"mwp4-01-mpack").hexdigest(),
215 "head": _fake_commit_id("mwp4-01-head"),
216 "branch": "main",
217 },
218 created_at=now,
219 )
220 db_session.add(prebuild)
221 db_session.add(index)
222 await db_session.commit()
223
224 # Claim the next job.
225 # Pre-fix: returns the prebuild (earliest created_at wins) → assertion fails.
226 # Post-fix: barrier skips the prebuild; returns the index instead.
227 claimed = await claim_next_job(db_session)
228 await db_session.commit()
229
230 assert claimed is not None, "Expected a job to be claimed — queue has two pending rows."
231 assert claimed.job_type == "mpack.index", (
232 f"claim_next_job must skip fetch.mpack.prebuild while mpack.index is "
233 f"pending for the same repo_id, but returned '{claimed.job_type}'. "
234 f"The prebuild must be gated behind all pending index jobs (RC-4 barrier)."
235 )
236
237
238 # ---------------------------------------------------------------------------
239 # MWP4_02 — prebuild run before index swallows FetchNotIndexedError (bug proof)
240 # ---------------------------------------------------------------------------
241
242
243 @pytest.mark.asyncio
244 async def test_mwp4_02_prebuild_swallows_fetch_not_indexed_error(
245 db_session: AsyncSession,
246 ) -> None:
247 """MWP4_02 (Phase 0 bug documentation): process_fetch_mpack_prebuild_job run
248 before mpack.index is processed swallows FetchNotIndexedError and writes no
249 MusehubFetchMPackCache row — the broken silent-no-cache outcome of RC-4.
250
251 Setup: push one commit so MusehubObject, MusehubCommit, and MusehubBranch
252 rows exist. wire_push_unpack_mpack writes MusehubMPackIndex rows at step 7d
253 synchronously, so we explicitly delete them after the push to simulate the
254 race state: DB has commits/snapshots but MusehubMPackIndex is empty (as if
255 the mpack.index job hasn't run yet).
256
257 Inside process_fetch_mpack_prebuild_job:
258 wire_fetch_mpack is called with force_build=True. It assembles the
259 object set from snapshots, checks index coverage at lines 681–694 of
260 musehub_wire_fetch.py, finds every object missing from
261 MusehubMPackIndex, and raises FetchNotIndexedError. The broad
262 ``except Exception`` at lines 1164–1167 catches and logs it; the
263 function returns a result dict with tips_built=0 and no cache row.
264
265 Pre-fix: this test PASSES — tips_built==0 and cache_count==0 confirm
266 the bug is present (FetchNotIndexedError was silently swallowed).
267
268 Post-fix (Phase 3 — narrow the swallow):
269 process_fetch_mpack_prebuild_job will re-raise FetchNotIndexedError
270 instead of returning quietly. Update this test at that point to use
271 ``pytest.raises(FetchNotIndexedError)`` rather than asserting tips_built.
272 With Phase 2's barrier in place, this path is never hit in production —
273 the prebuild only runs after the index has completed.
274 """
275 repo = await create_repo(db_session, owner="gabriel")
276 repo_id = str(repo.repo_id)
277
278 data = b"mwp4-02-blob-content"
279 oid = blob_id(data)
280 snap_id = hash_snapshot({"file.txt": oid})
281 cid = _fake_commit_id("mwp4-02-c1")
282
283 # Push one commit — creates MusehubObject, MusehubCommit, MusehubBranch,
284 # enqueues mpack.index + fetch.mpack.prebuild, AND writes MusehubMPackIndex
285 # rows at step 7d of wire_push_unpack_mpack (synchronous inline indexing).
286 mpack_key = await _e2e_push(db_session, repo_id, cid, None, snap_id, "file.txt", oid, data)
287
288 # Simulate the RC-4 race: delete the inline MusehubMPackIndex rows that the
289 # push wrote. This replicates the state where the prebuild job is claimed
290 # before the mpack.index job has had a chance to run — MusehubCommit,
291 # MusehubSnapshot, and MusehubBranch are populated, but MusehubMPackIndex
292 # is empty for this mpack_key.
293 await db_session.execute(
294 delete(MusehubMPackIndex).where(MusehubMPackIndex.mpack_id == mpack_key)
295 )
296 await db_session.commit()
297
298 # Locate the prebuild job enqueued by wire_push_unpack_mpack.
299 prebuild_job = (await db_session.execute(
300 select(MusehubBackgroundJob).where(
301 MusehubBackgroundJob.repo_id == repo_id,
302 MusehubBackgroundJob.job_type == "fetch.mpack.prebuild",
303 MusehubBackgroundJob.status == "pending",
304 ).limit(1)
305 )).scalar_one_or_none()
306 assert prebuild_job is not None, (
307 "wire_push_unpack_mpack must enqueue a fetch.mpack.prebuild job after a push."
308 )
309
310 # Run the prebuild handler with MusehubMPackIndex cleared (simulating RC-4 race).
311 # wire_fetch_mpack raises FetchNotIndexedError at musehub_wire_fetch.py:694;
312 # the broad except at lines 1164–1167 swallows it and returns tips_built=0.
313 result = await process_fetch_mpack_prebuild_job(db_session, prebuild_job.job_id)
314 await db_session.commit()
315
316 # --- Bug proof #1: FetchNotIndexedError was swallowed (not propagated) ---
317 # If this assertion fails, the swallow has already been narrowed (Phase 3 done).
318 assert result["tips_built"] == 0, (
319 f"tips_built must be 0 when MusehubMPackIndex is empty (FetchNotIndexedError "
320 f"swallowed by broad except). Got tips_built={result['tips_built']}. "
321 f"If this fails, Phase 3 is already applied — update to pytest.raises()."
322 )
323 assert result["tips_requested"] > 0, (
324 f"tips_requested must be > 0 (branch tip exists); got "
325 f"{result['tips_requested']}. Check that wire_push_unpack_mpack created a "
326 f"MusehubBranch row pointing to the pushed commit."
327 )
328
329 # --- Bug proof #2: no cache row written — clone would 503 ---
330 cache_count = (await db_session.execute(
331 select(func.count()).select_from(MusehubFetchMPackCache).where(
332 MusehubFetchMPackCache.repo_id == repo_id,
333 )
334 )).scalar()
335 assert cache_count == 0, (
336 f"No MusehubFetchMPackCache row must exist when the prebuild swallowed "
337 f"FetchNotIndexedError (the broken silent-no-cache outcome). "
338 f"Found {cache_count} row(s). If this fails, the bug is fixed — "
339 f"update to assert cache_count == 1."
340 )
File History 1 commit
sha256:2c523da45351334b5c4dbefed4dc3dd553b3faa8737a4e6caf301e5dc82141be test(mwp4): Phase 0 RED reproduction tests for RC-4 ordering race Sonnet 4.6 22 days ago