"""Push path — wire_refs, wire_repair_*, wire_push_mpack_presign, wire_push_unpack_mpack.""" from typing import TypedDict import asyncio import collections import hashlib import logging import msgpack as _msgpack import time as _time_module from datetime import datetime, timezone from sqlalchemy import func, select, text as _sa_text from sqlalchemy.dialects.postgresql import insert as _pg_insert from sqlalchemy.ext.asyncio import AsyncSession from musehub.db.musehub_abuse_models import MusehubBlockedHash, MusehubDailyPushBytes, MusehubPushAnomaly from musehub.db.musehub_jobs_models import MusehubBackgroundJob from musehub.db.musehub_collaborator_models import MusehubCollaborator from musehub.db.musehub_repo_models import ( MusehubBranch, MusehubCommit, MusehubCommitGraph, MusehubCommitRef, MusehubObject, MusehubObjectRef, MusehubMPackIndex, MusehubRepo, MusehubSnapshot, MusehubSnapshotRef, ) from musehub.models.wire import WireCommit, WireRefsResponse from muse.core.types import blob_id, split_id from muse.core.ids import hash_snapshot, long_id from musehub.core.genesis import compute_branch_id from musehub.types.json_types import IntDict, JSONObject, JSONValue, ReadOnlyJSONObject, StrDict from musehub.config import settings from musehub.storage import get_backend from musehub.storage.backends import read_object_bytes from musehub.services.musehub_gc import invalidate_fetch_mpack_cache from musehub.services.musehub_wire_shared import ( MPackValidationError, NonFastForwardError, ObjectHashMismatch, RepairResult, _ChildMap, _commit_identity_bytes, _is_ancestor_db, _is_fast_forward, _reconstruct_manifest, _reconstruct_manifest_validated, _to_wire_commit, _upsert_object_refs, _utc_now, _parse_iso, _str_list, _int_safe, logger, ) _ObjFetchMap = dict[str, tuple[str, str, str | None, bytes | None]] _SnapshotManifest = dict[str, str] _EMPTY_OID = "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" _COMMIT_BATCH = 50 async def wire_refs( session: AsyncSession, repo_id: str, ) -> WireRefsResponse | None: _t0 = _time_module.perf_counter() def _ms() -> float: return (_time_module.perf_counter() - _t0) * 1000 logger.info("[wire_refs] START repo_id=%s", repo_id) repo_row = await session.get(MusehubRepo, repo_id) logger.info("[wire_refs] repo_row loaded found=%s t=%.1fms", repo_row is not None, _ms()) if repo_row is None: return None branch_rows = ( await session.execute( select(MusehubBranch).where(MusehubBranch.repo_id == repo_id) ) ).scalars().all() logger.info("[wire_refs] branches loaded count=%d t=%.1fms", len(branch_rows), _ms()) branch_heads: StrDict = { b.name: b.head_commit_id for b in branch_rows if b.head_commit_id } domain_meta: JSONObject = ( repo_row.domain_meta if isinstance(repo_row.domain_meta, dict) else {} ) domain = str(domain_meta.get("domain", "code")) default_branch = getattr(repo_row, "default_branch", None) or "main" logger.info("[wire_refs] RETURN branches=%d TOTAL=%.1fms", len(branch_heads), _ms()) return WireRefsResponse( repo_id=repo_id, domain=domain, default_branch=default_branch, branch_heads=branch_heads, ) async def _enqueue_repair_prebuild(session: AsyncSession, repo_id: str) -> None: """Enqueue a fetch.mpack.prebuild job after repair so the next clone is a fast HIT. Repair invalidates the full repo cache (repo-wide, D1) — the next clone would be a synchronous MISS-build without this. Enqueueing a prebuild here closes the loop so the worker rebuilds the mpack from corrected rows before the next client request. Covers all current branch tips (any tip may reach the repaired content). Does NOT commit — the caller's existing session.commit() makes the row visible. Idempotent: enqueue_job skips insertion when a pending job already exists. """ from musehub.services.musehub_jobs import enqueue_job tips = (await session.execute( select(MusehubBranch.head_commit_id) .where(MusehubBranch.repo_id == repo_id) .where(MusehubBranch.head_commit_id.isnot(None)) )).scalars().all() if tips: await enqueue_job( session, repo_id, "fetch.mpack.prebuild", {"tip_commit_ids": list(tips)}, ) async def wire_repair_object( session: AsyncSession, repo_id: str, object_id: str, content: bytes, caller_id: str | None, ) -> RepairResult: repo_row = await session.get(MusehubRepo, repo_id) if repo_row is None: raise ValueError("repo not found") if not caller_id: raise PermissionError("repair rejected: unauthenticated") if caller_id != repo_row.owner: collab_row = (await session.execute( select(MusehubCollaborator).where( MusehubCollaborator.repo_id == repo_id, MusehubCollaborator.identity_handle == caller_id, MusehubCollaborator.accepted_at.isnot(None), MusehubCollaborator.permission.in_(["write", "admin"]), ) )).scalar_one_or_none() if collab_row is None: raise PermissionError("repair rejected: not authorized") actual_id = blob_id(content) if actual_id != object_id: _, expected_hex = split_id(object_id) _, actual = split_id(actual_id) raise ObjectHashMismatch( f"repair content hash mismatch for {object_id!r}: " f"declared={expected_hex[:16]}… actual={actual[:16]}…" ) backend = get_backend() uri = await backend.put(object_id, content) await session.merge(MusehubObject( object_id=object_id, path="", size_bytes=len(content), storage_uri=uri, content_cache=None, )) await _upsert_object_refs(session, repo_id, [object_id]) await invalidate_fetch_mpack_cache(session, repo_id) await _enqueue_repair_prebuild(session, repo_id) await session.commit() logger.info( "✅ repair-object repo=%s object_id=%s size=%d", repo_id, object_id, len(content), ) return {"repaired": True} async def wire_fetch_objects( session: AsyncSession, repo_id: str, object_ids: list[str], ) -> list[dict]: result: list[dict] = [] for oid in object_ids: obj_row = await session.get(MusehubObject, oid) if obj_row is None: continue content: bytes | None = await read_object_bytes(obj_row, session=session) if content is None: continue result.append({"object_id": oid, "content": content}) return result async def wire_repair_snapshot( session: AsyncSession, repo_id: str, snapshot_id: str, manifest: dict[str, str], directories: list[str], caller_id: str | None, ) -> RepairResult: from musehub.muse_cli.snapshot import compute_snapshot_id repo_row = await session.get(MusehubRepo, repo_id) if repo_row is None: raise ValueError("repo not found") if not caller_id: raise PermissionError("repair rejected: unauthenticated") if caller_id != repo_row.owner: collab_row = (await session.execute( select(MusehubCollaborator).where( MusehubCollaborator.repo_id == repo_id, MusehubCollaborator.identity_handle == caller_id, MusehubCollaborator.accepted_at.isnot(None), MusehubCollaborator.permission.in_(["write", "admin"]), ) )).scalar_one_or_none() if collab_row is None: raise PermissionError("repair rejected: not authorized") recomputed = compute_snapshot_id(manifest, directories or []) if recomputed != snapshot_id: raise ObjectHashMismatch( f"repair snapshot hash mismatch for {snapshot_id!r}: " f"declared={snapshot_id}… recomputed={recomputed}…" ) import msgpack as _msgpack blob = _msgpack.packb(manifest, use_bin_type=True) dirs_sorted = sorted(directories) if directories else [] await session.merge(MusehubSnapshot( snapshot_id=snapshot_id, directories=dirs_sorted, manifest_blob=blob, entry_count=len(manifest), created_at=_utc_now(), )) await session.execute( _pg_insert(MusehubSnapshotRef) .values([{"repo_id": repo_id, "snapshot_id": snapshot_id, "created_at": _utc_now()}]) .on_conflict_do_nothing(index_elements=["repo_id", "snapshot_id"]) ) await invalidate_fetch_mpack_cache(session, repo_id) await _enqueue_repair_prebuild(session, repo_id) await session.commit() logger.info( "✅ repair-snapshot repo=%s snapshot_id=%s entries=%d", repo_id, snapshot_id, len(manifest), ) return {"repaired": True} async def wire_repair_commit( session: AsyncSession, repo_id: str, commit: ReadOnlyJSONObject, caller_id: str | None, ) -> RepairResult: """Replace a stored commit's identity fields with verified-correct content. Commit-level analog of wire_repair_object / wire_repair_snapshot. Intended for operators to repair commits whose stored row no longer reproduces its commit_id — e.g. commits the rc10 object-store migration stamped with a signer_public_key (and signature) without recomputing the id. Because signer_public_key is part of the commit identity hash, the serve path then recomputes a different id, the client's hash check fails on clone, the commit is dropped, and every descendant fails "parent not in mpack", emptying the working tree. Request body (msgpack): commit dict — a wire commit record (WireCommit shape). Its identity fields (parent ids, snapshot_id, message, committed_at, author, signer_public_key) must reproduce commit_id. committed_at is round-tripped through ``timestamp.isoformat()`` exactly as the serve path reproduces it (via _to_wire_commit), so the identity we verify is the identity that will be re-served after we store the parsed datetime. A commit whose timestamp cannot survive that round-trip fails the hash check and is rejected rather than repaired into a row that would still fail to clone. Only the repo owner or a write/admin collaborator may call this. """ repo_row = await session.get(MusehubRepo, repo_id) if repo_row is None: raise ValueError("repo not found") if not caller_id: raise PermissionError("repair rejected: unauthenticated") if caller_id != repo_row.owner: collab_row = (await session.execute( select(MusehubCollaborator).where( MusehubCollaborator.repo_id == repo_id, MusehubCollaborator.identity_handle == caller_id, MusehubCollaborator.accepted_at.isnot(None), MusehubCollaborator.permission.in_(["write", "admin"]), ) )).scalar_one_or_none() if collab_row is None: raise PermissionError("repair rejected: not authorized") commit_id = str(commit.get("commit_id") or "") if not commit_id: raise ValueError("repair-commit requires commit.commit_id") row = await session.get(MusehubCommit, commit_id) if row is None: raise ValueError(f"commit {commit_id} not found") # A malformed payload must surface as a 422 (ObjectHashMismatch -> 422 at the # route), never a 500 — WireCommit's field validators raise on bad ids. try: wire = WireCommit(**commit) except Exception as exc: raise ObjectHashMismatch(f"repair commit invalid payload for {commit_id!r}: {exc}") # Pin committed_at to the SERVE representation so the identity we verify is exactly # the one the serve path will reproduce from the stored datetime. ts = _parse_iso(wire.committed_at) wire.committed_at = ts.isoformat() recomputed = long_id(hashlib.sha256(_commit_identity_bytes(wire)).hexdigest()) if recomputed != commit_id: raise ObjectHashMismatch( f"repair commit hash mismatch for {commit_id!r}: " f"declared={commit_id} recomputed={recomputed}" ) # Force-overwrite identity-bearing fields in place; preserve server-only # bookkeeping (push-target branch, created_at, provenance) on the existing row. parents = [p for p in (wire.parent_commit_id, wire.parent2_commit_id) if p] row.parent_ids = parents row.snapshot_id = wire.snapshot_id row.message = wire.message row.author = wire.author row.timestamp = ts row.signer_public_key = wire.signer_public_key row.signature = wire.signature row.signer_key_id = wire.signer_key_id await invalidate_fetch_mpack_cache(session, repo_id) await _enqueue_repair_prebuild(session, repo_id) await session.commit() logger.info( "✅ repair-commit repo=%s commit_id=%s signer=%r", repo_id, commit_id, wire.signer_public_key, ) return {"repaired": True} async def _fetch_commit( session: AsyncSession, commit_id: str, ) -> MusehubCommit | None: return await session.get(MusehubCommit, commit_id) _ObjFetchMap = dict[str, tuple[str, str, str | None, bytes | None]] async def _fetch_object_meta( session: AsyncSession, repo_id: str, object_ids: list[str], ) -> _ObjFetchMap: result: _ObjFetchMap = {} if _EMPTY_OID in object_ids: result[_EMPTY_OID] = (repo_id, "", None, None) remaining = [oid for oid in object_ids if oid != _EMPTY_OID] if remaining: obj_rows_q = await session.execute( select(MusehubObject).where(MusehubObject.object_id.in_(remaining)) ) result.update({ r.object_id: (repo_id, r.path or "", r.storage_uri, r.content_cache) for r in obj_rows_q.scalars().all() }) return result def compute_object_byte_offsets(wire_bytes: bytes) -> "dict[str, tuple[int, int]]": """Return {object_id: (absolute_byte_offset, byte_length)} for every object in the OBJECTS section of a wire mpack binary. Parses the section table to find the OBJECTS section, then walks the ``_build_pack`` layout to compute each object's content start position within *wire_bytes*. The returned offsets are absolute — callers can issue ``Range: bytes=offset-offset+length-1`` S3 GETs directly against the mpack. Returns an empty dict for wire mpacks with no OBJECTS section. """ import struct as _struct _WIRE_SEC_BLOBS = 1 _OID_BYTES = 71 # len("sha256:") + 64 hex chars _PACK_HEADER = 13 # magic(4) + version(1) + count(8) if len(wire_bytes) < 6 or wire_bytes[:4] != b"MUSE": return {} section_count = wire_bytes[5] cursor = 6 objects_section_offset: int | None = None for _ in range(section_count): sec_type = wire_bytes[cursor] (sec_offset,) = _struct.unpack_from(" list[WireCommit]: if not commits: return [] by_id = {c.commit_id: c for c in commits} in_degree: IntDict = {c.commit_id: 0 for c in commits} children: _ChildMap = {c.commit_id: [] for c in commits} for c in commits: for pid in filter(None, [c.parent_commit_id, c.parent2_commit_id]): if pid in by_id: in_degree[c.commit_id] += 1 children[pid].append(c.commit_id) queue: collections.deque[str] = collections.deque( cid for cid, deg in in_degree.items() if deg == 0 ) result: list[WireCommit] = [] while queue: cid = queue.popleft() result.append(by_id[cid]) for child_id in children.get(cid, []): in_degree[child_id] -= 1 if in_degree[child_id] == 0: queue.append(child_id) sorted_ids = {c.commit_id for c in result} for c in commits: if c.commit_id not in sorted_ids: result.append(c) return result async def record_mpack_bytes_uploaded( session: AsyncSession, identity_id: str, size_bytes: int, ) -> None: import datetime as _dt today = _dt.date.today() await session.execute( _pg_insert(MusehubDailyPushBytes) .values(identity_id=identity_id, date=today, bytes_uploaded=size_bytes) .on_conflict_do_update( index_elements=["identity_id", "date"], set_={"bytes_uploaded": MusehubDailyPushBytes.bytes_uploaded + size_bytes}, ) ) async def check_push_anomaly( session: AsyncSession, identity_id: str, bytes_today: int, ) -> bool: import datetime as _dt import uuid as _uuid today = _dt.date.today() cutoff = today - _dt.timedelta(days=30) result = await session.execute( select(func.avg(MusehubDailyPushBytes.bytes_uploaded)).where( MusehubDailyPushBytes.identity_id == identity_id, MusehubDailyPushBytes.date >= cutoff, MusehubDailyPushBytes.date < today, ) ) avg_bytes = result.scalar() if avg_bytes is None or avg_bytes == 0: return False ratio = bytes_today / float(avg_bytes) if ratio <= 10.0: return False anomaly_id = "sha256:" + hashlib.sha256( f"{identity_id}:{today.isoformat()}:{bytes_today}".encode() ).hexdigest() logger.warning( "[anomaly] push spike detected identity_id=%s bytes_today=%d avg_bytes=%.0f ratio=%.1fx", identity_id[:20], bytes_today, avg_bytes, ratio, ) session.add(MusehubPushAnomaly( anomaly_id=anomaly_id, identity_id=identity_id, bytes_today=bytes_today, rolling_avg_bytes=float(avg_bytes), ratio=ratio, )) return True async def wire_push_mpack_presign( mpack_key: str, size_bytes: int, ttl_seconds: int = 3600, ) -> JSONObject: logger.warning("[mpack-presign] START mpack_key=%s size_bytes=%d ttl=%ds", mpack_key, size_bytes, ttl_seconds) backend = get_backend() upload_url = await backend.presign_mpack_put(mpack_key, ttl_seconds) logger.warning("[mpack-presign] upload_url=%s", upload_url) return {"upload_url": upload_url, "mpack_key": mpack_key} async def wire_push_unpack_mpack( session: AsyncSession, repo_id: str, mpack_key: str, pusher_id: str | None, branch: str = "main", head_commit_id: str = "", commits_count: int = 0, blobs_count: int = 0, force: bool = False, ) -> JSONObject: t0 = _time_module.monotonic() def _ms(ref: float = 0.0) -> float: return (_time_module.monotonic() - (t0 if ref == 0.0 else ref)) * 1000 from musehub.storage.backends import get_backend backend = get_backend() logger.warning( "[SERVER step 0] receive repo=%s branch=%s head=%s", repo_id[:20], branch, head_commit_id[:20] if head_commit_id else "none", ) # step 1: fetch mpack bytes from MinIO by mpack_key wire_bytes = await backend.get_mpack(mpack_key) t_got = _time_module.monotonic() if not wire_bytes: raise ValueError(f"mpack {mpack_key[:20]} not found in storage") mpack_bytes_len = len(wire_bytes) logger.warning( "[SERVER step 1] fetched mpack from object store %.1fms size=%d bytes (%.4f MB) key=%s", _ms(), mpack_bytes_len, mpack_bytes_len / 1_048_576, mpack_key, ) from musehub.config import get_settings as _get_settings _max_bytes = _get_settings().mpack_max_bytes if mpack_bytes_len > _max_bytes: raise ValueError( f"mpack {mpack_key[:20]} size {mpack_bytes_len:,} bytes " f"exceeds limit {_max_bytes:,} bytes" ) # step 2: verify blob_id(bytes) == mpack_key actual_id = blob_id(wire_bytes) t_verify = _time_module.monotonic() _, expected_hex = split_id(mpack_key) _, got_hex = split_id(actual_id) match = actual_id == mpack_key logger.warning( "[SERVER step 2] integrity check %.1fms expected=%s got=%s match=%s", (t_verify - t_got) * 1000, expected_hex[:20], got_hex[:20], match, ) if not match: raise ValueError( f"mpack integrity failure: sha256={got_hex[:20]}… ≠ key={expected_hex[:20]}…" ) # ── 2. Inline index ─────────────────────────────────────────────────────── import zstandard as _zstd_sync if wire_bytes[:4] != b"MUSE": raise ValueError(f"mpack is not MUSE binary format (got {wire_bytes[:4]!r})") from muse.core.mpack import parse_wire_mpack as _parse_wire_mpack _mpack_sync: JSONObject = _parse_wire_mpack(wire_bytes) t_parse = _time_module.monotonic() _raw_blobs: list[JSONValue] = _mpack_sync.get("blobs") or [] _raw_snaps: list[JSONValue] = _mpack_sync.get("snapshots") or [] _raw_commits_inline: list[JSONValue] = _mpack_sync.get("commits") or [] logger.warning("[SERVER step 3+4] verify MUSE magic + parse %.1fms commits=%d snapshots=%d blobs=%d", (t_parse - t_verify) * 1000, len(_raw_commits_inline), len(_raw_snaps), len(_raw_blobs)) # 2b. zstd decompress all blobs — zip bomb guard runs inline _dctx_sync = _zstd_sync.ZstdDecompressor() _decompressed_sync: list[tuple[str, bytes]] = [] _total_blob_bytes = 0 _max_decompressed = settings.mpack_max_decompressed_bytes for _obj in _raw_blobs: _oid = _obj["object_id"] _raw = _obj.get("content") or b"" if _obj.get("encoding") == "zstd" and _raw: _raw = _dctx_sync.decompress(_raw) _total_blob_bytes += len(_raw) if _total_blob_bytes > _max_decompressed: _zip_err = MPackValidationError( f"decompressed size {_total_blob_bytes:,} exceeds limit " f"{_max_decompressed:,} — possible zip bomb" ) await backend.quarantine_mpack(mpack_key) raise _zip_err _decompressed_sync.append((_oid, _raw)) t_decomp = _time_module.monotonic() logger.warning("[SERVER step 5] decompress + zip-bomb check %.1fms blobs=%d uncompressed=%.2f MB", (t_decomp - t_parse) * 1000, len(_decompressed_sync), _total_blob_bytes / 1_048_576) # step 6: check blocked hashes if _decompressed_sync: _all_check_oids = [_oid for _oid, _ in _decompressed_sync] _blocked_oids = (await session.execute( select(MusehubBlockedHash.object_id) .where(MusehubBlockedHash.object_id.in_(_all_check_oids)) )).scalars().all() if _blocked_oids: _blocked_err = MPackValidationError( f"mpack contains {len(_blocked_oids)} blocked object(s): " + ", ".join(oid[:20] for oid in _blocked_oids[:3]) + ("..." if len(_blocked_oids) > 3 else "") ) await backend.quarantine_mpack(mpack_key) raise _blocked_err logger.warning("[SERVER step 6] check blocked hashes objects=%d blocked=0", len(_decompressed_sync)) # step 7: write objects _new_oids: list[str] = [] if _decompressed_sync: _cc_oids = [_oid for _oid, _ in _decompressed_sync] t_obj_sel0 = _time_module.monotonic() _existing_oids = set((await session.execute( _sa_text("SELECT object_id FROM musehub_objects WHERE object_id = ANY(:ids)"), {"ids": _cc_oids}, )).scalars()) t_obj_sel1 = _time_module.monotonic() logger.warning("[SERVER step 7a] objects: dedup SELECT %.1fms total=%d existing=%d new=%d", (t_obj_sel1 - t_obj_sel0) * 1000, len(_cc_oids), len(_existing_oids), len(_cc_oids) - len(_existing_oids)) session.add_all([ MusehubObject(object_id=_oid, path="", size_bytes=len(_data), storage_uri=f"mpack://{mpack_key}", content_cache=None) for _oid, _data in _decompressed_sync if _oid not in _existing_oids ]) t_obj_add = _time_module.monotonic() logger.warning("[SERVER step 7b] objects: INSERT new %.1fms", (t_obj_add - t_obj_sel1) * 1000) _new_oids = [_oid for _oid in _cc_oids if _oid not in _existing_oids] t_ref0 = _time_module.monotonic() await _upsert_object_refs(session, repo_id, _cc_oids) t_ref1 = _time_module.monotonic() logger.warning("[SERVER step 7c] objects: UPSERT object_refs %.1fms oids=%d", (t_ref1 - t_ref0) * 1000, len(_cc_oids)) if _cc_oids: _byte_offsets = compute_object_byte_offsets(wire_bytes) _mpack_idx_rows = [ { "entity_id": _oid, "mpack_id": mpack_key, "entity_type": "object", "created_at": _utc_now(), "byte_offset": _byte_offsets.get(_oid, (None, None))[0], "byte_length": _byte_offsets.get(_oid, (None, None))[1], } for _oid in _cc_oids ] # asyncpg caps query parameters at 32767; 6 columns per row → 5461 rows/batch _MPACK_IDX_BATCH = 5461 for _batch_start in range(0, len(_mpack_idx_rows), _MPACK_IDX_BATCH): _batch = _mpack_idx_rows[_batch_start:_batch_start + _MPACK_IDX_BATCH] _idx_stmt = _pg_insert(MusehubMPackIndex).values(_batch) await session.execute( _idx_stmt.on_conflict_do_update( index_elements=["entity_id", "mpack_id"], set_={ "byte_offset": _idx_stmt.excluded.byte_offset, "byte_length": _idx_stmt.excluded.byte_length, }, where=MusehubMPackIndex.byte_offset.is_(None), ) ) t_idx_add = _time_module.monotonic() logger.warning("[SERVER step 7d] objects: UPSERT mpack_index %.1fms total=%d", (t_idx_add - t_ref1) * 1000, len(_cc_oids)) else: t_idx_add = t_decomp # step 7: write snapshots _snap_ids_in_mpack = {_sd.get("snapshot_id") for _sd in _raw_snaps if _sd.get("snapshot_id")} _external_parent_sids = { _sd.get("parent_snapshot_id") for _sd in _raw_snaps if _sd.get("parent_snapshot_id") and _sd.get("parent_snapshot_id") not in _snap_ids_in_mpack } _parent_snap_manifests: dict[str, dict] = {} t_snap_ext0 = _time_module.monotonic() if _external_parent_sids: _psnap_rows = (await session.execute( select(MusehubSnapshot.snapshot_id, MusehubSnapshot.manifest_blob) .where(MusehubSnapshot.snapshot_id.in_(_external_parent_sids)) )).all() _psid_direct = set() for _psid, _pblob in _psnap_rows: if _pblob: _parent_snap_manifests[_psid] = dict(_msgpack.unpackb(_pblob, raw=False)) _psid_direct.add(_psid) # An external parent with no cached manifest_blob is delta-only (or, in the # worst case, missing a row entirely) — reconstruct it by walking its delta # chain instead of silently treating it as an empty manifest. A silent {} # here permanently corrupts every descendant snapshot (musehub#134). _psid_needs_reconstruct = _external_parent_sids - _psid_direct for _psid in _psid_needs_reconstruct: _reconstructed = await _reconstruct_manifest_validated(session, _psid) if _reconstructed: _parent_snap_manifests[_psid] = _reconstructed t_snap_ext1 = _time_module.monotonic() logger.warning("[SERVER step 8a] snapshots: fetch external parents %.1fms external=%d loaded=%d", (t_snap_ext1 - t_snap_ext0) * 1000, len(_external_parent_sids), len(_parent_snap_manifests)) _head_snap_id: str | None = None for _cd in _raw_commits_inline: if _cd.get("commit_id") == head_commit_id: _head_snap_id = _cd.get("snapshot_id") or None break _snap_resolved: dict[str, _SnapshotManifest] = {} _snap_rows_inline = [] _n_snap_root = 0 _n_snap_head = 0 _n_snap_delta_only = 0 _snap_manifest_bytes_total = 0 _snap_delta_bytes_total = 0 t_snap_loop0 = _time_module.monotonic() for _sd in _raw_snaps: _sid = _sd.get("snapshot_id", "") if not _sid: continue _parent_sid = _sd.get("parent_snapshot_id") _delta_upsert: dict[str, str] = _sd.get("delta_upsert") or {} _delta_remove: list[str] = _sd.get("delta_remove") or [] if _parent_sid and _parent_sid in _snap_resolved: _parent_base = _snap_resolved[_parent_sid] _null_inherited = [k for k, v in _parent_base.items() if not v] if _null_inherited: logger.warning("[PUSH-NULL-FILTER] snap=%s filtered %d null OIDs from in-flight parent: %s", _sid[:20], len(_null_inherited), _null_inherited) _base = {k: v for k, v in _parent_base.items() if v} elif _parent_sid: _parent_base = _parent_snap_manifests.get(_parent_sid) or {} _null_inherited = [k for k, v in _parent_base.items() if not v] if _null_inherited: logger.warning("[PUSH-NULL-FILTER] snap=%s filtered %d null OIDs from DB parent: %s", _sid[:20], len(_null_inherited), _null_inherited) _base = {k: v for k, v in _parent_base.items() if v} else: _base = {} _null_delta = [k for k, v in _delta_upsert.items() if not v] if _null_delta: logger.warning("[PUSH-NULL-FILTER] snap=%s filtered %d null OIDs from delta_upsert: %s", _sid[:20], len(_null_delta), _null_delta) _base.update({k: v for k, v in _delta_upsert.items() if v}) for _path in _delta_remove: _base.pop(_path, None) _snap_resolved[_sid] = _base _delta_encoded: JSONObject = {"add": _delta_upsert} if _delta_remove: _delta_encoded["rm"] = _delta_remove _delta_blob = _msgpack.packb(_delta_encoded, use_bin_type=True) if (_delta_upsert or _delta_remove) else None _is_root = not _parent_sid _is_head = _sid == _head_snap_id # Only store full manifest for root and head — delta-only snapshots # get manifest_blob=None here. They are NOT backfilled by any job; the fetch # path reconstructs their manifest on-demand by walking the delta chain # (_reconstruct_manifest in musehub_wire_shared.py). The only path that ever # persists a manifest_blob for a delta-only row is wire_repair_snapshot, and # only when a client explicitly repairs that snapshot (it validates the # recomputed snapshot_id before writing). The mpack.index job does NOT touch # manifest_blob — it only indexes blob byte-offsets. # Storing all 1000 manifests inline would push hundreds of MB into a single # transaction, blocking snapshot_refs inserts (regression at scale). _manifest_blob = _msgpack.packb(_base, use_bin_type=True) if (_is_root or _is_head) else None if _is_root: _n_snap_root += 1 if _is_head: _n_snap_head += 1 if not _is_root and not _is_head: _n_snap_delta_only += 1 if _manifest_blob: _snap_manifest_bytes_total += len(_manifest_blob) if _delta_blob: _snap_delta_bytes_total += len(_delta_blob) _snap_dirs = [str(d) for d in (_sd.get("directories") or []) if d] # The hash IS the proof: a base resolved from a missing/unreconstructable # external parent (see _psid_needs_reconstruct above) will never reproduce # _sid. Reject the push loudly rather than persist a manifest that can # never be verified again (musehub#134). if _parent_sid and hash_snapshot(_base, _snap_dirs or None) != _sid: raise ValueError( f"push rejected: snapshot {_sid[:20]}… cannot be reconstructed — " f"its parent {_parent_sid[:20]}… is missing or unrecoverable on the " f"server. Run 'muse verify' to audit store integrity before retrying." ) _snap_rows_inline.append({ "snapshot_id": _sid, "directories": _snap_dirs, "manifest_blob": _manifest_blob, "entry_count": len(_base), "created_at": _utc_now(), "parent_snapshot_id": _parent_sid or None, "delta_blob": _delta_blob, }) t_snap_loop1 = _time_module.monotonic() logger.warning( "[SERVER step 8b] snapshots: replay delta chain %.1fms " "total=%d root=%d head=%d delta_only=%d", (t_snap_loop1 - t_snap_loop0) * 1000, len(_snap_rows_inline), _n_snap_root, _n_snap_head, _n_snap_delta_only, ) if _snap_rows_inline: _snap_ids_to_check = [r["snapshot_id"] for r in _snap_rows_inline] t_snap_sel0 = _time_module.monotonic() _existing_snap_rows = (await session.execute( _sa_text( "SELECT snapshot_id, (delta_blob IS NOT NULL) AS has_delta " "FROM musehub_snapshots WHERE snapshot_id = ANY(:ids)" ), {"ids": _snap_ids_to_check}, )).all() _existing_sids: set[str] = {row[0] for row in _existing_snap_rows} _sids_already_healed: set[str] = {row[0] for row in _existing_snap_rows if row[1]} t_snap_sel1 = _time_module.monotonic() logger.warning("[SERVER step 8c] snapshots: dedup SELECT %.1fms existing=%d new=%d", (t_snap_sel1 - t_snap_sel0) * 1000, len(_existing_sids), len(_snap_rows_inline) - len(_existing_sids)) _new_snap_dicts = [r for r in _snap_rows_inline if r["snapshot_id"] not in _existing_sids] _heal_snap_dicts = [ r for r in _snap_rows_inline if r["snapshot_id"] in _existing_sids and r["snapshot_id"] not in _sids_already_healed and r.get("delta_blob") ] if _new_snap_dicts: session.add_all([MusehubSnapshot(**_r) for _r in _new_snap_dicts]) if _heal_snap_dicts: _HEAL_CHUNK = 2000 for _hi in range(0, len(_heal_snap_dicts), _HEAL_CHUNK): _chunk = _heal_snap_dicts[_hi : _hi + _HEAL_CHUNK] _heal_stmt = _pg_insert(MusehubSnapshot).values(_chunk) await session.execute( _heal_stmt.on_conflict_do_update( index_elements=["snapshot_id"], set_={ "delta_blob": _heal_stmt.excluded.delta_blob, "parent_snapshot_id": _heal_stmt.excluded.parent_snapshot_id, }, where=MusehubSnapshot.delta_blob.is_(None), ) ) t_snap_add = _time_module.monotonic() logger.warning("[SERVER step 8d] snapshots: INSERT new %.1fms new=%d healed=%d", (t_snap_add - t_snap_sel1) * 1000, len(_new_snap_dicts), len(_heal_snap_dicts)) _new_snap_rows = _new_snap_dicts if _new_snap_rows: await session.execute( _pg_insert(MusehubSnapshotRef) .values([{"repo_id": repo_id, "snapshot_id": r["snapshot_id"], "created_at": _utc_now()} for r in _new_snap_rows]) .on_conflict_do_nothing(index_elements=["repo_id", "snapshot_id"]) ) t_snap_ref = _time_module.monotonic() logger.warning("[SERVER step 8e] snapshots: UPSERT snapshot_refs %.1fms new=%d", (t_snap_ref - t_snap_add) * 1000, len(_new_snap_rows)) else: t_snap_ref = t_snap_loop1 # step 8: write commits from muse.core.commits import CommitRecord as _CR_inline _commit_rows_inline = [] _graph_rows_inline = [] _commit_ids_in_mpack = { _cd.get("commit_id") for _cd in _raw_commits_inline if _cd.get("commit_id") } _external_parent_cids = { pid for _cd in _raw_commits_inline for pid in ( ([_cd["parent_commit_id"]] if _cd.get("parent_commit_id") else []) + ([_cd["parent2_commit_id"]] if _cd.get("parent2_commit_id") else []) ) if pid not in _commit_ids_in_mpack } t_gen_sel0 = _time_module.monotonic() _db_parent_gens: dict[str, int] = {} if _external_parent_cids: _gen_rows = (await session.execute( select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) .where(MusehubCommitGraph.commit_id.in_(_external_parent_cids)) )).all() _db_parent_gens = {_cid: _gval for _cid, _gval in _gen_rows} t_gen_sel1 = _time_module.monotonic() logger.warning("[SERVER step 9a] commits: fetch parent generations %.1fms external_parents=%d found=%d", (t_gen_sel1 - t_gen_sel0) * 1000, len(_external_parent_cids), len(_db_parent_gens)) t_commit_parse0 = _time_module.monotonic() _cid_set_inline = {_cd.get("commit_id") for _cd in _raw_commits_inline if _cd.get("commit_id")} _cd_by_cid: dict[str, dict] = {_cd["commit_id"]: _cd for _cd in _raw_commits_inline if _cd.get("commit_id")} _children: dict[str, list[dict]] = {cid: [] for cid in _cid_set_inline} _in_degree: dict[str, int] = {cid: 0 for cid in _cid_set_inline} for _cd in _raw_commits_inline: for _pk in ("parent_commit_id", "parent2_commit_id"): _p = _cd.get(_pk) or "" if _p in _cid_set_inline: _children[_p].append(_cd) _in_degree[_cd["commit_id"]] = _in_degree.get(_cd["commit_id"], 0) + 1 from collections import deque as _deque _queue: _deque[dict] = _deque( _cd_by_cid[cid] for cid, deg in _in_degree.items() if deg == 0 ) _topo_sorted: list[dict] = [] while _queue: _cd = _queue.popleft() _topo_sorted.append(_cd) for _child in _children.get(_cd.get("commit_id", ""), []): _child_cid = _child.get("commit_id", "") _in_degree[_child_cid] -= 1 if _in_degree[_child_cid] == 0: _queue.append(_child) _topo_seen = {_cd.get("commit_id") for _cd in _topo_sorted} _topo_sorted.extend(_cd for _cd in _raw_commits_inline if _cd.get("commit_id") not in _topo_seen) _inline_gen: dict[str, int] = {} for _cd in _topo_sorted: # musehub#195: capture the exact client-original committed_at string # BEFORE CommitRecord.from_dict parses it into a datetime -- this is # the only point in the whole ingest path where the untouched wire # string still exists. Postgres timestamptz normalizes non-UTC # offsets on write, so reconstructing wire bytes from the stored # `timestamp` column alone would produce a different string (and a # different commit_id) than the one this string hashes to. _committed_at_raw = _cd.get("committed_at") or None try: _cr = _CR_inline.from_dict(_cd) except Exception: continue if not _cr.commit_id: continue _pids: list[str] = [] if _cr.parent_commit_id: _pids.append(_cr.parent_commit_id) if _cr.parent2_commit_id: _pids.append(_cr.parent2_commit_id) _parent_gens: list[int] = [] for _pid in _pids: if _pid in _inline_gen: _parent_gens.append(_inline_gen[_pid]) elif _pid in _db_parent_gens: _parent_gens.append(_db_parent_gens[_pid]) if not _pids: _gen = 0 # genuine root — no parents, generation 0 is correct elif len(_parent_gens) == len(_pids): _gen = max(_parent_gens) + 1 else: _unresolved_pids = [ p for p in _pids if p not in _inline_gen and p not in _db_parent_gens ] logger.warning( "[MWP1] unresolved parent generations — backfilling %d at " "wire_push_unpack_mpack commit=%s unresolved_parents=%s", len(_unresolved_pids), _cr.commit_id, _unresolved_pids, ) _max_parent_gen = await _resolve_generation_with_backfill( session, _pids, inline_gen=_inline_gen, db_gen=_db_parent_gens, ) _gen = _max_parent_gen + 1 _inline_gen[_cr.commit_id] = _gen _commit_rows_inline.append({ "commit_id": _cr.commit_id, "branch": _cr.branch or "main", "message": _cr.message or "", "author": _cr.author or "", "timestamp": _cr.committed_at or _utc_now(), "committed_at_raw": _committed_at_raw, "parent_ids": _pids, "snapshot_id": _cr.snapshot_id or None, "agent_id": _cr.agent_id or "", "model_id": _cr.model_id or "", "toolchain_id": _cr.toolchain_id or "", "commit_branch": _cr.branch or None, "signature": _cr.signature or "", "signer_public_key": _cr.signer_public_key or "", "signer_key_id": _cr.signer_key_id or "", "sem_ver_bump": _cr.sem_ver_bump or "none", "breaking_changes": _cr.breaking_changes or [], "reviewed_by": [], "test_runs": 0, "prompt_hash": _cr.prompt_hash or "", "structured_delta": _cr.structured_delta if isinstance(_cr.structured_delta, dict) else None, }) _graph_rows_inline.append({ "commit_id": _cr.commit_id, "parent_ids": _pids, "generation": _gen, "snapshot_id": _cr.snapshot_id or None, }) t_commit_parse1 = _time_module.monotonic() _gen_values = sorted(_inline_gen.values()) logger.warning( "[SERVER step 9b] commits: topo sort + parse %.1fms parsed=%d gen_min=%s gen_max=%s", (t_commit_parse1 - t_commit_parse0) * 1000, len(_commit_rows_inline), _gen_values[0] if _gen_values else "NONE", _gen_values[-1] if _gen_values else "NONE", ) if _commit_rows_inline: t_csel0 = _time_module.monotonic() _commit_ids_to_check = [r["commit_id"] for r in _commit_rows_inline] _existing_cids = set((await session.execute( _sa_text("SELECT commit_id FROM musehub_commits WHERE commit_id = ANY(:ids)"), {"ids": _commit_ids_to_check}, )).scalars()) t_csel1 = _time_module.monotonic() logger.warning("[SERVER step 9c] commits: dedup SELECT %.1fms existing=%d new=%d", (t_csel1 - t_csel0) * 1000, len(_existing_cids), len(_commit_rows_inline) - len(_existing_cids)) _new_commit_rows = [r for r in _commit_rows_inline if r["commit_id"] not in _existing_cids] session.add_all([MusehubCommit(**_r) for _r in _new_commit_rows]) t_cadd = _time_module.monotonic() logger.warning("[SERVER step 9d] commits: INSERT new %.1fms new=%d", (t_cadd - t_csel1) * 1000, len(_new_commit_rows)) # Upsert refs for ALL commits in this push, not just globally-new ones. # A commit may already exist in musehub_commits from another repo's push # but still be missing its ref for THIS repo — using _new_commit_rows here # silently drops those refs and leaves commit history invisible to intel. if _commit_rows_inline: await session.execute( _pg_insert(MusehubCommitRef) .values([{"repo_id": repo_id, "commit_id": r["commit_id"], "created_at": _utc_now()} for r in _commit_rows_inline]) .on_conflict_do_nothing(index_elements=["repo_id", "commit_id"]) ) t_cref = _time_module.monotonic() logger.warning("[SERVER step 9e] commits: UPSERT commit_refs %.1fms total=%d", (t_cref - t_cadd) * 1000, len(_commit_rows_inline)) if _graph_rows_inline: _graph_stmt = _pg_insert(MusehubCommitGraph).values(_graph_rows_inline) await session.execute(_graph_stmt.on_conflict_do_update( index_elements=["commit_id"], set_={"generation": _graph_stmt.excluded.generation, "snapshot_id": _graph_stmt.excluded.snapshot_id}, )) t_graph = _time_module.monotonic() logger.warning("[SERVER step 9f] commits: UPSERT commit_graph %.1fms rows=%d", (t_graph - t_cref) * 1000, len(_graph_rows_inline)) else: t_graph = t_commit_parse1 # step 9: advance branch pointer tip = head_commit_id if tip and branch: t_branch0 = _time_module.monotonic() branch_row = (await session.execute( select(MusehubBranch).where( MusehubBranch.repo_id == repo_id, MusehubBranch.name == branch, ).with_for_update() )).scalar_one_or_none() current_head = branch_row.head_commit_id if branch_row else None if not force and current_head: if not _is_fast_forward(tip, current_head, wire_bytes): # The mpack-only walk can't confirm ancestry when some ancestor # commits are already on the remote (absent from the mpack because # they were in the 'have' set). Fall back to the commit graph. if not await _is_ancestor_db(session, current_head, tip, repo_id): raise NonFastForwardError( f"push rejected: {tip[:20]}… is not a fast-forward of " f"{current_head[:20]}… — use --force to override" ) if branch_row is None: session.add(MusehubBranch( branch_id=compute_branch_id(repo_id, branch), repo_id=repo_id, name=branch, head_commit_id=tip, )) else: branch_row.head_commit_id = tip # bust cache on force-push: tip-scoped so sibling caches survive (D1 + D3) if force and current_head and current_head != tip: await invalidate_fetch_mpack_cache(session, repo_id, tips=[current_head, tip]) t_branch1 = _time_module.monotonic() logger.warning("[SERVER step 10] advance branch pointer (atomic CAS) %.1fms", (t_branch1 - t_branch0) * 1000) else: t_branch1 = t_graph # commit db transaction t_pre_commit = _time_module.monotonic() await session.commit() t_commit = _time_module.monotonic() logger.warning("[SERVER step 10] commit db transaction %.1fms", (t_commit - t_pre_commit) * 1000) # step 11: enqueue async jobs t_intel0 = _time_module.monotonic() from musehub.services.musehub_jobs import enqueue_push_intel await enqueue_push_intel( session, repo_id, head=tip, branch=branch, mpack_key=mpack_key, ) t_intel1 = _time_module.monotonic() logger.warning("[SERVER step 11] enqueue intel jobs %.1fms", (t_intel1 - t_intel0) * 1000) elapsed_ms = (t_intel1 - t0) * 1000 logger.warning("[SERVER ✅] TOTAL=%.1fms branch=%s head=%s", elapsed_ms, branch, head_commit_id[:20] if head_commit_id else "none") return { "head": tip, "branch": branch, "blobs_in_mpack": blobs_count, "commits_in_mpack": commits_count, "blobs_written": len(_new_oids), "commits_written": len(_new_commit_rows) if _commit_rows_inline else 0, "snapshots_written": len(_new_snap_dicts) if _snap_rows_inline else 0, } class PurgeResult(TypedDict): removed: int kept: int async def purge_stale_mpack_index_entries(session: AsyncSession) -> PurgeResult: """Delete MusehubMPackIndex rows for mpacks no longer in MinIO. Left behind when mpack.gc deleted old mpacks but didn't prune the index. Safe to delete: objects with stale entries have valid s3:// storage_uri and are served directly from S3 without needing the mpack path. Returns {"removed": N, "kept": M}. """ from musehub.storage.backends import get_backend backend = get_backend() # Find all distinct mpack_ids in the index. distinct_mpacks = (await session.execute( select(MusehubMPackIndex.mpack_id).distinct() )).scalars().all() removed = 0 kept = 0 for mpack_id in distinct_mpacks: alive = await backend.exists_mpack(mpack_id) if alive: kept_count = (await session.execute( select(MusehubMPackIndex.entity_id).where( MusehubMPackIndex.mpack_id == mpack_id ) )).scalars().all() kept += len(kept_count) else: # Delete all index entries for this dead mpack. rows = (await session.execute( select(MusehubMPackIndex).where( MusehubMPackIndex.mpack_id == mpack_id ) )).scalars().all() for row in rows: await session.delete(row) removed += len(rows) await session.flush() logger.info("purge_stale_mpack_index_entries: removed=%d kept=%d", removed, kept) return {"removed": removed, "kept": kept} async def _resolve_generation_with_backfill( session: AsyncSession, pids: list[str], *, inline_gen: dict[str, int] | None = None, db_gen: dict[str, int] | None = None, ) -> int: """Resolve the max generation across *pids* by backfilling missing graph rows. Walks ``musehub_commits.parent_ids`` toward roots, computing and upserting ``musehub_commit_graph`` rows for every ancestor whose generation is unknown. The walk is **bounded**: it stops at the first commit already in the graph, so steady-state cost is O(unresolved frontier), not O(full history). Returns ``max(generation for pid in pids)``. Caller adds +1 to get the child's generation. Raises ``ValueError`` if any pid is absent from ``musehub_commits``. A fast-forward push must not reference commits the server does not have. """ known: dict[str, int] = {} if inline_gen: known.update(inline_gen) if db_gen: known.update(db_gen) # parent_map: commits we discovered via the walk that need generation computed. parent_map: dict[str, list[str]] = {} snapshot_map: dict[str, str | None] = {} frontier = {p for p in pids if p not in known} while frontier: # Stop the walk at commits already in musehub_commit_graph (anchors). graph_rows = (await session.execute( select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) .where(MusehubCommitGraph.commit_id.in_(frontier)) )).all() for cid, gen in graph_rows: known[cid] = gen need_walk = {c for c in frontier if c not in known} if not need_walk: break # Fetch authoritative parent pointers from musehub_commits (source-of-truth DAG). commit_rows = (await session.execute( select(MusehubCommit.commit_id, MusehubCommit.parent_ids, MusehubCommit.snapshot_id) .where(MusehubCommit.commit_id.in_(need_walk)) )).all() found = {cid for cid, _, _ in commit_rows} missing = need_walk - found if missing: raise ValueError( "[MWP1] integrity error: parent commit(s) absent from musehub_commits " f"— fast-forward push invariant violated: {sorted(missing)}" ) for cid, parent_ids, snap_id in commit_rows: parent_map[cid] = parent_ids snapshot_map[cid] = snap_id # Expand frontier: parents not yet known or already queued for walk. frontier = { p for cid, parent_ids, _ in commit_rows for p in parent_ids if p not in known and p not in parent_map } if not parent_map: # All pids were already resolved (in inline_gen, db_gen, or graph table). return max(known[p] for p in pids) # Topological sort of parent_map (Kahn's algorithm — roots first). children_map: dict[str, list[str]] = {cid: [] for cid in parent_map} in_degree: dict[str, int] = {cid: 0 for cid in parent_map} for cid, pids_of_cid in parent_map.items(): for p in pids_of_cid: if p in parent_map: children_map[p].append(cid) in_degree[cid] += 1 queue: collections.deque[str] = collections.deque( cid for cid, deg in in_degree.items() if deg == 0 ) computed: dict[str, int] = {} new_rows: list[dict] = [] now = _utc_now() while queue: cid = queue.popleft() pids_of_cid = parent_map[cid] parent_gens = [ computed[p] if p in computed else known[p] for p in pids_of_cid if p in computed or p in known ] gen = (max(parent_gens) + 1) if parent_gens else 0 computed[cid] = gen new_rows.append({ "commit_id": cid, "parent_ids": pids_of_cid, "generation": gen, "snapshot_id": snapshot_map.get(cid), "created_at": now, }) for child in children_map.get(cid, []): in_degree[child] -= 1 if in_degree[child] == 0: queue.append(child) if new_rows: _CHUNK = 2000 for i in range(0, len(new_rows), _CHUNK): chunk = new_rows[i:i + _CHUNK] stmt = _pg_insert(MusehubCommitGraph).values(chunk) await session.execute(stmt.on_conflict_do_update( index_elements=["commit_id"], set_={ "generation": stmt.excluded.generation, "snapshot_id": stmt.excluded.snapshot_id, }, )) known.update(computed) return max(known[p] for p in pids if p in known) async def repair_corrupt_commit_generations(session: AsyncSession) -> dict: """Repair musehub_commit_graph rows where generation=0 but parent_ids is non-empty. These rows were written by the RC-1 bug before Phase 2 fixed the generation computation. The repair: 1. Finds every row with ``generation == 0 AND cardinality(parent_ids) > 0``. 2. Fetches external parent generations from the graph (parents not in the corrupt set already have correct, non-zero generations there). 3. Backfills any external parents missing from the graph via ``_resolve_generation_with_backfill``. 4. Topologically sorts corrupt commits (parents before children) and recomputes correct generations bottom-up. 5. Bulk-upserts only the ``generation`` column — all other fields preserved. Idempotent: a second call finds zero corrupt rows and returns immediately. Returns ``{"total_corrupt": N, "repaired": N}``. """ # Step 1 — find all corrupt rows (gen=0, non-empty parents). corrupt_result = (await session.execute( select( MusehubCommitGraph.commit_id, MusehubCommitGraph.parent_ids, MusehubCommitGraph.snapshot_id, ) .where(MusehubCommitGraph.generation == 0) .where(func.cardinality(MusehubCommitGraph.parent_ids) > 0) )).all() if not corrupt_result: logger.info("[MWP1] repair: no corrupt commit-graph rows found — nothing to do") return {"total_corrupt": 0, "repaired": 0} corrupt_map: dict[str, list[str]] = { cid: list(pids) for cid, pids, _ in corrupt_result } corrupt_snap: dict[str, str | None] = { cid: snap for cid, _, snap in corrupt_result } corrupt_set = set(corrupt_map.keys()) total_corrupt = len(corrupt_set) logger.warning("[MWP1] repair: found %d corrupt commit-graph row(s)", total_corrupt) # Step 2 — collect external parents (parents NOT in corrupt_set) and fetch # their generations from the graph. These are the walk anchors. external_parent_cids = { p for pids in corrupt_map.values() for p in pids if p not in corrupt_set } known_gens: dict[str, int] = {} if external_parent_cids: ext_rows = (await session.execute( select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) .where(MusehubCommitGraph.commit_id.in_(external_parent_cids)) )).all() known_gens = {cid: gen for cid, gen in ext_rows} # Step 3 — backfill any external parents absent from the graph entirely # (should not happen after Phase 2, but handle gracefully). missing_ext = external_parent_cids - set(known_gens.keys()) if missing_ext: logger.warning( "[MWP1] repair: %d external parents have no graph row — backfilling", len(missing_ext), ) try: await _resolve_generation_with_backfill( session, list(missing_ext), inline_gen={}, db_gen={} ) refetch = (await session.execute( select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) .where(MusehubCommitGraph.commit_id.in_(missing_ext)) )).all() for cid, gen in refetch: known_gens[cid] = gen except ValueError as exc: logger.error("[MWP1] repair: backfill for missing externals failed: %s", exc) # Step 4 — topological sort of corrupt commits (Kahn's, roots first). children_map: dict[str, list[str]] = {cid: [] for cid in corrupt_set} in_degree: dict[str, int] = {cid: 0 for cid in corrupt_set} for cid, pids in corrupt_map.items(): for p in pids: if p in corrupt_set: children_map[p].append(cid) in_degree[cid] += 1 queue: collections.deque[str] = collections.deque( cid for cid, deg in in_degree.items() if deg == 0 ) computed: dict[str, int] = {} corrected: list[dict] = [] now = _utc_now() while queue: cid = queue.popleft() pids = corrupt_map[cid] parent_gens = [] for p in pids: if p in computed: parent_gens.append(computed[p]) elif p in known_gens: parent_gens.append(known_gens[p]) else: logger.warning( "[MWP1] repair: parent %s of corrupt commit %s not found " "in graph or computed set — using 0 as fallback", p[:20], cid[:20], ) parent_gens.append(0) correct_gen = (max(parent_gens) + 1) if parent_gens else 0 computed[cid] = correct_gen corrected.append({ "commit_id": cid, "parent_ids": pids, "generation": correct_gen, "snapshot_id": corrupt_snap.get(cid), "created_at": now, }) for child in children_map.get(cid, []): in_degree[child] -= 1 if in_degree[child] == 0: queue.append(child) # Step 5 — bulk-upsert, updating only the generation column. if corrected: _CHUNK = 2000 for i in range(0, len(corrected), _CHUNK): chunk = corrected[i:i + _CHUNK] stmt = _pg_insert(MusehubCommitGraph).values(chunk) await session.execute(stmt.on_conflict_do_update( index_elements=["commit_id"], set_={"generation": stmt.excluded.generation}, )) logger.warning("[MWP1] repair: corrected %d row(s)", len(corrected)) return {"total_corrupt": total_corrupt, "repaired": len(corrected)} async def check_commit_graph_invariant(session: AsyncSession) -> dict: """Check that no musehub_commit_graph row has generation=0 with non-empty parents. Returns {"valid": bool, "violations": int}. A violation means the RC-1 bug wrote a corrupt generation that will poison the fetch range scan. Call repair_corrupt_commit_generations to fix any violations found. """ count_q = await session.execute( select(func.count()).select_from(MusehubCommitGraph) .where(MusehubCommitGraph.generation == 0) .where(func.cardinality(MusehubCommitGraph.parent_ids) > 0) ) violations = count_q.scalar_one() return {"valid": violations == 0, "violations": violations} async def _build_commit_graph_from_raw(session: AsyncSession, raw_commits: list[dict]) -> int: """Compute generation numbers and upsert MusehubCommitGraph rows. Mirrors the generation logic in wire_push_unpack_mpack but operates on an already-parsed commit list (used by process_mpack_index_job). Returns the number of rows upserted. """ if not raw_commits: return 0 from collections import deque as _deque cid_set = {cd.get("commit_id") for cd in raw_commits if cd.get("commit_id")} cd_by_cid = {cd["commit_id"]: cd for cd in raw_commits if cd.get("commit_id")} # Topological sort (Kahn's algorithm — oldest first so parents resolve before children). children: dict[str, list[dict]] = {cid: [] for cid in cid_set} in_degree: dict[str, int] = {cid: 0 for cid in cid_set} for cd in raw_commits: for pk in ("parent_commit_id", "parent2_commit_id"): p = cd.get(pk) or "" if p in cid_set: children[p].append(cd) in_degree[cd["commit_id"]] = in_degree.get(cd["commit_id"], 0) + 1 queue: _deque[dict] = _deque(cd_by_cid[cid] for cid, deg in in_degree.items() if deg == 0) topo: list[dict] = [] while queue: cd = queue.popleft() topo.append(cd) for child in children.get(cd.get("commit_id", ""), []): child_cid = child.get("commit_id", "") in_degree[child_cid] -= 1 if in_degree[child_cid] == 0: queue.append(child) seen = {cd.get("commit_id") for cd in topo} topo.extend(cd for cd in raw_commits if cd.get("commit_id") not in seen) # Fetch generation numbers for any external parents already in the DB. external_pids = { pid for cd in raw_commits for pid in ([cd.get("parent_commit_id") or "", cd.get("parent2_commit_id") or ""]) if pid and pid not in cid_set } db_parent_gens: dict[str, int] = {} if external_pids: gen_rows = (await session.execute( select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) .where(MusehubCommitGraph.commit_id.in_(external_pids)) )).all() db_parent_gens = {cid: gval for cid, gval in gen_rows} inline_gen: dict[str, int] = {} graph_rows: list[dict] = [] now = _utc_now() for cd in topo: cid = cd.get("commit_id", "") if not cid: continue pids = [p for p in [cd.get("parent_commit_id") or "", cd.get("parent2_commit_id") or ""] if p] parent_gens = [inline_gen[p] if p in inline_gen else db_parent_gens[p] for p in pids if p in inline_gen or p in db_parent_gens] if not pids: gen = 0 # genuine root — no parents, generation 0 is correct elif len(parent_gens) == len(pids): gen = max(parent_gens) + 1 else: unresolved_pids = [ p for p in pids if p not in inline_gen and p not in db_parent_gens ] logger.warning( "[MWP1] unresolved parent generations — backfilling %d at " "_build_commit_graph_from_raw commit=%s unresolved_parents=%s", len(unresolved_pids), cid, unresolved_pids, ) max_parent_gen = await _resolve_generation_with_backfill( session, pids, inline_gen=inline_gen, db_gen=db_parent_gens, ) gen = max_parent_gen + 1 inline_gen[cid] = gen graph_rows.append({ "commit_id": cid, "parent_ids": pids, "generation": gen, "snapshot_id": cd.get("snapshot_id") or None, "created_at": now, }) if not graph_rows: return 0 _CHUNK = 2000 written = 0 for i in range(0, len(graph_rows), _CHUNK): chunk = graph_rows[i: i + _CHUNK] stmt = _pg_insert(MusehubCommitGraph).values(chunk) await session.execute(stmt.on_conflict_do_update( index_elements=["commit_id"], set_={ "generation": stmt.excluded.generation, "snapshot_id": stmt.excluded.snapshot_id, }, )) written += len(chunk) return written async def process_mpack_index_job(session: AsyncSession, job_id: str) -> JSONObject: """Populate MusehubMPackIndex byte-ranges for every object in a pushed mpack. Called by the background worker after every push. Reads the mpack from MinIO, computes byte offsets for each blob, then bulk-upserts MusehubMPackIndex rows with byte_offset and byte_length so the fetch path can serve blobs via byte-range GETs instead of downloading the full mpack. Returns a result dict with counts for logging. """ t0 = _time_module.monotonic() job_row = (await session.execute( select(MusehubBackgroundJob).where(MusehubBackgroundJob.job_id == job_id) )).scalar_one_or_none() if job_row is None: raise ValueError(f"mpack.index job not found: {job_id}") payload = job_row.payload or {} repo_id = job_row.repo_id mpack_key = str(payload.get("mpack_key", "")) if not mpack_key: raise ValueError(f"mpack.index job {job_id} missing mpack_key in payload") from musehub.storage.backends import get_backend backend = get_backend() wire_bytes = await backend.get_mpack(mpack_key) if not wire_bytes: raise ValueError(f"mpack {mpack_key[:20]} not found in storage") t_fetched = _time_module.monotonic() # Parse: handle both MUSE binary and legacy msgpack formats. if wire_bytes[:4] == b"MUSE": from muse.core.mpack import parse_wire_mpack as _parse_wire mpack: JSONObject = _parse_wire(wire_bytes) else: mpack = _msgpack.unpackb(wire_bytes, raw=False) raw_blobs: list[JSONObject] = mpack.get("blobs") or mpack.get("objects") or [] t_parsed = _time_module.monotonic() logger.info( "[mpack.index] job=%s fetch=%.3fs parse=%.3fs blobs=%d mpack=%d KB", job_id[:16], t_fetched - t0, t_parsed - t_fetched, len(raw_blobs), len(wire_bytes) // 1024, ) # Compute byte offsets for every object in the MUSE binary mpack. byte_offsets: dict[str, tuple[int, int]] = {} if wire_bytes[:4] == b"MUSE": all_oids = [obj.get("object_id", "") for obj in raw_blobs if obj.get("object_id")] byte_offsets = compute_object_byte_offsets(wire_bytes) # Upsert MusehubMPackIndex rows — one per blob with byte_offset/byte_length. now = _utc_now() _CHUNK = 2000 index_rows = [] for oid, (offset, length) in byte_offsets.items(): index_rows.append({ "entity_id": oid, "mpack_id": mpack_key, "entity_type": "object", "created_at": now, "byte_offset": offset, "byte_length": length, }) # Also index any blobs without byte-range data (fallback: full-mpack serve). indexed_oids = set(byte_offsets.keys()) for obj in raw_blobs: oid = obj.get("object_id", "") if oid and oid not in indexed_oids: index_rows.append({ "entity_id": oid, "mpack_id": mpack_key, "entity_type": "object", "created_at": now, }) written = 0 for i in range(0, len(index_rows), _CHUNK): chunk = index_rows[i: i + _CHUNK] stmt = _pg_insert(MusehubMPackIndex).values(chunk) # On conflict: update byte_offset/byte_length if we now have them. await session.execute( stmt.on_conflict_do_update( index_elements=["entity_id", "mpack_id"], set_={ "byte_offset": stmt.excluded.byte_offset, "byte_length": stmt.excluded.byte_length, }, where=MusehubMPackIndex.byte_offset.is_(None), ) ) written += len(chunk) # Upsert MusehubObject rows — storage_uri points to covering mpack, no per-object S3 PUT. # Must happen before object_refs because object_refs has a FK on musehub_objects. all_blob_oids = [obj.get("object_id", "") for obj in raw_blobs if obj.get("object_id")] if all_blob_oids: mpack_uri = f"mpack://{mpack_key}" obj_rows = [ {"object_id": oid, "storage_uri": mpack_uri, "path": "", "created_at": now} for oid in all_blob_oids ] for i in range(0, len(obj_rows), _CHUNK): chunk = obj_rows[i: i + _CHUNK] obj_stmt = _pg_insert(MusehubObject).values(chunk) await session.execute( obj_stmt.on_conflict_do_update( index_elements=["object_id"], set_={"storage_uri": obj_stmt.excluded.storage_uri}, where=MusehubObject.storage_uri.is_(None), ) ) # Write object_refs after musehub_objects so the FK constraint is satisfied. if all_blob_oids and repo_id: await _upsert_object_refs(session, repo_id, all_blob_oids) # Build commit graph rows from mpack commits. raw_commits: list[dict] = mpack.get("commits") or [] graph_written = await _build_commit_graph_from_raw(session, raw_commits) t_done = _time_module.monotonic() logger.info( "✅ [mpack.index] done job=%s index_rows=%d graph_rows=%d elapsed=%.3fs", job_id[:16], written, graph_written, t_done - t0, ) return { "mpack_index_written": written, "byte_ranges_computed": len(byte_offsets), "commit_graph_written": graph_written, "mpack_size_bytes": len(wire_bytes), "elapsed_ms": (t_done - t0) * 1000, }