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