test_mwp3_job_idempotency.py
python
sha256:0032d6cfa33bc3c8367436ad768e7dd0e339b4332153160247da8266cb5fa352
Merge branch 'task/version-tags-phase3-server' into dev
Human
17 days ago
| 1 | """MWP-3 Phase 0 (RED) — Job-enqueue idempotency drops fresh payloads on back-to-back pushes. |
| 2 | |
| 3 | Sub-ticket: musehub#108 — https://staging.musehub.ai/gabriel/musehub/issues/108 |
| 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 | |
| 9 | Root cause (RC-3): ``enqueue_push_intel`` deduplicates pending jobs on the |
| 10 | coarse key ``(repo_id, job_type, status='pending')``. This assumption — that |
| 11 | any pending job of a given type covers all future work of that type — is false |
| 12 | for two of the three live job types: |
| 13 | |
| 14 | Face 1 — ``mpack.index``: each pushed mpack has a distinct ``mpack_key`` |
| 15 | (built as ``mpack_payload`` at musehub_jobs.py:128). If push A's index |
| 16 | job is still pending when push B lands, B's job is silently dropped by the |
| 17 | coarse dedup. Push B's mpack is never indexed. |
| 18 | |
| 19 | Face 2 — ``fetch.mpack.prebuild``: the payload captures branch tips *at |
| 20 | enqueue time* (musehub_jobs.py:131-136). The handler |
| 21 | ``process_fetch_mpack_prebuild_job`` trusts that stale list verbatim |
| 22 | (musehub_wire_fetch.py:1067). If push B advances a branch while A's |
| 23 | prebuild is still pending, the single deduped job carries push A's stale |
| 24 | tip set, and push B's new tip is never prebuilt. |
| 25 | |
| 26 | Worker scope reminder: only ``mpack.index`` and ``fetch.mpack.prebuild`` are |
| 27 | live today (musehub/worker.py:107-108: ``if job_type not in (...): pass``). |
| 28 | RC-3's entire blast radius is these two types — both gate clone correctness. |
| 29 | |
| 30 | Expected Phase 0 status: |
| 31 | MWP3_01 RED — two enqueue_push_intel calls with distinct mpack_keys → |
| 32 | coarse dedup drops the second; only one mpack.index exists. |
| 33 | MWP3_02 RED — prebuild handler runs with stale payload; live branch tip |
| 34 | tipB is absent from the wire_fetch_mpack want= list. |
| 35 | MWP3_03 GREEN — single enqueue_push_intel call → exactly one mpack.index |
| 36 | and one fetch.mpack.prebuild pending (no duplicate explosion). |
| 37 | |
| 38 | No production code is changed in Phase 0. |
| 39 | """ |
| 40 | |
| 41 | from __future__ import annotations |
| 42 | |
| 43 | import hashlib |
| 44 | from datetime import datetime, timezone |
| 45 | from unittest.mock import AsyncMock, patch |
| 46 | |
| 47 | import pytest |
| 48 | from sqlalchemy import func, select |
| 49 | from sqlalchemy.ext.asyncio import AsyncSession |
| 50 | |
| 51 | from muse.core.ids import hash_snapshot |
| 52 | from muse.core.mpack import build_wire_mpack |
| 53 | from muse.core.types import blob_id |
| 54 | from musehub.core.genesis import compute_job_id |
| 55 | from musehub.db.musehub_jobs_models import MusehubBackgroundJob |
| 56 | from musehub.db.musehub_repo_models import MusehubFetchMPackCache, MusehubMPackIndex |
| 57 | from musehub.services.musehub_jobs import complete_job, enqueue_push_intel |
| 58 | from musehub.services.musehub_wire_fetch import process_fetch_mpack_prebuild_job |
| 59 | from musehub.services.musehub_wire_push import process_mpack_index_job, wire_push_unpack_mpack |
| 60 | from tests.factories import create_branch, create_repo |
| 61 | |
| 62 | |
| 63 | # --------------------------------------------------------------------------- |
| 64 | # Helpers |
| 65 | # --------------------------------------------------------------------------- |
| 66 | |
| 67 | |
| 68 | def _fake_commit_id(seed: str) -> str: |
| 69 | return "sha256:" + hashlib.sha256(seed.encode()).hexdigest() |
| 70 | |
| 71 | |
| 72 | def _fake_mpack_key(seed: str) -> str: |
| 73 | return "sha256:" + hashlib.sha256(f"mpack-{seed}".encode()).hexdigest() |
| 74 | |
| 75 | |
| 76 | def _now() -> datetime: |
| 77 | return datetime.now(tz=timezone.utc) |
| 78 | |
| 79 | |
| 80 | async def _insert_prebuild_job( |
| 81 | session: AsyncSession, |
| 82 | repo_id: str, |
| 83 | tip_commit_ids: list[str], |
| 84 | ) -> str: |
| 85 | """Insert a fetch.mpack.prebuild job directly with a given (possibly stale) tip set. |
| 86 | |
| 87 | Used in MWP3_02 to simulate a job whose payload was captured at push A time, |
| 88 | before push B advanced a branch. |
| 89 | """ |
| 90 | now = _now() |
| 91 | job_id = compute_job_id(repo_id, "fetch.mpack.prebuild", now.isoformat()) |
| 92 | session.add(MusehubBackgroundJob( |
| 93 | job_id=job_id, |
| 94 | repo_id=repo_id, |
| 95 | job_type="fetch.mpack.prebuild", |
| 96 | payload={"tip_commit_ids": tip_commit_ids}, |
| 97 | status="pending", |
| 98 | created_at=now, |
| 99 | attempt=0, |
| 100 | )) |
| 101 | await session.commit() |
| 102 | return job_id |
| 103 | |
| 104 | |
| 105 | # --------------------------------------------------------------------------- |
| 106 | # MWP3_01 — Face 1: coarse dedup silently drops push B's mpack.index (RED) |
| 107 | # --------------------------------------------------------------------------- |
| 108 | |
| 109 | |
| 110 | @pytest.mark.asyncio |
| 111 | async def test_mwp3_01_distinct_mpack_keys_both_enqueued(db_session: AsyncSession) -> None: |
| 112 | """MWP3_01 RED: two back-to-back pushes with distinct mpack_keys must each |
| 113 | produce a pending mpack.index job. |
| 114 | |
| 115 | Simulation: push A lands and enqueues mpack.index with mpack_key=mpack_a. |
| 116 | The job is committed (pending). Push B lands before the worker drains — |
| 117 | enqueue_push_intel is called again with mpack_key=mpack_b. |
| 118 | |
| 119 | Current (buggy) behavior: the coarse SELECT finds mpack.index already |
| 120 | pending for this repo_id and skips the insert. Only mpack_a is indexed; |
| 121 | mpack_b is never indexed, leaving push B's blobs unreachable. |
| 122 | |
| 123 | After fix (Phase 2): enqueue_push_intel deduplicates mpack.index on |
| 124 | (repo_id, mpack.index, payload->>'mpack_key', pending), so distinct |
| 125 | mpack_keys always produce two rows. |
| 126 | """ |
| 127 | repo = await create_repo(db_session, owner="gabriel") |
| 128 | repo_id = str(repo.repo_id) |
| 129 | |
| 130 | tip_a = _fake_commit_id("mwp3-01-tip-a") |
| 131 | tip_b = _fake_commit_id("mwp3-01-tip-b") |
| 132 | mpack_a = _fake_mpack_key("mwp3-01-a") |
| 133 | mpack_b = _fake_mpack_key("mwp3-01-b") |
| 134 | |
| 135 | # Push A: create branch main→tipA, enqueue push intel for push A. |
| 136 | await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) |
| 137 | await enqueue_push_intel(db_session, repo_id, head=tip_a, mpack_key=mpack_a) |
| 138 | await db_session.commit() |
| 139 | |
| 140 | # Push B arrives before the worker drains push A's jobs. |
| 141 | # Its mpack_key is distinct — this is separate, non-substitutable index work. |
| 142 | await enqueue_push_intel(db_session, repo_id, head=tip_b, mpack_key=mpack_b) |
| 143 | await db_session.commit() |
| 144 | |
| 145 | # Both mpack_keys must have a pending mpack.index job. |
| 146 | pending_index_rows = (await db_session.execute( |
| 147 | select(MusehubBackgroundJob.job_id, MusehubBackgroundJob.payload).where( |
| 148 | MusehubBackgroundJob.repo_id == repo_id, |
| 149 | MusehubBackgroundJob.job_type == "mpack.index", |
| 150 | MusehubBackgroundJob.status == "pending", |
| 151 | ) |
| 152 | )).all() |
| 153 | |
| 154 | mpack_keys_in_db = [row[1].get("mpack_key") for row in pending_index_rows] |
| 155 | |
| 156 | assert len(pending_index_rows) == 2, ( |
| 157 | f"Expected 2 pending mpack.index jobs (one per distinct mpack_key), " |
| 158 | f"got {len(pending_index_rows)}. " |
| 159 | f"mpack_keys found in DB: {mpack_keys_in_db}. " |
| 160 | f"Root cause: coarse (repo_id, job_type, pending) dedup in " |
| 161 | f"enqueue_push_intel (musehub_jobs.py:158-164) silently dropped push " |
| 162 | f"B's mpack.index. Fix: dedup mpack.index on payload->>'mpack_key' " |
| 163 | f"instead of job_type alone." |
| 164 | ) |
| 165 | assert mpack_a in mpack_keys_in_db, ( |
| 166 | f"Push A mpack_key={mpack_a!r} missing from DB. Got: {mpack_keys_in_db}" |
| 167 | ) |
| 168 | assert mpack_b in mpack_keys_in_db, ( |
| 169 | f"Push B mpack_key={mpack_b!r} was dropped by coarse dedup. " |
| 170 | f"Got: {mpack_keys_in_db}" |
| 171 | ) |
| 172 | |
| 173 | |
| 174 | # --------------------------------------------------------------------------- |
| 175 | # MWP3_02 — Face 2: prebuild handler trusts stale payload tip set (RED) |
| 176 | # --------------------------------------------------------------------------- |
| 177 | |
| 178 | |
| 179 | @pytest.mark.asyncio |
| 180 | async def test_mwp3_02_prebuild_handler_uses_stale_payload_tips(db_session: AsyncSession) -> None: |
| 181 | """MWP3_02 RED: process_fetch_mpack_prebuild_job trusts payload["tip_commit_ids"] |
| 182 | verbatim, missing branch tips that advanced after the job was enqueued. |
| 183 | |
| 184 | Scenario: |
| 185 | - Branch 'main' points to tipA at enqueue time → payload captures {tipA}. |
| 186 | - Push B adds branch 'dev' pointing to tipB; push B is deduped (no new |
| 187 | prebuild job). |
| 188 | - The single pending prebuild job has payload={"tip_commit_ids": [tipA]}. |
| 189 | - MusehubBranch now has both tipA (main) and tipB (dev). |
| 190 | - Handler runs → reads only from payload → calls wire_fetch_mpack with |
| 191 | want=[tipA]; tipB is silently omitted from the prebuild. |
| 192 | |
| 193 | After fix (Phase 1): the handler queries MusehubBranch.head_commit_id at |
| 194 | run time, getting {tipA, tipB}, and passes both to wire_fetch_mpack. |
| 195 | The pending job becomes self-coalescing: it always reflects live state. |
| 196 | """ |
| 197 | repo = await create_repo(db_session, owner="gabriel") |
| 198 | repo_id = str(repo.repo_id) |
| 199 | |
| 200 | tip_a = _fake_commit_id("mwp3-02-tip-a") |
| 201 | tip_b = _fake_commit_id("mwp3-02-tip-b") |
| 202 | |
| 203 | # Live DB state at handler run time: two branches, both tips present. |
| 204 | # This reflects the situation after push B has committed its branch advance. |
| 205 | await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) |
| 206 | await create_branch(db_session, repo_id, name="dev", head_commit_id=tip_b) |
| 207 | |
| 208 | # The pending prebuild job carries a STALE payload — only tipA was known at |
| 209 | # push A's enqueue time. Push B was deduped; tipB is absent from the payload. |
| 210 | job_id = await _insert_prebuild_job(db_session, repo_id, tip_commit_ids=[tip_a]) |
| 211 | |
| 212 | fake_mpack_id = "sha256:" + hashlib.sha256(b"mwp3-02-combined-mpack").hexdigest() |
| 213 | |
| 214 | with patch( |
| 215 | "musehub.services.musehub_wire_fetch.wire_fetch_mpack", |
| 216 | new_callable=AsyncMock, |
| 217 | return_value={ |
| 218 | "mpack_id": fake_mpack_id, |
| 219 | "mpack_url": "https://r2.example/combined", |
| 220 | "commit_count": 2, |
| 221 | "blob_count": 4, |
| 222 | }, |
| 223 | ) as mock_build: |
| 224 | await process_fetch_mpack_prebuild_job(db_session, job_id) |
| 225 | await db_session.commit() |
| 226 | |
| 227 | assert mock_build.call_count == 1, ( |
| 228 | f"Expected wire_fetch_mpack to be called exactly once; " |
| 229 | f"got call_count={mock_build.call_count}" |
| 230 | ) |
| 231 | |
| 232 | # wire_fetch_mpack is called as: |
| 233 | # wire_fetch_mpack(session, repo_id, want=tips_to_build, have=[], force_build=True) |
| 234 | actual_want: list[str] = list(mock_build.call_args.kwargs.get("want", [])) |
| 235 | |
| 236 | assert tip_b in actual_want, ( |
| 237 | f"wire_fetch_mpack was called with want={actual_want!r}, " |
| 238 | f"but tipB={tip_b!r} is absent. " |
| 239 | f"Root cause: process_fetch_mpack_prebuild_job reads tip_commit_ids " |
| 240 | f"from payload at musehub_wire_fetch.py:1067, which was stale at " |
| 241 | f"enqueue time. Fix: query MusehubBranch.head_commit_id from the DB " |
| 242 | f"at run time so the handler is self-coalescing." |
| 243 | ) |
| 244 | assert tip_a in actual_want, ( |
| 245 | f"tipA={tip_a!r} must also be in the want list; got {actual_want!r}" |
| 246 | ) |
| 247 | |
| 248 | |
| 249 | # --------------------------------------------------------------------------- |
| 250 | # MWP3_03 — Guard: single push enqueues exactly one of each (GREEN) |
| 251 | # --------------------------------------------------------------------------- |
| 252 | |
| 253 | |
| 254 | @pytest.mark.asyncio |
| 255 | async def test_mwp3_03_single_push_enqueues_exactly_one_each(db_session: AsyncSession) -> None: |
| 256 | """MWP3_03 GREEN: a single enqueue_push_intel call enqueues exactly one |
| 257 | mpack.index job and exactly one fetch.mpack.prebuild job. |
| 258 | |
| 259 | This guard must remain GREEN through all MWP-3 phases. The per-key dedup |
| 260 | fix for mpack.index (Phase 2) must not cause duplicate jobs for the normal |
| 261 | single-push case — only distinct mpack_keys should produce multiple rows. |
| 262 | """ |
| 263 | repo = await create_repo(db_session, owner="gabriel") |
| 264 | repo_id = str(repo.repo_id) |
| 265 | |
| 266 | tip_a = _fake_commit_id("mwp3-03-tip-a") |
| 267 | mpack_a = _fake_mpack_key("mwp3-03-a") |
| 268 | |
| 269 | await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) |
| 270 | await enqueue_push_intel(db_session, repo_id, head=tip_a, mpack_key=mpack_a) |
| 271 | await db_session.commit() |
| 272 | |
| 273 | count_index = (await db_session.execute( |
| 274 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 275 | MusehubBackgroundJob.repo_id == repo_id, |
| 276 | MusehubBackgroundJob.job_type == "mpack.index", |
| 277 | MusehubBackgroundJob.status == "pending", |
| 278 | ) |
| 279 | )).scalar() |
| 280 | |
| 281 | count_prebuild = (await db_session.execute( |
| 282 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 283 | MusehubBackgroundJob.repo_id == repo_id, |
| 284 | MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", |
| 285 | MusehubBackgroundJob.status == "pending", |
| 286 | ) |
| 287 | )).scalar() |
| 288 | |
| 289 | assert count_index == 1, ( |
| 290 | f"Expected exactly 1 pending mpack.index job for a single push; " |
| 291 | f"got {count_index}. The per-key dedup must not produce duplicates " |
| 292 | f"when only one mpack_key is enqueued." |
| 293 | ) |
| 294 | assert count_prebuild == 1, ( |
| 295 | f"Expected exactly 1 pending fetch.mpack.prebuild job for a single push; " |
| 296 | f"got {count_prebuild}." |
| 297 | ) |
| 298 | |
| 299 | |
| 300 | # --------------------------------------------------------------------------- |
| 301 | # MWP3_04 — Phase 1: handler reads live tips even when payload is empty (GREEN after fix) |
| 302 | # --------------------------------------------------------------------------- |
| 303 | |
| 304 | |
| 305 | @pytest.mark.asyncio |
| 306 | async def test_mwp3_04_handler_reads_live_tips_when_payload_empty(db_session: AsyncSession) -> None: |
| 307 | """MWP3_04: process_fetch_mpack_prebuild_job must query MusehubBranch for live |
| 308 | tips at run time, even when payload["tip_commit_ids"] is empty. |
| 309 | |
| 310 | Scenario: a prebuild job was somehow enqueued with an empty tip list (e.g. |
| 311 | a race where the branch was advanced after enqueue_push_intel ran its branch |
| 312 | query). Two live branches exist in the DB pointing to tipA and tipB. |
| 313 | |
| 314 | Current (buggy) behavior: the handler returns early at line 1069 because |
| 315 | payload["tip_commit_ids"] is []. wire_fetch_mpack is never called. |
| 316 | |
| 317 | After fix: the handler queries MusehubBranch, gets {tipA, tipB}, and calls |
| 318 | wire_fetch_mpack with want=[tipA, tipB]. |
| 319 | """ |
| 320 | repo = await create_repo(db_session, owner="gabriel") |
| 321 | repo_id = str(repo.repo_id) |
| 322 | |
| 323 | tip_a = _fake_commit_id("mwp3-04-tip-a") |
| 324 | tip_b = _fake_commit_id("mwp3-04-tip-b") |
| 325 | |
| 326 | await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) |
| 327 | await create_branch(db_session, repo_id, name="dev", head_commit_id=tip_b) |
| 328 | |
| 329 | # Deliberately empty payload — branch tips not captured. |
| 330 | job_id = await _insert_prebuild_job(db_session, repo_id, tip_commit_ids=[]) |
| 331 | |
| 332 | fake_mpack_id = "sha256:" + hashlib.sha256(b"mwp3-04-mpack").hexdigest() |
| 333 | |
| 334 | with patch( |
| 335 | "musehub.services.musehub_wire_fetch.wire_fetch_mpack", |
| 336 | new_callable=AsyncMock, |
| 337 | return_value={ |
| 338 | "mpack_id": fake_mpack_id, |
| 339 | "mpack_url": "https://r2.example/combined", |
| 340 | "commit_count": 2, |
| 341 | "blob_count": 4, |
| 342 | }, |
| 343 | ) as mock_build: |
| 344 | result = await process_fetch_mpack_prebuild_job(db_session, job_id) |
| 345 | await db_session.commit() |
| 346 | |
| 347 | assert mock_build.call_count == 1, ( |
| 348 | f"wire_fetch_mpack must be called once — handler must query live branches " |
| 349 | f"instead of returning early on empty payload. call_count={mock_build.call_count}. " |
| 350 | f"result={result}" |
| 351 | ) |
| 352 | actual_want: list[str] = list(mock_build.call_args.kwargs.get("want", [])) |
| 353 | assert tip_a in actual_want, f"tipA must be in want; got {actual_want!r}" |
| 354 | assert tip_b in actual_want, f"tipB must be in want; got {actual_want!r}" |
| 355 | assert result["tips_requested"] == 2, ( |
| 356 | f"tips_requested must reflect the live tip count (2); got {result['tips_requested']}" |
| 357 | ) |
| 358 | |
| 359 | |
| 360 | # --------------------------------------------------------------------------- |
| 361 | # MWP3_05 — Phase 1: stale payload tips lose to live DB tips (GREEN after fix) |
| 362 | # --------------------------------------------------------------------------- |
| 363 | |
| 364 | |
| 365 | @pytest.mark.asyncio |
| 366 | async def test_mwp3_05_live_tips_supersede_stale_payload(db_session: AsyncSession) -> None: |
| 367 | """MWP3_05: when payload["tip_commit_ids"] carries only {tipA} but MusehubBranch |
| 368 | has {tipA, tipB} live, the handler must build for {tipA, tipB}. |
| 369 | |
| 370 | This directly exercises the self-coalescing property: a single pending |
| 371 | prebuild job whose payload was captured before push B advanced a branch must |
| 372 | still cover push B's tip when it finally runs. |
| 373 | |
| 374 | This is the same scenario as MWP3_02 (which became RED in Phase 0) — Phase 1 |
| 375 | makes both MWP3_05 and MWP3_02 GREEN by the same code change. |
| 376 | """ |
| 377 | repo = await create_repo(db_session, owner="gabriel") |
| 378 | repo_id = str(repo.repo_id) |
| 379 | |
| 380 | tip_a = _fake_commit_id("mwp3-05-tip-a") |
| 381 | tip_b = _fake_commit_id("mwp3-05-tip-b") |
| 382 | |
| 383 | await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) |
| 384 | await create_branch(db_session, repo_id, name="dev", head_commit_id=tip_b) |
| 385 | |
| 386 | # Stale payload: only tipA. tipB is live in MusehubBranch but absent from payload. |
| 387 | job_id = await _insert_prebuild_job(db_session, repo_id, tip_commit_ids=[tip_a]) |
| 388 | |
| 389 | fake_mpack_id = "sha256:" + hashlib.sha256(b"mwp3-05-mpack").hexdigest() |
| 390 | |
| 391 | with patch( |
| 392 | "musehub.services.musehub_wire_fetch.wire_fetch_mpack", |
| 393 | new_callable=AsyncMock, |
| 394 | return_value={ |
| 395 | "mpack_id": fake_mpack_id, |
| 396 | "mpack_url": "https://r2.example/combined", |
| 397 | "commit_count": 2, |
| 398 | "blob_count": 4, |
| 399 | }, |
| 400 | ) as mock_build: |
| 401 | await process_fetch_mpack_prebuild_job(db_session, job_id) |
| 402 | await db_session.commit() |
| 403 | |
| 404 | actual_want: list[str] = list(mock_build.call_args.kwargs.get("want", [])) |
| 405 | assert tip_b in actual_want, ( |
| 406 | f"tipB must be in want — live DB tip must supersede stale payload. " |
| 407 | f"want={actual_want!r}" |
| 408 | ) |
| 409 | assert tip_a in actual_want, ( |
| 410 | f"tipA must also be in want. want={actual_want!r}" |
| 411 | ) |
| 412 | |
| 413 | |
| 414 | # --------------------------------------------------------------------------- |
| 415 | # MWP3_06 — Phase 1 integration: both tips get cache rows (GREEN after fix) |
| 416 | # --------------------------------------------------------------------------- |
| 417 | |
| 418 | |
| 419 | @pytest.mark.asyncio |
| 420 | async def test_mwp3_06_cache_rows_written_for_all_live_tips(db_session: AsyncSession) -> None: |
| 421 | """MWP3_06 Integration: after the Phase 1 fix, running the single pending |
| 422 | prebuild job writes musehub_fetch_mpack_cache rows for ALL live branch tips, |
| 423 | even those absent from the stale payload. |
| 424 | |
| 425 | Simulates the real scenario: push A enqueues prebuild with payload={tipA}; |
| 426 | push B adds branch dev→tipB (deduped — no new prebuild job); the worker |
| 427 | drains by running the single pending job. Both tipA and tipB must end up with |
| 428 | a cache row pointing at the same combined mpack_id. |
| 429 | """ |
| 430 | repo = await create_repo(db_session, owner="gabriel") |
| 431 | repo_id = str(repo.repo_id) |
| 432 | |
| 433 | tip_a = _fake_commit_id("mwp3-06-tip-a") |
| 434 | tip_b = _fake_commit_id("mwp3-06-tip-b") |
| 435 | combined_mpack_id = _fake_mpack_key("mwp3-06-combined") |
| 436 | |
| 437 | await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) |
| 438 | await create_branch(db_session, repo_id, name="dev", head_commit_id=tip_b) |
| 439 | |
| 440 | # Stale payload — only tipA captured at push A's enqueue time. |
| 441 | job_id = await _insert_prebuild_job(db_session, repo_id, tip_commit_ids=[tip_a]) |
| 442 | |
| 443 | with patch( |
| 444 | "musehub.services.musehub_wire_fetch.wire_fetch_mpack", |
| 445 | new_callable=AsyncMock, |
| 446 | return_value={ |
| 447 | "mpack_id": combined_mpack_id, |
| 448 | "mpack_url": "https://r2.example/combined", |
| 449 | "commit_count": 4, |
| 450 | "blob_count": 8, |
| 451 | }, |
| 452 | ): |
| 453 | result = await process_fetch_mpack_prebuild_job(db_session, job_id) |
| 454 | await db_session.commit() |
| 455 | |
| 456 | # Both tips must have a cache row pointing at the same combined mpack_id. |
| 457 | cache_rows = (await db_session.execute( |
| 458 | select( |
| 459 | MusehubFetchMPackCache.tip_commit_id, |
| 460 | MusehubFetchMPackCache.mpack_id, |
| 461 | ).where(MusehubFetchMPackCache.repo_id == repo_id) |
| 462 | )).all() |
| 463 | cached = {row[0]: row[1] for row in cache_rows} |
| 464 | |
| 465 | assert tip_a in cached, ( |
| 466 | f"tipA={tip_a[:20]!r} must have a cache row; got keys: {[k[:20] for k in cached]}" |
| 467 | ) |
| 468 | assert tip_b in cached, ( |
| 469 | f"tipB={tip_b[:20]!r} must have a cache row — it was absent from the " |
| 470 | f"stale payload but is a live branch tip. Got keys: {[k[:20] for k in cached]}" |
| 471 | ) |
| 472 | assert cached[tip_a] == combined_mpack_id, ( |
| 473 | f"tipA cache must point at combined_mpack_id. Got: {cached.get(tip_a)}" |
| 474 | ) |
| 475 | assert cached[tip_b] == combined_mpack_id, ( |
| 476 | f"tipB cache must point at combined_mpack_id. Got: {cached.get(tip_b)}" |
| 477 | ) |
| 478 | assert result["tips_requested"] == 2, ( |
| 479 | f"tips_requested must be 2 (live tip count); got {result['tips_requested']}" |
| 480 | ) |
| 481 | |
| 482 | |
| 483 | # =========================================================================== |
| 484 | # Phase 2 — mpack.index per-mpack_key dedup |
| 485 | # =========================================================================== |
| 486 | |
| 487 | |
| 488 | # --------------------------------------------------------------------------- |
| 489 | # MWP3_07 — distinct mpack_keys must each produce a pending mpack.index (RED) |
| 490 | # --------------------------------------------------------------------------- |
| 491 | |
| 492 | |
| 493 | @pytest.mark.asyncio |
| 494 | async def test_mwp3_07_distinct_mpack_keys_both_produce_index_jobs( |
| 495 | db_session: AsyncSession, |
| 496 | ) -> None: |
| 497 | """MWP3_07 RED → GREEN after Phase 2 fix: two enqueue_push_intel calls with |
| 498 | distinct mpack_keys must both produce a pending mpack.index job. |
| 499 | |
| 500 | This is the same core assertion as MWP3_01 but written as a Phase 2 |
| 501 | deliverable — the canonical test that makes MWP3_01 GREEN and validates |
| 502 | the per-key dedup path. |
| 503 | |
| 504 | Current (buggy) behavior: the coarse SELECT finds mpack.index already |
| 505 | pending after push A; push B's job is silently dropped. Only one row. |
| 506 | After fix: enqueue_push_intel deduplicates mpack.index on |
| 507 | payload->>'mpack_key', so distinct keys always produce distinct rows. |
| 508 | """ |
| 509 | repo = await create_repo(db_session, owner="gabriel") |
| 510 | repo_id = str(repo.repo_id) |
| 511 | |
| 512 | tip_a = _fake_commit_id("mwp3-07-tip-a") |
| 513 | tip_b = _fake_commit_id("mwp3-07-tip-b") |
| 514 | mpack_a = _fake_mpack_key("mwp3-07-a") |
| 515 | mpack_b = _fake_mpack_key("mwp3-07-b") |
| 516 | |
| 517 | await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) |
| 518 | await enqueue_push_intel(db_session, repo_id, head=tip_a, mpack_key=mpack_a) |
| 519 | await db_session.commit() |
| 520 | |
| 521 | # Push B while push A's jobs are still pending. |
| 522 | await enqueue_push_intel(db_session, repo_id, head=tip_b, mpack_key=mpack_b) |
| 523 | await db_session.commit() |
| 524 | |
| 525 | rows = (await db_session.execute( |
| 526 | select(MusehubBackgroundJob.payload).where( |
| 527 | MusehubBackgroundJob.repo_id == repo_id, |
| 528 | MusehubBackgroundJob.job_type == "mpack.index", |
| 529 | MusehubBackgroundJob.status == "pending", |
| 530 | ) |
| 531 | )).scalars().all() |
| 532 | |
| 533 | keys_in_db = [r.get("mpack_key") for r in rows] |
| 534 | assert len(rows) == 2, ( |
| 535 | f"Expected 2 pending mpack.index rows (one per mpack_key), got {len(rows)}. " |
| 536 | f"mpack_keys in DB: {keys_in_db}" |
| 537 | ) |
| 538 | assert mpack_a in keys_in_db, f"Push A key {mpack_a!r} missing: {keys_in_db}" |
| 539 | assert mpack_b in keys_in_db, f"Push B key {mpack_b!r} missing: {keys_in_db}" |
| 540 | |
| 541 | |
| 542 | # --------------------------------------------------------------------------- |
| 543 | # MWP3_08 — same mpack_key is still deduped to one row (GREEN guard) |
| 544 | # --------------------------------------------------------------------------- |
| 545 | |
| 546 | |
| 547 | @pytest.mark.asyncio |
| 548 | async def test_mwp3_08_same_mpack_key_deduped_to_one_row( |
| 549 | db_session: AsyncSession, |
| 550 | ) -> None: |
| 551 | """MWP3_08 GREEN (guard): two enqueue_push_intel calls with the SAME mpack_key |
| 552 | must still produce exactly one pending mpack.index job. |
| 553 | |
| 554 | Per-key dedup must not cause a duplicate-job explosion for idempotent |
| 555 | re-pushes (same mpack content re-uploaded). This guard must remain GREEN |
| 556 | through Phase 2 and all subsequent phases. |
| 557 | """ |
| 558 | repo = await create_repo(db_session, owner="gabriel") |
| 559 | repo_id = str(repo.repo_id) |
| 560 | |
| 561 | tip_a = _fake_commit_id("mwp3-08-tip-a") |
| 562 | tip_b = _fake_commit_id("mwp3-08-tip-b") |
| 563 | mpack_same = _fake_mpack_key("mwp3-08-same") |
| 564 | |
| 565 | await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) |
| 566 | await enqueue_push_intel(db_session, repo_id, head=tip_a, mpack_key=mpack_same) |
| 567 | await db_session.commit() |
| 568 | |
| 569 | await enqueue_push_intel(db_session, repo_id, head=tip_b, mpack_key=mpack_same) |
| 570 | await db_session.commit() |
| 571 | |
| 572 | count = (await db_session.execute( |
| 573 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 574 | MusehubBackgroundJob.repo_id == repo_id, |
| 575 | MusehubBackgroundJob.job_type == "mpack.index", |
| 576 | MusehubBackgroundJob.status == "pending", |
| 577 | ) |
| 578 | )).scalar() |
| 579 | |
| 580 | assert count == 1, ( |
| 581 | f"Expected exactly 1 pending mpack.index for an identical re-push " |
| 582 | f"(same mpack_key); got {count}. Per-key dedup must collapse true duplicates." |
| 583 | ) |
| 584 | |
| 585 | |
| 586 | # --------------------------------------------------------------------------- |
| 587 | # MWP3_08b — enqueue_job dedup_payload_key param (distinct → 2 rows, RED) |
| 588 | # --------------------------------------------------------------------------- |
| 589 | |
| 590 | |
| 591 | @pytest.mark.asyncio |
| 592 | async def test_mwp3_08b_distinct_payload_key_produces_two_rows( |
| 593 | db_session: AsyncSession, |
| 594 | ) -> None: |
| 595 | """MWP3_08b RED → GREEN: enqueue_job with dedup_payload_key='mpack_key' and |
| 596 | distinct payload values must produce two pending rows. |
| 597 | |
| 598 | Without the parameter, enqueue_job uses coarse (repo_id, job_type, pending) |
| 599 | dedup and silently drops the second call. With dedup_payload_key, it keys on |
| 600 | payload->>'mpack_key', allowing distinct mpacks to each have a job. |
| 601 | """ |
| 602 | from musehub.services.musehub_jobs import enqueue_job |
| 603 | |
| 604 | repo = await create_repo(db_session, owner="gabriel") |
| 605 | repo_id = str(repo.repo_id) |
| 606 | |
| 607 | mpack_a = _fake_mpack_key("mwp3-08b-a") |
| 608 | mpack_b = _fake_mpack_key("mwp3-08b-b") |
| 609 | |
| 610 | await enqueue_job( |
| 611 | db_session, repo_id, "mpack.index", |
| 612 | {"mpack_key": mpack_a, "head": "sha256:aaa"}, |
| 613 | dedup_payload_key="mpack_key", |
| 614 | ) |
| 615 | await db_session.commit() |
| 616 | |
| 617 | await enqueue_job( |
| 618 | db_session, repo_id, "mpack.index", |
| 619 | {"mpack_key": mpack_b, "head": "sha256:bbb"}, |
| 620 | dedup_payload_key="mpack_key", |
| 621 | ) |
| 622 | await db_session.commit() |
| 623 | |
| 624 | rows = (await db_session.execute( |
| 625 | select(MusehubBackgroundJob.payload).where( |
| 626 | MusehubBackgroundJob.repo_id == repo_id, |
| 627 | MusehubBackgroundJob.job_type == "mpack.index", |
| 628 | MusehubBackgroundJob.status == "pending", |
| 629 | ) |
| 630 | )).scalars().all() |
| 631 | |
| 632 | keys_in_db = [r.get("mpack_key") for r in rows] |
| 633 | assert len(rows) == 2, ( |
| 634 | f"Expected 2 rows with dedup_payload_key and distinct mpack_keys; " |
| 635 | f"got {len(rows)}. keys in DB: {keys_in_db}" |
| 636 | ) |
| 637 | assert mpack_a in keys_in_db and mpack_b in keys_in_db, ( |
| 638 | f"Both keys must be present. Got: {keys_in_db}" |
| 639 | ) |
| 640 | |
| 641 | |
| 642 | @pytest.mark.asyncio |
| 643 | async def test_mwp3_08b_same_payload_key_deduped_to_one_row( |
| 644 | db_session: AsyncSession, |
| 645 | ) -> None: |
| 646 | """MWP3_08b guard GREEN: enqueue_job with dedup_payload_key and identical |
| 647 | payload values deduplicates to one row — per-key dedup collapses true duplicates. |
| 648 | """ |
| 649 | from musehub.services.musehub_jobs import enqueue_job |
| 650 | |
| 651 | repo = await create_repo(db_session, owner="gabriel") |
| 652 | repo_id = str(repo.repo_id) |
| 653 | |
| 654 | mpack_same = _fake_mpack_key("mwp3-08b-same") |
| 655 | |
| 656 | await enqueue_job( |
| 657 | db_session, repo_id, "mpack.index", |
| 658 | {"mpack_key": mpack_same, "head": "sha256:aaa"}, |
| 659 | dedup_payload_key="mpack_key", |
| 660 | ) |
| 661 | await db_session.commit() |
| 662 | |
| 663 | await enqueue_job( |
| 664 | db_session, repo_id, "mpack.index", |
| 665 | {"mpack_key": mpack_same, "head": "sha256:bbb"}, |
| 666 | dedup_payload_key="mpack_key", |
| 667 | ) |
| 668 | await db_session.commit() |
| 669 | |
| 670 | count = (await db_session.execute( |
| 671 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 672 | MusehubBackgroundJob.repo_id == repo_id, |
| 673 | MusehubBackgroundJob.job_type == "mpack.index", |
| 674 | MusehubBackgroundJob.status == "pending", |
| 675 | ) |
| 676 | )).scalar() |
| 677 | |
| 678 | assert count == 1, ( |
| 679 | f"Same mpack_key must dedup to 1 row with dedup_payload_key; got {count}" |
| 680 | ) |
| 681 | |
| 682 | |
| 683 | @pytest.mark.asyncio |
| 684 | async def test_mwp3_08b_default_coarse_behavior_unchanged( |
| 685 | db_session: AsyncSession, |
| 686 | ) -> None: |
| 687 | """MWP3_08b guard GREEN: enqueue_job without dedup_payload_key retains existing |
| 688 | coarse (repo_id, job_type, pending) dedup — second call returns None, one row. |
| 689 | """ |
| 690 | from musehub.services.musehub_jobs import enqueue_job |
| 691 | |
| 692 | repo = await create_repo(db_session, owner="gabriel") |
| 693 | repo_id = str(repo.repo_id) |
| 694 | |
| 695 | id1 = await enqueue_job( |
| 696 | db_session, repo_id, "mpack.index", |
| 697 | {"mpack_key": _fake_mpack_key("mwp3-08b-coarse-a"), "head": "sha256:aaa"}, |
| 698 | ) |
| 699 | await db_session.commit() |
| 700 | |
| 701 | id2 = await enqueue_job( |
| 702 | db_session, repo_id, "mpack.index", |
| 703 | {"mpack_key": _fake_mpack_key("mwp3-08b-coarse-b"), "head": "sha256:bbb"}, |
| 704 | ) |
| 705 | await db_session.commit() |
| 706 | |
| 707 | assert id1 is not None, "First call must return a job_id" |
| 708 | assert id2 is None, "Second call without dedup_payload_key must be deduped (return None)" |
| 709 | |
| 710 | count = (await db_session.execute( |
| 711 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 712 | MusehubBackgroundJob.repo_id == repo_id, |
| 713 | MusehubBackgroundJob.job_type == "mpack.index", |
| 714 | MusehubBackgroundJob.status == "pending", |
| 715 | ) |
| 716 | )).scalar() |
| 717 | assert count == 1, f"Coarse dedup must yield 1 row; got {count}" |
| 718 | |
| 719 | |
| 720 | # --------------------------------------------------------------------------- |
| 721 | # MWP3_09 — coarse types stay deduped; only mpack.index multiplied (GREEN guard) |
| 722 | # --------------------------------------------------------------------------- |
| 723 | |
| 724 | |
| 725 | @pytest.mark.asyncio |
| 726 | async def test_mwp3_09_coarse_types_remain_deduplicated( |
| 727 | db_session: AsyncSession, |
| 728 | ) -> None: |
| 729 | """MWP3_09 GREEN (guard): after two enqueue_push_intel calls with distinct |
| 730 | mpack_keys, all coarse job types (fetch.mpack.prebuild, intel.*, gc, …) must |
| 731 | each have exactly one pending row — the fix must not loosen coarse dedup. |
| 732 | |
| 733 | Only mpack.index is allowed to multiply (per distinct mpack_key). |
| 734 | """ |
| 735 | repo = await create_repo(db_session, owner="gabriel") |
| 736 | repo_id = str(repo.repo_id) |
| 737 | |
| 738 | tip_a = _fake_commit_id("mwp3-09-tip-a") |
| 739 | tip_b = _fake_commit_id("mwp3-09-tip-b") |
| 740 | mpack_a = _fake_mpack_key("mwp3-09-a") |
| 741 | mpack_b = _fake_mpack_key("mwp3-09-b") |
| 742 | |
| 743 | await create_branch(db_session, repo_id, name="main", head_commit_id=tip_a) |
| 744 | await enqueue_push_intel(db_session, repo_id, head=tip_a, mpack_key=mpack_a) |
| 745 | await db_session.commit() |
| 746 | |
| 747 | await enqueue_push_intel(db_session, repo_id, head=tip_b, mpack_key=mpack_b) |
| 748 | await db_session.commit() |
| 749 | |
| 750 | # Query all pending rows and group by job_type. |
| 751 | all_rows = (await db_session.execute( |
| 752 | select(MusehubBackgroundJob.job_type).where( |
| 753 | MusehubBackgroundJob.repo_id == repo_id, |
| 754 | MusehubBackgroundJob.status == "pending", |
| 755 | ) |
| 756 | )).scalars().all() |
| 757 | |
| 758 | from collections import Counter |
| 759 | counts = Counter(all_rows) |
| 760 | |
| 761 | # mpack.index should have 2 (one per distinct mpack_key). |
| 762 | # Every other type should have exactly 1. |
| 763 | duplicated_coarse = { |
| 764 | jt: cnt for jt, cnt in counts.items() |
| 765 | if jt != "mpack.index" and cnt > 1 |
| 766 | } |
| 767 | assert not duplicated_coarse, ( |
| 768 | f"Coarse job types must remain deduplicated (≤1 each) after two " |
| 769 | f"enqueue_push_intel calls. Found duplicates: {duplicated_coarse}" |
| 770 | ) |
| 771 | |
| 772 | |
| 773 | # --------------------------------------------------------------------------- |
| 774 | # Phase 3 — E2E helpers |
| 775 | # --------------------------------------------------------------------------- |
| 776 | |
| 777 | _E2E_OWNER = "gabriel" |
| 778 | |
| 779 | |
| 780 | def _e2e_raw_commit( |
| 781 | cid: str, |
| 782 | parent: str | None, |
| 783 | snap_id: str, |
| 784 | branch: str = "main", |
| 785 | ) -> dict: |
| 786 | return { |
| 787 | "commit_id": cid, |
| 788 | "branch": branch, |
| 789 | "message": f"commit {cid[:12]}", |
| 790 | "author": _E2E_OWNER, |
| 791 | "committed_at": _now().isoformat(), |
| 792 | "parent_commit_id": parent, |
| 793 | "parent2_commit_id": None, |
| 794 | "snapshot_id": snap_id, |
| 795 | "agent_id": "", |
| 796 | "model_id": "", |
| 797 | "toolchain_id": "", |
| 798 | "sem_ver_bump": "none", |
| 799 | "breaking_changes": [], |
| 800 | "signature": "", |
| 801 | "signer_key_id": "", |
| 802 | "signer_public_key": "", |
| 803 | "prompt_hash": "", |
| 804 | } |
| 805 | |
| 806 | |
| 807 | async def _e2e_push( |
| 808 | session: AsyncSession, |
| 809 | repo_id: str, |
| 810 | cid: str, |
| 811 | parent: str | None, |
| 812 | snap_id: str, |
| 813 | fpath: str, |
| 814 | oid: str, |
| 815 | data: bytes, |
| 816 | branch: str = "main", |
| 817 | ) -> str: |
| 818 | """Build a single-commit mpack, store it in the conftest MemoryBackend, push it. |
| 819 | |
| 820 | The conftest autouse fixture patches musehub.storage.backends.get_backend to a |
| 821 | MemoryBackend instance shared across all modules. Storing in _mpacks[key] here |
| 822 | is equivalent to pre-seeding MinIO — wire_push_unpack_mpack and |
| 823 | process_mpack_index_job both call get_backend() and see the same bytes. |
| 824 | |
| 825 | Returns the mpack_key so callers can assert on MPackIndex after draining. |
| 826 | """ |
| 827 | import musehub.storage.backends as _b_mod |
| 828 | backend = _b_mod.get_backend() |
| 829 | |
| 830 | mpack_bytes = build_wire_mpack({ |
| 831 | "commits": [_e2e_raw_commit(cid, parent, snap_id, branch=branch)], |
| 832 | "snapshots": [{ |
| 833 | "snapshot_id": snap_id, |
| 834 | "parent_snapshot_id": None, |
| 835 | "delta_upsert": {fpath: oid}, |
| 836 | "delta_remove": [], |
| 837 | "directories": [], |
| 838 | }], |
| 839 | "blobs": [{"object_id": oid, "content": data}], |
| 840 | "tags": [], |
| 841 | }) |
| 842 | mpack_key = blob_id(mpack_bytes) |
| 843 | backend._mpacks[mpack_key] = mpack_bytes |
| 844 | |
| 845 | await wire_push_unpack_mpack( |
| 846 | session, |
| 847 | repo_id, |
| 848 | mpack_key, |
| 849 | pusher_id=_E2E_OWNER, |
| 850 | branch=branch, |
| 851 | head_commit_id=cid, |
| 852 | commits_count=1, |
| 853 | blobs_count=1, |
| 854 | force=True, |
| 855 | ) |
| 856 | await session.commit() |
| 857 | return mpack_key |
| 858 | |
| 859 | |
| 860 | async def _drain_index_jobs(session: AsyncSession, repo_id: str, limit: int | None = None) -> list[str]: |
| 861 | """Drain pending mpack.index jobs in created_at order, return job_ids drained.""" |
| 862 | q = ( |
| 863 | select(MusehubBackgroundJob) |
| 864 | .where( |
| 865 | MusehubBackgroundJob.repo_id == repo_id, |
| 866 | MusehubBackgroundJob.job_type == "mpack.index", |
| 867 | MusehubBackgroundJob.status == "pending", |
| 868 | ) |
| 869 | .order_by(MusehubBackgroundJob.created_at) |
| 870 | ) |
| 871 | if limit is not None: |
| 872 | q = q.limit(limit) |
| 873 | jobs = (await session.execute(q)).scalars().all() |
| 874 | drained = [] |
| 875 | for job in jobs: |
| 876 | await process_mpack_index_job(session, job.job_id) |
| 877 | await complete_job(session, job.job_id) |
| 878 | await session.commit() |
| 879 | drained.append(job.job_id) |
| 880 | return drained |
| 881 | |
| 882 | |
| 883 | async def _drain_prebuild_job(session: AsyncSession, repo_id: str) -> None: |
| 884 | """Drain the single pending fetch.mpack.prebuild job.""" |
| 885 | row = (await session.execute( |
| 886 | select(MusehubBackgroundJob).where( |
| 887 | MusehubBackgroundJob.repo_id == repo_id, |
| 888 | MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", |
| 889 | MusehubBackgroundJob.status == "pending", |
| 890 | ).limit(1) |
| 891 | )).scalar_one_or_none() |
| 892 | assert row is not None, "Expected a pending fetch.mpack.prebuild job" |
| 893 | await process_fetch_mpack_prebuild_job(session, row.job_id) |
| 894 | await session.commit() |
| 895 | |
| 896 | |
| 897 | # --------------------------------------------------------------------------- |
| 898 | # MWP3_10 — E2E acceptance gate: two pushes + full drain → both indexed + cache |
| 899 | # --------------------------------------------------------------------------- |
| 900 | |
| 901 | |
| 902 | @pytest.mark.asyncio |
| 903 | async def test_mwp3_10_e2e_two_pushes_full_drain(db_session: AsyncSession) -> None: |
| 904 | """MWP3_10 GREEN (Phase 3 — E2E acceptance gate): two back-to-back pushes, |
| 905 | no drain between them, then full drain → both mpacks indexed and clone cache |
| 906 | updated to push B's tip. |
| 907 | |
| 908 | This is the canonical RC-3 scenario — two pushes before the worker wakes up. |
| 909 | |
| 910 | Before the MWP-3 fix (Phase 0 state): |
| 911 | - Push A enqueues mpack.index(key_a). |
| 912 | - Push B: coarse dedup finds mpack.index pending → drops key_b. Only key_a |
| 913 | ever gets indexed. clone cache never advances to c2. |
| 914 | |
| 915 | After the MWP-3 fix (Phases 1+2): |
| 916 | - Push A enqueues mpack.index(key_a) + fetch.mpack.prebuild. |
| 917 | - Push B enqueues mpack.index(key_b) (per-key dedup allows it). |
| 918 | The prebuild is deduped to one row but re-reads live branch tips (Phase 1). |
| 919 | - Drain order: index jobs first (populates MusehubMPackIndex so wire_fetch_mpack |
| 920 | does not raise FetchNotIndexedError), then prebuild. |
| 921 | - After drain: MPackIndex has rows for key_a AND key_b; clone cache has c2. |
| 922 | """ |
| 923 | repo = await create_repo(db_session, owner=_E2E_OWNER) |
| 924 | repo_id = str(repo.repo_id) |
| 925 | |
| 926 | data_a = b"mwp3-10-blob-a" |
| 927 | data_b = b"mwp3-10-blob-b" |
| 928 | oid_a = blob_id(data_a) |
| 929 | oid_b = blob_id(data_b) |
| 930 | snap_a = hash_snapshot({"file_a.txt": oid_a}) |
| 931 | snap_b = hash_snapshot({"file_b.txt": oid_b}) |
| 932 | c1 = _fake_commit_id("mwp3-10-c1") |
| 933 | c2 = _fake_commit_id("mwp3-10-c2") |
| 934 | |
| 935 | # Push A — creates branch main → c1; enqueues mpack.index(key_a) + prebuild. |
| 936 | mpack_key_a = await _e2e_push(db_session, repo_id, c1, None, snap_a, "file_a.txt", oid_a, data_a) |
| 937 | |
| 938 | # Push B — advances main → c2; enqueues mpack.index(key_b); prebuild deduped. |
| 939 | mpack_key_b = await _e2e_push(db_session, repo_id, c2, c1, snap_b, "file_b.txt", oid_b, data_b) |
| 940 | |
| 941 | # Both mpack.index keys must be pending — Phase 2 fix: per-key dedup. |
| 942 | pending_index = (await db_session.execute( |
| 943 | select(MusehubBackgroundJob.payload).where( |
| 944 | MusehubBackgroundJob.repo_id == repo_id, |
| 945 | MusehubBackgroundJob.job_type == "mpack.index", |
| 946 | MusehubBackgroundJob.status == "pending", |
| 947 | ) |
| 948 | )).scalars().all() |
| 949 | pending_keys = {pl.get("mpack_key") for pl in pending_index} |
| 950 | assert mpack_key_a in pending_keys and mpack_key_b in pending_keys, ( |
| 951 | f"Expected both mpack_keys pending before drain. " |
| 952 | f"Found: {pending_keys}" |
| 953 | ) |
| 954 | assert len(pending_index) == 2, ( |
| 955 | f"Expected 2 pending mpack.index jobs, got {len(pending_index)}" |
| 956 | ) |
| 957 | |
| 958 | # Drain: index jobs first (wire_fetch_mpack raises FetchNotIndexedError |
| 959 | # if index is absent), then prebuild. |
| 960 | await _drain_index_jobs(db_session, repo_id) |
| 961 | await _drain_prebuild_job(db_session, repo_id) |
| 962 | |
| 963 | # (a) Both mpack keys must appear in MusehubMPackIndex. |
| 964 | count_a = (await db_session.execute( |
| 965 | select(func.count()).select_from(MusehubMPackIndex).where( |
| 966 | MusehubMPackIndex.mpack_id == mpack_key_a, |
| 967 | MusehubMPackIndex.entity_type == "object", |
| 968 | ) |
| 969 | )).scalar() |
| 970 | count_b = (await db_session.execute( |
| 971 | select(func.count()).select_from(MusehubMPackIndex).where( |
| 972 | MusehubMPackIndex.mpack_id == mpack_key_b, |
| 973 | MusehubMPackIndex.entity_type == "object", |
| 974 | ) |
| 975 | )).scalar() |
| 976 | assert count_a > 0, ( |
| 977 | f"MusehubMPackIndex has no rows for mpack_key_a={mpack_key_a[:20]}. " |
| 978 | f"process_mpack_index_job for push A never ran or wrote no rows." |
| 979 | ) |
| 980 | assert count_b > 0, ( |
| 981 | f"MusehubMPackIndex has no rows for mpack_key_b={mpack_key_b[:20]}. " |
| 982 | f"process_mpack_index_job for push B was never enqueued (RC-3 Face 1 not fixed)." |
| 983 | ) |
| 984 | |
| 985 | # (b) Clone cache must point to c2 (push B's tip). |
| 986 | # process_fetch_mpack_prebuild_job reads live branch tip (c2 — Phase 1 fix). |
| 987 | cache_row = (await db_session.execute( |
| 988 | select(MusehubFetchMPackCache).where( |
| 989 | MusehubFetchMPackCache.repo_id == repo_id, |
| 990 | MusehubFetchMPackCache.tip_commit_id == c2, |
| 991 | ) |
| 992 | )).scalar_one_or_none() |
| 993 | assert cache_row is not None, ( |
| 994 | f"MusehubFetchMPackCache has no row for tip_commit_id=c2={c2[:20]}. " |
| 995 | f"The prebuild job used a stale tip (RC-3 Face 2 not fixed) or " |
| 996 | f"wire_fetch_mpack raised FetchNotIndexedError because push B's mpack " |
| 997 | f"was never indexed (RC-3 Face 1 not fixed)." |
| 998 | ) |
| 999 | |
| 1000 | |
| 1001 | # --------------------------------------------------------------------------- |
| 1002 | # MWP3_11 — Stress: N=10 pushes, partial drain mid-sequence, full final drain |
| 1003 | # --------------------------------------------------------------------------- |
| 1004 | |
| 1005 | |
| 1006 | @pytest.mark.asyncio |
| 1007 | async def test_mwp3_11_stress_partial_drain_n10(db_session: AsyncSession) -> None: |
| 1008 | """MWP3_11 GREEN (Phase 3 — stress): 10 rapid pushes with a partial drain |
| 1009 | mid-sequence. All 10 mpacks must be indexed and clone cache must point to |
| 1010 | the final tip after the full drain. |
| 1011 | |
| 1012 | Sequence: |
| 1013 | 1. Push commits 1-5 without draining → 5 pending mpack.index, 1 prebuild. |
| 1014 | 2. Partial drain: drain only the first 3 mpack.index jobs (chronological |
| 1015 | order). Commits 4 and 5 remain unindexed (2 pending index jobs). |
| 1016 | 3. Push commits 6-10 without draining → 5 new mpack.index jobs enqueued |
| 1017 | (dedup allows them because each has a distinct mpack_key). Prebuild |
| 1018 | stays at 1 pending row (still deduped from push 1). |
| 1019 | 4. Full drain: drain all 7 remaining mpack.index jobs (2 + 5), then drain |
| 1020 | the single pending prebuild job. |
| 1021 | |
| 1022 | Assertions: |
| 1023 | (a) MusehubMPackIndex has rows for all 10 mpack_keys. |
| 1024 | (b) MusehubFetchMPackCache has row for commit_10's tip (live branch tip |
| 1025 | at prebuild run time = Phase 1 fix; all 10 index jobs drained before |
| 1026 | prebuild = Phase 2 fix). |
| 1027 | """ |
| 1028 | repo = await create_repo(db_session, owner=_E2E_OWNER) |
| 1029 | repo_id = str(repo.repo_id) |
| 1030 | n = 10 |
| 1031 | |
| 1032 | cids: list[str] = [_fake_commit_id(f"mwp3-11-c{i}") for i in range(1, n + 1)] |
| 1033 | datas: list[bytes] = [f"mwp3-11-blob-{i}".encode() for i in range(1, n + 1)] |
| 1034 | oids: list[str] = [blob_id(d) for d in datas] |
| 1035 | snaps: list[str] = [hash_snapshot({f"f{i}.txt": oids[i - 1]}) for i in range(1, n + 1)] |
| 1036 | |
| 1037 | mpack_keys: list[str] = [] |
| 1038 | |
| 1039 | # Step 1 — push commits 1-5 without draining. |
| 1040 | for i in range(5): |
| 1041 | parent = cids[i - 1] if i > 0 else None |
| 1042 | key = await _e2e_push( |
| 1043 | db_session, repo_id, cids[i], parent, snaps[i], |
| 1044 | f"f{i + 1}.txt", oids[i], datas[i], |
| 1045 | ) |
| 1046 | mpack_keys.append(key) |
| 1047 | |
| 1048 | pending_after_5 = (await db_session.execute( |
| 1049 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 1050 | MusehubBackgroundJob.repo_id == repo_id, |
| 1051 | MusehubBackgroundJob.job_type == "mpack.index", |
| 1052 | MusehubBackgroundJob.status == "pending", |
| 1053 | ) |
| 1054 | )).scalar() |
| 1055 | assert pending_after_5 == 5, ( |
| 1056 | f"Expected 5 pending mpack.index jobs after 5 pushes, got {pending_after_5}" |
| 1057 | ) |
| 1058 | |
| 1059 | # Step 2 — partial drain: drain only the first 3 index jobs. |
| 1060 | await _drain_index_jobs(db_session, repo_id, limit=3) |
| 1061 | |
| 1062 | pending_after_partial = (await db_session.execute( |
| 1063 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 1064 | MusehubBackgroundJob.repo_id == repo_id, |
| 1065 | MusehubBackgroundJob.job_type == "mpack.index", |
| 1066 | MusehubBackgroundJob.status == "pending", |
| 1067 | ) |
| 1068 | )).scalar() |
| 1069 | assert pending_after_partial == 2, ( |
| 1070 | f"Expected 2 pending mpack.index jobs after draining 3 of 5, " |
| 1071 | f"got {pending_after_partial}" |
| 1072 | ) |
| 1073 | |
| 1074 | # Step 3 — push commits 6-10 without draining. |
| 1075 | for i in range(5, n): |
| 1076 | key = await _e2e_push( |
| 1077 | db_session, repo_id, cids[i], cids[i - 1], snaps[i], |
| 1078 | f"f{i + 1}.txt", oids[i], datas[i], |
| 1079 | ) |
| 1080 | mpack_keys.append(key) |
| 1081 | |
| 1082 | # 2 remaining from commits 4+5 + 5 new from commits 6-10 = 7 pending. |
| 1083 | pending_after_10 = (await db_session.execute( |
| 1084 | select(func.count()).select_from(MusehubBackgroundJob).where( |
| 1085 | MusehubBackgroundJob.repo_id == repo_id, |
| 1086 | MusehubBackgroundJob.job_type == "mpack.index", |
| 1087 | MusehubBackgroundJob.status == "pending", |
| 1088 | ) |
| 1089 | )).scalar() |
| 1090 | assert pending_after_10 == 7, ( |
| 1091 | f"Expected 7 pending mpack.index jobs after pushes 6-10 " |
| 1092 | f"(2 leftover + 5 new = 7), got {pending_after_10}" |
| 1093 | ) |
| 1094 | |
| 1095 | # Step 4 — full drain: all remaining index jobs, then prebuild. |
| 1096 | await _drain_index_jobs(db_session, repo_id) |
| 1097 | await _drain_prebuild_job(db_session, repo_id) |
| 1098 | |
| 1099 | # (a) All 10 mpack_keys must appear in MusehubMPackIndex. |
| 1100 | for i, mk in enumerate(mpack_keys, start=1): |
| 1101 | cnt = (await db_session.execute( |
| 1102 | select(func.count()).select_from(MusehubMPackIndex).where( |
| 1103 | MusehubMPackIndex.mpack_id == mk, |
| 1104 | MusehubMPackIndex.entity_type == "object", |
| 1105 | ) |
| 1106 | )).scalar() |
| 1107 | assert cnt > 0, ( |
| 1108 | f"MusehubMPackIndex has no rows for commit {i} (mpack_key={mk[:20]}). " |
| 1109 | f"process_mpack_index_job for this push was either never enqueued " |
| 1110 | f"(RC-3 Face 1) or silently skipped during partial drain." |
| 1111 | ) |
| 1112 | |
| 1113 | # (b) Clone cache must point to commit 10 (live branch tip at prebuild run time). |
| 1114 | tip_10 = cids[n - 1] |
| 1115 | cache_row = (await db_session.execute( |
| 1116 | select(MusehubFetchMPackCache).where( |
| 1117 | MusehubFetchMPackCache.repo_id == repo_id, |
| 1118 | MusehubFetchMPackCache.tip_commit_id == tip_10, |
| 1119 | ) |
| 1120 | )).scalar_one_or_none() |
| 1121 | assert cache_row is not None, ( |
| 1122 | f"MusehubFetchMPackCache has no row for tip_commit_id=commit_10={tip_10[:20]}. " |
| 1123 | f"The prebuild ran with a stale tip (RC-3 Face 2 not fixed) or " |
| 1124 | f"wire_fetch_mpack raised FetchNotIndexedError because some push mpack " |
| 1125 | f"was not indexed (RC-3 Face 1 not fixed)." |
| 1126 | ) |
File History
1 commit
sha256:0032d6cfa33bc3c8367436ad768e7dd0e339b4332153160247da8266cb5fa352
Merge branch 'task/version-tags-phase3-server' into dev
Human
17 days ago