test_mwp4_prebuild_ordering.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 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, complete_job, enqueue_push_intel |
| 48 | from musehub.services.musehub_wire_fetch import process_fetch_mpack_prebuild_job |
| 49 | from musehub.services.musehub_wire_push import process_mpack_index_job, wire_push_unpack_mpack |
| 50 | from musehub.services.musehub_wire_shared import FetchNotIndexedError |
| 51 | from tests.factories import create_repo |
| 52 | |
| 53 | |
| 54 | # --------------------------------------------------------------------------- |
| 55 | # Helpers |
| 56 | # --------------------------------------------------------------------------- |
| 57 | |
| 58 | |
| 59 | def _fake_commit_id(seed: str) -> str: |
| 60 | return "sha256:" + hashlib.sha256(seed.encode()).hexdigest() |
| 61 | |
| 62 | |
| 63 | def _now() -> datetime: |
| 64 | return datetime.now(tz=timezone.utc) |
| 65 | |
| 66 | |
| 67 | def _job_row( |
| 68 | repo_id: str, |
| 69 | job_type: str, |
| 70 | payload: dict, |
| 71 | created_at: datetime, |
| 72 | status: str = "pending", |
| 73 | ) -> MusehubBackgroundJob: |
| 74 | """Build a MusehubBackgroundJob row ready to be added to a session.""" |
| 75 | job_id = compute_job_id(repo_id, job_type, created_at.isoformat()) |
| 76 | return MusehubBackgroundJob( |
| 77 | job_id=job_id, |
| 78 | repo_id=repo_id, |
| 79 | job_type=job_type, |
| 80 | payload=payload, |
| 81 | status=status, |
| 82 | created_at=created_at, |
| 83 | attempt=0, |
| 84 | ) |
| 85 | |
| 86 | |
| 87 | _E2E_OWNER = "gabriel" |
| 88 | |
| 89 | |
| 90 | def _e2e_raw_commit( |
| 91 | cid: str, |
| 92 | parent: str | None, |
| 93 | snap_id: str, |
| 94 | branch: str = "main", |
| 95 | ) -> dict: |
| 96 | return { |
| 97 | "commit_id": cid, |
| 98 | "branch": branch, |
| 99 | "message": f"commit {cid[:12]}", |
| 100 | "author": _E2E_OWNER, |
| 101 | "committed_at": _now().isoformat(), |
| 102 | "parent_commit_id": parent, |
| 103 | "parent2_commit_id": None, |
| 104 | "snapshot_id": snap_id, |
| 105 | "agent_id": "", |
| 106 | "model_id": "", |
| 107 | "toolchain_id": "", |
| 108 | "sem_ver_bump": "none", |
| 109 | "breaking_changes": [], |
| 110 | "signature": "", |
| 111 | "signer_key_id": "", |
| 112 | "signer_public_key": "", |
| 113 | "prompt_hash": "", |
| 114 | } |
| 115 | |
| 116 | |
| 117 | async def _e2e_push( |
| 118 | session: AsyncSession, |
| 119 | repo_id: str, |
| 120 | cid: str, |
| 121 | parent: str | None, |
| 122 | snap_id: str, |
| 123 | fpath: str, |
| 124 | oid: str, |
| 125 | data: bytes, |
| 126 | branch: str = "main", |
| 127 | ) -> str: |
| 128 | """Build a single-commit mpack, store it in the conftest MemoryBackend, push it. |
| 129 | |
| 130 | The conftest autouse fixture patches ``musehub.storage.backends.get_backend`` |
| 131 | to a ``MemoryBackend`` instance shared across all modules. Pre-seeding |
| 132 | ``backend._mpacks[key]`` is equivalent to a MinIO PUT — both |
| 133 | ``wire_push_unpack_mpack`` and ``process_mpack_index_job`` call |
| 134 | ``get_backend()`` and see the same bytes. |
| 135 | |
| 136 | Returns the ``mpack_key`` so callers can locate the enqueued index job. |
| 137 | """ |
| 138 | import musehub.storage.backends as _b_mod |
| 139 | backend = _b_mod.get_backend() |
| 140 | |
| 141 | mpack_bytes = build_wire_mpack({ |
| 142 | "commits": [_e2e_raw_commit(cid, parent, snap_id, branch=branch)], |
| 143 | "snapshots": [{ |
| 144 | "snapshot_id": snap_id, |
| 145 | "parent_snapshot_id": None, |
| 146 | "delta_upsert": {fpath: oid}, |
| 147 | "delta_remove": [], |
| 148 | "directories": [], |
| 149 | }], |
| 150 | "blobs": [{"object_id": oid, "content": data}], |
| 151 | "tags": [], |
| 152 | }) |
| 153 | mpack_key = blob_id(mpack_bytes) |
| 154 | backend._mpacks[mpack_key] = mpack_bytes |
| 155 | |
| 156 | await wire_push_unpack_mpack( |
| 157 | session, |
| 158 | repo_id, |
| 159 | mpack_key, |
| 160 | pusher_id=_E2E_OWNER, |
| 161 | branch=branch, |
| 162 | head_commit_id=cid, |
| 163 | commits_count=1, |
| 164 | blobs_count=1, |
| 165 | force=True, |
| 166 | ) |
| 167 | await session.commit() |
| 168 | return mpack_key |
| 169 | |
| 170 | |
| 171 | # --------------------------------------------------------------------------- |
| 172 | # MWP4_01 — claim_next_job returns prebuild before a pending index (RED) |
| 173 | # --------------------------------------------------------------------------- |
| 174 | |
| 175 | |
| 176 | @pytest.mark.asyncio |
| 177 | async def test_mwp4_01_claim_returns_prebuild_before_index( |
| 178 | db_session: AsyncSession, |
| 179 | ) -> None: |
| 180 | """MWP4_01 RED: claim_next_job must NOT return fetch.mpack.prebuild while |
| 181 | an mpack.index for the same repo is pending. |
| 182 | |
| 183 | Setup: insert a fetch.mpack.prebuild with created_at = now-5s (earlier |
| 184 | than the mpack.index at now). This simulates the cross-push race where |
| 185 | push A's prebuild is enqueued before push B's index job arrives. |
| 186 | |
| 187 | Pre-fix (expected to FAIL): |
| 188 | claim_next_job orders solely by created_at; the prebuild (earlier |
| 189 | timestamp) is returned ahead of the index. The assertion |
| 190 | ``claimed.job_type == "mpack.index"`` therefore fails. |
| 191 | |
| 192 | Post-fix (Phase 2 — correlated NOT EXISTS barrier): |
| 193 | The barrier prevents fetch.mpack.prebuild from being claimed while any |
| 194 | mpack.index for the same repo_id is pending or running. The index is |
| 195 | returned instead, and the prebuild waits until the index drains. |
| 196 | """ |
| 197 | repo = await create_repo(db_session, owner="gabriel") |
| 198 | repo_id = str(repo.repo_id) |
| 199 | |
| 200 | now = _now() |
| 201 | earlier = now - timedelta(seconds=5) |
| 202 | |
| 203 | # Insert prebuild FIRST with an earlier timestamp to force it ahead in the |
| 204 | # ORDER BY created_at ordering — this is the race we're fixing. |
| 205 | prebuild = _job_row( |
| 206 | repo_id, |
| 207 | "fetch.mpack.prebuild", |
| 208 | {"tip_commit_ids": [_fake_commit_id("mwp4-01-tip")]}, |
| 209 | created_at=earlier, |
| 210 | ) |
| 211 | index = _job_row( |
| 212 | repo_id, |
| 213 | "mpack.index", |
| 214 | { |
| 215 | "mpack_key": "sha256:" + hashlib.sha256(b"mwp4-01-mpack").hexdigest(), |
| 216 | "head": _fake_commit_id("mwp4-01-head"), |
| 217 | "branch": "main", |
| 218 | }, |
| 219 | created_at=now, |
| 220 | ) |
| 221 | db_session.add(prebuild) |
| 222 | db_session.add(index) |
| 223 | await db_session.commit() |
| 224 | |
| 225 | # Claim the next job. |
| 226 | # Pre-fix: returns the prebuild (earliest created_at wins) → assertion fails. |
| 227 | # Post-fix: barrier skips the prebuild; returns the index instead. |
| 228 | claimed = await claim_next_job(db_session) |
| 229 | await db_session.commit() |
| 230 | |
| 231 | assert claimed is not None, "Expected a job to be claimed — queue has two pending rows." |
| 232 | assert claimed.job_type == "mpack.index", ( |
| 233 | f"claim_next_job must skip fetch.mpack.prebuild while mpack.index is " |
| 234 | f"pending for the same repo_id, but returned '{claimed.job_type}'. " |
| 235 | f"The prebuild must be gated behind all pending index jobs (RC-4 barrier)." |
| 236 | ) |
| 237 | |
| 238 | |
| 239 | # --------------------------------------------------------------------------- |
| 240 | # MWP4_02 — prebuild run before index swallows FetchNotIndexedError (bug proof) |
| 241 | # --------------------------------------------------------------------------- |
| 242 | |
| 243 | |
| 244 | @pytest.mark.asyncio |
| 245 | async def test_mwp4_02_prebuild_swallows_fetch_not_indexed_error( |
| 246 | db_session: AsyncSession, |
| 247 | ) -> None: |
| 248 | """MWP4_02 (updated for Phase 3): process_fetch_mpack_prebuild_job raises |
| 249 | FetchNotIndexedError when MusehubMPackIndex is empty — no longer swallowed. |
| 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 and has also terminally failed, bypassing |
| 256 | the Phase 2 barrier). |
| 257 | |
| 258 | Phase 0 behavior (pre-Phase 3): FetchNotIndexedError was silently swallowed |
| 259 | by the broad ``except Exception``; tips_built==0, no cache row. |
| 260 | |
| 261 | Phase 3 behavior: FetchNotIndexedError is caught specifically and re-raised |
| 262 | so the worker's fail_job records it as a failed job (retryable). The broad |
| 263 | ``except Exception`` still handles genuinely unexpected errors. |
| 264 | |
| 265 | With the Phase 2 barrier in place this path is only reachable when an |
| 266 | mpack.index has terminally failed — exactly the case that should surface as |
| 267 | a failed job, not a silent empty cache. |
| 268 | """ |
| 269 | repo = await create_repo(db_session, owner="gabriel") |
| 270 | repo_id = str(repo.repo_id) |
| 271 | |
| 272 | data = b"mwp4-02-blob-content" |
| 273 | oid = blob_id(data) |
| 274 | snap_id = hash_snapshot({"file.txt": oid}) |
| 275 | cid = _fake_commit_id("mwp4-02-c1") |
| 276 | |
| 277 | # Push one commit — creates MusehubObject, MusehubCommit, MusehubBranch, |
| 278 | # enqueues mpack.index + fetch.mpack.prebuild, AND writes MusehubMPackIndex |
| 279 | # rows at step 7d of wire_push_unpack_mpack (synchronous inline indexing). |
| 280 | mpack_key = await _e2e_push(db_session, repo_id, cid, None, snap_id, "file.txt", oid, data) |
| 281 | |
| 282 | # Simulate the RC-4 race: delete the inline MusehubMPackIndex rows that the |
| 283 | # push wrote. This replicates the state where the prebuild job is claimed |
| 284 | # before the mpack.index job has had a chance to run — MusehubCommit, |
| 285 | # MusehubSnapshot, and MusehubBranch are populated, but MusehubMPackIndex |
| 286 | # is empty for this mpack_key. |
| 287 | await db_session.execute( |
| 288 | delete(MusehubMPackIndex).where(MusehubMPackIndex.mpack_id == mpack_key) |
| 289 | ) |
| 290 | await db_session.commit() |
| 291 | |
| 292 | # Locate the prebuild job enqueued by wire_push_unpack_mpack. |
| 293 | prebuild_job = (await db_session.execute( |
| 294 | select(MusehubBackgroundJob).where( |
| 295 | MusehubBackgroundJob.repo_id == repo_id, |
| 296 | MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", |
| 297 | MusehubBackgroundJob.status == "pending", |
| 298 | ).limit(1) |
| 299 | )).scalar_one_or_none() |
| 300 | assert prebuild_job is not None, ( |
| 301 | "wire_push_unpack_mpack must enqueue a fetch.mpack.prebuild job after a push." |
| 302 | ) |
| 303 | |
| 304 | # Phase 3: FetchNotIndexedError must propagate (no longer swallowed). |
| 305 | # The worker's fail_job path will record this as a retryable failure. |
| 306 | with pytest.raises(FetchNotIndexedError): |
| 307 | await process_fetch_mpack_prebuild_job(db_session, prebuild_job.job_id) |
| 308 | await db_session.rollback() |
| 309 | |
| 310 | |
| 311 | # --------------------------------------------------------------------------- |
| 312 | # MWP4_03 — fetch.mpack.prebuild created_at is strictly after mpack.index |
| 313 | # --------------------------------------------------------------------------- |
| 314 | |
| 315 | |
| 316 | @pytest.mark.asyncio |
| 317 | async def test_mwp4_03_prebuild_created_at_after_index( |
| 318 | db_session: AsyncSession, |
| 319 | ) -> None: |
| 320 | """MWP4_03: enqueue_push_intel must assign fetch.mpack.prebuild a created_at |
| 321 | strictly greater than mpack.index and strictly less than intel.code.* subtypes. |
| 322 | |
| 323 | This is the Phase 1 defense-in-depth stagger: |
| 324 | |
| 325 | t+0s mpack.index, intel.code, intel.structural, gc, ... (foundation) |
| 326 | t+1s fetch.mpack.prebuild (NEW middle tier) |
| 327 | t+2s intel.code.* subtypes (subtype tier) |
| 328 | |
| 329 | Pre-fix: FAILS — fetch.mpack.prebuild and mpack.index share the same |
| 330 | created_at (both are in _FOUNDATION_TYPES at t+0s). |
| 331 | |
| 332 | Post-fix (Phase 1): PASSES — fetch.mpack.prebuild gets created_at_prebuild |
| 333 | (t+1s) while mpack.index stays at the foundation tier (t+0s). |
| 334 | """ |
| 335 | repo = await create_repo(db_session, owner="gabriel") |
| 336 | repo_id = str(repo.repo_id) |
| 337 | fake_head = _fake_commit_id("mwp4-03-head") |
| 338 | fake_mpack_key = "sha256:" + hashlib.sha256(b"mwp4-03-mpack").hexdigest() |
| 339 | |
| 340 | await enqueue_push_intel( |
| 341 | db_session, |
| 342 | repo_id=repo_id, |
| 343 | head=fake_head, |
| 344 | domain_id=None, # triggers code-domain subtypes (intel.code.*) |
| 345 | branch="main", |
| 346 | owner=None, |
| 347 | mpack_key=fake_mpack_key, |
| 348 | ) |
| 349 | await db_session.commit() |
| 350 | |
| 351 | # Fetch created_at for mpack.index |
| 352 | index_row = (await db_session.execute( |
| 353 | select(MusehubBackgroundJob.created_at).where( |
| 354 | MusehubBackgroundJob.repo_id == repo_id, |
| 355 | MusehubBackgroundJob.job_type == "mpack.index", |
| 356 | ).limit(1) |
| 357 | )).scalar_one_or_none() |
| 358 | assert index_row is not None, "mpack.index job must be enqueued by enqueue_push_intel" |
| 359 | |
| 360 | # Fetch created_at for fetch.mpack.prebuild |
| 361 | prebuild_row = (await db_session.execute( |
| 362 | select(MusehubBackgroundJob.created_at).where( |
| 363 | MusehubBackgroundJob.repo_id == repo_id, |
| 364 | MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", |
| 365 | ).limit(1) |
| 366 | )).scalar_one_or_none() |
| 367 | assert prebuild_row is not None, "fetch.mpack.prebuild job must be enqueued by enqueue_push_intel" |
| 368 | |
| 369 | # Fetch created_at for any intel.code.* subtype (the t+2s tier) |
| 370 | subtype_row = (await db_session.execute( |
| 371 | select(MusehubBackgroundJob.created_at).where( |
| 372 | MusehubBackgroundJob.repo_id == repo_id, |
| 373 | MusehubBackgroundJob.job_type == "intel.code.coupling", |
| 374 | ).limit(1) |
| 375 | )).scalar_one_or_none() |
| 376 | assert subtype_row is not None, ( |
| 377 | "intel.code.coupling job must be enqueued when domain_id=None — " |
| 378 | "needed to verify the three-tier ordering" |
| 379 | ) |
| 380 | |
| 381 | index_at = index_row |
| 382 | prebuild_at = prebuild_row |
| 383 | subtype_at = subtype_row |
| 384 | |
| 385 | # Tier 1 → Tier 2: prebuild strictly after index |
| 386 | assert prebuild_at > index_at, ( |
| 387 | f"fetch.mpack.prebuild created_at ({prebuild_at.isoformat()}) must be " |
| 388 | f"strictly greater than mpack.index created_at ({index_at.isoformat()}). " |
| 389 | f"Pre-fix: they share the same timestamp (both in _FOUNDATION_TYPES at t+0s). " |
| 390 | f"Phase 1 fix: assign fetch.mpack.prebuild created_at_prebuild = t+1s." |
| 391 | ) |
| 392 | |
| 393 | # Tier 2 → Tier 3: prebuild strictly before intel.code.* subtypes |
| 394 | assert prebuild_at < subtype_at, ( |
| 395 | f"fetch.mpack.prebuild created_at ({prebuild_at.isoformat()}) must be " |
| 396 | f"strictly less than intel.code.* subtype created_at ({subtype_at.isoformat()}). " |
| 397 | f"The three-tier ordering must be: foundation < prebuild < subtypes." |
| 398 | ) |
| 399 | |
| 400 | |
| 401 | # --------------------------------------------------------------------------- |
| 402 | # Phase 2 — claim-layer dependency barrier (MWP4_04 – MWP4_09) |
| 403 | # --------------------------------------------------------------------------- |
| 404 | |
| 405 | |
| 406 | @pytest.mark.asyncio |
| 407 | async def test_mwp4_04_prebuild_not_claimed_while_index_pending( |
| 408 | db_session: AsyncSession, |
| 409 | ) -> None: |
| 410 | """MWP4_04: claim_next_job must NOT return fetch.mpack.prebuild while any |
| 411 | mpack.index for the same repo is pending — even when the prebuild has an |
| 412 | earlier created_at (the exact cross-push race from MWP4_01). |
| 413 | |
| 414 | Pre-fix: FAILS — claim returns the prebuild (earlier timestamp wins). |
| 415 | Post-fix (Phase 2 barrier): PASSES — the correlated NOT EXISTS filter |
| 416 | blocks the prebuild; claim returns the mpack.index instead. |
| 417 | """ |
| 418 | repo = await create_repo(db_session, owner="gabriel") |
| 419 | repo_id = str(repo.repo_id) |
| 420 | |
| 421 | now = _now() |
| 422 | earlier = now - timedelta(seconds=5) |
| 423 | |
| 424 | prebuild = _job_row( |
| 425 | repo_id, "fetch.mpack.prebuild", |
| 426 | {"tip_commit_ids": [_fake_commit_id("mwp4-04-tip")]}, |
| 427 | created_at=earlier, |
| 428 | ) |
| 429 | index = _job_row( |
| 430 | repo_id, "mpack.index", |
| 431 | {"mpack_key": "sha256:" + hashlib.sha256(b"mwp4-04").hexdigest(), "head": _fake_commit_id("h"), "branch": "main"}, |
| 432 | created_at=now, |
| 433 | ) |
| 434 | db_session.add(prebuild) |
| 435 | db_session.add(index) |
| 436 | await db_session.commit() |
| 437 | |
| 438 | claimed = await claim_next_job(db_session) |
| 439 | await db_session.commit() |
| 440 | |
| 441 | assert claimed is not None, "A claimable job must exist (mpack.index is pending)" |
| 442 | assert claimed.job_type == "mpack.index", ( |
| 443 | f"claim_next_job must skip fetch.mpack.prebuild while mpack.index is " |
| 444 | f"pending for the same repo — the Phase 2 barrier. Got: {claimed.job_type}" |
| 445 | ) |
| 446 | |
| 447 | |
| 448 | @pytest.mark.asyncio |
| 449 | async def test_mwp4_05_prebuild_claimable_after_index_done( |
| 450 | db_session: AsyncSession, |
| 451 | ) -> None: |
| 452 | """MWP4_05: once mpack.index is in terminal state 'done', the prebuild |
| 453 | must be claimable on the next claim_next_job call. |
| 454 | |
| 455 | Pre-fix: PASSES (no barrier means prebuild is always claimable by timestamp). |
| 456 | Post-fix: must still PASS — 'done' is not a blocking state. |
| 457 | """ |
| 458 | repo = await create_repo(db_session, owner="gabriel") |
| 459 | repo_id = str(repo.repo_id) |
| 460 | |
| 461 | now = _now() |
| 462 | index_done = _job_row( |
| 463 | repo_id, "mpack.index", |
| 464 | {"mpack_key": "sha256:" + hashlib.sha256(b"mwp4-05").hexdigest(), "head": _fake_commit_id("h"), "branch": "main"}, |
| 465 | created_at=now - timedelta(seconds=10), |
| 466 | status="done", |
| 467 | ) |
| 468 | prebuild = _job_row( |
| 469 | repo_id, "fetch.mpack.prebuild", |
| 470 | {"tip_commit_ids": [_fake_commit_id("mwp4-05-tip")]}, |
| 471 | created_at=now, |
| 472 | ) |
| 473 | db_session.add(index_done) |
| 474 | db_session.add(prebuild) |
| 475 | await db_session.commit() |
| 476 | |
| 477 | claimed = await claim_next_job(db_session) |
| 478 | await db_session.commit() |
| 479 | |
| 480 | assert claimed is not None, "fetch.mpack.prebuild must be claimable when mpack.index is done" |
| 481 | assert claimed.job_type == "fetch.mpack.prebuild", ( |
| 482 | f"mpack.index in 'done' state must not block the prebuild. Got: {claimed.job_type}" |
| 483 | ) |
| 484 | |
| 485 | |
| 486 | @pytest.mark.asyncio |
| 487 | async def test_mwp4_06_prebuild_blocked_while_index_running( |
| 488 | db_session: AsyncSession, |
| 489 | ) -> None: |
| 490 | """MWP4_06: an mpack.index in 'running' state (already claimed by another |
| 491 | worker) must also block the prebuild — the barrier covers both pending and |
| 492 | running. |
| 493 | |
| 494 | Setup: mpack.index is running (claimed), fetch.mpack.prebuild is pending. |
| 495 | Expected: claim_next_job returns None (the only pending job is blocked). |
| 496 | |
| 497 | Pre-fix: FAILS — prebuild is returned (no barrier). |
| 498 | Post-fix: PASSES — running mpack.index blocks the prebuild. |
| 499 | """ |
| 500 | repo = await create_repo(db_session, owner="gabriel") |
| 501 | repo_id = str(repo.repo_id) |
| 502 | |
| 503 | now = _now() |
| 504 | index_running = _job_row( |
| 505 | repo_id, "mpack.index", |
| 506 | {"mpack_key": "sha256:" + hashlib.sha256(b"mwp4-06").hexdigest(), "head": _fake_commit_id("h"), "branch": "main"}, |
| 507 | created_at=now - timedelta(seconds=10), |
| 508 | status="running", |
| 509 | ) |
| 510 | prebuild = _job_row( |
| 511 | repo_id, "fetch.mpack.prebuild", |
| 512 | {"tip_commit_ids": [_fake_commit_id("mwp4-06-tip")]}, |
| 513 | created_at=now, |
| 514 | ) |
| 515 | db_session.add(index_running) |
| 516 | db_session.add(prebuild) |
| 517 | await db_session.commit() |
| 518 | |
| 519 | claimed = await claim_next_job(db_session) |
| 520 | await db_session.commit() |
| 521 | |
| 522 | assert claimed is None, ( |
| 523 | f"claim_next_job must return None when the only pending job is a " |
| 524 | f"fetch.mpack.prebuild blocked by a running mpack.index. Got: " |
| 525 | f"{claimed.job_type if claimed else None}" |
| 526 | ) |
| 527 | |
| 528 | |
| 529 | @pytest.mark.asyncio |
| 530 | async def test_mwp4_07_terminal_index_does_not_block_prebuild( |
| 531 | db_session: AsyncSession, |
| 532 | ) -> None: |
| 533 | """MWP4_07: mpack.index rows in terminal states (failed, quarantined) must |
| 534 | NOT block the prebuild. A permanently-failed index should surface via |
| 535 | fail_job, not silently hold up the prebuild forever. |
| 536 | |
| 537 | Pre-fix: PASSES (no barrier means prebuild is always claimable). |
| 538 | Post-fix: must still PASS — only pending/running are blocking states. |
| 539 | """ |
| 540 | repo = await create_repo(db_session, owner="gabriel") |
| 541 | repo_id = str(repo.repo_id) |
| 542 | |
| 543 | now = _now() |
| 544 | for i, status in enumerate(("failed", "quarantined")): |
| 545 | index_terminal = _job_row( |
| 546 | repo_id, "mpack.index", |
| 547 | {"mpack_key": "sha256:" + hashlib.sha256(f"mwp4-07-{status}".encode()).hexdigest(), |
| 548 | "head": _fake_commit_id(f"h-{status}"), "branch": "main"}, |
| 549 | # Distinct created_at per row to avoid PK collision in compute_job_id |
| 550 | created_at=now - timedelta(seconds=10 + i), |
| 551 | status=status, |
| 552 | ) |
| 553 | db_session.add(index_terminal) |
| 554 | |
| 555 | prebuild = _job_row( |
| 556 | repo_id, "fetch.mpack.prebuild", |
| 557 | {"tip_commit_ids": [_fake_commit_id("mwp4-07-tip")]}, |
| 558 | created_at=now, |
| 559 | ) |
| 560 | db_session.add(prebuild) |
| 561 | await db_session.commit() |
| 562 | |
| 563 | claimed = await claim_next_job(db_session) |
| 564 | await db_session.commit() |
| 565 | |
| 566 | assert claimed is not None, ( |
| 567 | "fetch.mpack.prebuild must be claimable when all mpack.index rows are terminal" |
| 568 | ) |
| 569 | assert claimed.job_type == "fetch.mpack.prebuild", ( |
| 570 | f"Terminal mpack.index states (failed/quarantined) must not gate the " |
| 571 | f"prebuild. Got: {claimed.job_type}" |
| 572 | ) |
| 573 | |
| 574 | |
| 575 | @pytest.mark.asyncio |
| 576 | async def test_mwp4_08_barrier_is_per_repo( |
| 577 | db_session: AsyncSession, |
| 578 | ) -> None: |
| 579 | """MWP4_08: the barrier is scoped per repo_id. A pending mpack.index in |
| 580 | repo X must not block a fetch.mpack.prebuild in repo Y. |
| 581 | |
| 582 | Pre-fix: PASSES (no barrier means per-repo isolation is trivially true). |
| 583 | Post-fix: must still PASS — the correlated subquery joins on repo_id. |
| 584 | """ |
| 585 | repo_x = await create_repo(db_session, owner="gabriel") |
| 586 | repo_y = await create_repo(db_session, owner="gabriel") |
| 587 | repo_x_id = str(repo_x.repo_id) |
| 588 | repo_y_id = str(repo_y.repo_id) |
| 589 | |
| 590 | now = _now() |
| 591 | # Repo X: mpack.index pending (should block repo X's prebuild, not repo Y's) |
| 592 | index_x = _job_row( |
| 593 | repo_x_id, "mpack.index", |
| 594 | {"mpack_key": "sha256:" + hashlib.sha256(b"mwp4-08-x").hexdigest(), "head": _fake_commit_id("hx"), "branch": "main"}, |
| 595 | created_at=now - timedelta(seconds=5), |
| 596 | ) |
| 597 | # Repo Y: fetch.mpack.prebuild with no pending mpack.index — must be claimable |
| 598 | prebuild_y = _job_row( |
| 599 | repo_y_id, "fetch.mpack.prebuild", |
| 600 | {"tip_commit_ids": [_fake_commit_id("mwp4-08-y-tip")]}, |
| 601 | created_at=now, |
| 602 | ) |
| 603 | db_session.add(index_x) |
| 604 | db_session.add(prebuild_y) |
| 605 | await db_session.commit() |
| 606 | |
| 607 | claimed = await claim_next_job(db_session) |
| 608 | await db_session.commit() |
| 609 | |
| 610 | # index_x has an earlier created_at than prebuild_y; without the per-repo |
| 611 | # scope, index_x would be claimed. With the scope, index_x is also claimable |
| 612 | # (it's not a prebuild), so it gets claimed first — that's fine. The key |
| 613 | # assertion: claim must not return None (prebuild_y must not be spuriously |
| 614 | # blocked by repo X's index). |
| 615 | assert claimed is not None, ( |
| 616 | "A job must be claimable — repo Y's prebuild must not be blocked by " |
| 617 | "repo X's pending mpack.index" |
| 618 | ) |
| 619 | # Either index_x (earlier timestamp) or prebuild_y — both are acceptable. |
| 620 | # The important thing is that prebuild_y is NOT blocked by a different repo's index. |
| 621 | assert claimed.job_type in ("mpack.index", "fetch.mpack.prebuild"), ( |
| 622 | f"Unexpected job_type claimed: {claimed.job_type}" |
| 623 | ) |
| 624 | assert claimed.repo_id != repo_x_id or claimed.job_type == "mpack.index", ( |
| 625 | "If repo X's job is claimed it must be the index, not any blocked prebuild" |
| 626 | ) |
| 627 | |
| 628 | |
| 629 | @pytest.mark.asyncio |
| 630 | async def test_mwp4_09_barrier_does_not_affect_other_job_types( |
| 631 | db_session: AsyncSession, |
| 632 | ) -> None: |
| 633 | """MWP4_09: the barrier filters only fetch.mpack.prebuild. A pending |
| 634 | mpack.index is itself claimable, and other job types (e.g. intel.structural) |
| 635 | are unaffected by the barrier. |
| 636 | |
| 637 | Pre-fix: PASSES (no barrier — all types always claimable by timestamp). |
| 638 | Post-fix: must still PASS — the filter condition is |
| 639 | job_type == 'fetch.mpack.prebuild' AND EXISTS(pending index for same repo). |
| 640 | """ |
| 641 | repo = await create_repo(db_session, owner="gabriel") |
| 642 | repo_id = str(repo.repo_id) |
| 643 | |
| 644 | now = _now() |
| 645 | # A pending mpack.index — must still be claimable (barrier does not block index itself) |
| 646 | index_job = _job_row( |
| 647 | repo_id, "mpack.index", |
| 648 | {"mpack_key": "sha256:" + hashlib.sha256(b"mwp4-09").hexdigest(), "head": _fake_commit_id("h"), "branch": "main"}, |
| 649 | created_at=now - timedelta(seconds=10), |
| 650 | ) |
| 651 | # intel.structural — unrelated type, must be claimable regardless |
| 652 | intel_job = _job_row( |
| 653 | repo_id, "intel.structural", |
| 654 | {"head": _fake_commit_id("h2"), "branch": "main"}, |
| 655 | created_at=now - timedelta(seconds=5), |
| 656 | ) |
| 657 | db_session.add(index_job) |
| 658 | db_session.add(intel_job) |
| 659 | await db_session.commit() |
| 660 | |
| 661 | # Claim twice — both jobs must be claimable (neither is a blocked prebuild) |
| 662 | first = await claim_next_job(db_session) |
| 663 | await db_session.commit() |
| 664 | second = await claim_next_job(db_session) |
| 665 | await db_session.commit() |
| 666 | |
| 667 | assert first is not None, "First claim must succeed" |
| 668 | assert second is not None, "Second claim must succeed — barrier must not block non-prebuild types" |
| 669 | claimed_types = {first.job_type, second.job_type} |
| 670 | assert "mpack.index" in claimed_types, "mpack.index must be claimable" |
| 671 | assert "intel.structural" in claimed_types, "intel.structural must be claimable" |
| 672 | |
| 673 | |
| 674 | # --------------------------------------------------------------------------- |
| 675 | # Phase 3 — stop masking the index gap (MWP4_10) |
| 676 | # --------------------------------------------------------------------------- |
| 677 | |
| 678 | |
| 679 | @pytest.mark.asyncio |
| 680 | async def test_mwp4_10_fetch_not_indexed_error_propagates( |
| 681 | db_session: AsyncSession, |
| 682 | ) -> None: |
| 683 | """MWP4_10: when wire_fetch_mpack raises FetchNotIndexedError inside |
| 684 | process_fetch_mpack_prebuild_job, the handler must re-raise it instead of |
| 685 | returning a success result with tips_built=0. |
| 686 | |
| 687 | This is the Phase 3 narrowing of the broad except Exception at |
| 688 | musehub_wire_fetch.py:1164-1167. After the fix, FetchNotIndexedError is |
| 689 | caught specifically, logged, and re-raised so the worker's fail_job path |
| 690 | records it as a retryable failure. The broad except Exception still handles |
| 691 | genuinely unexpected errors — it must NOT catch FetchNotIndexedError. |
| 692 | |
| 693 | Setup: push one commit (creates MusehubCommit, MusehubBranch, MusehubMPackIndex), |
| 694 | then delete MusehubMPackIndex rows to simulate the state where a terminally- |
| 695 | failed mpack.index left the index gap (the Phase 2 barrier allows the prebuild |
| 696 | to run only after index work drains — terminal failures bypass the barrier). |
| 697 | |
| 698 | Pre-fix (Phase 0 behavior): FAILS — process_fetch_mpack_prebuild_job swallows |
| 699 | FetchNotIndexedError and returns tips_built=0 (no propagation). |
| 700 | |
| 701 | Post-fix (Phase 3): PASSES — FetchNotIndexedError propagates to the caller. |
| 702 | """ |
| 703 | repo = await create_repo(db_session, owner="gabriel") |
| 704 | repo_id = str(repo.repo_id) |
| 705 | |
| 706 | data = b"mwp4-10-blob-content" |
| 707 | oid = blob_id(data) |
| 708 | snap_id = hash_snapshot({"file.txt": oid}) |
| 709 | cid = _fake_commit_id("mwp4-10-c1") |
| 710 | |
| 711 | mpack_key = await _e2e_push(db_session, repo_id, cid, None, snap_id, "file.txt", oid, data) |
| 712 | |
| 713 | # Delete the inline MusehubMPackIndex rows to simulate an empty index |
| 714 | # (replicates the state after a terminally-failed mpack.index job). |
| 715 | await db_session.execute( |
| 716 | delete(MusehubMPackIndex).where(MusehubMPackIndex.mpack_id == mpack_key) |
| 717 | ) |
| 718 | await db_session.commit() |
| 719 | |
| 720 | prebuild_job = (await db_session.execute( |
| 721 | select(MusehubBackgroundJob).where( |
| 722 | MusehubBackgroundJob.repo_id == repo_id, |
| 723 | MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", |
| 724 | MusehubBackgroundJob.status == "pending", |
| 725 | ).limit(1) |
| 726 | )).scalar_one_or_none() |
| 727 | assert prebuild_job is not None, "A fetch.mpack.prebuild job must be enqueued after push" |
| 728 | |
| 729 | # Phase 3: FetchNotIndexedError must propagate — not swallowed. |
| 730 | # The worker calls fail_job on this exception, giving it retry chances. |
| 731 | with pytest.raises(FetchNotIndexedError): |
| 732 | await process_fetch_mpack_prebuild_job(db_session, prebuild_job.job_id) |
| 733 | await db_session.rollback() |
| 734 | |
| 735 | |
| 736 | # --------------------------------------------------------------------------- |
| 737 | # Phase 4 — integration + stress (real worker claim loop) |
| 738 | # --------------------------------------------------------------------------- |
| 739 | |
| 740 | |
| 741 | async def _drain_via_claim_loop( |
| 742 | session: AsyncSession, |
| 743 | repo_id: str, |
| 744 | max_iters: int = 60, |
| 745 | ) -> tuple[list[str], int]: |
| 746 | """Drive claim_next_job → dispatch → complete_job, mirroring _process_one. |
| 747 | |
| 748 | Returns: |
| 749 | claim_log: job_types in the order they were claimed |
| 750 | fetch_not_indexed_count: number of FetchNotIndexedError raised during drain |
| 751 | """ |
| 752 | claim_log: list[str] = [] |
| 753 | fetch_not_indexed_count = 0 |
| 754 | |
| 755 | for _ in range(max_iters): |
| 756 | job = await claim_next_job(session) |
| 757 | await session.commit() |
| 758 | if job is None: |
| 759 | break |
| 760 | |
| 761 | jid = job.job_id |
| 762 | jtype = job.job_type |
| 763 | claim_log.append(jtype) |
| 764 | |
| 765 | if jtype == "mpack.index": |
| 766 | await process_mpack_index_job(session, jid) |
| 767 | await session.commit() |
| 768 | elif jtype == "fetch.mpack.prebuild": |
| 769 | try: |
| 770 | await process_fetch_mpack_prebuild_job(session, jid) |
| 771 | await session.commit() |
| 772 | except FetchNotIndexedError: |
| 773 | fetch_not_indexed_count += 1 |
| 774 | await session.rollback() |
| 775 | # all other types: pass (disabled in the worker) |
| 776 | |
| 777 | await complete_job(session, jid) |
| 778 | await session.commit() |
| 779 | |
| 780 | return claim_log, fetch_not_indexed_count |
| 781 | |
| 782 | |
| 783 | @pytest.mark.asyncio |
| 784 | async def test_mwp4_11_e2e_claim_loop_orders_index_before_prebuild( |
| 785 | db_session: AsyncSession, |
| 786 | ) -> None: |
| 787 | """MWP4_11 E2E: one push then real claim loop — mpack.index claimed and |
| 788 | completed before fetch.mpack.prebuild is ever claimed. A |
| 789 | MusehubFetchMPackCache row must land at the pushed tip after drain. |
| 790 | |
| 791 | This mirrors the exact sequence musehub/worker.py::_process_one runs: |
| 792 | claim_next_job → dispatch → complete_job, repeat until queue empty. |
| 793 | |
| 794 | The Phase 2 claim-layer barrier is the mechanism that enforces the ordering. |
| 795 | Without it, PostgreSQL is free to return the prebuild first (RC-4). |
| 796 | |
| 797 | Assertions: |
| 798 | - mpack.index appears in the claim log before any fetch.mpack.prebuild |
| 799 | - A MusehubFetchMPackCache row exists for the pushed tip commit |
| 800 | """ |
| 801 | repo = await create_repo(db_session, owner="gabriel") |
| 802 | repo_id = str(repo.repo_id) |
| 803 | |
| 804 | data = b"mwp4-11-blob-content" |
| 805 | oid = blob_id(data) |
| 806 | snap_id = hash_snapshot({"file.txt": oid}) |
| 807 | cid = _fake_commit_id("mwp4-11-c1") |
| 808 | |
| 809 | await _e2e_push(db_session, repo_id, cid, None, snap_id, "file.txt", oid, data) |
| 810 | |
| 811 | claim_log, fetch_not_indexed_count = await _drain_via_claim_loop(db_session, repo_id) |
| 812 | |
| 813 | assert "mpack.index" in claim_log, ( |
| 814 | "mpack.index must be claimed during the drain" |
| 815 | ) |
| 816 | assert "fetch.mpack.prebuild" in claim_log, ( |
| 817 | "fetch.mpack.prebuild must be claimed during the drain" |
| 818 | ) |
| 819 | |
| 820 | first_prebuild_pos = claim_log.index("fetch.mpack.prebuild") |
| 821 | last_index_pos = max(i for i, t in enumerate(claim_log) if t == "mpack.index") |
| 822 | assert last_index_pos < first_prebuild_pos, ( |
| 823 | f"Every mpack.index claim must complete before the first fetch.mpack.prebuild " |
| 824 | f"is claimed. claim_log={claim_log}" |
| 825 | ) |
| 826 | |
| 827 | assert fetch_not_indexed_count == 0, ( |
| 828 | f"Zero FetchNotIndexedError must occur — the barrier ensures the prebuild " |
| 829 | f"only runs after the index drains. Got {fetch_not_indexed_count} error(s)." |
| 830 | ) |
| 831 | |
| 832 | cache_count = (await db_session.execute( |
| 833 | select(func.count()).select_from(MusehubFetchMPackCache).where( |
| 834 | MusehubFetchMPackCache.repo_id == repo_id, |
| 835 | MusehubFetchMPackCache.tip_commit_id == cid, |
| 836 | ) |
| 837 | )).scalar_one() |
| 838 | assert cache_count >= 1, ( |
| 839 | f"MusehubFetchMPackCache must have at least one row for tip={cid[:20]} " |
| 840 | f"after the prebuild runs. Got {cache_count} rows." |
| 841 | ) |
| 842 | |
| 843 | |
| 844 | @pytest.mark.asyncio |
| 845 | async def test_mwp4_12_stress_n10_pushes_drain_with_no_ordering_violation( |
| 846 | db_session: AsyncSession, |
| 847 | ) -> None: |
| 848 | """MWP4_12 stress: N=10 back-to-back pushes with no intermediate drain, then |
| 849 | claim-loop drain. Three assertions mirror the musehub#109 acceptance criteria: |
| 850 | |
| 851 | (a) Every fetch.mpack.prebuild claim happens only when zero mpack.index jobs |
| 852 | are pending or running for the same repo. |
| 853 | (b) Zero FetchNotIndexedError raised during the entire drain. |
| 854 | (c) Final clone cache covers the latest tip: all cache rows point to the same |
| 855 | mpack_id (len == 1) and a row exists for the latest pushed commit. |
| 856 | |
| 857 | N=10 creates N distinct mpack.index jobs (per-key dedup allows them all) and |
| 858 | a single coalesced fetch.mpack.prebuild (coarse dedup folds them to one row). |
| 859 | The MWP-3 self-coalescing fix makes the prebuild re-read live branch tips at |
| 860 | run time, so it always targets the latest tip regardless of which push enqueued |
| 861 | it. RC-4 closes the remaining gap: the prebuild cannot be claimed until every |
| 862 | mpack.index for the repo has drained. |
| 863 | """ |
| 864 | repo = await create_repo(db_session, owner="gabriel") |
| 865 | repo_id = str(repo.repo_id) |
| 866 | |
| 867 | N = 10 |
| 868 | cids = [_fake_commit_id(f"mwp4-12-c{i}") for i in range(N)] |
| 869 | |
| 870 | for i in range(N): |
| 871 | data = f"mwp4-12-blob-{i}-unique-content".encode() |
| 872 | oid = blob_id(data) |
| 873 | snap_id = hash_snapshot({f"file_{i}.txt": oid}) |
| 874 | parent = cids[i - 1] if i > 0 else None |
| 875 | await _e2e_push( |
| 876 | db_session, repo_id, cids[i], parent, |
| 877 | snap_id, f"file_{i}.txt", oid, data, |
| 878 | ) |
| 879 | |
| 880 | # Verify N mpack.index jobs and exactly 1 fetch.mpack.prebuild were enqueued. |
| 881 | index_pending = (await db_session.execute( |
| 882 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 883 | MusehubBackgroundJob.repo_id == repo_id, |
| 884 | MusehubBackgroundJob.job_type == "mpack.index", |
| 885 | MusehubBackgroundJob.status == "pending", |
| 886 | ) |
| 887 | )).scalar_one() |
| 888 | assert index_pending == N, ( |
| 889 | f"Expected {N} pending mpack.index jobs after {N} pushes, got {index_pending}. " |
| 890 | f"Per-key dedup (MWP-3) must allow each distinct mpack_key through." |
| 891 | ) |
| 892 | |
| 893 | prebuild_pending = (await db_session.execute( |
| 894 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 895 | MusehubBackgroundJob.repo_id == repo_id, |
| 896 | MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", |
| 897 | MusehubBackgroundJob.status == "pending", |
| 898 | ) |
| 899 | )).scalar_one() |
| 900 | assert prebuild_pending == 1, ( |
| 901 | f"Coarse dedup must collapse {N} prebuild enqueues to 1 pending row. " |
| 902 | f"Got {prebuild_pending}." |
| 903 | ) |
| 904 | |
| 905 | # Drive the claim loop and check barrier invariant at each prebuild claim. |
| 906 | prebuild_barrier_violations: list[int] = [] |
| 907 | |
| 908 | for _ in range(N + 30): # N indexes + 1 prebuild + ~20 intel.* subtypes |
| 909 | job = await claim_next_job(db_session) |
| 910 | await db_session.commit() |
| 911 | if job is None: |
| 912 | break |
| 913 | |
| 914 | jid = job.job_id |
| 915 | jtype = job.job_type |
| 916 | |
| 917 | if jtype == "fetch.mpack.prebuild": |
| 918 | # (a) At the moment of claim, no mpack.index must be pending/running |
| 919 | # for this repo. claim_next_job committed above — DB state is stable. |
| 920 | blocking = (await db_session.execute( |
| 921 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 922 | MusehubBackgroundJob.repo_id == repo_id, |
| 923 | MusehubBackgroundJob.job_type == "mpack.index", |
| 924 | MusehubBackgroundJob.status.in_(("pending", "running")), |
| 925 | ) |
| 926 | )).scalar_one() |
| 927 | if blocking > 0: |
| 928 | prebuild_barrier_violations.append(blocking) |
| 929 | |
| 930 | try: |
| 931 | await process_fetch_mpack_prebuild_job(db_session, jid) |
| 932 | await db_session.commit() |
| 933 | except FetchNotIndexedError: |
| 934 | await db_session.rollback() |
| 935 | |
| 936 | elif jtype == "mpack.index": |
| 937 | await process_mpack_index_job(db_session, jid) |
| 938 | await db_session.commit() |
| 939 | |
| 940 | # all other types: pass (mirrors the worker's disabled-jobs path) |
| 941 | |
| 942 | await complete_job(db_session, jid) |
| 943 | await db_session.commit() |
| 944 | |
| 945 | # (a) No prebuild was ever claimed while mpack.index was pending/running |
| 946 | assert not prebuild_barrier_violations, ( |
| 947 | f"The RC-4 barrier was violated: fetch.mpack.prebuild was claimed while " |
| 948 | f"{prebuild_barrier_violations} mpack.index job(s) were still pending/running. " |
| 949 | f"The Phase 2 correlated NOT EXISTS barrier must prevent this." |
| 950 | ) |
| 951 | |
| 952 | # (b) Zero FetchNotIndexedError raised during drain |
| 953 | # Re-verify by checking all prebuild jobs are 'done' (not 'failed') |
| 954 | failed_prebuild = (await db_session.execute( |
| 955 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 956 | MusehubBackgroundJob.repo_id == repo_id, |
| 957 | MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", |
| 958 | MusehubBackgroundJob.status == "failed", |
| 959 | ) |
| 960 | )).scalar_one() |
| 961 | assert failed_prebuild == 0, ( |
| 962 | f"{failed_prebuild} fetch.mpack.prebuild job(s) ended in 'failed' state. " |
| 963 | f"The barrier must ensure the prebuild runs only after all index work drains, " |
| 964 | f"producing zero FetchNotIndexedError occurrences." |
| 965 | ) |
| 966 | |
| 967 | # (c) Final clone cache covers the latest tip |
| 968 | latest_tip = cids[-1] |
| 969 | cache_for_tip = (await db_session.execute( |
| 970 | select(func.count()).select_from(MusehubFetchMPackCache).where( |
| 971 | MusehubFetchMPackCache.repo_id == repo_id, |
| 972 | MusehubFetchMPackCache.tip_commit_id == latest_tip, |
| 973 | ) |
| 974 | )).scalar_one() |
| 975 | assert cache_for_tip >= 1, ( |
| 976 | f"MusehubFetchMPackCache must have a row for the latest tip={latest_tip[:20]} " |
| 977 | f"after a full N={N} drain. Got {cache_for_tip} rows." |
| 978 | ) |
| 979 | |
| 980 | all_mpack_ids = (await db_session.execute( |
| 981 | select(MusehubFetchMPackCache.mpack_id).where( |
| 982 | MusehubFetchMPackCache.repo_id == repo_id, |
| 983 | ) |
| 984 | )).scalars().all() |
| 985 | assert len(all_mpack_ids) >= 1, "At least one cache row must exist after full drain" |
| 986 | unique_mpacks = set(all_mpack_ids) |
| 987 | assert len(unique_mpacks) == 1, ( |
| 988 | f"All MusehubFetchMPackCache rows for the repo must point to the same mpack_id " |
| 989 | f"after a full drain (single combined mpack covers all tips). " |
| 990 | f"Got {len(unique_mpacks)} distinct mpack_ids." |
| 991 | ) |