test_fetch_mpack_native_phase1.py
python
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595
fix: typing audit — 0 violations, 0 untyped defs across all…
Sonnet 4.6
minor
⚠ breaking
49 days ago
| 1 | """TDD — Phase 1: mpack-native object serving in wire_fetch_mpack (issue #69). |
| 2 | |
| 3 | Problem: the current fetch path calls backend.get(oid) once per object. |
| 4 | For 25,000 objects that is 25,000 individual MinIO GETs. This only works |
| 5 | because the push background job writes per-object MinIO keys — a 40s |
| 6 | synchronous cost that creates an availability window after every large push. |
| 7 | |
| 8 | Goal: serve objects by downloading the covering mpack(s) and extracting |
| 9 | in-memory. One mpack GET replaces N object GETs. Per-object keys are |
| 10 | never written; the mpack is the source of truth. |
| 11 | |
| 12 | Test IDs: |
| 13 | FN-1 content_cache hit → no mpack or per-object GET needed |
| 14 | FN-2 object stored only in mpack (no per-object key) → extracted correctly |
| 15 | FN-3 objects across two distinct mpacks → each mpack fetched exactly once |
| 16 | FN-4 object in MPackIndex but mpack is missing from storage → falls back to per-object GET |
| 17 | FN-5 object absent from MPackIndex → FetchNotIndexedError (unchanged gate) |
| 18 | FN-6 mixed: content_cache + mpack + legacy per-object → all three sources merged |
| 19 | """ |
| 20 | from __future__ import annotations |
| 21 | |
| 22 | import hashlib |
| 23 | from collections.abc import Mapping |
| 24 | from datetime import datetime, timezone |
| 25 | from unittest.mock import AsyncMock, call |
| 26 | |
| 27 | import msgpack |
| 28 | import pytest |
| 29 | from sqlalchemy.dialects.postgresql import insert as pg_insert |
| 30 | from sqlalchemy.ext.asyncio import AsyncSession |
| 31 | |
| 32 | from muse.core.types import blob_id, fake_id |
| 33 | from musehub.db import musehub_repo_models as db |
| 34 | from musehub.services.musehub_wire import FetchNotIndexedError, wire_fetch_mpack |
| 35 | from tests.factories import create_repo |
| 36 | |
| 37 | |
| 38 | # ── fixtures & helpers ──────────────────────────────────────────────────────── |
| 39 | |
| 40 | def _now() -> datetime: |
| 41 | return datetime.now(tz=timezone.utc) |
| 42 | |
| 43 | |
| 44 | def _make_mpack_bytes(objects: Mapping[str, bytes]) -> bytes: |
| 45 | """Encode objects into a wire mpack payload (uncompressed).""" |
| 46 | return msgpack.packb( |
| 47 | { |
| 48 | "commits": [], |
| 49 | "snapshots": [], |
| 50 | "blobs": [{"object_id": oid, "content": data} for oid, data in objects.items()], |
| 51 | "branch_heads": {}, |
| 52 | }, |
| 53 | use_bin_type=True, |
| 54 | ) |
| 55 | |
| 56 | |
| 57 | def _make_mpack_bytes_zstd(objects: Mapping[str, bytes]) -> bytes: |
| 58 | """Encode objects into a wire mpack payload with zstd-compressed content. |
| 59 | |
| 60 | This mimics the real push mpack format: each object entry has |
| 61 | encoding='zstd' and content=compressed_bytes, but object_id is the |
| 62 | sha256 of the *uncompressed* bytes (content-addressed). |
| 63 | """ |
| 64 | import zstandard as _zstd |
| 65 | cctx = _zstd.ZstdCompressor() |
| 66 | entries = [] |
| 67 | for oid, data in objects.items(): |
| 68 | entries.append({ |
| 69 | "object_id": oid, |
| 70 | "encoding": "zstd", |
| 71 | "content": cctx.compress(data), |
| 72 | }) |
| 73 | return msgpack.packb( |
| 74 | { |
| 75 | "commits": [], |
| 76 | "snapshots": [], |
| 77 | "blobs": entries, |
| 78 | "branch_heads": {}, |
| 79 | }, |
| 80 | use_bin_type=True, |
| 81 | ) |
| 82 | |
| 83 | |
| 84 | def _mpack_key(payload: bytes) -> str: |
| 85 | return "sha256:" + hashlib.sha256(payload).hexdigest() |
| 86 | |
| 87 | |
| 88 | class _FakeBackend: |
| 89 | """In-memory backend with separate per-object and mpack namespaces. |
| 90 | |
| 91 | Tracks how many times get() and get_mpack() are called so tests can |
| 92 | assert that the mpack-native path is taken instead of per-object GETs. |
| 93 | """ |
| 94 | |
| 95 | def __init__(self) -> None: |
| 96 | self._objects: dict[str, bytes] = {} |
| 97 | self._mpacks: dict[str, bytes] = {} |
| 98 | self.get_calls: list[str] = [] |
| 99 | self.get_mpack_calls: list[str] = [] |
| 100 | |
| 101 | async def put(self, oid: str, data: bytes) -> str: |
| 102 | self._objects[oid] = data |
| 103 | return f"mem://{oid}" |
| 104 | |
| 105 | async def get(self, oid: str) -> bytes | None: |
| 106 | self.get_calls.append(oid) |
| 107 | return self._objects.get(oid) |
| 108 | |
| 109 | async def get_mpack(self, mpack_id: str) -> bytes | None: |
| 110 | self.get_mpack_calls.append(mpack_id) |
| 111 | return self._mpacks.get(mpack_id) |
| 112 | |
| 113 | async def put_mpack(self, mpack_id: str, data: bytes) -> None: |
| 114 | self._mpacks[mpack_id] = data |
| 115 | |
| 116 | async def exists(self, oid: str) -> bool: |
| 117 | return oid in self._objects |
| 118 | |
| 119 | async def delete(self, oid: str) -> None: |
| 120 | self._objects.pop(oid, None) |
| 121 | |
| 122 | async def presign_get(self, oid: str, ttl: int) -> str: |
| 123 | return f"https://minio.test/{oid}?ttl={ttl}" |
| 124 | |
| 125 | async def presign_mpack_get(self, mpack_id: str, ttl: int) -> str: |
| 126 | return f"https://minio.test/mpacks/{mpack_id}?ttl={ttl}" |
| 127 | |
| 128 | def uri_for(self, oid: str) -> str: |
| 129 | return f"mem://{oid}" |
| 130 | |
| 131 | supports_presign: bool = True |
| 132 | |
| 133 | |
| 134 | async def _make_commit( |
| 135 | session: AsyncSession, |
| 136 | repo_id: str, |
| 137 | *, |
| 138 | manifest: dict[str, str], |
| 139 | seed: str, |
| 140 | parent_ids: list[str] | None = None, |
| 141 | generation: int = 0, |
| 142 | ) -> db.MusehubCommit: |
| 143 | from muse.core.ids import hash_snapshot as _hash_snapshot |
| 144 | snap_id = _hash_snapshot(manifest, []) |
| 145 | snap = db.MusehubSnapshot( |
| 146 | snapshot_id=snap_id, |
| 147 | directories=[], |
| 148 | manifest_blob=msgpack.packb(manifest, use_bin_type=True), |
| 149 | entry_count=len(manifest), |
| 150 | created_at=_now(), |
| 151 | ) |
| 152 | session.add(snap) |
| 153 | await session.execute( |
| 154 | pg_insert(db.MusehubSnapshotRef) |
| 155 | .values(repo_id=repo_id, snapshot_id=snap_id) |
| 156 | .on_conflict_do_nothing() |
| 157 | ) |
| 158 | commit_id = fake_id(f"commit-{seed}") |
| 159 | commit = db.MusehubCommit( |
| 160 | commit_id=commit_id, |
| 161 | branch="main", |
| 162 | parent_ids=parent_ids or [], |
| 163 | message=f"commit {seed}", |
| 164 | author="gabriel", |
| 165 | timestamp=_now(), |
| 166 | snapshot_id=snap_id, |
| 167 | ) |
| 168 | session.add(commit) |
| 169 | await session.execute( |
| 170 | pg_insert(db.MusehubCommitRef) |
| 171 | .values(repo_id=repo_id, commit_id=commit_id) |
| 172 | .on_conflict_do_nothing() |
| 173 | ) |
| 174 | await session.execute( |
| 175 | pg_insert(db.MusehubCommitGraph) |
| 176 | .values( |
| 177 | commit_id=commit_id, |
| 178 | parent_ids=parent_ids or [], |
| 179 | generation=generation, |
| 180 | snapshot_id=snap_id, |
| 181 | ) |
| 182 | .on_conflict_do_nothing() |
| 183 | ) |
| 184 | await session.commit() |
| 185 | return commit |
| 186 | |
| 187 | |
| 188 | async def _index_object_in_mpack( |
| 189 | session: AsyncSession, |
| 190 | oid: str, |
| 191 | mpack_id: str, |
| 192 | ) -> None: |
| 193 | """Write an MPackIndex row mapping oid → mpack_id (no per-object MinIO key).""" |
| 194 | await session.execute( |
| 195 | pg_insert(db.MusehubObject) |
| 196 | .values( |
| 197 | object_id=oid, |
| 198 | path="file.dat", |
| 199 | size_bytes=8, |
| 200 | storage_uri=f"mpack://{mpack_id}", # mpack URI, not per-object |
| 201 | ) |
| 202 | .on_conflict_do_nothing(index_elements=["object_id"]) |
| 203 | ) |
| 204 | await session.execute( |
| 205 | pg_insert(db.MusehubMPackIndex) |
| 206 | .values(entity_id=oid, mpack_id=mpack_id, entity_type="object", created_at=_now()) |
| 207 | .on_conflict_do_nothing(index_elements=["entity_id", "mpack_id"]) |
| 208 | ) |
| 209 | await session.commit() |
| 210 | |
| 211 | |
| 212 | # ── FN-1: content_cache hit ─────────────────────────────────────────────────── |
| 213 | |
| 214 | @pytest.mark.asyncio |
| 215 | async def test_fn1_content_cache_served_without_mpack_get( |
| 216 | db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch |
| 217 | ) -> None: |
| 218 | """Objects in content_cache are served from DB; no mpack or per-object GET fired.""" |
| 219 | backend = _FakeBackend() |
| 220 | monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) |
| 221 | monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend) |
| 222 | |
| 223 | repo = await create_repo(db_session, owner="gabriel", visibility="public") |
| 224 | raw = b"cached content" |
| 225 | oid = blob_id(raw) |
| 226 | |
| 227 | await db_session.execute( |
| 228 | pg_insert(db.MusehubObject) |
| 229 | .values( |
| 230 | object_id=oid, |
| 231 | path="cached.txt", |
| 232 | size_bytes=len(raw), |
| 233 | storage_uri=f"mem://{oid}", |
| 234 | content_cache=raw, |
| 235 | ) |
| 236 | .on_conflict_do_nothing(index_elements=["object_id"]) |
| 237 | ) |
| 238 | await db_session.execute( |
| 239 | pg_insert(db.MusehubObjectRef) |
| 240 | .values(repo_id=repo.repo_id, object_id=oid) |
| 241 | .on_conflict_do_nothing() |
| 242 | ) |
| 243 | # Also add MPackIndex so the coverage check passes |
| 244 | mpack_bytes = _make_mpack_bytes({oid: raw}) |
| 245 | mpack_id = _mpack_key(mpack_bytes) |
| 246 | await backend.put_mpack(mpack_id, mpack_bytes) |
| 247 | await _index_object_in_mpack(db_session, oid, mpack_id) |
| 248 | |
| 249 | commit = await _make_commit( |
| 250 | db_session, repo.repo_id, manifest={"cached.txt": oid}, seed="fn1" |
| 251 | ) |
| 252 | |
| 253 | result = await wire_fetch_mpack( |
| 254 | db_session, repo.repo_id, want=[commit.commit_id], have=[], force_build=True |
| 255 | ) |
| 256 | |
| 257 | assert result["blob_count"] == 1 |
| 258 | assert backend.get_calls == [], "content_cache hit must not trigger per-object GET" |
| 259 | assert backend.get_mpack_calls == [], "content_cache hit must not trigger mpack GET" |
| 260 | |
| 261 | |
| 262 | # ── FN-2: object stored only in mpack ──────────────────────────────────────── |
| 263 | |
| 264 | @pytest.mark.asyncio |
| 265 | async def test_fn2_object_served_from_mpack_no_per_object_key( |
| 266 | db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch |
| 267 | ) -> None: |
| 268 | """Object exists only in mpack storage (no per-object key) → correctly extracted.""" |
| 269 | backend = _FakeBackend() |
| 270 | monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) |
| 271 | monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend) |
| 272 | |
| 273 | repo = await create_repo(db_session, owner="gabriel", visibility="public") |
| 274 | raw = b"mpack-only content" |
| 275 | oid = blob_id(raw) |
| 276 | |
| 277 | mpack_bytes = _make_mpack_bytes({oid: raw}) |
| 278 | mpack_id = _mpack_key(mpack_bytes) |
| 279 | await backend.put_mpack(mpack_id, mpack_bytes) |
| 280 | await _index_object_in_mpack(db_session, oid, mpack_id) |
| 281 | await db_session.execute( |
| 282 | pg_insert(db.MusehubObjectRef) |
| 283 | .values(repo_id=repo.repo_id, object_id=oid) |
| 284 | .on_conflict_do_nothing() |
| 285 | ) |
| 286 | await db_session.commit() |
| 287 | |
| 288 | commit = await _make_commit( |
| 289 | db_session, repo.repo_id, manifest={"f.txt": oid}, seed="fn2" |
| 290 | ) |
| 291 | |
| 292 | result = await wire_fetch_mpack( |
| 293 | db_session, repo.repo_id, want=[commit.commit_id], have=[], force_build=True |
| 294 | ) |
| 295 | |
| 296 | assert result["blob_count"] == 1 |
| 297 | assert result["mpack_url"] is not None, "should return a presigned mpack URL" |
| 298 | # Per-object key was never written — get() must not have been called |
| 299 | assert backend.get_calls == [], "must not fall back to per-object GET when mpack covers the object" |
| 300 | assert mpack_id in backend.get_mpack_calls, "must fetch the covering mpack" |
| 301 | |
| 302 | # Verify the assembled fetch mpack contains the correct bytes. |
| 303 | from muse.core.mpack import parse_wire_mpack |
| 304 | fetch_mpack_id = result["mpack_id"] |
| 305 | fetch_raw = backend._mpacks.get(fetch_mpack_id) |
| 306 | assert fetch_raw is not None |
| 307 | payload = parse_wire_mpack(fetch_raw) |
| 308 | obj_map = {o["object_id"]: o["content"] for o in payload.get("blobs", [])} |
| 309 | assert obj_map.get(oid) == raw |
| 310 | |
| 311 | |
| 312 | # ── FN-3: objects across two mpacks → each fetched exactly once ─────────────── |
| 313 | |
| 314 | @pytest.mark.asyncio |
| 315 | async def test_fn3_two_mpacks_each_fetched_once( |
| 316 | db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch |
| 317 | ) -> None: |
| 318 | """Objects span two mpacks. Each mpack is downloaded exactly once (no redundant GETs).""" |
| 319 | backend = _FakeBackend() |
| 320 | monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) |
| 321 | monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend) |
| 322 | |
| 323 | repo = await create_repo(db_session, owner="gabriel", visibility="public") |
| 324 | |
| 325 | raw_a = b"object from mpack A" |
| 326 | raw_b = b"object from mpack B" |
| 327 | oid_a = blob_id(raw_a) |
| 328 | oid_b = blob_id(raw_b) |
| 329 | |
| 330 | mpack_a_bytes = _make_mpack_bytes({oid_a: raw_a}) |
| 331 | mpack_b_bytes = _make_mpack_bytes({oid_b: raw_b}) |
| 332 | mpack_a_id = _mpack_key(mpack_a_bytes) |
| 333 | mpack_b_id = _mpack_key(mpack_b_bytes) |
| 334 | |
| 335 | await backend.put_mpack(mpack_a_id, mpack_a_bytes) |
| 336 | await backend.put_mpack(mpack_b_id, mpack_b_bytes) |
| 337 | await _index_object_in_mpack(db_session, oid_a, mpack_a_id) |
| 338 | await _index_object_in_mpack(db_session, oid_b, mpack_b_id) |
| 339 | for oid in (oid_a, oid_b): |
| 340 | await db_session.execute( |
| 341 | pg_insert(db.MusehubObjectRef) |
| 342 | .values(repo_id=repo.repo_id, object_id=oid) |
| 343 | .on_conflict_do_nothing() |
| 344 | ) |
| 345 | await db_session.commit() |
| 346 | |
| 347 | commit = await _make_commit( |
| 348 | db_session, repo.repo_id, |
| 349 | manifest={"a.txt": oid_a, "b.txt": oid_b}, |
| 350 | seed="fn3", |
| 351 | ) |
| 352 | |
| 353 | result = await wire_fetch_mpack( |
| 354 | db_session, repo.repo_id, want=[commit.commit_id], have=[], force_build=True |
| 355 | ) |
| 356 | |
| 357 | assert result["blob_count"] == 2 |
| 358 | assert backend.get_calls == [], "no per-object GETs" |
| 359 | assert sorted(backend.get_mpack_calls) == sorted([mpack_a_id, mpack_b_id]) |
| 360 | assert backend.get_mpack_calls.count(mpack_a_id) == 1, "mpack A fetched exactly once" |
| 361 | assert backend.get_mpack_calls.count(mpack_b_id) == 1, "mpack B fetched exactly once" |
| 362 | |
| 363 | |
| 364 | # ── FN-4: mpack missing from storage → per-object fallback ─────────────────── |
| 365 | |
| 366 | @pytest.mark.asyncio |
| 367 | async def test_fn4_missing_mpack_falls_back_to_per_object_get( |
| 368 | db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch |
| 369 | ) -> None: |
| 370 | """MPackIndex points at a mpack that is missing from storage → falls back to backend.get().""" |
| 371 | backend = _FakeBackend() |
| 372 | monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) |
| 373 | monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend) |
| 374 | |
| 375 | repo = await create_repo(db_session, owner="gabriel", visibility="public") |
| 376 | raw = b"fallback content" |
| 377 | oid = blob_id(raw) |
| 378 | |
| 379 | # MPackIndex row exists but mpack is NOT in backend._mpacks |
| 380 | ghost_mpack_id = fake_id("ghost-mpack") |
| 381 | await _index_object_in_mpack(db_session, oid, ghost_mpack_id) |
| 382 | await db_session.execute( |
| 383 | pg_insert(db.MusehubObjectRef) |
| 384 | .values(repo_id=repo.repo_id, object_id=oid) |
| 385 | .on_conflict_do_nothing() |
| 386 | ) |
| 387 | await db_session.commit() |
| 388 | # Per-object key IS available (legacy fallback) |
| 389 | await backend.put(oid, raw) |
| 390 | |
| 391 | commit = await _make_commit( |
| 392 | db_session, repo.repo_id, manifest={"f.txt": oid}, seed="fn4" |
| 393 | ) |
| 394 | |
| 395 | result = await wire_fetch_mpack( |
| 396 | db_session, repo.repo_id, want=[commit.commit_id], have=[], force_build=True |
| 397 | ) |
| 398 | |
| 399 | assert result["blob_count"] == 1 |
| 400 | assert ghost_mpack_id in backend.get_mpack_calls, "attempted mpack GET" |
| 401 | assert oid in backend.get_calls, "fell back to per-object GET after mpack miss" |
| 402 | |
| 403 | |
| 404 | # ── FN-5: object absent from MPackIndex → FetchNotIndexedError ─────────────── |
| 405 | |
| 406 | @pytest.mark.asyncio |
| 407 | async def test_fn5_unindexed_object_raises_fetch_not_indexed_error( |
| 408 | db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch |
| 409 | ) -> None: |
| 410 | """Object not present in MPackIndex at all → FetchNotIndexedError (background job pending).""" |
| 411 | backend = _FakeBackend() |
| 412 | monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) |
| 413 | monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend) |
| 414 | |
| 415 | repo = await create_repo(db_session, owner="gabriel", visibility="public") |
| 416 | raw = b"not indexed yet" |
| 417 | oid = blob_id(raw) |
| 418 | |
| 419 | # Object row exists in DB but NO MPackIndex entry and NO per-object key |
| 420 | await db_session.execute( |
| 421 | pg_insert(db.MusehubObject) |
| 422 | .values(object_id=oid, path="f.txt", size_bytes=len(raw), storage_uri=f"mem://{oid}") |
| 423 | .on_conflict_do_nothing(index_elements=["object_id"]) |
| 424 | ) |
| 425 | await db_session.execute( |
| 426 | pg_insert(db.MusehubObjectRef) |
| 427 | .values(repo_id=repo.repo_id, object_id=oid) |
| 428 | .on_conflict_do_nothing() |
| 429 | ) |
| 430 | await db_session.commit() |
| 431 | |
| 432 | commit = await _make_commit( |
| 433 | db_session, repo.repo_id, manifest={"f.txt": oid}, seed="fn5" |
| 434 | ) |
| 435 | |
| 436 | with pytest.raises(FetchNotIndexedError): |
| 437 | await wire_fetch_mpack( |
| 438 | db_session, repo.repo_id, want=[commit.commit_id], have=[], force_build=True |
| 439 | ) |
| 440 | |
| 441 | |
| 442 | # ── FN-6: mixed sources — cache + mpack + legacy ───────────────────────────── |
| 443 | |
| 444 | @pytest.mark.asyncio |
| 445 | async def test_fn6_mixed_sources_all_merged( |
| 446 | db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch |
| 447 | ) -> None: |
| 448 | """Three objects: one from content_cache, one from mpack, one from legacy per-object key.""" |
| 449 | backend = _FakeBackend() |
| 450 | monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) |
| 451 | monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend) |
| 452 | |
| 453 | repo = await create_repo(db_session, owner="gabriel", visibility="public") |
| 454 | |
| 455 | raw_cached = b"in content_cache" |
| 456 | raw_mpack = b"in mpack storage" |
| 457 | raw_legacy = b"in per-object storage" |
| 458 | |
| 459 | oid_cached = blob_id(raw_cached) |
| 460 | oid_mpack = blob_id(raw_mpack) |
| 461 | oid_legacy = blob_id(raw_legacy) |
| 462 | |
| 463 | # oid_cached: stored in content_cache column |
| 464 | await db_session.execute( |
| 465 | pg_insert(db.MusehubObject) |
| 466 | .values( |
| 467 | object_id=oid_cached, path="cached.txt", |
| 468 | size_bytes=len(raw_cached), storage_uri=f"mem://{oid_cached}", |
| 469 | content_cache=raw_cached, |
| 470 | ) |
| 471 | .on_conflict_do_nothing(index_elements=["object_id"]) |
| 472 | ) |
| 473 | # oid_mpack: stored only in mpack, indexed via MPackIndex |
| 474 | mpack_bytes = _make_mpack_bytes({oid_mpack: raw_mpack}) |
| 475 | mpack_id = _mpack_key(mpack_bytes) |
| 476 | await backend.put_mpack(mpack_id, mpack_bytes) |
| 477 | await _index_object_in_mpack(db_session, oid_mpack, mpack_id) |
| 478 | |
| 479 | # oid_legacy: stored as per-object key in backend, has MPackIndex (mpack present too) |
| 480 | # but for the legacy case we'll make the mpack missing so it falls back to get() |
| 481 | ghost_id = fake_id("ghost") |
| 482 | await _index_object_in_mpack(db_session, oid_legacy, ghost_id) |
| 483 | await backend.put(oid_legacy, raw_legacy) |
| 484 | |
| 485 | for oid in (oid_cached, oid_mpack, oid_legacy): |
| 486 | await db_session.execute( |
| 487 | pg_insert(db.MusehubObjectRef) |
| 488 | .values(repo_id=repo.repo_id, object_id=oid) |
| 489 | .on_conflict_do_nothing() |
| 490 | ) |
| 491 | |
| 492 | # Also need MPackIndex for oid_cached so the coverage check passes |
| 493 | cache_mpack_bytes = _make_mpack_bytes({oid_cached: raw_cached}) |
| 494 | cache_mpack_id = _mpack_key(cache_mpack_bytes) |
| 495 | await backend.put_mpack(cache_mpack_id, cache_mpack_bytes) |
| 496 | await _index_object_in_mpack(db_session, oid_cached, cache_mpack_id) |
| 497 | |
| 498 | await db_session.commit() |
| 499 | |
| 500 | commit = await _make_commit( |
| 501 | db_session, repo.repo_id, |
| 502 | manifest={"cached.txt": oid_cached, "mpack.txt": oid_mpack, "legacy.txt": oid_legacy}, |
| 503 | seed="fn6", |
| 504 | ) |
| 505 | |
| 506 | result = await wire_fetch_mpack( |
| 507 | db_session, repo.repo_id, want=[commit.commit_id], have=[], force_build=True |
| 508 | ) |
| 509 | |
| 510 | assert result["blob_count"] == 3 |
| 511 | |
| 512 | # Verify assembled mpack contains all three objects with correct bytes |
| 513 | from muse.core.mpack import parse_wire_mpack |
| 514 | fetch_raw = backend._mpacks.get(result["mpack_id"]) |
| 515 | assert fetch_raw is not None |
| 516 | payload = parse_wire_mpack(fetch_raw) |
| 517 | obj_map = {o["object_id"]: bytes(o["content"]) for o in payload.get("blobs", [])} |
| 518 | |
| 519 | assert obj_map[oid_cached] == raw_cached |
| 520 | assert obj_map[oid_mpack] == raw_mpack |
| 521 | assert obj_map[oid_legacy] == raw_legacy |
| 522 | |
| 523 | # content_cache object must not trigger any GET |
| 524 | assert oid_cached not in backend.get_calls |
| 525 | # mpack object must not trigger per-object GET |
| 526 | assert oid_mpack not in backend.get_calls |
| 527 | # legacy fallback fired for oid_legacy |
| 528 | assert oid_legacy in backend.get_calls |
| 529 | |
| 530 | |
| 531 | # ── FN-7: zstd-compressed objects in mpack → decompressed before serving ────── |
| 532 | |
| 533 | @pytest.mark.asyncio |
| 534 | async def test_fn7_zstd_compressed_mpack_objects_decompressed( |
| 535 | db_session: AsyncSession, monkeypatch: pytest.MonkeyPatch |
| 536 | ) -> None: |
| 537 | """Objects stored with encoding='zstd' in the covering mpack must be |
| 538 | decompressed before being placed in the assembled fetch mpack. |
| 539 | |
| 540 | Regression test for the bug where _extract_from_mpack served raw compressed |
| 541 | bytes, causing the muse client's sha256 integrity check to fail: |
| 542 | expected sha256(decompressed) but got sha256(compressed) |
| 543 | This caused apply_mpack to skip all objects, leaving HEAD unadvanced and |
| 544 | muse log returning 0 commits. |
| 545 | """ |
| 546 | backend = _FakeBackend() |
| 547 | monkeypatch.setattr("musehub.services.musehub_wire.get_backend", lambda: backend) |
| 548 | monkeypatch.setattr("musehub.services.musehub_wire_fetch.get_backend", lambda: backend) |
| 549 | |
| 550 | repo = await create_repo(db_session, owner="gabriel", visibility="public") |
| 551 | |
| 552 | # Object ID is sha256 of DECOMPRESSED bytes (content-addressed). |
| 553 | raw = b"decompressed object content for fn7" |
| 554 | oid = blob_id(raw) |
| 555 | |
| 556 | # Push mpack stores the object with encoding='zstd'. |
| 557 | mpack_bytes_zstd = _make_mpack_bytes_zstd({oid: raw}) |
| 558 | mpack_id = _mpack_key(mpack_bytes_zstd) |
| 559 | await backend.put_mpack(mpack_id, mpack_bytes_zstd) |
| 560 | await _index_object_in_mpack(db_session, oid, mpack_id) |
| 561 | await db_session.execute( |
| 562 | pg_insert(db.MusehubObjectRef) |
| 563 | .values(repo_id=repo.repo_id, object_id=oid) |
| 564 | .on_conflict_do_nothing() |
| 565 | ) |
| 566 | await db_session.commit() |
| 567 | |
| 568 | commit = await _make_commit( |
| 569 | db_session, repo.repo_id, manifest={"file.txt": oid}, seed="fn7" |
| 570 | ) |
| 571 | |
| 572 | result = await wire_fetch_mpack( |
| 573 | db_session, repo.repo_id, want=[commit.commit_id], have=[], force_build=True |
| 574 | ) |
| 575 | |
| 576 | assert result["blob_count"] == 1, "expected 1 object in fetch mpack" |
| 577 | |
| 578 | # The assembled fetch mpack must contain the DECOMPRESSED bytes. |
| 579 | # If still compressed, the client's sha256(content) != oid integrity check fails. |
| 580 | from muse.core.mpack import parse_wire_mpack |
| 581 | fetch_raw = backend._mpacks.get(result["mpack_id"]) |
| 582 | assert fetch_raw is not None, "fetch mpack not stored in backend" |
| 583 | payload = parse_wire_mpack(fetch_raw) |
| 584 | obj_map = {o["object_id"]: bytes(o["content"]) for o in payload.get("blobs", [])} |
| 585 | |
| 586 | assert oid in obj_map, f"object {oid[:20]} missing from assembled fetch mpack" |
| 587 | assert obj_map[oid] == raw, ( |
| 588 | "object content in assembled mpack must be DECOMPRESSED bytes matching oid; " |
| 589 | "got compressed bytes — integrity check would fail on client" |
| 590 | ) |
File History
1 commit
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595
fix: typing audit — 0 violations, 0 untyped defs across all…
Sonnet 4.6
minor
⚠
49 days ago