musehub_wire_shared.py
python
sha256:981c31e3af9c23f9fecb2385134f30419b6e77a8e678fcdf8ff3c02945949bbf
fix: apply hash-verified dirs to delta-only snapshot path i…
Sonnet 4.6
35 days ago
| 1 | """Shared types, helpers, and error classes for the wire protocol service.""" |
| 2 | |
| 3 | import asyncio |
| 4 | import collections |
| 5 | import hashlib |
| 6 | import logging |
| 7 | import msgpack as _msgpack |
| 8 | import time as _time_module |
| 9 | from datetime import datetime, timezone |
| 10 | |
| 11 | _BLOB_PUT_SEM: asyncio.Semaphore | None = None |
| 12 | |
| 13 | def _get_blob_put_sem() -> asyncio.Semaphore: |
| 14 | global _BLOB_PUT_SEM |
| 15 | if _BLOB_PUT_SEM is None: |
| 16 | _BLOB_PUT_SEM = asyncio.Semaphore(100) |
| 17 | return _BLOB_PUT_SEM |
| 18 | |
| 19 | from sqlalchemy import func, select, text as _sa_text |
| 20 | from sqlalchemy.dialects.postgresql import insert as _pg_insert |
| 21 | from sqlalchemy.ext.asyncio import AsyncSession |
| 22 | |
| 23 | from musehub.db.musehub_abuse_models import MusehubBlockedHash, MusehubDailyPushBytes, MusehubPushAnomaly |
| 24 | from musehub.db.musehub_collaborator_models import MusehubCollaborator |
| 25 | from musehub.db.musehub_jobs_models import MusehubBackgroundJob |
| 26 | from musehub.db.musehub_repo_models import ( |
| 27 | MusehubBranch, |
| 28 | MusehubCommit, |
| 29 | MusehubCommitGraph, |
| 30 | MusehubCommitRef, |
| 31 | MusehubObject, |
| 32 | MusehubObjectRef, |
| 33 | MusehubMPackIndex, |
| 34 | MusehubPackIndex, |
| 35 | MusehubRepo, |
| 36 | MusehubSnapshot, |
| 37 | MusehubSnapshotRef, |
| 38 | ) |
| 39 | from musehub.models.wire import ( |
| 40 | WireCommit, |
| 41 | WireMPack, |
| 42 | WireFetchRequest, |
| 43 | WireObject, |
| 44 | WireRefsResponse, |
| 45 | WireSnapshot, |
| 46 | ) |
| 47 | from muse.core.types import blob_id, decode_pubkey, split_id |
| 48 | from muse.core.ids import commit_identity_bytes as _muse_commit_identity_bytes |
| 49 | from muse.core.provenance import provenance_payload, verify_commit_ed25519 |
| 50 | from muse.core.compression import ZSTD_AVAILABLE, compress_and_encode, compute_delta |
| 51 | from musehub.core.genesis import compute_branch_id |
| 52 | from musehub.types.json_types import IntDict, JSONObject, JSONValue, StrDict |
| 53 | from musehub.types.pydantic_types import PydanticJson, unwrap |
| 54 | from musehub.config import settings |
| 55 | from musehub.storage import get_backend |
| 56 | from musehub.storage.backends import read_object_bytes, StorageBackend |
| 57 | from musehub.services.musehub_jobs import enqueue_push_intel, enqueue_profile_snapshot |
| 58 | from collections.abc import AsyncIterator |
| 59 | from typing import TypedDict |
| 60 | |
| 61 | _monotonic = _time_module.monotonic |
| 62 | |
| 63 | class RepairResult(TypedDict): |
| 64 | repaired: bool |
| 65 | |
| 66 | class FetchPresignResult(TypedDict): |
| 67 | presign: bool |
| 68 | blob_urls: dict[str, str] |
| 69 | commits: list[dict] |
| 70 | snapshots: list[dict] |
| 71 | branch_heads: dict[str, str] |
| 72 | repo_id: str |
| 73 | domain: str |
| 74 | default_branch: str |
| 75 | expires_at: str | None |
| 76 | commit_count: int |
| 77 | blob_count: int |
| 78 | |
| 79 | class FetchMPackResult(TypedDict): |
| 80 | mpack_url: str | None |
| 81 | mpack_id: str | None |
| 82 | commit_count: int |
| 83 | blob_count: int |
| 84 | |
| 85 | class FetchNotIndexedError(Exception): |
| 86 | def __init__(self, missing_count: int) -> None: |
| 87 | super().__init__(f"{missing_count} object(s) not yet indexed") |
| 88 | self.missing_count = missing_count |
| 89 | |
| 90 | |
| 91 | class MPackNotReadyError(Exception): |
| 92 | """Raised when no prebuild mpack exists for the requested tips yet. |
| 93 | |
| 94 | The worker is building it — callers should retry after a short delay. |
| 95 | """ |
| 96 | |
| 97 | type _ObjFetchMap = dict[str, tuple[str, str, str | None, bytes | None]] |
| 98 | type _ObjectMap = dict[str, MusehubObject] |
| 99 | type _BytesMap = dict[str, bytes] |
| 100 | type _ChildMap = dict[str, list[str]] |
| 101 | |
| 102 | logger = logging.getLogger(__name__) |
| 103 | |
| 104 | def _commit_identity_bytes(wire_commit: "WireCommit") -> bytes: |
| 105 | parent_ids: list[str] = [] |
| 106 | if wire_commit.parent_commit_id: |
| 107 | parent_ids.append(wire_commit.parent_commit_id) |
| 108 | if wire_commit.parent2_commit_id: |
| 109 | parent_ids.append(wire_commit.parent2_commit_id) |
| 110 | return _muse_commit_identity_bytes( |
| 111 | parent_ids=parent_ids, |
| 112 | snapshot_id=wire_commit.snapshot_id or "", |
| 113 | message=wire_commit.message, |
| 114 | committed_at_iso=wire_commit.committed_at, |
| 115 | author=wire_commit.author, |
| 116 | signer_public_key=wire_commit.signer_public_key, |
| 117 | ) |
| 118 | |
| 119 | |
| 120 | async def _reconstruct_manifest(session: AsyncSession, snapshot_id: str) -> StrDict: |
| 121 | import msgpack as _mp |
| 122 | |
| 123 | # Single recursive CTE walks the entire delta chain in one round-trip. |
| 124 | # On RDS the old per-ancestor query loop was O(N) round-trips × latency; |
| 125 | # this reduces it to 1 regardless of chain depth. |
| 126 | sql = _sa_text(""" |
| 127 | WITH RECURSIVE chain AS ( |
| 128 | SELECT snapshot_id, manifest_blob, delta_blob, parent_snapshot_id, 0 AS depth |
| 129 | FROM musehub_snapshots |
| 130 | WHERE snapshot_id = :start_id |
| 131 | UNION ALL |
| 132 | SELECT s.snapshot_id, s.manifest_blob, s.delta_blob, s.parent_snapshot_id, c.depth + 1 |
| 133 | FROM musehub_snapshots s |
| 134 | JOIN chain c ON s.snapshot_id = c.parent_snapshot_id |
| 135 | WHERE c.manifest_blob IS NULL AND c.parent_snapshot_id IS NOT NULL |
| 136 | ) |
| 137 | SELECT manifest_blob, delta_blob |
| 138 | FROM chain |
| 139 | ORDER BY depth |
| 140 | """) |
| 141 | rows = (await session.execute(sql, {"start_id": snapshot_id})).fetchall() |
| 142 | |
| 143 | base: dict[str, str] = {} |
| 144 | deltas: list[bytes] = [] |
| 145 | for manifest_blob, delta_blob in rows: |
| 146 | if manifest_blob is not None: |
| 147 | base = dict(_mp.unpackb(manifest_blob, raw=False)) |
| 148 | break |
| 149 | if delta_blob is not None: |
| 150 | deltas.append(delta_blob) |
| 151 | |
| 152 | for d_blob in reversed(deltas): |
| 153 | d = _mp.unpackb(d_blob, raw=False) |
| 154 | if isinstance(d, dict) and "add" in d: |
| 155 | base.update(d.get("add") or {}) |
| 156 | for p in d.get("rm") or []: |
| 157 | base.pop(p, None) |
| 158 | else: |
| 159 | base.update(d) |
| 160 | return base |
| 161 | |
| 162 | |
| 163 | def _cached_manifest_reproduces(manifest: StrDict, directories: list[str] | None, snapshot_id: str) -> bool: |
| 164 | """True if a cached file-manifest reproduces its snapshot_id with any plausible |
| 165 | directory set (stored, derived, or empty). Used to detect a corrupt manifest_blob |
| 166 | so the serving path can fall back to delta-chain reconstruction.""" |
| 167 | # A manifest with null/empty OIDs was stored in a buggy state — the hash may still |
| 168 | # match (it was computed with the nulls) but the data is unusable. Force reconstruction |
| 169 | # so the delta chain supplies the real blob IDs. |
| 170 | if any(not v for v in manifest.values()): |
| 171 | return False |
| 172 | from muse.core.ids import hash_snapshot as _hash_snapshot |
| 173 | from muse.core.snapshot import directories_from_manifest as _dirs_from_manifest |
| 174 | for dirs in (list(directories or []), _dirs_from_manifest(manifest), []): |
| 175 | if _hash_snapshot(manifest, dirs) == snapshot_id: |
| 176 | return True |
| 177 | return False |
| 178 | |
| 179 | |
| 180 | async def _reconstruct_manifest_validated(session: AsyncSession, snapshot_id: str) -> StrDict: |
| 181 | """Reconstruct a snapshot's file-manifest, VALIDATING cached manifests and skipping |
| 182 | corrupt ones — defense against legacy data-integrity corruption. |
| 183 | |
| 184 | Walks the parent chain from ``snapshot_id``. The nearest ancestor whose cached |
| 185 | ``manifest_blob`` reproduces *its* snapshot_id becomes the base; the deltas of every |
| 186 | node between that ancestor and the target are then applied forward. Unlike |
| 187 | :func:`_reconstruct_manifest`, a corrupt (non-reproducing) ``manifest_blob`` does NOT |
| 188 | terminate the walk — it is treated like a delta-only node so the walk continues to a |
| 189 | trustworthy ancestor (e.g. the root, or a head whose full manifest is intact). |
| 190 | |
| 191 | Per-node ``session.get`` walk (not the single CTE) because validity depends on |
| 192 | ``hash_snapshot`` and cannot be expressed in SQL. Depth to the nearest valid base is |
| 193 | normally small (roots/heads carry full manifests), so the round-trip cost is bounded. |
| 194 | """ |
| 195 | import msgpack as _mp |
| 196 | |
| 197 | deltas: list[bytes] = [] # collected target-first |
| 198 | base: dict[str, str] = {} |
| 199 | cur: str | None = snapshot_id |
| 200 | seen: set[str] = set() |
| 201 | while cur and cur not in seen: |
| 202 | seen.add(cur) |
| 203 | row = await session.get(MusehubSnapshot, cur) |
| 204 | if row is None: |
| 205 | break |
| 206 | if row.manifest_blob is not None: |
| 207 | candidate = dict(_mp.unpackb(row.manifest_blob, raw=False)) |
| 208 | if _cached_manifest_reproduces(candidate, row.directories, cur): |
| 209 | base = candidate # trustworthy base found |
| 210 | break |
| 211 | # else: corrupt cache — ignore it, treat node as delta-only and keep walking |
| 212 | if row.delta_blob is not None: |
| 213 | deltas.append(row.delta_blob) |
| 214 | cur = row.parent_snapshot_id |
| 215 | |
| 216 | base = dict(base) |
| 217 | total_null_skipped = 0 |
| 218 | for d_blob in reversed(deltas): |
| 219 | d = _mp.unpackb(d_blob, raw=False) |
| 220 | if isinstance(d, dict) and "add" in d: |
| 221 | # Skip null OIDs — delta_blobs stored during a push bug may carry null values |
| 222 | # for files whose blobs weren't yet indexed. Applying them would overwrite a |
| 223 | # valid base OID with null, causing the blob to be omitted from the mpack. |
| 224 | adds = d.get("add") or {} |
| 225 | nulls = [k for k, v in adds.items() if not v] |
| 226 | if nulls: |
| 227 | total_null_skipped += len(nulls) |
| 228 | logger.warning("[BLOB-DEBUG] _reconstruct snap=%s skipping %d null OIDs in delta add: %s", |
| 229 | snapshot_id, len(nulls), nulls) |
| 230 | base.update({k: v for k, v in adds.items() if v}) |
| 231 | for p in d.get("rm") or []: |
| 232 | base.pop(p, None) |
| 233 | else: |
| 234 | nulls = [k for k, v in d.items() if not v] |
| 235 | if nulls: |
| 236 | total_null_skipped += len(nulls) |
| 237 | logger.warning("[BLOB-DEBUG] _reconstruct snap=%s skipping %d null OIDs in flat delta: %s", |
| 238 | snapshot_id, len(nulls), nulls) |
| 239 | base.update({k: v for k, v in d.items() if v}) |
| 240 | if total_null_skipped: |
| 241 | logger.warning("[BLOB-DEBUG] _reconstruct snap=%s DONE base_size=%d total_null_skipped=%d", |
| 242 | snapshot_id, len(base), total_null_skipped) |
| 243 | return base |
| 244 | |
| 245 | |
| 246 | async def _upsert_object_refs( |
| 247 | session: AsyncSession, |
| 248 | repo_id: str, |
| 249 | object_ids: list[str], |
| 250 | ) -> None: |
| 251 | if not object_ids: |
| 252 | return |
| 253 | _CHUNK = 5000 |
| 254 | for i in range(0, len(object_ids), _CHUNK): |
| 255 | chunk = object_ids[i : i + _CHUNK] |
| 256 | await session.execute( |
| 257 | _pg_insert(MusehubObjectRef) |
| 258 | .values([{"repo_id": repo_id, "object_id": oid} for oid in chunk]) |
| 259 | .on_conflict_do_nothing(index_elements=["repo_id", "object_id"]) |
| 260 | ) |
| 261 | |
| 262 | class MPackValidationError(ValueError): |
| 263 | pass |
| 264 | |
| 265 | class ObjectHashMismatch(ValueError): |
| 266 | pass |
| 267 | |
| 268 | class NonFastForwardError(Exception): |
| 269 | pass |
| 270 | |
| 271 | |
| 272 | def _is_fast_forward(incoming_head: str, current_head: str, wire_bytes: bytes) -> bool: |
| 273 | if not current_head or incoming_head == current_head: |
| 274 | return True |
| 275 | |
| 276 | if wire_bytes[:4] != b"MUSE": |
| 277 | raise ValueError(f"mpack is not MUSE binary format (got {wire_bytes[:4]!r})") |
| 278 | from muse.core.mpack import parse_wire_mpack as _parse_wire_mpack |
| 279 | mpack: JSONObject = _parse_wire_mpack(wire_bytes) |
| 280 | commits: list[JSONValue] = mpack.get("commits") or [] |
| 281 | |
| 282 | parent_map: dict[str, list[str]] = {} |
| 283 | for c in commits: |
| 284 | cid = c.get("commit_id", "") |
| 285 | parents: list[str] = [] |
| 286 | if c.get("parent_commit_id"): |
| 287 | parents.append(c["parent_commit_id"]) |
| 288 | if c.get("parent2_commit_id"): |
| 289 | parents.append(c["parent2_commit_id"]) |
| 290 | parent_map[cid] = parents |
| 291 | |
| 292 | seen: set[str] = set() |
| 293 | queue: list[str] = [incoming_head] |
| 294 | while queue: |
| 295 | cid = queue.pop() |
| 296 | if cid in seen: |
| 297 | continue |
| 298 | seen.add(cid) |
| 299 | if cid == current_head: |
| 300 | return True |
| 301 | for pid in parent_map.get(cid, []): |
| 302 | queue.append(pid) |
| 303 | |
| 304 | return False |
| 305 | |
| 306 | |
| 307 | async def _is_ancestor_db( |
| 308 | session: "AsyncSession", |
| 309 | ancestor_id: str, |
| 310 | descendant_id: str, |
| 311 | repo_id: str, |
| 312 | max_hops: int = 1000, |
| 313 | ) -> bool: |
| 314 | from musehub.graph.walk import walk_dag_async |
| 315 | |
| 316 | async def _adj(cid: str) -> list[str]: |
| 317 | row = (await session.execute( |
| 318 | select(MusehubCommitGraph).where(MusehubCommitGraph.commit_id == cid) |
| 319 | )).scalar_one_or_none() |
| 320 | if row is None: |
| 321 | return [] |
| 322 | return list(row.parent_ids or []) |
| 323 | |
| 324 | async for cid in walk_dag_async(descendant_id, _adj, max_nodes=max_hops): |
| 325 | if cid == ancestor_id: |
| 326 | return True |
| 327 | return False |
| 328 | |
| 329 | |
| 330 | class PolyglotObjectError(ValueError): |
| 331 | def __init__(self, object_id: str, detail: str) -> None: |
| 332 | super().__init__(f"polyglot object rejected ({object_id!r}): {detail}") |
| 333 | self.object_id = object_id |
| 334 | |
| 335 | def _verify_object_hash(object_id: str, content: bytes) -> None: |
| 336 | if not object_id.startswith("sha256:"): |
| 337 | raise ObjectHashMismatch( |
| 338 | f"object_id {object_id!r} is not sha256-prefixed — " |
| 339 | "all content-addressed IDs must use the sha256: prefix" |
| 340 | ) |
| 341 | actual_id = blob_id(content) |
| 342 | if actual_id != object_id: |
| 343 | _, declared = split_id(object_id) |
| 344 | _, actual = split_id(actual_id) |
| 345 | raise ObjectHashMismatch( |
| 346 | f"object_id hash mismatch for {object_id!r}: " |
| 347 | f"declared={declared[:16]}… actual={actual[:16]}…" |
| 348 | ) |
| 349 | |
| 350 | def _utc_now() -> datetime: |
| 351 | return datetime.now(tz=timezone.utc) |
| 352 | |
| 353 | def _parse_iso(s: str) -> datetime: |
| 354 | try: |
| 355 | return datetime.fromisoformat(s.replace("Z", "+00:00")) |
| 356 | except (ValueError, AttributeError): |
| 357 | return _utc_now() |
| 358 | |
| 359 | def _str_values(d: JSONValue) -> StrDict: |
| 360 | if not isinstance(d, dict): |
| 361 | return {} |
| 362 | return {str(k): str(v) for k, v in d.items()} |
| 363 | |
| 364 | def _str_list(v: JSONValue) -> list[str]: |
| 365 | if not isinstance(v, list): |
| 366 | return [] |
| 367 | return [str(x) for x in v] |
| 368 | |
| 369 | def _int_safe(v: JSONValue, default: int = 0) -> int: |
| 370 | return int(v) if isinstance(v, (int, float)) else default |
| 371 | |
| 372 | def _to_wire_commit(row: MusehubCommit) -> WireCommit: |
| 373 | parent_ids: list[str] = row.parent_ids if isinstance(row.parent_ids, list) else [] |
| 374 | return WireCommit( |
| 375 | commit_id=row.commit_id, |
| 376 | branch=row.branch or "", |
| 377 | snapshot_id=row.snapshot_id, |
| 378 | message=row.message or "", |
| 379 | committed_at=row.timestamp.isoformat() if row.timestamp else "", |
| 380 | parent_commit_id=parent_ids[0] if len(parent_ids) >= 1 else None, |
| 381 | parent2_commit_id=parent_ids[1] if len(parent_ids) >= 2 else None, |
| 382 | author=row.author or "", |
| 383 | metadata={}, |
| 384 | structured_delta=row.structured_delta if isinstance(row.structured_delta, dict) else None, |
| 385 | sem_ver_bump=str(row.sem_ver_bump or "none"), |
| 386 | breaking_changes=_str_list(row.breaking_changes), |
| 387 | agent_id=str(row.agent_id or ""), |
| 388 | model_id=str(row.model_id or ""), |
| 389 | toolchain_id=str(row.toolchain_id or ""), |
| 390 | prompt_hash=str(row.prompt_hash or ""), |
| 391 | signature=str(row.signature or ""), |
| 392 | signer_public_key=str(row.signer_public_key or ""), |
| 393 | signer_key_id=str(row.signer_key_id or ""), |
| 394 | format_version=1, |
| 395 | reviewed_by=_str_list(row.reviewed_by), |
| 396 | test_runs=_int_safe(row.test_runs), |
| 397 | ) |
| 398 | |
| 399 | def _snap_row_to_wire(row: MusehubSnapshot) -> JSONObject: |
| 400 | """Serve snapshot in full-manifest format from the DB manifest_blob cache. |
| 401 | |
| 402 | Sends {snapshot_id, manifest, directories} so the client applies the |
| 403 | manifest directly without delta reconstruction. This avoids two bugs |
| 404 | that the old pseudo-delta format had: |
| 405 | 1. delta_remove was always [] — files removed since the parent were |
| 406 | never subtracted, producing a manifest with extra paths. |
| 407 | 2. parent_snapshot_id was set but the client used it as a delta base, |
| 408 | meaning the reconstructed manifest was parent ∪ current, not current. |
| 409 | |
| 410 | Delta-only rows (manifest_blob=None) send an empty manifest and log a |
| 411 | warning — callers should use _snap_row_to_wire_s3 with a session so the |
| 412 | manifest can be reconstructed from the delta chain. |
| 413 | |
| 414 | Directories selection — two legacy scenarios for rows with directories=[]: |
| 415 | |
| 416 | Scenario A (before rc12): the push client computed snapshot_id using |
| 417 | hash_snapshot(manifest, derived_dirs) but the server wasn't storing dirs. |
| 418 | Fix: derive dirs from the manifest; they reproduce the stored snapshot_id. |
| 419 | |
| 420 | Scenario B (even older): snapshot_id was hashed with no directories at |
| 421 | all — hash_snapshot(manifest, None). Sending derived dirs to the client |
| 422 | causes a hash mismatch and aborts the clone. |
| 423 | |
| 424 | To distinguish the two, we verify the candidate against the stored |
| 425 | snapshot_id. This is O(1) in the number of paths and is the only |
| 426 | approach that is guaranteed correct — anything else is a guess. |
| 427 | """ |
| 428 | import msgpack as _mp |
| 429 | from muse.core.snapshot import directories_from_manifest as _dirs_from_manifest |
| 430 | from muse.core.ids import hash_snapshot as _hash_snapshot |
| 431 | if row.manifest_blob: |
| 432 | manifest: dict[str, str] = dict(_mp.unpackb(row.manifest_blob, raw=False)) |
| 433 | if row.directories: |
| 434 | dirs = list(row.directories) |
| 435 | dirs_source = "db" |
| 436 | else: |
| 437 | derived = _dirs_from_manifest(manifest) |
| 438 | if _hash_snapshot(manifest, derived) == row.snapshot_id: |
| 439 | # Scenario A: hashed with dirs, server just wasn't storing them. |
| 440 | dirs = derived |
| 441 | dirs_source = "derived" |
| 442 | else: |
| 443 | # Scenario B: snapshot_id was hashed without directories. |
| 444 | dirs = [] |
| 445 | dirs_source = "empty" |
| 446 | if _hash_snapshot(manifest, []) != row.snapshot_id: |
| 447 | logger.error( |
| 448 | "[_snap_row_to_wire] HASH INCONSISTENCY snap=%s — " |
| 449 | "neither dirs=derived (%d) nor dirs=[] reproduces snapshot_id. " |
| 450 | "Data integrity violation.", |
| 451 | row.snapshot_id[:20], len(derived), |
| 452 | ) |
| 453 | logger.debug( |
| 454 | "[_snap_row_to_wire] snap=%s format=full manifest=%d dirs=%d dirs_source=%s", |
| 455 | row.snapshot_id[:20], len(manifest), len(dirs), dirs_source, |
| 456 | ) |
| 457 | else: |
| 458 | manifest = {} |
| 459 | dirs = [] |
| 460 | logger.warning( |
| 461 | "[_snap_row_to_wire] snap=%s manifest_blob=None (delta-only row) — " |
| 462 | "sending empty manifest; use _snap_row_to_wire_s3(session=session)", |
| 463 | row.snapshot_id[:20], |
| 464 | ) |
| 465 | return { |
| 466 | "snapshot_id": row.snapshot_id, |
| 467 | "manifest": manifest, |
| 468 | "directories": dirs, |
| 469 | "created_at": row.created_at.isoformat() if row.created_at else "", |
| 470 | } |
| 471 | |
| 472 | |
| 473 | async def _snap_row_to_wire_s3( |
| 474 | row: MusehubSnapshot, |
| 475 | backend: "StorageBackend", |
| 476 | session: "AsyncSession | None" = None, |
| 477 | ) -> JSONObject: |
| 478 | """Serve snapshot wire dict from DB. |
| 479 | |
| 480 | Three cases: |
| 481 | 1. manifest_blob present AND it reproduces snapshot_id → fast path (serve cache). |
| 482 | 2. manifest_blob absent OR corrupt (does not reproduce snapshot_id), session given |
| 483 | → reconstruct the manifest from the delta chain via the *validating* walker, |
| 484 | which skips corrupt cached ancestors. This is the legacy-corruption defense: |
| 485 | a snapshot whose cached manifest_blob was written empty/incomplete by an older |
| 486 | version is recovered on read instead of poisoning the clone. |
| 487 | 3. No session → best-effort _snap_row_to_wire (cannot reconstruct). |
| 488 | """ |
| 489 | import msgpack as _mp |
| 490 | from muse.core.snapshot import directories_from_manifest as _dirs_from_manifest |
| 491 | from muse.core.ids import hash_snapshot as _hash_snapshot |
| 492 | |
| 493 | cache_ok = False |
| 494 | if row.manifest_blob is not None: |
| 495 | cached = dict(_mp.unpackb(row.manifest_blob, raw=False)) |
| 496 | cache_ok = _cached_manifest_reproduces(cached, row.directories, row.snapshot_id) |
| 497 | |
| 498 | if cache_ok or session is None: |
| 499 | # Trustworthy cache, or no session to reconstruct with → existing behavior. |
| 500 | return _snap_row_to_wire(row) |
| 501 | |
| 502 | # manifest_blob is absent or corrupt → reconstruct from the (validated) delta chain. |
| 503 | reconstructed_from = "delta-only" if row.manifest_blob is None else "corrupt-cache" |
| 504 | manifest = await _reconstruct_manifest_validated(session, row.snapshot_id) |
| 505 | if row.directories and _hash_snapshot(manifest, list(row.directories)) == row.snapshot_id: |
| 506 | dirs, dirs_source = list(row.directories), "db" |
| 507 | else: |
| 508 | derived = _dirs_from_manifest(manifest) |
| 509 | if _hash_snapshot(manifest, derived) == row.snapshot_id: |
| 510 | dirs, dirs_source = derived, "derived" |
| 511 | elif _hash_snapshot(manifest, []) == row.snapshot_id: |
| 512 | dirs, dirs_source = [], "empty" |
| 513 | else: |
| 514 | # Reconstruction still does not reproduce snapshot_id — the delta chain |
| 515 | # itself is unrecoverable (no intact ancestor). Surface loudly. |
| 516 | dirs, dirs_source = derived, "derived-UNVERIFIED" |
| 517 | logger.error( |
| 518 | "[_snap_row_to_wire_s3] UNRECOVERABLE snap=%s (%s) — reconstructed " |
| 519 | "manifest=%d does not reproduce snapshot_id. Delta chain has no intact " |
| 520 | "ancestor. Data integrity violation.", |
| 521 | row.snapshot_id[:20], reconstructed_from, len(manifest), |
| 522 | ) |
| 523 | logger.info( |
| 524 | "[_snap_row_to_wire_s3] snap=%s reconstructed (%s) manifest=%d dirs=%d dirs_source=%s", |
| 525 | row.snapshot_id[:20], reconstructed_from, len(manifest), len(dirs), dirs_source, |
| 526 | ) |
| 527 | return { |
| 528 | "snapshot_id": row.snapshot_id, |
| 529 | "manifest": manifest, |
| 530 | "directories": dirs, |
| 531 | "created_at": row.created_at.isoformat() if row.created_at else "", |
| 532 | } |
| 533 | |
| 534 | |
| 535 | async def _commit_to_wire_s3(row: MusehubCommit, backend: "StorageBackend") -> WireCommit: |
| 536 | """Serve commit wire object from DB. |
| 537 | |
| 538 | Commits are canonical in the DB. S3 is for blobs only. |
| 539 | storage_uri=None is the correct expected state — not a warning condition. |
| 540 | """ |
| 541 | return _to_wire_commit(row) |
File History
13 commits
sha256:981c31e3af9c23f9fecb2385134f30419b6e77a8e678fcdf8ff3c02945949bbf
fix: apply hash-verified dirs to delta-only snapshot path i…
Sonnet 4.6
35 days ago
sha256:e9f33a1593cab77c39f8aa7c876316b2f9543a177d8f8b71598656699b5fcf6f
fix: derive directories from manifest when DB column is emp…
Sonnet 4.6
35 days ago
sha256:9acbc3291c00dc6dbad9685f88df517c29f130ac0238283278c1f2b46e6bc10b
fix: send snapshots in full-manifest format; handle delta-o…
Sonnet 4.6
patch
35 days ago
sha256:52b29c48bd61a50edb216d2dfc4f71d30ec2e0ab37326b3ab970826d5c9c0145
fix: send snapshot manifest under 'manifest' key not 'delta…
Sonnet 4.6
39 days ago
sha256:009b5a222314f47640a58d75ce5a1f428f1624cf0b51384dfcdfbdfab3cc42a4
feat: migration idempotency, file attribution DAG walk, mpa…
Sonnet 4.6
minor
⚠
44 days ago
sha256:f3995ec2c05c9c34b0e4d6e96349a811d0117a1c51d78096d757998ccb3c0520
fix: blobs only in S3/mpack — remove commit/snapshot indivi…
Sonnet 4.6
patch
46 days ago
sha256:ad616c6113d6c00f4efed6b2993734ca46d3e9b5bee25addd4ce8ae6b57136e5
chore: bump version to 0.2.0rc11; typing audit clean + all …
Sonnet 4.6
minor
⚠
48 days ago
sha256:d33e88196634b22b27811fa156c7d32a2311173b9daa7c7be91e2829372a24f5
fix: sign ContentType into presigned PUT URLs + wire fetch …
Sonnet 4.6
minor
⚠
48 days ago
sha256:ecca645d94c5f39c88f4bc1283447ba0f4635ef3cbb11d0cd9b3759cba289d00
fix: compute_snapshot_id uses typed-object formula and wire…
Sonnet 4.6
minor
⚠
48 days ago
sha256:e35be48854f182f7bf02dc6cc0f58d22b3de3a544b570c0e2bc53f9e75a3607d
feat(phase6): remove delta_blob path, dead imports, add fal…
Sonnet 4.6
minor
⚠
49 days ago
sha256:9d0ffea20e344782dc6a969d4a240b3d7c96392b5dc03bbd9421890cb78c6f19
feat(phase4): wire fetch serves commits and snapshots from …
Sonnet 4.6
patch
49 days ago
sha256:ab9eda7b6479e1c35cdba9a54f62bacd2825de8faacec3ba67a9a8ef45914b7d
fix: migration and wire protocol alignment
Sonnet 4.6
minor
⚠
50 days ago
sha256:ef10830ce231e0a20efcb0e2586cb879471247e916616e6fdd0d51df459e2595
fix: typing audit — 0 violations, 0 untyped defs across all…
Sonnet 4.6
minor
⚠
50 days ago