gabriel / musehub public
test_mwp9_running_job_dedup.py python
161 lines 6.7 KB
Raw
sha256:195787044debd61f93cc31b92754ee0f78e1b6f66757ee07ea5058ed5ce5197d fix: dedup fetch.mpack.prebuild against running jobs, not j… Sonnet 5 minor ⚠ breaking 1 day ago
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, 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 enqueue_push_intel
41 from musehub.services.musehub_wire_push import _enqueue_repair_prebuild
42 from tests.factories import create_branch, create_repo
43
44
45 def _fake_commit_id(seed: str) -> str:
46 return "sha256:" + hashlib.sha256(seed.encode()).hexdigest()
47
48
49 def _now() -> datetime:
50 return datetime.now(tz=timezone.utc)
51
52
53 async def _insert_prebuild_job(session: AsyncSession, repo_id: str, status: str) -> str:
54 now = _now()
55 job_id = compute_job_id(repo_id, "fetch.mpack.prebuild", now.isoformat() + status)
56 session.add(MusehubBackgroundJob(
57 job_id=job_id,
58 repo_id=repo_id,
59 job_type="fetch.mpack.prebuild",
60 payload={"tip_commit_ids": [_fake_commit_id("a")]},
61 status=status,
62 created_at=now,
63 attempt=0,
64 ))
65 await session.commit()
66 return job_id
67
68
69 async def _count_prebuild_jobs(session: AsyncSession, repo_id: str) -> int:
70 rows = (await session.execute(
71 select(MusehubBackgroundJob.job_id).where(
72 MusehubBackgroundJob.repo_id == repo_id,
73 MusehubBackgroundJob.job_type == "fetch.mpack.prebuild",
74 )
75 )).scalars().all()
76 return len(rows)
77
78
79 @pytest.mark.asyncio
80 async def test_mwp9_01_pending_prebuild_still_deduped_on_repair(db_session: AsyncSession) -> None:
81 """GREEN guard: existing 'pending' dedup must keep working unchanged."""
82 repo = await create_repo(db_session)
83 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip"))
84 await _insert_prebuild_job(db_session, repo.repo_id, status="pending")
85
86 await _enqueue_repair_prebuild(db_session, repo.repo_id)
87 await db_session.commit()
88
89 assert await _count_prebuild_jobs(db_session, repo.repo_id) == 1
90
91
92 @pytest.mark.asyncio
93 async def test_mwp9_02_running_prebuild_deduped_on_repair(db_session: AsyncSession) -> None:
94 """RED before the fix: a 'running' prebuild job is not covered by the
95 current dedup, so repair-commit enqueues a redundant second job."""
96 repo = await create_repo(db_session)
97 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip"))
98 await _insert_prebuild_job(db_session, repo.repo_id, status="running")
99
100 await _enqueue_repair_prebuild(db_session, repo.repo_id)
101 await db_session.commit()
102
103 assert await _count_prebuild_jobs(db_session, repo.repo_id) == 1, (
104 "repair-commit enqueued a redundant fetch.mpack.prebuild job while one "
105 "was already running — the exact flooding mechanism that caused live "
106 "clone failures during a bulk repair run"
107 )
108
109
110 @pytest.mark.asyncio
111 async def test_mwp9_03_running_prebuild_deduped_on_regular_push(db_session: AsyncSession) -> None:
112 """RED before the fix: the regular post-push path has the identical gap --
113 confirms this isn't a repair-script-only issue, it's the baseline
114 push-time behavior under any rapid repeated push load."""
115 repo = await create_repo(db_session)
116 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip"))
117 await _insert_prebuild_job(db_session, repo.repo_id, status="running")
118
119 await enqueue_push_intel(
120 db_session, repo.repo_id, head=_fake_commit_id("tip"),
121 domain_id="code", branch="main",
122 )
123 await db_session.commit()
124
125 assert await _count_prebuild_jobs(db_session, repo.repo_id) == 1, (
126 "a regular push enqueued a redundant fetch.mpack.prebuild job while "
127 "one was already running"
128 )
129
130
131 @pytest.mark.asyncio
132 async def test_mwp9_04_other_coarse_job_types_unaffected(db_session: AsyncSession) -> None:
133 """Scope guard: the fix must be narrow to fetch.mpack.prebuild. A
134 'running' gc job (not self-coalescing by any documented design) must
135 still allow a fresh 'pending' gc job to be enqueued, exactly as before."""
136 repo = await create_repo(db_session)
137 await create_branch(db_session, repo.repo_id, name="main", head_commit_id=_fake_commit_id("tip"))
138 now = _now()
139 db_session.add(MusehubBackgroundJob(
140 job_id=compute_job_id(repo.repo_id, "gc", now.isoformat()),
141 repo_id=repo.repo_id, job_type="gc", payload={},
142 status="running", created_at=now, attempt=0,
143 ))
144 await db_session.commit()
145
146 await enqueue_push_intel(
147 db_session, repo.repo_id, head=_fake_commit_id("tip"),
148 domain_id="code", branch="main",
149 )
150 await db_session.commit()
151
152 gc_rows = (await db_session.execute(
153 select(MusehubBackgroundJob.job_id).where(
154 MusehubBackgroundJob.repo_id == repo.repo_id,
155 MusehubBackgroundJob.job_type == "gc",
156 )
157 )).scalars().all()
158 assert len(gc_rows) == 2, (
159 "gc's dedup window must remain pending-only -- broadening it to "
160 "'running' for every coarse job type was explicitly out of scope"
161 )
File History 1 commit
sha256:195787044debd61f93cc31b92754ee0f78e1b6f66757ee07ea5058ed5ce5197d fix: dedup fetch.mpack.prebuild against running jobs, not j… Sonnet 5 minor ⚠ 1 day ago