musehub_wire_push.py
python
sha256:7d6dd8f4a89e2d1fef2d84f6e65feaff51385d382f466766b7f690a22ec18e32
fix: fall back to DB ancestry check when mpack-only fast-fo…
Sonnet 4.6
patch
134 days ago
| 1 | """Push path — wire_refs, wire_repair_*, wire_push_mpack_presign, wire_push_unpack_mpack.""" |
| 2 | |
| 3 | from typing import TypedDict |
| 4 | |
| 5 | import asyncio |
| 6 | import collections |
| 7 | import hashlib |
| 8 | import logging |
| 9 | import msgpack as _msgpack |
| 10 | import time as _time_module |
| 11 | from datetime import datetime, timezone |
| 12 | |
| 13 | from sqlalchemy import func, select, text as _sa_text |
| 14 | from sqlalchemy.dialects.postgresql import insert as _pg_insert |
| 15 | from sqlalchemy.ext.asyncio import AsyncSession |
| 16 | |
| 17 | from musehub.db.musehub_abuse_models import MusehubBlockedHash, MusehubDailyPushBytes, MusehubPushAnomaly |
| 18 | from musehub.db.musehub_jobs_models import MusehubBackgroundJob |
| 19 | from musehub.db.musehub_collaborator_models import MusehubCollaborator |
| 20 | from musehub.db.musehub_repo_models import ( |
| 21 | MusehubBranch, |
| 22 | MusehubCommit, |
| 23 | MusehubCommitGraph, |
| 24 | MusehubCommitRef, |
| 25 | MusehubObject, |
| 26 | MusehubObjectRef, |
| 27 | MusehubMPackIndex, |
| 28 | MusehubRepo, |
| 29 | MusehubSnapshot, |
| 30 | MusehubSnapshotRef, |
| 31 | ) |
| 32 | from musehub.models.wire import WireCommit, WireRefsResponse |
| 33 | from muse.core.types import blob_id, split_id |
| 34 | from muse.core.ids import hash_snapshot, long_id |
| 35 | from musehub.core.genesis import compute_branch_id |
| 36 | from musehub.types.json_types import IntDict, JSONObject, JSONValue, ReadOnlyJSONObject, StrDict |
| 37 | from musehub.config import settings |
| 38 | from musehub.storage import get_backend |
| 39 | from musehub.storage.backends import read_object_bytes |
| 40 | |
| 41 | from musehub.services.musehub_gc import invalidate_fetch_mpack_cache |
| 42 | from musehub.services.musehub_wire_shared import ( |
| 43 | MPackValidationError, |
| 44 | NonFastForwardError, |
| 45 | ObjectHashMismatch, |
| 46 | RepairResult, |
| 47 | _ChildMap, |
| 48 | _commit_identity_bytes, |
| 49 | _is_ancestor_db, |
| 50 | _is_fast_forward, |
| 51 | _reconstruct_manifest, |
| 52 | _reconstruct_manifest_validated, |
| 53 | _to_wire_commit, |
| 54 | _upsert_object_refs, |
| 55 | _utc_now, |
| 56 | _parse_iso, |
| 57 | _str_list, |
| 58 | _int_safe, |
| 59 | logger, |
| 60 | ) |
| 61 | |
| 62 | _ObjFetchMap = dict[str, tuple[str, str, str | None, bytes | None]] |
| 63 | _SnapshotManifest = dict[str, str] |
| 64 | _EMPTY_OID = "sha256:e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855" |
| 65 | _COMMIT_BATCH = 50 |
| 66 | |
| 67 | |
| 68 | async def wire_refs( |
| 69 | session: AsyncSession, |
| 70 | repo_id: str, |
| 71 | ) -> WireRefsResponse | None: |
| 72 | _t0 = _time_module.perf_counter() |
| 73 | def _ms() -> float: |
| 74 | return (_time_module.perf_counter() - _t0) * 1000 |
| 75 | |
| 76 | logger.info("[wire_refs] START repo_id=%s", repo_id) |
| 77 | |
| 78 | repo_row = await session.get(MusehubRepo, repo_id) |
| 79 | logger.info("[wire_refs] repo_row loaded found=%s t=%.1fms", repo_row is not None, _ms()) |
| 80 | if repo_row is None: |
| 81 | return None |
| 82 | |
| 83 | branch_rows = ( |
| 84 | await session.execute( |
| 85 | select(MusehubBranch).where(MusehubBranch.repo_id == repo_id) |
| 86 | ) |
| 87 | ).scalars().all() |
| 88 | logger.info("[wire_refs] branches loaded count=%d t=%.1fms", len(branch_rows), _ms()) |
| 89 | |
| 90 | branch_heads: StrDict = { |
| 91 | b.name: b.head_commit_id |
| 92 | for b in branch_rows |
| 93 | if b.head_commit_id |
| 94 | } |
| 95 | |
| 96 | domain_meta: JSONObject = ( |
| 97 | repo_row.domain_meta if isinstance(repo_row.domain_meta, dict) else {} |
| 98 | ) |
| 99 | domain = str(domain_meta.get("domain", "code")) |
| 100 | default_branch = getattr(repo_row, "default_branch", None) or "main" |
| 101 | |
| 102 | logger.info("[wire_refs] RETURN branches=%d TOTAL=%.1fms", len(branch_heads), _ms()) |
| 103 | return WireRefsResponse( |
| 104 | repo_id=repo_id, |
| 105 | domain=domain, |
| 106 | default_branch=default_branch, |
| 107 | branch_heads=branch_heads, |
| 108 | ) |
| 109 | |
| 110 | async def _enqueue_repair_prebuild(session: AsyncSession, repo_id: str) -> None: |
| 111 | """Enqueue a fetch.mpack.prebuild job after repair so the next clone is a fast HIT. |
| 112 | |
| 113 | Repair invalidates the full repo cache (repo-wide, D1) — the next clone would be a |
| 114 | synchronous MISS-build without this. Enqueueing a prebuild here closes the loop so |
| 115 | the worker rebuilds the mpack from corrected rows before the next client request. |
| 116 | |
| 117 | Covers all current branch tips (any tip may reach the repaired content). |
| 118 | Does NOT commit — the caller's existing session.commit() makes the row visible. |
| 119 | Idempotent: enqueue_job skips insertion when a pending OR running job |
| 120 | already exists — fetch.mpack.prebuild is self-coalescing (it re-reads |
| 121 | live branch tips at run time), so a job already in flight when a repair |
| 122 | lands will still pick up the repaired content. Without this, repeated |
| 123 | repair-commit calls arriving faster than one rebuild cycle (confirmed |
| 124 | live: 30-100+s per rebuild) flood the queue with redundant full rebuilds |
| 125 | and starve ordinary clone/fetch requests. |
| 126 | """ |
| 127 | from musehub.services.musehub_jobs import enqueue_job |
| 128 | tips = (await session.execute( |
| 129 | select(MusehubBranch.head_commit_id) |
| 130 | .where(MusehubBranch.repo_id == repo_id) |
| 131 | .where(MusehubBranch.head_commit_id.isnot(None)) |
| 132 | )).scalars().all() |
| 133 | if tips: |
| 134 | await enqueue_job( |
| 135 | session, repo_id, "fetch.mpack.prebuild", |
| 136 | {"tip_commit_ids": list(tips)}, |
| 137 | dedup_statuses=("pending", "running"), |
| 138 | ) |
| 139 | |
| 140 | |
| 141 | async def wire_repair_object( |
| 142 | session: AsyncSession, |
| 143 | repo_id: str, |
| 144 | object_id: str, |
| 145 | content: bytes, |
| 146 | caller_id: str | None, |
| 147 | ) -> RepairResult: |
| 148 | repo_row = await session.get(MusehubRepo, repo_id) |
| 149 | if repo_row is None: |
| 150 | raise ValueError("repo not found") |
| 151 | |
| 152 | if not caller_id: |
| 153 | raise PermissionError("repair rejected: unauthenticated") |
| 154 | if caller_id != repo_row.owner: |
| 155 | collab_row = (await session.execute( |
| 156 | select(MusehubCollaborator).where( |
| 157 | MusehubCollaborator.repo_id == repo_id, |
| 158 | MusehubCollaborator.identity_handle == caller_id, |
| 159 | MusehubCollaborator.accepted_at.isnot(None), |
| 160 | MusehubCollaborator.permission.in_(["write", "admin"]), |
| 161 | ) |
| 162 | )).scalar_one_or_none() |
| 163 | if collab_row is None: |
| 164 | raise PermissionError("repair rejected: not authorized") |
| 165 | |
| 166 | actual_id = blob_id(content) |
| 167 | if actual_id != object_id: |
| 168 | _, expected_hex = split_id(object_id) |
| 169 | _, actual = split_id(actual_id) |
| 170 | raise ObjectHashMismatch( |
| 171 | f"repair content hash mismatch for {object_id!r}: " |
| 172 | f"declared={expected_hex[:16]}… actual={actual[:16]}…" |
| 173 | ) |
| 174 | |
| 175 | backend = get_backend() |
| 176 | uri = await backend.put(object_id, content) |
| 177 | |
| 178 | await session.merge(MusehubObject( |
| 179 | object_id=object_id, path="", size_bytes=len(content), |
| 180 | storage_uri=uri, content_cache=None, |
| 181 | )) |
| 182 | await _upsert_object_refs(session, repo_id, [object_id]) |
| 183 | await invalidate_fetch_mpack_cache(session, repo_id) |
| 184 | await _enqueue_repair_prebuild(session, repo_id) |
| 185 | await session.commit() |
| 186 | |
| 187 | logger.info( |
| 188 | "✅ repair-object repo=%s object_id=%s size=%d", |
| 189 | repo_id, object_id, len(content), |
| 190 | ) |
| 191 | return {"repaired": True} |
| 192 | |
| 193 | async def wire_fetch_objects( |
| 194 | session: AsyncSession, |
| 195 | repo_id: str, |
| 196 | object_ids: list[str], |
| 197 | ) -> list[dict]: |
| 198 | result: list[dict] = [] |
| 199 | for oid in object_ids: |
| 200 | obj_row = await session.get(MusehubObject, oid) |
| 201 | if obj_row is None: |
| 202 | continue |
| 203 | content: bytes | None = await read_object_bytes(obj_row, session=session) |
| 204 | if content is None: |
| 205 | continue |
| 206 | result.append({"object_id": oid, "content": content}) |
| 207 | return result |
| 208 | |
| 209 | async def wire_repair_snapshot( |
| 210 | session: AsyncSession, |
| 211 | repo_id: str, |
| 212 | snapshot_id: str, |
| 213 | manifest: dict[str, str], |
| 214 | directories: list[str], |
| 215 | caller_id: str | None, |
| 216 | ) -> RepairResult: |
| 217 | from musehub.muse_cli.snapshot import compute_snapshot_id |
| 218 | |
| 219 | repo_row = await session.get(MusehubRepo, repo_id) |
| 220 | if repo_row is None: |
| 221 | raise ValueError("repo not found") |
| 222 | |
| 223 | if not caller_id: |
| 224 | raise PermissionError("repair rejected: unauthenticated") |
| 225 | if caller_id != repo_row.owner: |
| 226 | collab_row = (await session.execute( |
| 227 | select(MusehubCollaborator).where( |
| 228 | MusehubCollaborator.repo_id == repo_id, |
| 229 | MusehubCollaborator.identity_handle == caller_id, |
| 230 | MusehubCollaborator.accepted_at.isnot(None), |
| 231 | MusehubCollaborator.permission.in_(["write", "admin"]), |
| 232 | ) |
| 233 | )).scalar_one_or_none() |
| 234 | if collab_row is None: |
| 235 | raise PermissionError("repair rejected: not authorized") |
| 236 | |
| 237 | recomputed = compute_snapshot_id(manifest, directories or []) |
| 238 | if recomputed != snapshot_id: |
| 239 | raise ObjectHashMismatch( |
| 240 | f"repair snapshot hash mismatch for {snapshot_id!r}: " |
| 241 | f"declared={snapshot_id}… recomputed={recomputed}…" |
| 242 | ) |
| 243 | |
| 244 | import msgpack as _msgpack |
| 245 | blob = _msgpack.packb(manifest, use_bin_type=True) |
| 246 | dirs_sorted = sorted(directories) if directories else [] |
| 247 | await session.merge(MusehubSnapshot( |
| 248 | snapshot_id=snapshot_id, |
| 249 | directories=dirs_sorted, |
| 250 | manifest_blob=blob, |
| 251 | entry_count=len(manifest), |
| 252 | created_at=_utc_now(), |
| 253 | )) |
| 254 | await session.execute( |
| 255 | _pg_insert(MusehubSnapshotRef) |
| 256 | .values([{"repo_id": repo_id, "snapshot_id": snapshot_id, "created_at": _utc_now()}]) |
| 257 | .on_conflict_do_nothing(index_elements=["repo_id", "snapshot_id"]) |
| 258 | ) |
| 259 | await invalidate_fetch_mpack_cache(session, repo_id) |
| 260 | await _enqueue_repair_prebuild(session, repo_id) |
| 261 | await session.commit() |
| 262 | |
| 263 | logger.info( |
| 264 | "✅ repair-snapshot repo=%s snapshot_id=%s entries=%d", |
| 265 | repo_id, snapshot_id, len(manifest), |
| 266 | ) |
| 267 | return {"repaired": True} |
| 268 | |
| 269 | async def wire_repair_commit( |
| 270 | session: AsyncSession, |
| 271 | repo_id: str, |
| 272 | commit: ReadOnlyJSONObject, |
| 273 | caller_id: str | None, |
| 274 | ) -> RepairResult: |
| 275 | """Replace a stored commit's identity fields with verified-correct content. |
| 276 | |
| 277 | Commit-level analog of wire_repair_object / wire_repair_snapshot. Intended for |
| 278 | operators to repair commits whose stored row no longer reproduces its commit_id — |
| 279 | e.g. commits the rc10 object-store migration stamped with a signer_public_key |
| 280 | (and signature) without recomputing the id. Because signer_public_key is part of |
| 281 | the commit identity hash, the serve path then recomputes a different id, the |
| 282 | client's hash check fails on clone, the commit is dropped, and every descendant |
| 283 | fails "parent not in mpack", emptying the working tree. |
| 284 | |
| 285 | Request body (msgpack): |
| 286 | commit dict — a wire commit record (WireCommit shape). Its identity fields |
| 287 | (parent ids, snapshot_id, message, committed_at, author, |
| 288 | signer_public_key) must reproduce commit_id. |
| 289 | |
| 290 | committed_at is round-tripped through ``timestamp.isoformat()`` exactly as the |
| 291 | serve path reproduces it (via _to_wire_commit), so the identity we verify is the |
| 292 | identity that will be re-served after we store the parsed datetime. A commit whose |
| 293 | timestamp cannot survive that round-trip fails the hash check and is rejected |
| 294 | rather than repaired into a row that would still fail to clone. Only the repo |
| 295 | owner or a write/admin collaborator may call this. |
| 296 | """ |
| 297 | repo_row = await session.get(MusehubRepo, repo_id) |
| 298 | if repo_row is None: |
| 299 | raise ValueError("repo not found") |
| 300 | |
| 301 | if not caller_id: |
| 302 | raise PermissionError("repair rejected: unauthenticated") |
| 303 | if caller_id != repo_row.owner: |
| 304 | collab_row = (await session.execute( |
| 305 | select(MusehubCollaborator).where( |
| 306 | MusehubCollaborator.repo_id == repo_id, |
| 307 | MusehubCollaborator.identity_handle == caller_id, |
| 308 | MusehubCollaborator.accepted_at.isnot(None), |
| 309 | MusehubCollaborator.permission.in_(["write", "admin"]), |
| 310 | ) |
| 311 | )).scalar_one_or_none() |
| 312 | if collab_row is None: |
| 313 | raise PermissionError("repair rejected: not authorized") |
| 314 | |
| 315 | commit_id = str(commit.get("commit_id") or "") |
| 316 | if not commit_id: |
| 317 | raise ValueError("repair-commit requires commit.commit_id") |
| 318 | row = await session.get(MusehubCommit, commit_id) |
| 319 | if row is None: |
| 320 | raise ValueError(f"commit {commit_id} not found") |
| 321 | |
| 322 | # A malformed payload must surface as a 422 (ObjectHashMismatch -> 422 at the |
| 323 | # route), never a 500 — WireCommit's field validators raise on bad ids. |
| 324 | try: |
| 325 | wire = WireCommit(**commit) |
| 326 | except Exception as exc: |
| 327 | raise ObjectHashMismatch(f"repair commit invalid payload for {commit_id!r}: {exc}") |
| 328 | |
| 329 | # Pin committed_at to the SERVE representation so the identity we verify is exactly |
| 330 | # the one the serve path will reproduce from the stored datetime. |
| 331 | ts = _parse_iso(wire.committed_at) |
| 332 | wire.committed_at = ts.isoformat() |
| 333 | |
| 334 | recomputed = long_id(hashlib.sha256(_commit_identity_bytes(wire)).hexdigest()) |
| 335 | if recomputed != commit_id: |
| 336 | raise ObjectHashMismatch( |
| 337 | f"repair commit hash mismatch for {commit_id!r}: " |
| 338 | f"declared={commit_id} recomputed={recomputed}" |
| 339 | ) |
| 340 | |
| 341 | # Force-overwrite identity-bearing fields in place; preserve server-only |
| 342 | # bookkeeping (push-target branch, created_at, provenance) on the existing row. |
| 343 | parents = [p for p in (wire.parent_commit_id, wire.parent2_commit_id) if p] |
| 344 | row.parent_ids = parents |
| 345 | row.snapshot_id = wire.snapshot_id |
| 346 | row.message = wire.message |
| 347 | row.author = wire.author |
| 348 | row.timestamp = ts |
| 349 | row.signer_public_key = wire.signer_public_key |
| 350 | row.signature = wire.signature |
| 351 | row.signer_key_id = wire.signer_key_id |
| 352 | await invalidate_fetch_mpack_cache(session, repo_id) |
| 353 | await _enqueue_repair_prebuild(session, repo_id) |
| 354 | await session.commit() |
| 355 | |
| 356 | logger.info( |
| 357 | "✅ repair-commit repo=%s commit_id=%s signer=%r", |
| 358 | repo_id, commit_id, wire.signer_public_key, |
| 359 | ) |
| 360 | return {"repaired": True} |
| 361 | |
| 362 | async def _fetch_commit( |
| 363 | session: AsyncSession, |
| 364 | commit_id: str, |
| 365 | ) -> MusehubCommit | None: |
| 366 | return await session.get(MusehubCommit, commit_id) |
| 367 | |
| 368 | _ObjFetchMap = dict[str, tuple[str, str, str | None, bytes | None]] |
| 369 | |
| 370 | async def _fetch_object_meta( |
| 371 | session: AsyncSession, |
| 372 | repo_id: str, |
| 373 | object_ids: list[str], |
| 374 | ) -> _ObjFetchMap: |
| 375 | result: _ObjFetchMap = {} |
| 376 | if _EMPTY_OID in object_ids: |
| 377 | result[_EMPTY_OID] = (repo_id, "", None, None) |
| 378 | remaining = [oid for oid in object_ids if oid != _EMPTY_OID] |
| 379 | if remaining: |
| 380 | obj_rows_q = await session.execute( |
| 381 | select(MusehubObject).where(MusehubObject.object_id.in_(remaining)) |
| 382 | ) |
| 383 | result.update({ |
| 384 | r.object_id: (repo_id, r.path or "", r.storage_uri, r.content_cache) |
| 385 | for r in obj_rows_q.scalars().all() |
| 386 | }) |
| 387 | return result |
| 388 | |
| 389 | def compute_object_byte_offsets(wire_bytes: bytes) -> "dict[str, tuple[int, int]]": |
| 390 | """Return {object_id: (absolute_byte_offset, byte_length)} for every object |
| 391 | in the OBJECTS section of a wire mpack binary. |
| 392 | |
| 393 | Parses the section table to find the OBJECTS section, then walks the |
| 394 | ``_build_pack`` layout to compute each object's content start position |
| 395 | within *wire_bytes*. The returned offsets are absolute — callers can issue |
| 396 | ``Range: bytes=offset-offset+length-1`` S3 GETs directly against the mpack. |
| 397 | |
| 398 | Returns an empty dict for wire mpacks with no OBJECTS section. |
| 399 | """ |
| 400 | import struct as _struct |
| 401 | |
| 402 | _WIRE_SEC_BLOBS = 1 |
| 403 | _OID_BYTES = 71 # len("sha256:") + 64 hex chars |
| 404 | _PACK_HEADER = 13 # magic(4) + version(1) + count(8) |
| 405 | |
| 406 | if len(wire_bytes) < 6 or wire_bytes[:4] != b"MUSE": |
| 407 | return {} |
| 408 | |
| 409 | section_count = wire_bytes[5] |
| 410 | cursor = 6 |
| 411 | objects_section_offset: int | None = None |
| 412 | for _ in range(section_count): |
| 413 | sec_type = wire_bytes[cursor] |
| 414 | (sec_offset,) = _struct.unpack_from("<Q", wire_bytes, cursor + 1) |
| 415 | if sec_type == _WIRE_SEC_BLOBS: |
| 416 | objects_section_offset = sec_offset |
| 417 | break |
| 418 | cursor += 17 |
| 419 | |
| 420 | if objects_section_offset is None: |
| 421 | return {} |
| 422 | |
| 423 | # Walk the _build_pack layout within the OBJECTS section |
| 424 | pack_start = objects_section_offset |
| 425 | if wire_bytes[pack_start : pack_start + 4] != b"MUSE": |
| 426 | return {} # not a valid pack section |
| 427 | |
| 428 | (object_count,) = _struct.unpack_from("<Q", wire_bytes, pack_start + 5) |
| 429 | pos = pack_start + _PACK_HEADER |
| 430 | result: dict[str, tuple[int, int]] = {} |
| 431 | |
| 432 | for _ in range(object_count): |
| 433 | oid = wire_bytes[pos : pos + _OID_BYTES].decode("ascii", errors="replace") |
| 434 | pos += _OID_BYTES |
| 435 | (length,) = _struct.unpack_from("<Q", wire_bytes, pos) |
| 436 | pos += 8 # skip length field |
| 437 | result[oid] = (pos, length) # pos is now at content start |
| 438 | pos += length |
| 439 | |
| 440 | return result |
| 441 | |
| 442 | |
| 443 | def _topological_sort(commits: list[WireCommit]) -> list[WireCommit]: |
| 444 | if not commits: |
| 445 | return [] |
| 446 | by_id = {c.commit_id: c for c in commits} |
| 447 | in_degree: IntDict = {c.commit_id: 0 for c in commits} |
| 448 | children: _ChildMap = {c.commit_id: [] for c in commits} |
| 449 | |
| 450 | for c in commits: |
| 451 | for pid in filter(None, [c.parent_commit_id, c.parent2_commit_id]): |
| 452 | if pid in by_id: |
| 453 | in_degree[c.commit_id] += 1 |
| 454 | children[pid].append(c.commit_id) |
| 455 | |
| 456 | queue: collections.deque[str] = collections.deque( |
| 457 | cid for cid, deg in in_degree.items() if deg == 0 |
| 458 | ) |
| 459 | result: list[WireCommit] = [] |
| 460 | while queue: |
| 461 | cid = queue.popleft() |
| 462 | result.append(by_id[cid]) |
| 463 | for child_id in children.get(cid, []): |
| 464 | in_degree[child_id] -= 1 |
| 465 | if in_degree[child_id] == 0: |
| 466 | queue.append(child_id) |
| 467 | |
| 468 | sorted_ids = {c.commit_id for c in result} |
| 469 | for c in commits: |
| 470 | if c.commit_id not in sorted_ids: |
| 471 | result.append(c) |
| 472 | |
| 473 | return result |
| 474 | |
| 475 | async def record_mpack_bytes_uploaded( |
| 476 | session: AsyncSession, |
| 477 | identity_id: str, |
| 478 | size_bytes: int, |
| 479 | ) -> None: |
| 480 | import datetime as _dt |
| 481 | today = _dt.date.today() |
| 482 | await session.execute( |
| 483 | _pg_insert(MusehubDailyPushBytes) |
| 484 | .values(identity_id=identity_id, date=today, bytes_uploaded=size_bytes) |
| 485 | .on_conflict_do_update( |
| 486 | index_elements=["identity_id", "date"], |
| 487 | set_={"bytes_uploaded": MusehubDailyPushBytes.bytes_uploaded + size_bytes}, |
| 488 | ) |
| 489 | ) |
| 490 | |
| 491 | |
| 492 | async def check_push_anomaly( |
| 493 | session: AsyncSession, |
| 494 | identity_id: str, |
| 495 | bytes_today: int, |
| 496 | ) -> bool: |
| 497 | import datetime as _dt |
| 498 | import uuid as _uuid |
| 499 | |
| 500 | today = _dt.date.today() |
| 501 | cutoff = today - _dt.timedelta(days=30) |
| 502 | |
| 503 | result = await session.execute( |
| 504 | select(func.avg(MusehubDailyPushBytes.bytes_uploaded)).where( |
| 505 | MusehubDailyPushBytes.identity_id == identity_id, |
| 506 | MusehubDailyPushBytes.date >= cutoff, |
| 507 | MusehubDailyPushBytes.date < today, |
| 508 | ) |
| 509 | ) |
| 510 | avg_bytes = result.scalar() |
| 511 | if avg_bytes is None or avg_bytes == 0: |
| 512 | return False |
| 513 | |
| 514 | ratio = bytes_today / float(avg_bytes) |
| 515 | if ratio <= 10.0: |
| 516 | return False |
| 517 | |
| 518 | anomaly_id = "sha256:" + hashlib.sha256( |
| 519 | f"{identity_id}:{today.isoformat()}:{bytes_today}".encode() |
| 520 | ).hexdigest() |
| 521 | logger.warning( |
| 522 | "[anomaly] push spike detected identity_id=%s bytes_today=%d avg_bytes=%.0f ratio=%.1fx", |
| 523 | identity_id[:20], bytes_today, avg_bytes, ratio, |
| 524 | ) |
| 525 | session.add(MusehubPushAnomaly( |
| 526 | anomaly_id=anomaly_id, |
| 527 | identity_id=identity_id, |
| 528 | bytes_today=bytes_today, |
| 529 | rolling_avg_bytes=float(avg_bytes), |
| 530 | ratio=ratio, |
| 531 | )) |
| 532 | return True |
| 533 | |
| 534 | |
| 535 | async def wire_push_mpack_presign( |
| 536 | mpack_key: str, |
| 537 | size_bytes: int, |
| 538 | ttl_seconds: int = 3600, |
| 539 | ) -> JSONObject: |
| 540 | logger.warning("[mpack-presign] START mpack_key=%s size_bytes=%d ttl=%ds", mpack_key, size_bytes, ttl_seconds) |
| 541 | backend = get_backend() |
| 542 | upload_url = await backend.presign_mpack_put(mpack_key, ttl_seconds) |
| 543 | logger.warning("[mpack-presign] upload_url=%s", upload_url) |
| 544 | return {"upload_url": upload_url, "mpack_key": mpack_key} |
| 545 | |
| 546 | |
| 547 | async def wire_push_unpack_mpack( |
| 548 | session: AsyncSession, |
| 549 | repo_id: str, |
| 550 | mpack_key: str, |
| 551 | pusher_id: str | None, |
| 552 | branch: str = "main", |
| 553 | head_commit_id: str = "", |
| 554 | commits_count: int = 0, |
| 555 | blobs_count: int = 0, |
| 556 | force: bool = False, |
| 557 | ) -> JSONObject: |
| 558 | t0 = _time_module.monotonic() |
| 559 | def _ms(ref: float = 0.0) -> float: |
| 560 | return (_time_module.monotonic() - (t0 if ref == 0.0 else ref)) * 1000 |
| 561 | |
| 562 | from musehub.storage.backends import get_backend |
| 563 | backend = get_backend() |
| 564 | |
| 565 | logger.warning( |
| 566 | "[SERVER step 0] receive repo=%s branch=%s head=%s", |
| 567 | repo_id[:20], branch, head_commit_id[:20] if head_commit_id else "none", |
| 568 | ) |
| 569 | |
| 570 | # step 1: fetch mpack bytes from MinIO by mpack_key |
| 571 | wire_bytes = await backend.get_mpack(mpack_key) |
| 572 | t_got = _time_module.monotonic() |
| 573 | if not wire_bytes: |
| 574 | raise ValueError(f"mpack {mpack_key[:20]} not found in storage") |
| 575 | mpack_bytes_len = len(wire_bytes) |
| 576 | logger.warning( |
| 577 | "[SERVER step 1] fetched mpack from object store %.1fms size=%d bytes (%.4f MB) key=%s", |
| 578 | _ms(), mpack_bytes_len, mpack_bytes_len / 1_048_576, mpack_key, |
| 579 | ) |
| 580 | |
| 581 | from musehub.config import get_settings as _get_settings |
| 582 | _max_bytes = _get_settings().mpack_max_bytes |
| 583 | if mpack_bytes_len > _max_bytes: |
| 584 | raise ValueError( |
| 585 | f"mpack {mpack_key[:20]} size {mpack_bytes_len:,} bytes " |
| 586 | f"exceeds limit {_max_bytes:,} bytes" |
| 587 | ) |
| 588 | |
| 589 | # step 2: verify blob_id(bytes) == mpack_key |
| 590 | actual_id = blob_id(wire_bytes) |
| 591 | t_verify = _time_module.monotonic() |
| 592 | _, expected_hex = split_id(mpack_key) |
| 593 | _, got_hex = split_id(actual_id) |
| 594 | match = actual_id == mpack_key |
| 595 | logger.warning( |
| 596 | "[SERVER step 2] integrity check %.1fms expected=%s got=%s match=%s", |
| 597 | (t_verify - t_got) * 1000, expected_hex[:20], got_hex[:20], match, |
| 598 | ) |
| 599 | if not match: |
| 600 | raise ValueError( |
| 601 | f"mpack integrity failure: sha256={got_hex[:20]}… ≠ key={expected_hex[:20]}…" |
| 602 | ) |
| 603 | |
| 604 | # ── 2. Inline index ─────────────────────────────────────────────────────── |
| 605 | import zstandard as _zstd_sync |
| 606 | |
| 607 | if wire_bytes[:4] != b"MUSE": |
| 608 | raise ValueError(f"mpack is not MUSE binary format (got {wire_bytes[:4]!r})") |
| 609 | from muse.core.mpack import parse_wire_mpack as _parse_wire_mpack |
| 610 | _mpack_sync: JSONObject = _parse_wire_mpack(wire_bytes) |
| 611 | t_parse = _time_module.monotonic() |
| 612 | _raw_blobs: list[JSONValue] = _mpack_sync.get("blobs") or [] |
| 613 | _raw_snaps: list[JSONValue] = _mpack_sync.get("snapshots") or [] |
| 614 | _raw_commits_inline: list[JSONValue] = _mpack_sync.get("commits") or [] |
| 615 | logger.warning("[SERVER step 3+4] verify MUSE magic + parse %.1fms commits=%d snapshots=%d blobs=%d", |
| 616 | (t_parse - t_verify) * 1000, len(_raw_commits_inline), len(_raw_snaps), len(_raw_blobs)) |
| 617 | |
| 618 | # 2b. zstd decompress all blobs — zip bomb guard runs inline |
| 619 | _dctx_sync = _zstd_sync.ZstdDecompressor() |
| 620 | _decompressed_sync: list[tuple[str, bytes]] = [] |
| 621 | _total_blob_bytes = 0 |
| 622 | _max_decompressed = settings.mpack_max_decompressed_bytes |
| 623 | for _obj in _raw_blobs: |
| 624 | _oid = _obj["object_id"] |
| 625 | _raw = _obj.get("content") or b"" |
| 626 | if _obj.get("encoding") == "zstd" and _raw: |
| 627 | _raw = _dctx_sync.decompress(_raw) |
| 628 | _total_blob_bytes += len(_raw) |
| 629 | if _total_blob_bytes > _max_decompressed: |
| 630 | _zip_err = MPackValidationError( |
| 631 | f"decompressed size {_total_blob_bytes:,} exceeds limit " |
| 632 | f"{_max_decompressed:,} — possible zip bomb" |
| 633 | ) |
| 634 | await backend.quarantine_mpack(mpack_key) |
| 635 | raise _zip_err |
| 636 | _decompressed_sync.append((_oid, _raw)) |
| 637 | t_decomp = _time_module.monotonic() |
| 638 | logger.warning("[SERVER step 5] decompress + zip-bomb check %.1fms blobs=%d uncompressed=%.2f MB", |
| 639 | (t_decomp - t_parse) * 1000, len(_decompressed_sync), _total_blob_bytes / 1_048_576) |
| 640 | |
| 641 | # step 6: check blocked hashes |
| 642 | if _decompressed_sync: |
| 643 | _all_check_oids = [_oid for _oid, _ in _decompressed_sync] |
| 644 | _blocked_oids = (await session.execute( |
| 645 | select(MusehubBlockedHash.object_id) |
| 646 | .where(MusehubBlockedHash.object_id.in_(_all_check_oids)) |
| 647 | )).scalars().all() |
| 648 | if _blocked_oids: |
| 649 | _blocked_err = MPackValidationError( |
| 650 | f"mpack contains {len(_blocked_oids)} blocked object(s): " |
| 651 | + ", ".join(oid[:20] for oid in _blocked_oids[:3]) |
| 652 | + ("..." if len(_blocked_oids) > 3 else "") |
| 653 | ) |
| 654 | await backend.quarantine_mpack(mpack_key) |
| 655 | raise _blocked_err |
| 656 | logger.warning("[SERVER step 6] check blocked hashes objects=%d blocked=0", len(_decompressed_sync)) |
| 657 | |
| 658 | # step 7: write objects |
| 659 | _new_oids: list[str] = [] |
| 660 | if _decompressed_sync: |
| 661 | _cc_oids = [_oid for _oid, _ in _decompressed_sync] |
| 662 | |
| 663 | t_obj_sel0 = _time_module.monotonic() |
| 664 | _existing_oids = set((await session.execute( |
| 665 | _sa_text("SELECT object_id FROM musehub_objects WHERE object_id = ANY(:ids)"), |
| 666 | {"ids": _cc_oids}, |
| 667 | )).scalars()) |
| 668 | t_obj_sel1 = _time_module.monotonic() |
| 669 | logger.warning("[SERVER step 7a] objects: dedup SELECT %.1fms total=%d existing=%d new=%d", |
| 670 | (t_obj_sel1 - t_obj_sel0) * 1000, len(_cc_oids), len(_existing_oids), |
| 671 | len(_cc_oids) - len(_existing_oids)) |
| 672 | |
| 673 | session.add_all([ |
| 674 | MusehubObject(object_id=_oid, path="", size_bytes=len(_data), |
| 675 | storage_uri=f"mpack://{mpack_key}", content_cache=None) |
| 676 | for _oid, _data in _decompressed_sync |
| 677 | if _oid not in _existing_oids |
| 678 | ]) |
| 679 | t_obj_add = _time_module.monotonic() |
| 680 | logger.warning("[SERVER step 7b] objects: INSERT new %.1fms", (t_obj_add - t_obj_sel1) * 1000) |
| 681 | |
| 682 | _new_oids = [_oid for _oid in _cc_oids if _oid not in _existing_oids] |
| 683 | t_ref0 = _time_module.monotonic() |
| 684 | await _upsert_object_refs(session, repo_id, _cc_oids) |
| 685 | t_ref1 = _time_module.monotonic() |
| 686 | logger.warning("[SERVER step 7c] objects: UPSERT object_refs %.1fms oids=%d", |
| 687 | (t_ref1 - t_ref0) * 1000, len(_cc_oids)) |
| 688 | |
| 689 | if _cc_oids: |
| 690 | _byte_offsets = compute_object_byte_offsets(wire_bytes) |
| 691 | _mpack_idx_rows = [ |
| 692 | { |
| 693 | "entity_id": _oid, "mpack_id": mpack_key, |
| 694 | "entity_type": "object", "created_at": _utc_now(), |
| 695 | "byte_offset": _byte_offsets.get(_oid, (None, None))[0], |
| 696 | "byte_length": _byte_offsets.get(_oid, (None, None))[1], |
| 697 | } |
| 698 | for _oid in _cc_oids |
| 699 | ] |
| 700 | # asyncpg caps query parameters at 32767; 6 columns per row → 5461 rows/batch |
| 701 | _MPACK_IDX_BATCH = 5461 |
| 702 | for _batch_start in range(0, len(_mpack_idx_rows), _MPACK_IDX_BATCH): |
| 703 | _batch = _mpack_idx_rows[_batch_start:_batch_start + _MPACK_IDX_BATCH] |
| 704 | _idx_stmt = _pg_insert(MusehubMPackIndex).values(_batch) |
| 705 | await session.execute( |
| 706 | _idx_stmt.on_conflict_do_update( |
| 707 | index_elements=["entity_id", "mpack_id"], |
| 708 | set_={ |
| 709 | "byte_offset": _idx_stmt.excluded.byte_offset, |
| 710 | "byte_length": _idx_stmt.excluded.byte_length, |
| 711 | }, |
| 712 | where=MusehubMPackIndex.byte_offset.is_(None), |
| 713 | ) |
| 714 | ) |
| 715 | t_idx_add = _time_module.monotonic() |
| 716 | logger.warning("[SERVER step 7d] objects: UPSERT mpack_index %.1fms total=%d", |
| 717 | (t_idx_add - t_ref1) * 1000, len(_cc_oids)) |
| 718 | else: |
| 719 | t_idx_add = t_decomp |
| 720 | |
| 721 | # step 7: write snapshots |
| 722 | _snap_ids_in_mpack = {_sd.get("snapshot_id") for _sd in _raw_snaps if _sd.get("snapshot_id")} |
| 723 | _external_parent_sids = { |
| 724 | _sd.get("parent_snapshot_id") |
| 725 | for _sd in _raw_snaps |
| 726 | if _sd.get("parent_snapshot_id") and _sd.get("parent_snapshot_id") not in _snap_ids_in_mpack |
| 727 | } |
| 728 | _parent_snap_manifests: dict[str, dict] = {} |
| 729 | t_snap_ext0 = _time_module.monotonic() |
| 730 | if _external_parent_sids: |
| 731 | _psnap_rows = (await session.execute( |
| 732 | select(MusehubSnapshot.snapshot_id, MusehubSnapshot.manifest_blob) |
| 733 | .where(MusehubSnapshot.snapshot_id.in_(_external_parent_sids)) |
| 734 | )).all() |
| 735 | _psid_direct = set() |
| 736 | for _psid, _pblob in _psnap_rows: |
| 737 | if _pblob: |
| 738 | _parent_snap_manifests[_psid] = dict(_msgpack.unpackb(_pblob, raw=False)) |
| 739 | _psid_direct.add(_psid) |
| 740 | # An external parent with no cached manifest_blob is delta-only (or, in the |
| 741 | # worst case, missing a row entirely) — reconstruct it by walking its delta |
| 742 | # chain instead of silently treating it as an empty manifest. A silent {} |
| 743 | # here permanently corrupts every descendant snapshot (musehub#134). |
| 744 | _psid_needs_reconstruct = _external_parent_sids - _psid_direct |
| 745 | for _psid in _psid_needs_reconstruct: |
| 746 | _reconstructed = await _reconstruct_manifest_validated(session, _psid) |
| 747 | if _reconstructed: |
| 748 | _parent_snap_manifests[_psid] = _reconstructed |
| 749 | t_snap_ext1 = _time_module.monotonic() |
| 750 | logger.warning("[SERVER step 8a] snapshots: fetch external parents %.1fms external=%d loaded=%d", |
| 751 | (t_snap_ext1 - t_snap_ext0) * 1000, len(_external_parent_sids), len(_parent_snap_manifests)) |
| 752 | |
| 753 | _head_snap_id: str | None = None |
| 754 | for _cd in _raw_commits_inline: |
| 755 | if _cd.get("commit_id") == head_commit_id: |
| 756 | _head_snap_id = _cd.get("snapshot_id") or None |
| 757 | break |
| 758 | |
| 759 | _snap_resolved: dict[str, _SnapshotManifest] = {} |
| 760 | _snap_rows_inline = [] |
| 761 | _n_snap_root = 0 |
| 762 | _n_snap_head = 0 |
| 763 | _n_snap_delta_only = 0 |
| 764 | _snap_manifest_bytes_total = 0 |
| 765 | _snap_delta_bytes_total = 0 |
| 766 | |
| 767 | t_snap_loop0 = _time_module.monotonic() |
| 768 | for _sd in _raw_snaps: |
| 769 | _sid = _sd.get("snapshot_id", "") |
| 770 | if not _sid: |
| 771 | continue |
| 772 | _parent_sid = _sd.get("parent_snapshot_id") |
| 773 | _delta_upsert: dict[str, str] = _sd.get("delta_upsert") or {} |
| 774 | _delta_remove: list[str] = _sd.get("delta_remove") or [] |
| 775 | if _parent_sid and _parent_sid in _snap_resolved: |
| 776 | _parent_base = _snap_resolved[_parent_sid] |
| 777 | _null_inherited = [k for k, v in _parent_base.items() if not v] |
| 778 | if _null_inherited: |
| 779 | logger.warning("[PUSH-NULL-FILTER] snap=%s filtered %d null OIDs from in-flight parent: %s", |
| 780 | _sid[:20], len(_null_inherited), _null_inherited) |
| 781 | _base = {k: v for k, v in _parent_base.items() if v} |
| 782 | elif _parent_sid: |
| 783 | _parent_base = _parent_snap_manifests.get(_parent_sid) or {} |
| 784 | _null_inherited = [k for k, v in _parent_base.items() if not v] |
| 785 | if _null_inherited: |
| 786 | logger.warning("[PUSH-NULL-FILTER] snap=%s filtered %d null OIDs from DB parent: %s", |
| 787 | _sid[:20], len(_null_inherited), _null_inherited) |
| 788 | _base = {k: v for k, v in _parent_base.items() if v} |
| 789 | else: |
| 790 | _base = {} |
| 791 | _null_delta = [k for k, v in _delta_upsert.items() if not v] |
| 792 | if _null_delta: |
| 793 | logger.warning("[PUSH-NULL-FILTER] snap=%s filtered %d null OIDs from delta_upsert: %s", |
| 794 | _sid[:20], len(_null_delta), _null_delta) |
| 795 | _base.update({k: v for k, v in _delta_upsert.items() if v}) |
| 796 | for _path in _delta_remove: |
| 797 | _base.pop(_path, None) |
| 798 | _snap_resolved[_sid] = _base |
| 799 | |
| 800 | _delta_encoded: JSONObject = {"add": _delta_upsert} |
| 801 | if _delta_remove: |
| 802 | _delta_encoded["rm"] = _delta_remove |
| 803 | _delta_blob = _msgpack.packb(_delta_encoded, use_bin_type=True) if (_delta_upsert or _delta_remove) else None |
| 804 | |
| 805 | _is_root = not _parent_sid |
| 806 | _is_head = _sid == _head_snap_id |
| 807 | # Only store full manifest for root and head — delta-only snapshots |
| 808 | # get manifest_blob=None here. They are NOT backfilled by any job; the fetch |
| 809 | # path reconstructs their manifest on-demand by walking the delta chain |
| 810 | # (_reconstruct_manifest in musehub_wire_shared.py). The only path that ever |
| 811 | # persists a manifest_blob for a delta-only row is wire_repair_snapshot, and |
| 812 | # only when a client explicitly repairs that snapshot (it validates the |
| 813 | # recomputed snapshot_id before writing). The mpack.index job does NOT touch |
| 814 | # manifest_blob — it only indexes blob byte-offsets. |
| 815 | # Storing all 1000 manifests inline would push hundreds of MB into a single |
| 816 | # transaction, blocking snapshot_refs inserts (regression at scale). |
| 817 | _manifest_blob = _msgpack.packb(_base, use_bin_type=True) if (_is_root or _is_head) else None |
| 818 | |
| 819 | if _is_root: _n_snap_root += 1 |
| 820 | if _is_head: _n_snap_head += 1 |
| 821 | if not _is_root and not _is_head: _n_snap_delta_only += 1 |
| 822 | if _manifest_blob: _snap_manifest_bytes_total += len(_manifest_blob) |
| 823 | if _delta_blob: _snap_delta_bytes_total += len(_delta_blob) |
| 824 | |
| 825 | _snap_dirs = [str(d) for d in (_sd.get("directories") or []) if d] |
| 826 | # The hash IS the proof: a base resolved from a missing/unreconstructable |
| 827 | # external parent (see _psid_needs_reconstruct above) will never reproduce |
| 828 | # _sid — and, just as importantly, this ALSO catches wire/transport bugs |
| 829 | # that lose or mangle any part of the reconstructed (manifest, directories) |
| 830 | # pair even when the parent chain resolved fine (musehub#93: root-of-push |
| 831 | # entries were never checked at all here, since this used to run only |
| 832 | # `if _parent_sid`, silently accepting corrupt directories for exactly |
| 833 | # the entries — parent_snapshot_id=None — where nothing else in this |
| 834 | # function re-derives or re-checks them later). Run unconditionally: |
| 835 | # every snapshot entering the store must reproduce its own declared ID |
| 836 | # from what was actually reconstructed, root or not. Reject the push |
| 837 | # loudly rather than persist a snapshot that can never be verified again |
| 838 | # (musehub#134, musehub#93). |
| 839 | if hash_snapshot(_base, _snap_dirs or None) != _sid: |
| 840 | if _parent_sid: |
| 841 | raise ValueError( |
| 842 | f"push rejected: snapshot {_sid[:20]}… cannot be reconstructed — " |
| 843 | f"its parent {_parent_sid[:20]}… is missing or unrecoverable on the " |
| 844 | f"server. Run 'muse verify' to audit store integrity before retrying." |
| 845 | ) |
| 846 | raise ValueError( |
| 847 | f"push rejected: root snapshot {_sid[:20]}… does not reproduce its own " |
| 848 | f"declared ID from the manifest and directories received on the wire " |
| 849 | f"(hash_snapshot mismatch). This indicates data loss or corruption in " |
| 850 | f"transit, not a missing-parent issue. Run 'muse verify' to audit local " |
| 851 | f"store integrity before retrying." |
| 852 | ) |
| 853 | _snap_rows_inline.append({ |
| 854 | "snapshot_id": _sid, |
| 855 | "directories": _snap_dirs, |
| 856 | "manifest_blob": _manifest_blob, |
| 857 | "entry_count": len(_base), |
| 858 | "created_at": _utc_now(), |
| 859 | "parent_snapshot_id": _parent_sid or None, |
| 860 | "delta_blob": _delta_blob, |
| 861 | }) |
| 862 | t_snap_loop1 = _time_module.monotonic() |
| 863 | logger.warning( |
| 864 | "[SERVER step 8b] snapshots: replay delta chain %.1fms " |
| 865 | "total=%d root=%d head=%d delta_only=%d", |
| 866 | (t_snap_loop1 - t_snap_loop0) * 1000, len(_snap_rows_inline), |
| 867 | _n_snap_root, _n_snap_head, _n_snap_delta_only, |
| 868 | ) |
| 869 | |
| 870 | # ── identity-domain invariant enforcement (musehub#230) ───────────────── |
| 871 | # push_validator.py's IdentityPushValidator previously had zero production |
| 872 | # call sites -- I1 (acyclicity), I2 (root distance), and I3 (authorization) |
| 873 | # were never enforced on any real push, contrary to the module's own |
| 874 | # docstring. Run it here, before any object/snapshot/commit row is |
| 875 | # committed, for identity-domain repos only. |
| 876 | if _head_snap_id is not None: |
| 877 | _repo_row_for_domain = await session.get(MusehubRepo, repo_id) |
| 878 | if _repo_row_for_domain is not None and _repo_row_for_domain.domain_id == "identity": |
| 879 | from muse.plugins.identity.records import ( |
| 880 | parse_identity_path, |
| 881 | parse_relationship_path, |
| 882 | record_from_bytes, |
| 883 | ) |
| 884 | from musehub.graph.push_validator import IdentityPushValidator |
| 885 | |
| 886 | _head_manifest = _snap_resolved.get(_head_snap_id, {}) |
| 887 | _blob_bytes_by_oid: dict[str, bytes] = dict(_decompressed_sync) |
| 888 | _identity_paths = [p for p in _head_manifest if parse_identity_path(p) is not None] |
| 889 | _relationship_paths = [p for p in _head_manifest if parse_relationship_path(p) is not None] |
| 890 | _record_oids = [_head_manifest[p] for p in (_identity_paths + _relationship_paths)] |
| 891 | |
| 892 | _missing_oids = [oid for oid in _record_oids if oid not in _blob_bytes_by_oid] |
| 893 | if _missing_oids: |
| 894 | _missing_obj_rows = (await session.execute( |
| 895 | select(MusehubObject).where(MusehubObject.object_id.in_(_missing_oids)) |
| 896 | )).scalars().all() |
| 897 | for _obj_row in _missing_obj_rows: |
| 898 | _obj_bytes = await read_object_bytes(_obj_row, session) |
| 899 | if _obj_bytes is not None: |
| 900 | _blob_bytes_by_oid[_obj_row.object_id] = _obj_bytes |
| 901 | |
| 902 | _identities = [ |
| 903 | record_from_bytes(_blob_bytes_by_oid[_head_manifest[p]]) |
| 904 | for p in _identity_paths |
| 905 | if _head_manifest[p] in _blob_bytes_by_oid |
| 906 | ] |
| 907 | _relationships = [ |
| 908 | record_from_bytes(_blob_bytes_by_oid[_head_manifest[p]]) |
| 909 | for p in _relationship_paths |
| 910 | if _head_manifest[p] in _blob_bytes_by_oid |
| 911 | ] |
| 912 | |
| 913 | _push_validation = IdentityPushValidator().validate(_identities, _relationships) |
| 914 | if not _push_validation.valid: |
| 915 | raise ValueError( |
| 916 | "push rejected: identity domain invariant violation(s): " |
| 917 | + "; ".join(_push_validation.errors) |
| 918 | ) |
| 919 | |
| 920 | if _snap_rows_inline: |
| 921 | _snap_ids_to_check = [r["snapshot_id"] for r in _snap_rows_inline] |
| 922 | t_snap_sel0 = _time_module.monotonic() |
| 923 | _existing_snap_rows = (await session.execute( |
| 924 | _sa_text( |
| 925 | "SELECT snapshot_id, (delta_blob IS NOT NULL) AS has_delta " |
| 926 | "FROM musehub_snapshots WHERE snapshot_id = ANY(:ids)" |
| 927 | ), |
| 928 | {"ids": _snap_ids_to_check}, |
| 929 | )).all() |
| 930 | _existing_sids: set[str] = {row[0] for row in _existing_snap_rows} |
| 931 | _sids_already_healed: set[str] = {row[0] for row in _existing_snap_rows if row[1]} |
| 932 | t_snap_sel1 = _time_module.monotonic() |
| 933 | logger.warning("[SERVER step 8c] snapshots: dedup SELECT %.1fms existing=%d new=%d", |
| 934 | (t_snap_sel1 - t_snap_sel0) * 1000, len(_existing_sids), |
| 935 | len(_snap_rows_inline) - len(_existing_sids)) |
| 936 | |
| 937 | _new_snap_dicts = [r for r in _snap_rows_inline if r["snapshot_id"] not in _existing_sids] |
| 938 | _heal_snap_dicts = [ |
| 939 | r for r in _snap_rows_inline |
| 940 | if r["snapshot_id"] in _existing_sids |
| 941 | and r["snapshot_id"] not in _sids_already_healed |
| 942 | and r.get("delta_blob") |
| 943 | ] |
| 944 | if _new_snap_dicts: |
| 945 | session.add_all([MusehubSnapshot(**_r) for _r in _new_snap_dicts]) |
| 946 | if _heal_snap_dicts: |
| 947 | _HEAL_CHUNK = 2000 |
| 948 | for _hi in range(0, len(_heal_snap_dicts), _HEAL_CHUNK): |
| 949 | _chunk = _heal_snap_dicts[_hi : _hi + _HEAL_CHUNK] |
| 950 | _heal_stmt = _pg_insert(MusehubSnapshot).values(_chunk) |
| 951 | await session.execute( |
| 952 | _heal_stmt.on_conflict_do_update( |
| 953 | index_elements=["snapshot_id"], |
| 954 | set_={ |
| 955 | "delta_blob": _heal_stmt.excluded.delta_blob, |
| 956 | "parent_snapshot_id": _heal_stmt.excluded.parent_snapshot_id, |
| 957 | }, |
| 958 | where=MusehubSnapshot.delta_blob.is_(None), |
| 959 | ) |
| 960 | ) |
| 961 | t_snap_add = _time_module.monotonic() |
| 962 | logger.warning("[SERVER step 8d] snapshots: INSERT new %.1fms new=%d healed=%d", |
| 963 | (t_snap_add - t_snap_sel1) * 1000, len(_new_snap_dicts), len(_heal_snap_dicts)) |
| 964 | |
| 965 | _new_snap_rows = _new_snap_dicts |
| 966 | if _new_snap_rows: |
| 967 | await session.execute( |
| 968 | _pg_insert(MusehubSnapshotRef) |
| 969 | .values([{"repo_id": repo_id, "snapshot_id": r["snapshot_id"], "created_at": _utc_now()} for r in _new_snap_rows]) |
| 970 | .on_conflict_do_nothing(index_elements=["repo_id", "snapshot_id"]) |
| 971 | ) |
| 972 | t_snap_ref = _time_module.monotonic() |
| 973 | logger.warning("[SERVER step 8e] snapshots: UPSERT snapshot_refs %.1fms new=%d", (t_snap_ref - t_snap_add) * 1000, len(_new_snap_rows)) |
| 974 | else: |
| 975 | t_snap_ref = t_snap_loop1 |
| 976 | |
| 977 | # step 8: write commits |
| 978 | from muse.core.commits import CommitRecord as _CR_inline |
| 979 | _commit_rows_inline = [] |
| 980 | _graph_rows_inline = [] |
| 981 | |
| 982 | _commit_ids_in_mpack = { |
| 983 | _cd.get("commit_id") for _cd in _raw_commits_inline if _cd.get("commit_id") |
| 984 | } |
| 985 | _external_parent_cids = { |
| 986 | pid |
| 987 | for _cd in _raw_commits_inline |
| 988 | for pid in ( |
| 989 | ([_cd["parent_commit_id"]] if _cd.get("parent_commit_id") else []) |
| 990 | + ([_cd["parent2_commit_id"]] if _cd.get("parent2_commit_id") else []) |
| 991 | ) |
| 992 | if pid not in _commit_ids_in_mpack |
| 993 | } |
| 994 | t_gen_sel0 = _time_module.monotonic() |
| 995 | _db_parent_gens: dict[str, int] = {} |
| 996 | if _external_parent_cids: |
| 997 | _gen_rows = (await session.execute( |
| 998 | select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) |
| 999 | .where(MusehubCommitGraph.commit_id.in_(_external_parent_cids)) |
| 1000 | )).all() |
| 1001 | _db_parent_gens = {_cid: _gval for _cid, _gval in _gen_rows} |
| 1002 | t_gen_sel1 = _time_module.monotonic() |
| 1003 | logger.warning("[SERVER step 9a] commits: fetch parent generations %.1fms external_parents=%d found=%d", |
| 1004 | (t_gen_sel1 - t_gen_sel0) * 1000, len(_external_parent_cids), len(_db_parent_gens)) |
| 1005 | |
| 1006 | t_commit_parse0 = _time_module.monotonic() |
| 1007 | |
| 1008 | _cid_set_inline = {_cd.get("commit_id") for _cd in _raw_commits_inline if _cd.get("commit_id")} |
| 1009 | _cd_by_cid: dict[str, dict] = {_cd["commit_id"]: _cd for _cd in _raw_commits_inline if _cd.get("commit_id")} |
| 1010 | _children: dict[str, list[dict]] = {cid: [] for cid in _cid_set_inline} |
| 1011 | _in_degree: dict[str, int] = {cid: 0 for cid in _cid_set_inline} |
| 1012 | for _cd in _raw_commits_inline: |
| 1013 | for _pk in ("parent_commit_id", "parent2_commit_id"): |
| 1014 | _p = _cd.get(_pk) or "" |
| 1015 | if _p in _cid_set_inline: |
| 1016 | _children[_p].append(_cd) |
| 1017 | _in_degree[_cd["commit_id"]] = _in_degree.get(_cd["commit_id"], 0) + 1 |
| 1018 | from collections import deque as _deque |
| 1019 | _queue: _deque[dict] = _deque( |
| 1020 | _cd_by_cid[cid] for cid, deg in _in_degree.items() if deg == 0 |
| 1021 | ) |
| 1022 | _topo_sorted: list[dict] = [] |
| 1023 | while _queue: |
| 1024 | _cd = _queue.popleft() |
| 1025 | _topo_sorted.append(_cd) |
| 1026 | for _child in _children.get(_cd.get("commit_id", ""), []): |
| 1027 | _child_cid = _child.get("commit_id", "") |
| 1028 | _in_degree[_child_cid] -= 1 |
| 1029 | if _in_degree[_child_cid] == 0: |
| 1030 | _queue.append(_child) |
| 1031 | _topo_seen = {_cd.get("commit_id") for _cd in _topo_sorted} |
| 1032 | _topo_sorted.extend(_cd for _cd in _raw_commits_inline if _cd.get("commit_id") not in _topo_seen) |
| 1033 | |
| 1034 | _inline_gen: dict[str, int] = {} |
| 1035 | for _cd in _topo_sorted: |
| 1036 | # musehub#195: capture the exact client-original committed_at string |
| 1037 | # BEFORE CommitRecord.from_dict parses it into a datetime -- this is |
| 1038 | # the only point in the whole ingest path where the untouched wire |
| 1039 | # string still exists. Postgres timestamptz normalizes non-UTC |
| 1040 | # offsets on write, so reconstructing wire bytes from the stored |
| 1041 | # `timestamp` column alone would produce a different string (and a |
| 1042 | # different commit_id) than the one this string hashes to. |
| 1043 | _committed_at_raw = _cd.get("committed_at") or None |
| 1044 | try: |
| 1045 | _cr = _CR_inline.from_dict(_cd) |
| 1046 | except Exception: |
| 1047 | continue |
| 1048 | if not _cr.commit_id: |
| 1049 | continue |
| 1050 | _pids: list[str] = [] |
| 1051 | if _cr.parent_commit_id: |
| 1052 | _pids.append(_cr.parent_commit_id) |
| 1053 | if _cr.parent2_commit_id: |
| 1054 | _pids.append(_cr.parent2_commit_id) |
| 1055 | _parent_gens: list[int] = [] |
| 1056 | for _pid in _pids: |
| 1057 | if _pid in _inline_gen: |
| 1058 | _parent_gens.append(_inline_gen[_pid]) |
| 1059 | elif _pid in _db_parent_gens: |
| 1060 | _parent_gens.append(_db_parent_gens[_pid]) |
| 1061 | if not _pids: |
| 1062 | _gen = 0 # genuine root — no parents, generation 0 is correct |
| 1063 | elif len(_parent_gens) == len(_pids): |
| 1064 | _gen = max(_parent_gens) + 1 |
| 1065 | else: |
| 1066 | _unresolved_pids = [ |
| 1067 | p for p in _pids if p not in _inline_gen and p not in _db_parent_gens |
| 1068 | ] |
| 1069 | logger.warning( |
| 1070 | "[MWP1] unresolved parent generations — backfilling %d at " |
| 1071 | "wire_push_unpack_mpack commit=%s unresolved_parents=%s", |
| 1072 | len(_unresolved_pids), _cr.commit_id, _unresolved_pids, |
| 1073 | ) |
| 1074 | _max_parent_gen = await _resolve_generation_with_backfill( |
| 1075 | session, _pids, |
| 1076 | inline_gen=_inline_gen, |
| 1077 | db_gen=_db_parent_gens, |
| 1078 | ) |
| 1079 | _gen = _max_parent_gen + 1 |
| 1080 | _inline_gen[_cr.commit_id] = _gen |
| 1081 | _commit_rows_inline.append({ |
| 1082 | "commit_id": _cr.commit_id, |
| 1083 | "branch": _cr.branch or "main", |
| 1084 | "message": _cr.message or "", |
| 1085 | "author": _cr.author or "", |
| 1086 | "timestamp": _cr.committed_at or _utc_now(), |
| 1087 | "committed_at_raw": _committed_at_raw, |
| 1088 | "parent_ids": _pids, |
| 1089 | "snapshot_id": _cr.snapshot_id or None, |
| 1090 | "agent_id": _cr.agent_id or "", |
| 1091 | "model_id": _cr.model_id or "", |
| 1092 | "toolchain_id": _cr.toolchain_id or "", |
| 1093 | "commit_branch": _cr.branch or None, |
| 1094 | "signature": _cr.signature or "", |
| 1095 | "signer_public_key": _cr.signer_public_key or "", |
| 1096 | "signer_key_id": _cr.signer_key_id or "", |
| 1097 | "sem_ver_bump": _cr.sem_ver_bump or "none", |
| 1098 | "breaking_changes": _cr.breaking_changes or [], |
| 1099 | "reviewed_by": [], |
| 1100 | "test_runs": 0, |
| 1101 | "prompt_hash": _cr.prompt_hash or "", |
| 1102 | "structured_delta": _cr.structured_delta if isinstance(_cr.structured_delta, dict) else None, |
| 1103 | }) |
| 1104 | _graph_rows_inline.append({ |
| 1105 | "commit_id": _cr.commit_id, |
| 1106 | "parent_ids": _pids, |
| 1107 | "generation": _gen, |
| 1108 | "snapshot_id": _cr.snapshot_id or None, |
| 1109 | }) |
| 1110 | t_commit_parse1 = _time_module.monotonic() |
| 1111 | _gen_values = sorted(_inline_gen.values()) |
| 1112 | logger.warning( |
| 1113 | "[SERVER step 9b] commits: topo sort + parse %.1fms parsed=%d gen_min=%s gen_max=%s", |
| 1114 | (t_commit_parse1 - t_commit_parse0) * 1000, len(_commit_rows_inline), |
| 1115 | _gen_values[0] if _gen_values else "NONE", |
| 1116 | _gen_values[-1] if _gen_values else "NONE", |
| 1117 | ) |
| 1118 | |
| 1119 | if _commit_rows_inline: |
| 1120 | t_csel0 = _time_module.monotonic() |
| 1121 | _commit_ids_to_check = [r["commit_id"] for r in _commit_rows_inline] |
| 1122 | _existing_cids = set((await session.execute( |
| 1123 | _sa_text("SELECT commit_id FROM musehub_commits WHERE commit_id = ANY(:ids)"), |
| 1124 | {"ids": _commit_ids_to_check}, |
| 1125 | )).scalars()) |
| 1126 | t_csel1 = _time_module.monotonic() |
| 1127 | logger.warning("[SERVER step 9c] commits: dedup SELECT %.1fms existing=%d new=%d", |
| 1128 | (t_csel1 - t_csel0) * 1000, len(_existing_cids), |
| 1129 | len(_commit_rows_inline) - len(_existing_cids)) |
| 1130 | |
| 1131 | _new_commit_rows = [r for r in _commit_rows_inline if r["commit_id"] not in _existing_cids] |
| 1132 | session.add_all([MusehubCommit(**_r) for _r in _new_commit_rows]) |
| 1133 | t_cadd = _time_module.monotonic() |
| 1134 | logger.warning("[SERVER step 9d] commits: INSERT new %.1fms new=%d", (t_cadd - t_csel1) * 1000, len(_new_commit_rows)) |
| 1135 | # Upsert refs for ALL commits in this push, not just globally-new ones. |
| 1136 | # A commit may already exist in musehub_commits from another repo's push |
| 1137 | # but still be missing its ref for THIS repo — using _new_commit_rows here |
| 1138 | # silently drops those refs and leaves commit history invisible to intel. |
| 1139 | if _commit_rows_inline: |
| 1140 | await session.execute( |
| 1141 | _pg_insert(MusehubCommitRef) |
| 1142 | .values([{"repo_id": repo_id, "commit_id": r["commit_id"], "created_at": _utc_now()} for r in _commit_rows_inline]) |
| 1143 | .on_conflict_do_nothing(index_elements=["repo_id", "commit_id"]) |
| 1144 | ) |
| 1145 | t_cref = _time_module.monotonic() |
| 1146 | logger.warning("[SERVER step 9e] commits: UPSERT commit_refs %.1fms total=%d", (t_cref - t_cadd) * 1000, len(_commit_rows_inline)) |
| 1147 | |
| 1148 | if _graph_rows_inline: |
| 1149 | _graph_stmt = _pg_insert(MusehubCommitGraph).values(_graph_rows_inline) |
| 1150 | await session.execute(_graph_stmt.on_conflict_do_update( |
| 1151 | index_elements=["commit_id"], |
| 1152 | set_={"generation": _graph_stmt.excluded.generation, |
| 1153 | "snapshot_id": _graph_stmt.excluded.snapshot_id}, |
| 1154 | )) |
| 1155 | t_graph = _time_module.monotonic() |
| 1156 | logger.warning("[SERVER step 9f] commits: UPSERT commit_graph %.1fms rows=%d", |
| 1157 | (t_graph - t_cref) * 1000, len(_graph_rows_inline)) |
| 1158 | else: |
| 1159 | t_graph = t_commit_parse1 |
| 1160 | |
| 1161 | # step 9: advance branch pointer |
| 1162 | tip = head_commit_id |
| 1163 | if tip and branch: |
| 1164 | t_branch0 = _time_module.monotonic() |
| 1165 | branch_row = (await session.execute( |
| 1166 | select(MusehubBranch).where( |
| 1167 | MusehubBranch.repo_id == repo_id, |
| 1168 | MusehubBranch.name == branch, |
| 1169 | ).with_for_update() |
| 1170 | )).scalar_one_or_none() |
| 1171 | |
| 1172 | current_head = branch_row.head_commit_id if branch_row else None |
| 1173 | if not force and current_head: |
| 1174 | if not _is_fast_forward(tip, current_head, wire_bytes): |
| 1175 | # The mpack-only walk can't confirm ancestry when some ancestor |
| 1176 | # commits are already on the remote (absent from the mpack because |
| 1177 | # they were in the 'have' set). Fall back to the commit graph. |
| 1178 | if not await _is_ancestor_db(session, current_head, tip, repo_id): |
| 1179 | raise NonFastForwardError( |
| 1180 | f"push rejected: {tip[:20]}… is not a fast-forward of " |
| 1181 | f"{current_head[:20]}… — use --force to override" |
| 1182 | ) |
| 1183 | |
| 1184 | if branch_row is None: |
| 1185 | session.add(MusehubBranch( |
| 1186 | branch_id=compute_branch_id(repo_id, branch), |
| 1187 | repo_id=repo_id, |
| 1188 | name=branch, |
| 1189 | head_commit_id=tip, |
| 1190 | )) |
| 1191 | else: |
| 1192 | branch_row.head_commit_id = tip |
| 1193 | # bust cache on force-push: tip-scoped so sibling caches survive (D1 + D3) |
| 1194 | if force and current_head and current_head != tip: |
| 1195 | await invalidate_fetch_mpack_cache(session, repo_id, tips=[current_head, tip]) |
| 1196 | t_branch1 = _time_module.monotonic() |
| 1197 | logger.warning("[SERVER step 10] advance branch pointer (atomic CAS) %.1fms", (t_branch1 - t_branch0) * 1000) |
| 1198 | else: |
| 1199 | t_branch1 = t_graph |
| 1200 | |
| 1201 | # commit db transaction |
| 1202 | t_pre_commit = _time_module.monotonic() |
| 1203 | await session.commit() |
| 1204 | t_commit = _time_module.monotonic() |
| 1205 | logger.warning("[SERVER step 10] commit db transaction %.1fms", (t_commit - t_pre_commit) * 1000) |
| 1206 | |
| 1207 | # step 11: enqueue async jobs |
| 1208 | t_intel0 = _time_module.monotonic() |
| 1209 | from musehub.services.musehub_jobs import enqueue_push_intel |
| 1210 | await enqueue_push_intel( |
| 1211 | session, |
| 1212 | repo_id, |
| 1213 | head=tip, |
| 1214 | branch=branch, |
| 1215 | mpack_key=mpack_key, |
| 1216 | ) |
| 1217 | t_intel1 = _time_module.monotonic() |
| 1218 | logger.warning("[SERVER step 11] enqueue intel jobs %.1fms", (t_intel1 - t_intel0) * 1000) |
| 1219 | |
| 1220 | elapsed_ms = (t_intel1 - t0) * 1000 |
| 1221 | logger.warning("[SERVER ✅] TOTAL=%.1fms branch=%s head=%s", elapsed_ms, branch, head_commit_id[:20] if head_commit_id else "none") |
| 1222 | return { |
| 1223 | "head": tip, |
| 1224 | "branch": branch, |
| 1225 | "blobs_in_mpack": blobs_count, |
| 1226 | "commits_in_mpack": commits_count, |
| 1227 | "blobs_written": len(_new_oids), |
| 1228 | "commits_written": len(_new_commit_rows) if _commit_rows_inline else 0, |
| 1229 | "snapshots_written": len(_new_snap_dicts) if _snap_rows_inline else 0, |
| 1230 | } |
| 1231 | |
| 1232 | |
| 1233 | class PurgeResult(TypedDict): |
| 1234 | removed: int |
| 1235 | kept: int |
| 1236 | |
| 1237 | |
| 1238 | async def purge_stale_mpack_index_entries(session: AsyncSession) -> PurgeResult: |
| 1239 | """Delete MusehubMPackIndex rows for mpacks no longer in MinIO. |
| 1240 | |
| 1241 | Left behind when mpack.gc deleted old mpacks but didn't prune the index. |
| 1242 | Safe to delete: objects with stale entries have valid s3:// storage_uri |
| 1243 | and are served directly from S3 without needing the mpack path. |
| 1244 | |
| 1245 | Returns {"removed": N, "kept": M}. |
| 1246 | """ |
| 1247 | from musehub.storage.backends import get_backend |
| 1248 | backend = get_backend() |
| 1249 | |
| 1250 | # Find all distinct mpack_ids in the index. |
| 1251 | distinct_mpacks = (await session.execute( |
| 1252 | select(MusehubMPackIndex.mpack_id).distinct() |
| 1253 | )).scalars().all() |
| 1254 | |
| 1255 | removed = 0 |
| 1256 | kept = 0 |
| 1257 | for mpack_id in distinct_mpacks: |
| 1258 | alive = await backend.exists_mpack(mpack_id) |
| 1259 | if alive: |
| 1260 | kept_count = (await session.execute( |
| 1261 | select(MusehubMPackIndex.entity_id).where( |
| 1262 | MusehubMPackIndex.mpack_id == mpack_id |
| 1263 | ) |
| 1264 | )).scalars().all() |
| 1265 | kept += len(kept_count) |
| 1266 | else: |
| 1267 | # Delete all index entries for this dead mpack. |
| 1268 | rows = (await session.execute( |
| 1269 | select(MusehubMPackIndex).where( |
| 1270 | MusehubMPackIndex.mpack_id == mpack_id |
| 1271 | ) |
| 1272 | )).scalars().all() |
| 1273 | for row in rows: |
| 1274 | await session.delete(row) |
| 1275 | removed += len(rows) |
| 1276 | |
| 1277 | await session.flush() |
| 1278 | logger.info("purge_stale_mpack_index_entries: removed=%d kept=%d", removed, kept) |
| 1279 | return {"removed": removed, "kept": kept} |
| 1280 | |
| 1281 | |
| 1282 | async def _resolve_generation_with_backfill( |
| 1283 | session: AsyncSession, |
| 1284 | pids: list[str], |
| 1285 | *, |
| 1286 | inline_gen: dict[str, int] | None = None, |
| 1287 | db_gen: dict[str, int] | None = None, |
| 1288 | ) -> int: |
| 1289 | """Resolve the max generation across *pids* by backfilling missing graph rows. |
| 1290 | |
| 1291 | Walks ``musehub_commits.parent_ids`` toward roots, computing and upserting |
| 1292 | ``musehub_commit_graph`` rows for every ancestor whose generation is unknown. |
| 1293 | The walk is **bounded**: it stops at the first commit already in the graph, |
| 1294 | so steady-state cost is O(unresolved frontier), not O(full history). |
| 1295 | |
| 1296 | Returns ``max(generation for pid in pids)``. Caller adds +1 to get the |
| 1297 | child's generation. |
| 1298 | |
| 1299 | Raises ``ValueError`` if any pid is absent from ``musehub_commits``. A |
| 1300 | fast-forward push must not reference commits the server does not have. |
| 1301 | """ |
| 1302 | known: dict[str, int] = {} |
| 1303 | if inline_gen: |
| 1304 | known.update(inline_gen) |
| 1305 | if db_gen: |
| 1306 | known.update(db_gen) |
| 1307 | |
| 1308 | # parent_map: commits we discovered via the walk that need generation computed. |
| 1309 | parent_map: dict[str, list[str]] = {} |
| 1310 | snapshot_map: dict[str, str | None] = {} |
| 1311 | |
| 1312 | frontier = {p for p in pids if p not in known} |
| 1313 | while frontier: |
| 1314 | # Stop the walk at commits already in musehub_commit_graph (anchors). |
| 1315 | graph_rows = (await session.execute( |
| 1316 | select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) |
| 1317 | .where(MusehubCommitGraph.commit_id.in_(frontier)) |
| 1318 | )).all() |
| 1319 | for cid, gen in graph_rows: |
| 1320 | known[cid] = gen |
| 1321 | |
| 1322 | need_walk = {c for c in frontier if c not in known} |
| 1323 | if not need_walk: |
| 1324 | break |
| 1325 | |
| 1326 | # Fetch authoritative parent pointers from musehub_commits (source-of-truth DAG). |
| 1327 | commit_rows = (await session.execute( |
| 1328 | select(MusehubCommit.commit_id, MusehubCommit.parent_ids, MusehubCommit.snapshot_id) |
| 1329 | .where(MusehubCommit.commit_id.in_(need_walk)) |
| 1330 | )).all() |
| 1331 | |
| 1332 | found = {cid for cid, _, _ in commit_rows} |
| 1333 | missing = need_walk - found |
| 1334 | if missing: |
| 1335 | raise ValueError( |
| 1336 | "[MWP1] integrity error: parent commit(s) absent from musehub_commits " |
| 1337 | f"— fast-forward push invariant violated: {sorted(missing)}" |
| 1338 | ) |
| 1339 | |
| 1340 | for cid, parent_ids, snap_id in commit_rows: |
| 1341 | parent_map[cid] = parent_ids |
| 1342 | snapshot_map[cid] = snap_id |
| 1343 | |
| 1344 | # Expand frontier: parents not yet known or already queued for walk. |
| 1345 | frontier = { |
| 1346 | p |
| 1347 | for cid, parent_ids, _ in commit_rows |
| 1348 | for p in parent_ids |
| 1349 | if p not in known and p not in parent_map |
| 1350 | } |
| 1351 | |
| 1352 | if not parent_map: |
| 1353 | # All pids were already resolved (in inline_gen, db_gen, or graph table). |
| 1354 | return max(known[p] for p in pids) |
| 1355 | |
| 1356 | # Topological sort of parent_map (Kahn's algorithm — roots first). |
| 1357 | children_map: dict[str, list[str]] = {cid: [] for cid in parent_map} |
| 1358 | in_degree: dict[str, int] = {cid: 0 for cid in parent_map} |
| 1359 | for cid, pids_of_cid in parent_map.items(): |
| 1360 | for p in pids_of_cid: |
| 1361 | if p in parent_map: |
| 1362 | children_map[p].append(cid) |
| 1363 | in_degree[cid] += 1 |
| 1364 | |
| 1365 | queue: collections.deque[str] = collections.deque( |
| 1366 | cid for cid, deg in in_degree.items() if deg == 0 |
| 1367 | ) |
| 1368 | computed: dict[str, int] = {} |
| 1369 | new_rows: list[dict] = [] |
| 1370 | now = _utc_now() |
| 1371 | |
| 1372 | while queue: |
| 1373 | cid = queue.popleft() |
| 1374 | pids_of_cid = parent_map[cid] |
| 1375 | parent_gens = [ |
| 1376 | computed[p] if p in computed else known[p] |
| 1377 | for p in pids_of_cid |
| 1378 | if p in computed or p in known |
| 1379 | ] |
| 1380 | gen = (max(parent_gens) + 1) if parent_gens else 0 |
| 1381 | computed[cid] = gen |
| 1382 | new_rows.append({ |
| 1383 | "commit_id": cid, |
| 1384 | "parent_ids": pids_of_cid, |
| 1385 | "generation": gen, |
| 1386 | "snapshot_id": snapshot_map.get(cid), |
| 1387 | "created_at": now, |
| 1388 | }) |
| 1389 | for child in children_map.get(cid, []): |
| 1390 | in_degree[child] -= 1 |
| 1391 | if in_degree[child] == 0: |
| 1392 | queue.append(child) |
| 1393 | |
| 1394 | if new_rows: |
| 1395 | _CHUNK = 2000 |
| 1396 | for i in range(0, len(new_rows), _CHUNK): |
| 1397 | chunk = new_rows[i:i + _CHUNK] |
| 1398 | stmt = _pg_insert(MusehubCommitGraph).values(chunk) |
| 1399 | await session.execute(stmt.on_conflict_do_update( |
| 1400 | index_elements=["commit_id"], |
| 1401 | set_={ |
| 1402 | "generation": stmt.excluded.generation, |
| 1403 | "snapshot_id": stmt.excluded.snapshot_id, |
| 1404 | }, |
| 1405 | )) |
| 1406 | |
| 1407 | known.update(computed) |
| 1408 | return max(known[p] for p in pids if p in known) |
| 1409 | |
| 1410 | |
| 1411 | async def repair_corrupt_commit_generations(session: AsyncSession) -> dict: |
| 1412 | """Repair musehub_commit_graph rows where generation=0 but parent_ids is non-empty. |
| 1413 | |
| 1414 | These rows were written by the RC-1 bug before Phase 2 fixed the generation |
| 1415 | computation. The repair: |
| 1416 | |
| 1417 | 1. Finds every row with ``generation == 0 AND cardinality(parent_ids) > 0``. |
| 1418 | 2. Fetches external parent generations from the graph (parents not in the |
| 1419 | corrupt set already have correct, non-zero generations there). |
| 1420 | 3. Backfills any external parents missing from the graph via |
| 1421 | ``_resolve_generation_with_backfill``. |
| 1422 | 4. Topologically sorts corrupt commits (parents before children) and |
| 1423 | recomputes correct generations bottom-up. |
| 1424 | 5. Bulk-upserts only the ``generation`` column — all other fields preserved. |
| 1425 | |
| 1426 | Idempotent: a second call finds zero corrupt rows and returns immediately. |
| 1427 | Returns ``{"total_corrupt": N, "repaired": N}``. |
| 1428 | """ |
| 1429 | # Step 1 — find all corrupt rows (gen=0, non-empty parents). |
| 1430 | corrupt_result = (await session.execute( |
| 1431 | select( |
| 1432 | MusehubCommitGraph.commit_id, |
| 1433 | MusehubCommitGraph.parent_ids, |
| 1434 | MusehubCommitGraph.snapshot_id, |
| 1435 | ) |
| 1436 | .where(MusehubCommitGraph.generation == 0) |
| 1437 | .where(func.cardinality(MusehubCommitGraph.parent_ids) > 0) |
| 1438 | )).all() |
| 1439 | |
| 1440 | if not corrupt_result: |
| 1441 | logger.info("[MWP1] repair: no corrupt commit-graph rows found — nothing to do") |
| 1442 | return {"total_corrupt": 0, "repaired": 0} |
| 1443 | |
| 1444 | corrupt_map: dict[str, list[str]] = { |
| 1445 | cid: list(pids) for cid, pids, _ in corrupt_result |
| 1446 | } |
| 1447 | corrupt_snap: dict[str, str | None] = { |
| 1448 | cid: snap for cid, _, snap in corrupt_result |
| 1449 | } |
| 1450 | corrupt_set = set(corrupt_map.keys()) |
| 1451 | total_corrupt = len(corrupt_set) |
| 1452 | logger.warning("[MWP1] repair: found %d corrupt commit-graph row(s)", total_corrupt) |
| 1453 | |
| 1454 | # Step 2 — collect external parents (parents NOT in corrupt_set) and fetch |
| 1455 | # their generations from the graph. These are the walk anchors. |
| 1456 | external_parent_cids = { |
| 1457 | p for pids in corrupt_map.values() for p in pids if p not in corrupt_set |
| 1458 | } |
| 1459 | known_gens: dict[str, int] = {} |
| 1460 | if external_parent_cids: |
| 1461 | ext_rows = (await session.execute( |
| 1462 | select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) |
| 1463 | .where(MusehubCommitGraph.commit_id.in_(external_parent_cids)) |
| 1464 | )).all() |
| 1465 | known_gens = {cid: gen for cid, gen in ext_rows} |
| 1466 | |
| 1467 | # Step 3 — backfill any external parents absent from the graph entirely |
| 1468 | # (should not happen after Phase 2, but handle gracefully). |
| 1469 | missing_ext = external_parent_cids - set(known_gens.keys()) |
| 1470 | if missing_ext: |
| 1471 | logger.warning( |
| 1472 | "[MWP1] repair: %d external parents have no graph row — backfilling", |
| 1473 | len(missing_ext), |
| 1474 | ) |
| 1475 | try: |
| 1476 | await _resolve_generation_with_backfill( |
| 1477 | session, list(missing_ext), inline_gen={}, db_gen={} |
| 1478 | ) |
| 1479 | refetch = (await session.execute( |
| 1480 | select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) |
| 1481 | .where(MusehubCommitGraph.commit_id.in_(missing_ext)) |
| 1482 | )).all() |
| 1483 | for cid, gen in refetch: |
| 1484 | known_gens[cid] = gen |
| 1485 | except ValueError as exc: |
| 1486 | logger.error("[MWP1] repair: backfill for missing externals failed: %s", exc) |
| 1487 | |
| 1488 | # Step 4 — topological sort of corrupt commits (Kahn's, roots first). |
| 1489 | children_map: dict[str, list[str]] = {cid: [] for cid in corrupt_set} |
| 1490 | in_degree: dict[str, int] = {cid: 0 for cid in corrupt_set} |
| 1491 | for cid, pids in corrupt_map.items(): |
| 1492 | for p in pids: |
| 1493 | if p in corrupt_set: |
| 1494 | children_map[p].append(cid) |
| 1495 | in_degree[cid] += 1 |
| 1496 | |
| 1497 | queue: collections.deque[str] = collections.deque( |
| 1498 | cid for cid, deg in in_degree.items() if deg == 0 |
| 1499 | ) |
| 1500 | computed: dict[str, int] = {} |
| 1501 | corrected: list[dict] = [] |
| 1502 | now = _utc_now() |
| 1503 | |
| 1504 | while queue: |
| 1505 | cid = queue.popleft() |
| 1506 | pids = corrupt_map[cid] |
| 1507 | parent_gens = [] |
| 1508 | for p in pids: |
| 1509 | if p in computed: |
| 1510 | parent_gens.append(computed[p]) |
| 1511 | elif p in known_gens: |
| 1512 | parent_gens.append(known_gens[p]) |
| 1513 | else: |
| 1514 | logger.warning( |
| 1515 | "[MWP1] repair: parent %s of corrupt commit %s not found " |
| 1516 | "in graph or computed set — using 0 as fallback", |
| 1517 | p[:20], cid[:20], |
| 1518 | ) |
| 1519 | parent_gens.append(0) |
| 1520 | |
| 1521 | correct_gen = (max(parent_gens) + 1) if parent_gens else 0 |
| 1522 | computed[cid] = correct_gen |
| 1523 | corrected.append({ |
| 1524 | "commit_id": cid, |
| 1525 | "parent_ids": pids, |
| 1526 | "generation": correct_gen, |
| 1527 | "snapshot_id": corrupt_snap.get(cid), |
| 1528 | "created_at": now, |
| 1529 | }) |
| 1530 | for child in children_map.get(cid, []): |
| 1531 | in_degree[child] -= 1 |
| 1532 | if in_degree[child] == 0: |
| 1533 | queue.append(child) |
| 1534 | |
| 1535 | # Step 5 — bulk-upsert, updating only the generation column. |
| 1536 | if corrected: |
| 1537 | _CHUNK = 2000 |
| 1538 | for i in range(0, len(corrected), _CHUNK): |
| 1539 | chunk = corrected[i:i + _CHUNK] |
| 1540 | stmt = _pg_insert(MusehubCommitGraph).values(chunk) |
| 1541 | await session.execute(stmt.on_conflict_do_update( |
| 1542 | index_elements=["commit_id"], |
| 1543 | set_={"generation": stmt.excluded.generation}, |
| 1544 | )) |
| 1545 | logger.warning("[MWP1] repair: corrected %d row(s)", len(corrected)) |
| 1546 | |
| 1547 | return {"total_corrupt": total_corrupt, "repaired": len(corrected)} |
| 1548 | |
| 1549 | |
| 1550 | async def check_commit_graph_invariant(session: AsyncSession) -> dict: |
| 1551 | """Check that no musehub_commit_graph row has generation=0 with non-empty parents. |
| 1552 | |
| 1553 | Returns {"valid": bool, "violations": int}. A violation means the RC-1 bug |
| 1554 | wrote a corrupt generation that will poison the fetch range scan. Call |
| 1555 | repair_corrupt_commit_generations to fix any violations found. |
| 1556 | """ |
| 1557 | count_q = await session.execute( |
| 1558 | select(func.count()).select_from(MusehubCommitGraph) |
| 1559 | .where(MusehubCommitGraph.generation == 0) |
| 1560 | .where(func.cardinality(MusehubCommitGraph.parent_ids) > 0) |
| 1561 | ) |
| 1562 | violations = count_q.scalar_one() |
| 1563 | return {"valid": violations == 0, "violations": violations} |
| 1564 | |
| 1565 | |
| 1566 | async def _build_commit_graph_from_raw(session: AsyncSession, raw_commits: list[dict]) -> int: |
| 1567 | """Compute generation numbers and upsert MusehubCommitGraph rows. |
| 1568 | |
| 1569 | Mirrors the generation logic in wire_push_unpack_mpack but operates on |
| 1570 | an already-parsed commit list (used by process_mpack_index_job). |
| 1571 | Returns the number of rows upserted. |
| 1572 | """ |
| 1573 | if not raw_commits: |
| 1574 | return 0 |
| 1575 | |
| 1576 | from collections import deque as _deque |
| 1577 | |
| 1578 | cid_set = {cd.get("commit_id") for cd in raw_commits if cd.get("commit_id")} |
| 1579 | cd_by_cid = {cd["commit_id"]: cd for cd in raw_commits if cd.get("commit_id")} |
| 1580 | |
| 1581 | # Topological sort (Kahn's algorithm — oldest first so parents resolve before children). |
| 1582 | children: dict[str, list[dict]] = {cid: [] for cid in cid_set} |
| 1583 | in_degree: dict[str, int] = {cid: 0 for cid in cid_set} |
| 1584 | for cd in raw_commits: |
| 1585 | for pk in ("parent_commit_id", "parent2_commit_id"): |
| 1586 | p = cd.get(pk) or "" |
| 1587 | if p in cid_set: |
| 1588 | children[p].append(cd) |
| 1589 | in_degree[cd["commit_id"]] = in_degree.get(cd["commit_id"], 0) + 1 |
| 1590 | |
| 1591 | queue: _deque[dict] = _deque(cd_by_cid[cid] for cid, deg in in_degree.items() if deg == 0) |
| 1592 | topo: list[dict] = [] |
| 1593 | while queue: |
| 1594 | cd = queue.popleft() |
| 1595 | topo.append(cd) |
| 1596 | for child in children.get(cd.get("commit_id", ""), []): |
| 1597 | child_cid = child.get("commit_id", "") |
| 1598 | in_degree[child_cid] -= 1 |
| 1599 | if in_degree[child_cid] == 0: |
| 1600 | queue.append(child) |
| 1601 | seen = {cd.get("commit_id") for cd in topo} |
| 1602 | topo.extend(cd for cd in raw_commits if cd.get("commit_id") not in seen) |
| 1603 | |
| 1604 | # Fetch generation numbers for any external parents already in the DB. |
| 1605 | external_pids = { |
| 1606 | pid |
| 1607 | for cd in raw_commits |
| 1608 | for pid in ([cd.get("parent_commit_id") or "", cd.get("parent2_commit_id") or ""]) |
| 1609 | if pid and pid not in cid_set |
| 1610 | } |
| 1611 | db_parent_gens: dict[str, int] = {} |
| 1612 | if external_pids: |
| 1613 | gen_rows = (await session.execute( |
| 1614 | select(MusehubCommitGraph.commit_id, MusehubCommitGraph.generation) |
| 1615 | .where(MusehubCommitGraph.commit_id.in_(external_pids)) |
| 1616 | )).all() |
| 1617 | db_parent_gens = {cid: gval for cid, gval in gen_rows} |
| 1618 | |
| 1619 | inline_gen: dict[str, int] = {} |
| 1620 | graph_rows: list[dict] = [] |
| 1621 | now = _utc_now() |
| 1622 | for cd in topo: |
| 1623 | cid = cd.get("commit_id", "") |
| 1624 | if not cid: |
| 1625 | continue |
| 1626 | pids = [p for p in [cd.get("parent_commit_id") or "", cd.get("parent2_commit_id") or ""] if p] |
| 1627 | parent_gens = [inline_gen[p] if p in inline_gen else db_parent_gens[p] |
| 1628 | for p in pids if p in inline_gen or p in db_parent_gens] |
| 1629 | if not pids: |
| 1630 | gen = 0 # genuine root — no parents, generation 0 is correct |
| 1631 | elif len(parent_gens) == len(pids): |
| 1632 | gen = max(parent_gens) + 1 |
| 1633 | else: |
| 1634 | unresolved_pids = [ |
| 1635 | p for p in pids if p not in inline_gen and p not in db_parent_gens |
| 1636 | ] |
| 1637 | logger.warning( |
| 1638 | "[MWP1] unresolved parent generations — backfilling %d at " |
| 1639 | "_build_commit_graph_from_raw commit=%s unresolved_parents=%s", |
| 1640 | len(unresolved_pids), cid, unresolved_pids, |
| 1641 | ) |
| 1642 | max_parent_gen = await _resolve_generation_with_backfill( |
| 1643 | session, pids, |
| 1644 | inline_gen=inline_gen, |
| 1645 | db_gen=db_parent_gens, |
| 1646 | ) |
| 1647 | gen = max_parent_gen + 1 |
| 1648 | inline_gen[cid] = gen |
| 1649 | graph_rows.append({ |
| 1650 | "commit_id": cid, |
| 1651 | "parent_ids": pids, |
| 1652 | "generation": gen, |
| 1653 | "snapshot_id": cd.get("snapshot_id") or None, |
| 1654 | "created_at": now, |
| 1655 | }) |
| 1656 | |
| 1657 | if not graph_rows: |
| 1658 | return 0 |
| 1659 | |
| 1660 | _CHUNK = 2000 |
| 1661 | written = 0 |
| 1662 | for i in range(0, len(graph_rows), _CHUNK): |
| 1663 | chunk = graph_rows[i: i + _CHUNK] |
| 1664 | stmt = _pg_insert(MusehubCommitGraph).values(chunk) |
| 1665 | await session.execute(stmt.on_conflict_do_update( |
| 1666 | index_elements=["commit_id"], |
| 1667 | set_={ |
| 1668 | "generation": stmt.excluded.generation, |
| 1669 | "snapshot_id": stmt.excluded.snapshot_id, |
| 1670 | }, |
| 1671 | )) |
| 1672 | written += len(chunk) |
| 1673 | return written |
| 1674 | |
| 1675 | |
| 1676 | async def process_mpack_index_job(session: AsyncSession, job_id: str) -> JSONObject: |
| 1677 | """Populate MusehubMPackIndex byte-ranges for every object in a pushed mpack. |
| 1678 | |
| 1679 | Called by the background worker after every push. Reads the mpack from |
| 1680 | MinIO, computes byte offsets for each blob, then bulk-upserts |
| 1681 | MusehubMPackIndex rows with byte_offset and byte_length so the fetch path |
| 1682 | can serve blobs via byte-range GETs instead of downloading the full mpack. |
| 1683 | |
| 1684 | Returns a result dict with counts for logging. |
| 1685 | """ |
| 1686 | t0 = _time_module.monotonic() |
| 1687 | |
| 1688 | job_row = (await session.execute( |
| 1689 | select(MusehubBackgroundJob).where(MusehubBackgroundJob.job_id == job_id) |
| 1690 | )).scalar_one_or_none() |
| 1691 | if job_row is None: |
| 1692 | raise ValueError(f"mpack.index job not found: {job_id}") |
| 1693 | |
| 1694 | payload = job_row.payload or {} |
| 1695 | repo_id = job_row.repo_id |
| 1696 | mpack_key = str(payload.get("mpack_key", "")) |
| 1697 | if not mpack_key: |
| 1698 | raise ValueError(f"mpack.index job {job_id} missing mpack_key in payload") |
| 1699 | |
| 1700 | from musehub.storage.backends import get_backend |
| 1701 | backend = get_backend() |
| 1702 | wire_bytes = await backend.get_mpack(mpack_key) |
| 1703 | if not wire_bytes: |
| 1704 | raise ValueError(f"mpack {mpack_key[:20]} not found in storage") |
| 1705 | |
| 1706 | t_fetched = _time_module.monotonic() |
| 1707 | |
| 1708 | # Parse: handle both MUSE binary and legacy msgpack formats. |
| 1709 | if wire_bytes[:4] == b"MUSE": |
| 1710 | from muse.core.mpack import parse_wire_mpack as _parse_wire |
| 1711 | mpack: JSONObject = _parse_wire(wire_bytes) |
| 1712 | else: |
| 1713 | mpack = _msgpack.unpackb(wire_bytes, raw=False) |
| 1714 | |
| 1715 | raw_blobs: list[JSONObject] = mpack.get("blobs") or mpack.get("objects") or [] |
| 1716 | t_parsed = _time_module.monotonic() |
| 1717 | |
| 1718 | logger.info( |
| 1719 | "[mpack.index] job=%s fetch=%.3fs parse=%.3fs blobs=%d mpack=%d KB", |
| 1720 | job_id[:16], t_fetched - t0, t_parsed - t_fetched, |
| 1721 | len(raw_blobs), len(wire_bytes) // 1024, |
| 1722 | ) |
| 1723 | |
| 1724 | # Compute byte offsets for every object in the MUSE binary mpack. |
| 1725 | byte_offsets: dict[str, tuple[int, int]] = {} |
| 1726 | if wire_bytes[:4] == b"MUSE": |
| 1727 | all_oids = [obj.get("object_id", "") for obj in raw_blobs if obj.get("object_id")] |
| 1728 | byte_offsets = compute_object_byte_offsets(wire_bytes) |
| 1729 | |
| 1730 | # Upsert MusehubMPackIndex rows — one per blob with byte_offset/byte_length. |
| 1731 | now = _utc_now() |
| 1732 | _CHUNK = 2000 |
| 1733 | index_rows = [] |
| 1734 | for oid, (offset, length) in byte_offsets.items(): |
| 1735 | index_rows.append({ |
| 1736 | "entity_id": oid, |
| 1737 | "mpack_id": mpack_key, |
| 1738 | "entity_type": "object", |
| 1739 | "created_at": now, |
| 1740 | "byte_offset": offset, |
| 1741 | "byte_length": length, |
| 1742 | }) |
| 1743 | |
| 1744 | # Also index any blobs without byte-range data (fallback: full-mpack serve). |
| 1745 | indexed_oids = set(byte_offsets.keys()) |
| 1746 | for obj in raw_blobs: |
| 1747 | oid = obj.get("object_id", "") |
| 1748 | if oid and oid not in indexed_oids: |
| 1749 | index_rows.append({ |
| 1750 | "entity_id": oid, |
| 1751 | "mpack_id": mpack_key, |
| 1752 | "entity_type": "object", |
| 1753 | "created_at": now, |
| 1754 | }) |
| 1755 | |
| 1756 | written = 0 |
| 1757 | for i in range(0, len(index_rows), _CHUNK): |
| 1758 | chunk = index_rows[i: i + _CHUNK] |
| 1759 | stmt = _pg_insert(MusehubMPackIndex).values(chunk) |
| 1760 | # On conflict: update byte_offset/byte_length if we now have them. |
| 1761 | await session.execute( |
| 1762 | stmt.on_conflict_do_update( |
| 1763 | index_elements=["entity_id", "mpack_id"], |
| 1764 | set_={ |
| 1765 | "byte_offset": stmt.excluded.byte_offset, |
| 1766 | "byte_length": stmt.excluded.byte_length, |
| 1767 | }, |
| 1768 | where=MusehubMPackIndex.byte_offset.is_(None), |
| 1769 | ) |
| 1770 | ) |
| 1771 | written += len(chunk) |
| 1772 | |
| 1773 | # Upsert MusehubObject rows — storage_uri points to covering mpack, no per-object S3 PUT. |
| 1774 | # Must happen before object_refs because object_refs has a FK on musehub_objects. |
| 1775 | all_blob_oids = [obj.get("object_id", "") for obj in raw_blobs if obj.get("object_id")] |
| 1776 | if all_blob_oids: |
| 1777 | mpack_uri = f"mpack://{mpack_key}" |
| 1778 | obj_rows = [ |
| 1779 | {"object_id": oid, "storage_uri": mpack_uri, "path": "", "created_at": now} |
| 1780 | for oid in all_blob_oids |
| 1781 | ] |
| 1782 | for i in range(0, len(obj_rows), _CHUNK): |
| 1783 | chunk = obj_rows[i: i + _CHUNK] |
| 1784 | obj_stmt = _pg_insert(MusehubObject).values(chunk) |
| 1785 | await session.execute( |
| 1786 | obj_stmt.on_conflict_do_update( |
| 1787 | index_elements=["object_id"], |
| 1788 | set_={"storage_uri": obj_stmt.excluded.storage_uri}, |
| 1789 | where=MusehubObject.storage_uri.is_(None), |
| 1790 | ) |
| 1791 | ) |
| 1792 | |
| 1793 | # Write object_refs after musehub_objects so the FK constraint is satisfied. |
| 1794 | if all_blob_oids and repo_id: |
| 1795 | await _upsert_object_refs(session, repo_id, all_blob_oids) |
| 1796 | |
| 1797 | # Build commit graph rows from mpack commits. |
| 1798 | raw_commits: list[dict] = mpack.get("commits") or [] |
| 1799 | graph_written = await _build_commit_graph_from_raw(session, raw_commits) |
| 1800 | |
| 1801 | t_done = _time_module.monotonic() |
| 1802 | logger.info( |
| 1803 | "✅ [mpack.index] done job=%s index_rows=%d graph_rows=%d elapsed=%.3fs", |
| 1804 | job_id[:16], written, graph_written, t_done - t0, |
| 1805 | ) |
| 1806 | return { |
| 1807 | "mpack_index_written": written, |
| 1808 | "byte_ranges_computed": len(byte_offsets), |
| 1809 | "commit_graph_written": graph_written, |
| 1810 | "mpack_size_bytes": len(wire_bytes), |
| 1811 | "elapsed_ms": (t_done - t0) * 1000, |
| 1812 | } |
File History
20 commits
sha256:7d6dd8f4a89e2d1fef2d84f6e65feaff51385d382f466766b7f690a22ec18e32
fix: fall back to DB ancestry check when mpack-only fast-fo…
Sonnet 4.6
patch
134 days ago
sha256:1c94edd45e7ed3bf7012c2742e755bb7608890f0964aa2ba9ea7f8cf83807013
revert: undo unauthorized wire protocol change
Sonnet 4.6
patch
136 days ago
sha256:a1d7cf2d9e9c9f4edc566e56afbde3f6b3194ca712209beb02eb3c838d1f47e6
fix: fast-forward push validation falls back to DB ancestor walk
Sonnet 4.6
patch
136 days ago
sha256:80c35083d2abe19ef730693f2b35ad9a9342d2abbe81e802e6f397fe9ae75eb9
merge: pull staging/dev — asyncpg batch fix + content-addre…
Sonnet 4.6
patch
136 days ago
sha256:5667a3e21bf16fd2e6d6bd4a769bd1c0cf7634afa12cef6450cc77573196b7f9
asyncpg caps query parameters
Human
patch
136 days ago
sha256:c3e12fae919d99c0e030595f58eb0a7b7df9ae4bd4ed6ce582a46c33784a54d0
fix(mpack-index-job): write object_refs after musehub_objec…
Sonnet 4.6
patch
139 days ago
sha256:8ba2515ec3cb2a4089d54422f342b5a47161df29d5db72cb7356591d128fba6a
fix(wire-push): write object_refs and mpack_index for all m…
Sonnet 4.6
patch
139 days ago
sha256:009b5a222314f47640a58d75ce5a1f428f1624cf0b51384dfcdfbdfab3cc42a4
feat: migration idempotency, file attribution DAG walk, mpa…
Sonnet 4.6
minor
⚠
143 days ago
sha256:92528ae07d0e1239d87fd5fd1f439e8fbb49c9778a9a400bc4a736073fb28316
feat: byte-range blob reads, file attribution DAG walk, bra…
Sonnet 4.6
minor
⚠
143 days ago
sha256:f3995ec2c05c9c34b0e4d6e96349a811d0117a1c51d78096d757998ccb3c0520
fix: blobs only in S3/mpack — remove commit/snapshot indivi…
Sonnet 4.6
patch
145 days ago
sha256:8a7ff43f27504c1f6abba59ffbab3dc89bd1d8bcfa4c57f6e280286d1a58195a
feat: resurrect process_mpack_index_job with byte-range ind…
Sonnet 4.6
patch
145 days ago
sha256:ad616c6113d6c00f4efed6b2993734ca46d3e9b5bee25addd4ce8ae6b57136e5
chore: bump version to 0.2.0rc11; typing audit clean + all …
Sonnet 4.6
minor
⚠
147 days ago
sha256:ecca645d94c5f39c88f4bc1283447ba0f4635ef3cbb11d0cd9b3759cba289d00
fix: compute_snapshot_id uses typed-object formula and wire…
Sonnet 4.6
minor
⚠
147 days ago
sha256:e35be48854f182f7bf02dc6cc0f58d22b3de3a544b570c0e2bc53f9e75a3607d
feat(phase6): remove delta_blob path, dead imports, add fal…
Sonnet 4.6
minor
⚠
148 days ago
sha256:450998d182617fa93b737cbbdb3fe956c61566051739acec8c63ec5e7b4587f8
feat(phase3): write snapshot objects to S3 at all 3 write s…
Sonnet 4.6
patch
148 days ago
sha256:e597c0b97ade9c3c52ac4735ceb437ee69d1b6f0db61b8d7caa6467c5866566d
feat(phase2): write commit objects to S3 at all 5 write sit…
Sonnet 4.6
patch
148 days ago
sha256:6b1949fc2797ca4c1936a637a4cbfec828ef56cf52398a2e74ca3c4f494e728f
fix: use wire_bytes not mpack_bytes_raw in compute_object_b…
Sonnet 4.6
patch
148 days ago
sha256:302574ddba13c9a20694c0fb051176eef4896f943b63bc458df886633b1bfcd6
feat: mpack byte-range index — store byte_offset/byte_lengt…
Sonnet 4.6
minor
⚠
148 days ago
sha256:ba425baae57a4223a95f2cf4841181d790d04d02b9c98d0371f7b43046d92ff4
fix: store directories from incoming snapshot deltas instea…
Sonnet 4.6
patch
148 days ago
sha256:4c039010d7b35b3bb7e9242df3e73ff50336264cb89a85a7509dc959e8516fc1
fix: store signer_public_key from incoming commits instead …
Sonnet 4.6
patch
148 days ago