gabriel / musehub public
test_mpack_phase2.py python
304 lines 11.3 KB
Raw
sha256:009b5a222314f47640a58d75ce5a1f428f1624cf0b51384dfcdfbdfab3cc42a4 feat: migration idempotency, file attribution DAG walk, mpa… Sonnet 4.6 minor ⚠ breaking 44 days ago
1 """TDD — Phase 2: remove per-object MinIO writes from background job (issue #69).
2
3 After Phase 1 the fetch path reads objects from the covering mpack.
4 Phase 2 removes the bottleneck: the background job's parallel MinIO PUTs
5 that took ~25 s of the ~40 s XL job latency.
6
7 After this change:
8 - process_mpack_index_job must NOT call backend.put(oid, data) for any object
9 - MusehubObject rows must have storage_uri='mpack://{mpack_key}' (not a MinIO URI)
10 - MPackIndex rows must exist for all objects (required by Phase 1 fetch)
11 - wire_fetch_mpack must still serve correct data via the mpack path
12
13 Test IDs:
14 P2-1 process_mpack_index_job makes zero per-object backend.put() calls
15 P2-2 MusehubObject.storage_uri set to mpack URI after background job
16 P2-3 full round-trip: push-style mpack → bg job → fetch mpack → correct objects served
17 """
18 from __future__ import annotations
19
20 import hashlib
21 from collections.abc import Mapping
22 from datetime import datetime, timezone
23 from unittest.mock import AsyncMock, MagicMock, call
24
25 import msgpack
26 import pytest
27 import zstandard
28 from sqlalchemy.dialects.postgresql import insert as pg_insert
29 from sqlalchemy.ext.asyncio import AsyncSession
30 from sqlalchemy import select
31
32 from muse.core.types import blob_id, fake_id
33 from musehub.db import musehub_repo_models as db
34 from musehub.db.musehub_jobs_models import MusehubBackgroundJob
35 from musehub.services.musehub_wire import wire_fetch_mpack
36 from musehub.services.musehub_wire import process_mpack_index_job
37 from tests.factories import create_repo
38
39
40 def _now() -> datetime:
41 return datetime.now(tz=timezone.utc)
42
43
44 def _mpack_id(raw: bytes) -> str:
45 return "sha256:" + hashlib.sha256(raw).hexdigest()
46
47
48 def _build_push_mpack(objects: Mapping[str, bytes]) -> bytes:
49 """Build a realistic push mpack with zstd-compressed objects."""
50 cctx = zstandard.ZstdCompressor()
51 entries = [
52 {"object_id": oid, "encoding": "zstd", "content": cctx.compress(data)}
53 for oid, data in objects.items()
54 ]
55 return msgpack.packb(
56 {"commits": [], "snapshots": [], "blobs": entries, "branch_heads": {}},
57 use_bin_type=True,
58 )
59
60
61 class _FakeBackend:
62 """Records all put() calls so tests can assert they were NOT made."""
63
64 def __init__(self, mpack_store: Mapping[str, bytes]) -> None:
65 self._mpacks = mpack_store
66 self._objects: dict[str, bytes] = {}
67 self.put_calls: list[str] = []
68
69 async def put(self, oid: str, data: bytes) -> str:
70 self.put_calls.append(oid)
71 self._objects[oid] = data
72 return f"mem://{oid}"
73
74 async def get(self, oid: str) -> bytes | None:
75 return self._objects.get(oid)
76
77 async def get_mpack(self, mpack_id: str) -> bytes | None:
78 return self._mpacks.get(mpack_id)
79
80 async def put_mpack(self, mpack_id: str, data: bytes) -> None:
81 self._mpacks[mpack_id] = data
82
83 async def exists(self, oid: str) -> bool:
84 return oid in self._objects
85
86 async def delete(self, oid: str) -> None:
87 self._objects.pop(oid, None)
88
89 async def presign_get(self, oid: str, ttl: int) -> str:
90 return f"https://minio.test/{oid}"
91
92 async def presign_mpack_get(self, mpack_id: str, ttl: int) -> str:
93 return f"https://minio.test/mpacks/{mpack_id}"
94
95 async def quarantine_mpack(self, mpack_key: str) -> None:
96 pass
97
98 def uri_for(self, oid: str) -> str:
99 return f"mem://{oid}"
100
101 supports_presign: bool = True
102
103
104 async def _make_job(
105 session: AsyncSession,
106 repo_id: str,
107 mpack_key: str,
108 n_objects: int,
109 ) -> str:
110 """Insert a mpack.index background job row and return its job_id."""
111 from musehub.core.genesis import compute_job_id
112 now = datetime.now(tz=timezone.utc)
113 job_id = compute_job_id(repo_id, "mpack.index", now.isoformat())
114 session.add(MusehubBackgroundJob(
115 job_id=job_id,
116 repo_id=repo_id,
117 job_type="mpack.index",
118 payload={
119 "mpack_key": mpack_key,
120 "branch": "main",
121 "head": "",
122 "pusher_id": "gabriel",
123 "declared_objects_count": n_objects,
124 "declared_commits_count": 0,
125 },
126 status="pending",
127 created_at=now,
128 attempt=0,
129 ))
130 await session.commit()
131 return job_id
132
133
134 # ── P2-1: no per-object backend.put() calls ───────────────────────────────────
135
136 @pytest.mark.asyncio
137 async def test_p2_1_no_per_object_put_calls(
138 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
139 ) -> None:
140 """process_mpack_index_job must make zero per-object backend.put() calls.
141
142 After Phase 2, objects are served from the covering mpack. Writing
143 per-object MinIO keys is the bottleneck (~25s of ~40s XL job). Removing
144 them eliminates the wait without breaking fetch (Phase 1 serves from mpack).
145 """
146 raw_a = b"object content alpha"
147 raw_b = b"object content beta"
148 oid_a = blob_id(raw_a)
149 oid_b = blob_id(raw_b)
150
151 mpack_bytes = _build_push_mpack({oid_a: raw_a, oid_b: raw_b})
152 mpack_key = _mpack_id(mpack_bytes)
153
154 mpack_store: dict[str, bytes] = {mpack_key: mpack_bytes}
155 backend = _FakeBackend(mpack_store)
156 monkeypatch.setattr("musehub.storage.backends.get_backend", lambda: backend)
157 monkeypatch.setattr("musehub.storage.get_backend", lambda: backend)
158 monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend)
159
160 repo = await create_repo(db_session, owner="gabriel", visibility="public")
161 job_id = await _make_job(db_session, repo.repo_id, mpack_key, n_objects=2)
162
163 await process_mpack_index_job(db_session, job_id)
164 await db_session.commit()
165
166 # No per-object puts at all — Phase 2 removes them.
167 assert backend.put_calls == [], (
168 f"process_mpack_index_job called backend.put() for objects: {backend.put_calls}. "
169 f"Phase 2 must not write per-object MinIO keys."
170 )
171
172
173 # ── P2-2: storage_uri set to mpack URI ────────────────────────────────────────
174
175 @pytest.mark.asyncio
176 async def test_p2_2_storage_uri_is_mpack_uri(
177 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
178 ) -> None:
179 """After process_mpack_index_job, MusehubObject.storage_uri must be mpack://... URI.
180
181 Previously it was set to backend.uri_for(oid) (a MinIO object URI) after
182 the per-object PUT. After Phase 2 there is no PUT, so the URI must reflect
183 the actual storage location: the covering mpack.
184 """
185 raw = b"phase2 storage uri test"
186 oid = blob_id(raw)
187
188 mpack_bytes = _build_push_mpack({oid: raw})
189 mpack_key = _mpack_id(mpack_bytes)
190
191 backend = _FakeBackend({mpack_key: mpack_bytes})
192 monkeypatch.setattr("musehub.storage.backends.get_backend", lambda: backend)
193 monkeypatch.setattr("musehub.storage.get_backend", lambda: backend)
194 monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend)
195
196 repo = await create_repo(db_session, owner="gabriel", visibility="public")
197 job_id = await _make_job(db_session, repo.repo_id, mpack_key, n_objects=1)
198
199 await process_mpack_index_job(db_session, job_id)
200 await db_session.commit()
201
202 row = await db_session.get(db.MusehubObject, oid)
203 assert row is not None, f"MusehubObject row missing for {oid[:20]}"
204 assert row.storage_uri.startswith("mpack://"), (
205 f"storage_uri should be mpack://... but got {row.storage_uri!r}. "
206 f"Per-object MinIO key no longer exists after Phase 2."
207 )
208 assert mpack_key in row.storage_uri, (
209 f"storage_uri {row.storage_uri!r} does not reference mpack_key {mpack_key[:20]}"
210 )
211
212
213 # ── P2-3: full round-trip — push → bg job → fetch → correct objects ───────────
214
215 @pytest.mark.asyncio
216 async def test_p2_3_full_roundtrip_no_per_object_keys(
217 db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch
218 ) -> None:
219 """Full round-trip: mpack-in → background job (no MinIO PUT) → wire_fetch_mpack returns correct bytes.
220
221 This is the end-to-end Phase 2 test: push mpack stored in MinIO,
222 background job indexes without per-object puts, fetch serves from mpack.
223 """
224 raw_content = b"roundtrip content for phase2"
225 oid = blob_id(raw_content)
226
227 mpack_bytes = _build_push_mpack({oid: raw_content})
228 mpack_key = _mpack_id(mpack_bytes)
229
230 backend = _FakeBackend({mpack_key: mpack_bytes})
231 monkeypatch.setattr("musehub.storage.backends.get_backend", lambda: backend)
232 monkeypatch.setattr("musehub.storage.get_backend", lambda: backend)
233 monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend)
234
235 repo = await create_repo(db_session, owner="gabriel", visibility="public")
236 job_id = await _make_job(db_session, repo.repo_id, mpack_key, n_objects=1)
237
238 await process_mpack_index_job(db_session, job_id)
239 await db_session.commit()
240
241 # Confirm no per-object MinIO key was written.
242 assert backend.put_calls == [], "Phase 2: no per-object puts expected"
243 assert oid not in backend._objects, "per-object key must not exist after Phase 2 job"
244
245 # Now set up a commit/snapshot referencing this object so wire_fetch_mpack
246 # has something to serve.
247 snap_id = fake_id("snap-p2-roundtrip")
248 snap = db.MusehubSnapshot(
249 snapshot_id=snap_id,
250 manifest_blob=msgpack.packb({"file.txt": oid}, use_bin_type=True),
251 directories=[],
252 entry_count=1,
253 created_at=_now(),
254 )
255 db_session.add(snap)
256 commit_id = fake_id("commit-p2-roundtrip")
257 commit = db.MusehubCommit(
258 commit_id=commit_id,
259 branch="main",
260 parent_ids=[],
261 message="p2 roundtrip",
262 author="gabriel",
263 timestamp=_now(),
264 snapshot_id=snap_id,
265 )
266 db_session.add(commit)
267 await db_session.execute(
268 pg_insert(db.MusehubCommitGraph)
269 .values(commit_id=commit_id, parent_ids=[], generation=0, snapshot_id=snap_id)
270 .on_conflict_do_nothing()
271 )
272 await db_session.execute(
273 pg_insert(db.MusehubCommitRef)
274 .values(repo_id=repo.repo_id, commit_id=commit_id)
275 .on_conflict_do_nothing()
276 )
277 await db_session.commit()
278
279 # wire_fetch_mpack must serve the object from the mpack, not per-object GET.
280 result = await wire_fetch_mpack(
281 db_session, repo.repo_id, want=[commit_id], have=[]
282 )
283
284 assert result["mpack_url"] is not None, "fetch returned up-to-date but should have data"
285 assert result.get("blob_count", result.get("object_count", 0)) == 1
286
287 # Verify the assembled fetch mpack contains the correct bytes.
288 fetch_mpack_id = result["mpack_id"]
289 fetch_raw = backend._mpacks.get(fetch_mpack_id)
290 assert fetch_raw is not None, "assembled fetch mpack not found in backend"
291 if fetch_raw[:4] == b"MUSE":
292 from muse.core.mpack import parse_wire_mpack as _parse_wm
293 payload = _parse_wm(fetch_raw)
294 else:
295 payload = msgpack.unpackb(fetch_raw, raw=False)
296 obj_map = {o["object_id"]: bytes(o["content"]) for o in (payload.get("blobs") or payload.get("objects") or [])}
297
298 assert oid in obj_map, f"object {oid[:20]} missing from fetch mpack"
299 assert obj_map[oid] == raw_content, (
300 "object content from Phase 2 mpack-only path must match original bytes"
301 )
302
303 # Per-object GET must not have been called — the mpack path served it.
304 assert oid not in backend.put_calls
File History 1 commit
sha256:009b5a222314f47640a58d75ce5a1f428f1624cf0b51384dfcdfbdfab3cc42a4 feat: migration idempotency, file attribution DAG walk, mpa… Sonnet 4.6 minor ⚠ 44 days ago