test_mwp9_running_job_dedup.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """MWP-9 — fetch.mpack.prebuild dedup must cover 'running', not just 'pending'. |
| 2 | |
| 3 | Root cause (found live, while running a bulk commit-signature repair against |
| 4 | staging.musehub.ai, during the muse#108 wire-protocol investigation): |
| 5 | ``fetch.mpack.prebuild`` is explicitly self-coalescing by design -- |
| 6 | ``process_fetch_mpack_prebuild_job`` re-reads live branch tips at run time |
| 7 | rather than trusting its enqueue-time payload (see its own docstring). But |
| 8 | every enqueue site (the regular post-push path in ``enqueue_push_intel``, |
| 9 | and the repair path in ``wire_repair_commit`` -> ``_enqueue_repair_prebuild``) |
| 10 | only dedups against jobs with ``status == "pending"``. |
| 11 | |
| 12 | For a repo whose full rebuild takes longer than the gap between two triggers |
| 13 | (confirmed live: ~1624 commits / 162MB took 30-100+s to rebuild under load), |
| 14 | the first job transitions pending -> running, and every subsequent trigger |
| 15 | (another push, another repair-commit call) sees no "pending" row and happily |
| 16 | enqueues a brand new one -- even though the currently-running job will, |
| 17 | by its own self-coalescing design, already produce a mpack covering |
| 18 | whatever triggered the new enqueue. Each of those redundant jobs is a full, |
| 19 | independent rebuild. Under rapid repeated triggers (a bulk repair script |
| 20 | issuing 180+ calls in a few minutes, or just several pushes in quick |
| 21 | succession) this floods the single-worker queue and starves ordinary |
| 22 | clone/fetch requests past their 120s retry budget -- confirmed live: 3 of 5 |
| 23 | consecutive clone attempts hit 4x 503 retries and failed outright during a |
| 24 | repair batch run. |
| 25 | |
| 26 | Fix: for this specific, explicitly self-coalescing job type, dedup against |
| 27 | both 'pending' and 'running' at both enqueue sites. |
| 28 | """ |
| 29 | from __future__ import annotations |
| 30 | |
| 31 | import hashlib |
| 32 | from datetime import datetime, timedelta, timezone |
| 33 | |
| 34 | import pytest |
| 35 | from sqlalchemy import select |
| 36 | from sqlalchemy.ext.asyncio import AsyncSession |
| 37 | |
| 38 | from musehub.core.genesis import compute_job_id |
| 39 | from musehub.db.musehub_jobs_models import MusehubBackgroundJob |
| 40 | from musehub.services.musehub_jobs import ( |
| 41 | _MAX_ATTEMPTS, |
| 42 | enqueue_push_intel, |
| 43 | reclaim_stale_jobs, |
| 44 | ) |
| 45 | from musehub.services.musehub_wire_push import _enqueue_repair_prebuild |
| 46 | from tests.factories import create_branch, create_repo |
| 47 | |
| 48 | |
| 49 | def _fake_commit_id(seed: str) -> str: |
| 50 | return "sha256:" + hashlib.sha256(seed.encode()).hexdigest() |
| 51 | |
| 52 | |
| 53 | def _now() -> datetime: |
| 54 | return datetime.now(tz=timezone.utc) |
| 55 | |
| 56 | |
| 57 | async def _insert_prebuild_job(session: AsyncSession, repo_id: str, status: str) -> str: |
| 58 | now = _now() |
| 59 | job_id = compute_job_id(repo_id, "fetch.mpack.prebuild", now.isoformat() + status) |
| 60 | session.add(MusehubBackgroundJob( |
| 61 | job_id=job_id, |
| 62 | repo_id=repo_id, |
| 63 | job_type="fetch.mpack.prebuild", |
| 64 | payload={"tip_commit_ids": [_fake_commit_id("a")]}, |
| 65 | status=status, |
| 66 | created_at=now, |
| 67 | attempt=0, |
| 68 | )) |
| 69 | await session.commit() |
| 70 | return job_id |
| 71 | |
| 72 | |
| 73 | async def _count_prebuild_jobs(session: AsyncSession, repo_id: str) -> int: |
| 74 | rows = (await session.execute( |
| 75 | select(MusehubBackgroundJob.job_id).where( |
| 76 | MusehubBackgroundJob.repo_id == repo_id, |
| 77 | MusehubBackgroundJob.job_type == "fetch.mpack.prebuild", |
| 78 | ) |
| 79 | )).scalars().all() |
| 80 | return len(rows) |
| 81 | |
| 82 | |
| 83 | @pytest.mark.asyncio |
| 84 | async def test_mwp9_01_pending_prebuild_still_deduped_on_repair(db_session: AsyncSession) -> None: |
| 85 | """GREEN guard: existing 'pending' dedup must keep working unchanged.""" |
| 86 | repo = await create_repo(db_session) |
| 87 | await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip")) |
| 88 | await _insert_prebuild_job(db_session, repo.repo_id, status="pending") |
| 89 | |
| 90 | await _enqueue_repair_prebuild(db_session, repo.repo_id) |
| 91 | await db_session.commit() |
| 92 | |
| 93 | assert await _count_prebuild_jobs(db_session, repo.repo_id) == 1 |
| 94 | |
| 95 | |
| 96 | @pytest.mark.asyncio |
| 97 | async def test_mwp9_07_dedup_skip_against_running_job_stamps_superseded_at( |
| 98 | db_session: AsyncSession, |
| 99 | ) -> None: |
| 100 | """A dedup-skip against a 'running' row must stamp its superseded_at so |
| 101 | the handler can detect that its in-progress data read may predate this |
| 102 | new trigger's write (see test_fetch_mpack_prebuild.py's MWP9 tests for |
| 103 | the handler-side half of this fix).""" |
| 104 | repo = await create_repo(db_session) |
| 105 | await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip")) |
| 106 | job_id = await _insert_prebuild_job(db_session, repo.repo_id, status="running") |
| 107 | |
| 108 | await _enqueue_repair_prebuild(db_session, repo.repo_id) |
| 109 | await db_session.commit() |
| 110 | |
| 111 | row = (await db_session.execute( |
| 112 | select(MusehubBackgroundJob).where(MusehubBackgroundJob.job_id == job_id) |
| 113 | )).scalar_one() |
| 114 | assert row.superseded_at is not None |
| 115 | |
| 116 | |
| 117 | @pytest.mark.asyncio |
| 118 | async def test_mwp9_02_running_prebuild_deduped_on_repair(db_session: AsyncSession) -> None: |
| 119 | """RED before the fix: a 'running' prebuild job is not covered by the |
| 120 | current dedup, so repair-commit enqueues a redundant second job.""" |
| 121 | repo = await create_repo(db_session) |
| 122 | await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip")) |
| 123 | await _insert_prebuild_job(db_session, repo.repo_id, status="running") |
| 124 | |
| 125 | await _enqueue_repair_prebuild(db_session, repo.repo_id) |
| 126 | await db_session.commit() |
| 127 | |
| 128 | assert await _count_prebuild_jobs(db_session, repo.repo_id) == 1, ( |
| 129 | "repair-commit enqueued a redundant fetch.mpack.prebuild job while one " |
| 130 | "was already running — the exact flooding mechanism that caused live " |
| 131 | "clone failures during a bulk repair run" |
| 132 | ) |
| 133 | |
| 134 | |
| 135 | @pytest.mark.asyncio |
| 136 | async def test_mwp9_03_running_prebuild_deduped_on_regular_push(db_session: AsyncSession) -> None: |
| 137 | """RED before the fix: the regular post-push path has the identical gap -- |
| 138 | confirms this isn't a repair-script-only issue, it's the baseline |
| 139 | push-time behavior under any rapid repeated push load.""" |
| 140 | repo = await create_repo(db_session) |
| 141 | await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip")) |
| 142 | await _insert_prebuild_job(db_session, repo.repo_id, status="running") |
| 143 | |
| 144 | await enqueue_push_intel( |
| 145 | db_session, repo.repo_id, head=_fake_commit_id("tip"), |
| 146 | domain_id="code", branch="main", |
| 147 | ) |
| 148 | await db_session.commit() |
| 149 | |
| 150 | assert await _count_prebuild_jobs(db_session, repo.repo_id) == 1, ( |
| 151 | "a regular push enqueued a redundant fetch.mpack.prebuild job while " |
| 152 | "one was already running" |
| 153 | ) |
| 154 | |
| 155 | |
| 156 | @pytest.mark.asyncio |
| 157 | async def test_mwp9_04_other_coarse_job_types_unaffected(db_session: AsyncSession) -> None: |
| 158 | """Scope guard: the fix must be narrow to fetch.mpack.prebuild. A |
| 159 | 'running' gc job (not self-coalescing by any documented design) must |
| 160 | still allow a fresh 'pending' gc job to be enqueued, exactly as before.""" |
| 161 | repo = await create_repo(db_session) |
| 162 | await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip")) |
| 163 | now = _now() |
| 164 | db_session.add(MusehubBackgroundJob( |
| 165 | job_id=compute_job_id(repo.repo_id, "gc", now.isoformat()), |
| 166 | repo_id=repo.repo_id, job_type="gc", payload={}, |
| 167 | status="running", created_at=now, attempt=0, |
| 168 | )) |
| 169 | await db_session.commit() |
| 170 | |
| 171 | await enqueue_push_intel( |
| 172 | db_session, repo.repo_id, head=_fake_commit_id("tip"), |
| 173 | domain_id="code", branch="main", |
| 174 | ) |
| 175 | await db_session.commit() |
| 176 | |
| 177 | gc_rows = (await db_session.execute( |
| 178 | select(MusehubBackgroundJob.job_id).where( |
| 179 | MusehubBackgroundJob.repo_id == repo.repo_id, |
| 180 | MusehubBackgroundJob.job_type == "gc", |
| 181 | ) |
| 182 | )).scalars().all() |
| 183 | assert len(gc_rows) == 2, ( |
| 184 | "gc's dedup window must remain pending-only -- broadening it to " |
| 185 | "'running' for every coarse job type was explicitly out of scope" |
| 186 | ) |
| 187 | |
| 188 | |
| 189 | @pytest.mark.asyncio |
| 190 | async def test_mwp9_05_exhausted_stale_running_job_marked_failed_not_left_running( |
| 191 | db_session: AsyncSession, |
| 192 | ) -> None: |
| 193 | """RED before the fix (found live on staging, 2026-10-21): two |
| 194 | fetch.mpack.prebuild rows have sat at status='running', attempt=3 |
| 195 | (== _MAX_ATTEMPTS) since July and September -- a worker crash/restart |
| 196 | mid-job never called fail_job, and reclaim_stale_jobs explicitly |
| 197 | excludes attempt >= _MAX_ATTEMPTS from its reset-to-pending query, so |
| 198 | these rows are permanently stuck in 'running'. |
| 199 | |
| 200 | This was invisible before MWP9_02/03 because the old dedup only ever |
| 201 | looked at 'pending'. Once dedup also treats 'running' as coverage (the |
| 202 | whole point of this file), an exhausted zombie 'running' row permanently |
| 203 | blocks every future trigger for that job type on that repo -- staging's |
| 204 | clone has been wedged on exactly this since this fix went live. |
| 205 | |
| 206 | reclaim_stale_jobs must terminally fail exhausted stale jobs (not just |
| 207 | silently skip them) so they stop blocking dedup forever, while jobs |
| 208 | still under budget keep retrying via the existing reset-to-pending path. |
| 209 | """ |
| 210 | repo = await create_repo(db_session) |
| 211 | await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip")) |
| 212 | |
| 213 | stale_claimed_at = _now() - timedelta(minutes=30) |
| 214 | job_id = compute_job_id(repo.repo_id, "fetch.mpack.prebuild", "zombie") |
| 215 | db_session.add(MusehubBackgroundJob( |
| 216 | job_id=job_id, |
| 217 | repo_id=repo.repo_id, |
| 218 | job_type="fetch.mpack.prebuild", |
| 219 | payload={"tip_commit_ids": [_fake_commit_id("a")]}, |
| 220 | status="running", |
| 221 | created_at=stale_claimed_at, |
| 222 | claimed_at=stale_claimed_at, |
| 223 | attempt=_MAX_ATTEMPTS, |
| 224 | )) |
| 225 | await db_session.commit() |
| 226 | |
| 227 | await reclaim_stale_jobs(db_session) |
| 228 | await db_session.commit() |
| 229 | |
| 230 | row = (await db_session.execute( |
| 231 | select(MusehubBackgroundJob).where(MusehubBackgroundJob.job_id == job_id) |
| 232 | )).scalar_one() |
| 233 | assert row.status == "failed", ( |
| 234 | "an exhausted stale job must be terminally failed, not left in " |
| 235 | "'running' forever -- a permanent zombie that blocks dedup for this " |
| 236 | "job type/repo indefinitely" |
| 237 | ) |
| 238 | |
| 239 | |
| 240 | @pytest.mark.asyncio |
| 241 | async def test_mwp9_06_stale_running_job_under_budget_still_resets_to_pending( |
| 242 | db_session: AsyncSession, |
| 243 | ) -> None: |
| 244 | """GREEN guard: the existing retry behavior for stale jobs that still |
| 245 | have attempts remaining must be unchanged by the exhausted-attempt fix.""" |
| 246 | repo = await create_repo(db_session) |
| 247 | await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip")) |
| 248 | |
| 249 | stale_claimed_at = _now() - timedelta(minutes=30) |
| 250 | job_id = compute_job_id(repo.repo_id, "fetch.mpack.prebuild", "retryable") |
| 251 | db_session.add(MusehubBackgroundJob( |
| 252 | job_id=job_id, |
| 253 | repo_id=repo.repo_id, |
| 254 | job_type="fetch.mpack.prebuild", |
| 255 | payload={"tip_commit_ids": [_fake_commit_id("a")]}, |
| 256 | status="running", |
| 257 | created_at=stale_claimed_at, |
| 258 | claimed_at=stale_claimed_at, |
| 259 | attempt=_MAX_ATTEMPTS - 1, |
| 260 | )) |
| 261 | await db_session.commit() |
| 262 | |
| 263 | await reclaim_stale_jobs(db_session) |
| 264 | await db_session.commit() |
| 265 | |
| 266 | row = (await db_session.execute( |
| 267 | select(MusehubBackgroundJob).where(MusehubBackgroundJob.job_id == job_id) |
| 268 | )).scalar_one() |
| 269 | assert row.status == "pending" |