"""MWP-4 — Gate fetch.mpack.prebuild behind mpack.index (fixes RC-4). Sub-ticket: musehub#109 — https://staging.musehub.ai/gabriel/musehub/issues/109 Master: muse#58 — https://staging.musehub.ai/gabriel/muse/issues/58 Predecessors: musehub#106 (MWP-1, generation authority) — closed. musehub#107 (MWP-2, walk fallback) — closed. musehub#108 (MWP-3, job-enqueue idempotency) — closed. Root cause (RC-4): ``claim_next_job`` orders only by ``created_at`` with no tiebreaker. When ``mpack.index`` and ``fetch.mpack.prebuild`` share the same ``created_at`` (the normal case from ``enqueue_push_intel``), or the prebuild lands with an earlier timestamp due to cross-push interleaving, PostgreSQL may return the prebuild first. The prebuild then calls ``wire_fetch_mpack`` with ``force_build=True``, hits the index-coverage check at ``musehub_wire_fetch.py:681–694``, and raises ``FetchNotIndexedError`` because ``MusehubMPackIndex`` is still empty. The broad ``except Exception`` at ``musehub_wire_fetch.py:1164–1167`` swallows the error silently: ``tips_built`` stays ``0``, no ``MusehubFetchMPackCache`` row is written, and the next clone returns 503. Phase 0 — RED reproduction (no production code changes): MWP4_01 RED — ``claim_next_job`` returns the prebuild before the index when the prebuild has an earlier ``created_at``. Asserts the desired correct behavior → FAILS before Phase 2 fix. MWP4_02 — handler-level proof: prebuild run before index swallows FetchNotIndexedError and writes no cache row. Asserts the broken outcome → PASSES before fix (proves the bug). Must be updated in Phase 3 when the swallow is narrowed. """ from __future__ import annotations import hashlib from datetime import datetime, timedelta, timezone import pytest from sqlalchemy import delete, func, select from sqlalchemy.ext.asyncio import AsyncSession from muse.core.ids import hash_snapshot from muse.core.mpack import build_wire_mpack from muse.core.types import blob_id from musehub.core.genesis import compute_job_id from musehub.db.musehub_jobs_models import MusehubBackgroundJob from musehub.db.musehub_repo_models import MusehubFetchMPackCache, MusehubMPackIndex from musehub.services.musehub_jobs import claim_next_job from musehub.services.musehub_wire_fetch import process_fetch_mpack_prebuild_job from musehub.services.musehub_wire_push import wire_push_unpack_mpack from tests.factories import create_repo # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _fake_commit_id(seed: str) -> str: return "sha256:" + hashlib.sha256(seed.encode()).hexdigest() def _now() -> datetime: return datetime.now(tz=timezone.utc) def _job_row( repo_id: str, job_type: str, payload: dict, created_at: datetime, status: str = "pending", ) -> MusehubBackgroundJob: """Build a MusehubBackgroundJob row ready to be added to a session.""" job_id = compute_job_id(repo_id, job_type, created_at.isoformat()) return MusehubBackgroundJob( job_id=job_id, repo_id=repo_id, job_type=job_type, payload=payload, status=status, created_at=created_at, attempt=0, ) _E2E_OWNER = "gabriel" def _e2e_raw_commit( cid: str, parent: str | None, snap_id: str, branch: str = "main", ) -> dict: return { "commit_id": cid, "branch": branch, "message": f"commit {cid[:12]}", "author": _E2E_OWNER, "committed_at": _now().isoformat(), "parent_commit_id": parent, "parent2_commit_id": None, "snapshot_id": snap_id, "agent_id": "", "model_id": "", "toolchain_id": "", "sem_ver_bump": "none", "breaking_changes": [], "signature": "", "signer_key_id": "", "signer_public_key": "", "prompt_hash": "", } async def _e2e_push( session: AsyncSession, repo_id: str, cid: str, parent: str | None, snap_id: str, fpath: str, oid: str, data: bytes, branch: str = "main", ) -> str: """Build a single-commit mpack, store it in the conftest MemoryBackend, push it. The conftest autouse fixture patches ``musehub.storage.backends.get_backend`` to a ``MemoryBackend`` instance shared across all modules. Pre-seeding ``backend._mpacks[key]`` is equivalent to a MinIO PUT — both ``wire_push_unpack_mpack`` and ``process_mpack_index_job`` call ``get_backend()`` and see the same bytes. Returns the ``mpack_key`` so callers can locate the enqueued index job. """ import musehub.storage.backends as _b_mod backend = _b_mod.get_backend() mpack_bytes = build_wire_mpack({ "commits": [_e2e_raw_commit(cid, parent, snap_id, branch=branch)], "snapshots": [{ "snapshot_id": snap_id, "parent_snapshot_id": None, "delta_upsert": {fpath: oid}, "delta_remove": [], "directories": [], }], "blobs": [{"object_id": oid, "content": data}], "tags": [], }) mpack_key = blob_id(mpack_bytes) backend._mpacks[mpack_key] = mpack_bytes await wire_push_unpack_mpack( session, repo_id, mpack_key, pusher_id=_E2E_OWNER, branch=branch, head_commit_id=cid, commits_count=1, blobs_count=1, force=True, ) await session.commit() return mpack_key # --------------------------------------------------------------------------- # MWP4_01 — claim_next_job returns prebuild before a pending index (RED) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp4_01_claim_returns_prebuild_before_index( db_session: AsyncSession, ) -> None: """MWP4_01 RED: claim_next_job must NOT return fetch.mpack.prebuild while an mpack.index for the same repo is pending. Setup: insert a fetch.mpack.prebuild with created_at = now-5s (earlier than the mpack.index at now). This simulates the cross-push race where push A's prebuild is enqueued before push B's index job arrives. Pre-fix (expected to FAIL): claim_next_job orders solely by created_at; the prebuild (earlier timestamp) is returned ahead of the index. The assertion ``claimed.job_type == "mpack.index"`` therefore fails. Post-fix (Phase 2 — correlated NOT EXISTS barrier): The barrier prevents fetch.mpack.prebuild from being claimed while any mpack.index for the same repo_id is pending or running. The index is returned instead, and the prebuild waits until the index drains. """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) now = _now() earlier = now - timedelta(seconds=5) # Insert prebuild FIRST with an earlier timestamp to force it ahead in the # ORDER BY created_at ordering — this is the race we're fixing. prebuild = _job_row( repo_id, "fetch.mpack.prebuild", {"tip_commit_ids": [_fake_commit_id("mwp4-01-tip")]}, created_at=earlier, ) index = _job_row( repo_id, "mpack.index", { "mpack_key": "sha256:" + hashlib.sha256(b"mwp4-01-mpack").hexdigest(), "head": _fake_commit_id("mwp4-01-head"), "branch": "main", }, created_at=now, ) db_session.add(prebuild) db_session.add(index) await db_session.commit() # Claim the next job. # Pre-fix: returns the prebuild (earliest created_at wins) → assertion fails. # Post-fix: barrier skips the prebuild; returns the index instead. claimed = await claim_next_job(db_session) await db_session.commit() assert claimed is not None, "Expected a job to be claimed — queue has two pending rows." assert claimed.job_type == "mpack.index", ( f"claim_next_job must skip fetch.mpack.prebuild while mpack.index is " f"pending for the same repo_id, but returned '{claimed.job_type}'. " f"The prebuild must be gated behind all pending index jobs (RC-4 barrier)." ) # --------------------------------------------------------------------------- # MWP4_02 — prebuild run before index swallows FetchNotIndexedError (bug proof) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_mwp4_02_prebuild_swallows_fetch_not_indexed_error( db_session: AsyncSession, ) -> None: """MWP4_02 (Phase 0 bug documentation): process_fetch_mpack_prebuild_job run before mpack.index is processed swallows FetchNotIndexedError and writes no MusehubFetchMPackCache row — the broken silent-no-cache outcome of RC-4. Setup: push one commit so MusehubObject, MusehubCommit, and MusehubBranch rows exist. wire_push_unpack_mpack writes MusehubMPackIndex rows at step 7d synchronously, so we explicitly delete them after the push to simulate the race state: DB has commits/snapshots but MusehubMPackIndex is empty (as if the mpack.index job hasn't run yet). Inside process_fetch_mpack_prebuild_job: wire_fetch_mpack is called with force_build=True. It assembles the object set from snapshots, checks index coverage at lines 681–694 of musehub_wire_fetch.py, finds every object missing from MusehubMPackIndex, and raises FetchNotIndexedError. The broad ``except Exception`` at lines 1164–1167 catches and logs it; the function returns a result dict with tips_built=0 and no cache row. Pre-fix: this test PASSES — tips_built==0 and cache_count==0 confirm the bug is present (FetchNotIndexedError was silently swallowed). Post-fix (Phase 3 — narrow the swallow): process_fetch_mpack_prebuild_job will re-raise FetchNotIndexedError instead of returning quietly. Update this test at that point to use ``pytest.raises(FetchNotIndexedError)`` rather than asserting tips_built. With Phase 2's barrier in place, this path is never hit in production — the prebuild only runs after the index has completed. """ repo = await create_repo(db_session, owner="gabriel") repo_id = str(repo.repo_id) data = b"mwp4-02-blob-content" oid = blob_id(data) snap_id = hash_snapshot({"file.txt": oid}) cid = _fake_commit_id("mwp4-02-c1") # Push one commit — creates MusehubObject, MusehubCommit, MusehubBranch, # enqueues mpack.index + fetch.mpack.prebuild, AND writes MusehubMPackIndex # rows at step 7d of wire_push_unpack_mpack (synchronous inline indexing). mpack_key = await _e2e_push(db_session, repo_id, cid, None, snap_id, "file.txt", oid, data) # Simulate the RC-4 race: delete the inline MusehubMPackIndex rows that the # push wrote. This replicates the state where the prebuild job is claimed # before the mpack.index job has had a chance to run — MusehubCommit, # MusehubSnapshot, and MusehubBranch are populated, but MusehubMPackIndex # is empty for this mpack_key. await db_session.execute( delete(MusehubMPackIndex).where(MusehubMPackIndex.mpack_id == mpack_key) ) await db_session.commit() # Locate the prebuild job enqueued by wire_push_unpack_mpack. prebuild_job = (await db_session.execute( select(MusehubBackgroundJob).where( MusehubBackgroundJob.repo_id == repo_id, MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", MusehubBackgroundJob.status == "pending", ).limit(1) )).scalar_one_or_none() assert prebuild_job is not None, ( "wire_push_unpack_mpack must enqueue a fetch.mpack.prebuild job after a push." ) # Run the prebuild handler with MusehubMPackIndex cleared (simulating RC-4 race). # wire_fetch_mpack raises FetchNotIndexedError at musehub_wire_fetch.py:694; # the broad except at lines 1164–1167 swallows it and returns tips_built=0. result = await process_fetch_mpack_prebuild_job(db_session, prebuild_job.job_id) await db_session.commit() # --- Bug proof #1: FetchNotIndexedError was swallowed (not propagated) --- # If this assertion fails, the swallow has already been narrowed (Phase 3 done). assert result["tips_built"] == 0, ( f"tips_built must be 0 when MusehubMPackIndex is empty (FetchNotIndexedError " f"swallowed by broad except). Got tips_built={result['tips_built']}. " f"If this fails, Phase 3 is already applied — update to pytest.raises()." ) assert result["tips_requested"] > 0, ( f"tips_requested must be > 0 (branch tip exists); got " f"{result['tips_requested']}. Check that wire_push_unpack_mpack created a " f"MusehubBranch row pointing to the pushed commit." ) # --- Bug proof #2: no cache row written — clone would 503 --- cache_count = (await db_session.execute( select(func.count()).select_from(MusehubFetchMPackCache).where( MusehubFetchMPackCache.repo_id == repo_id, ) )).scalar() assert cache_count == 0, ( f"No MusehubFetchMPackCache row must exist when the prebuild swallowed " f"FetchNotIndexedError (the broken silent-no-cache outcome). " f"Found {cache_count} row(s). If this fails, the bug is fixed — " f"update to assert cache_count == 1." )