pack.py
python
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
33 days ago
| 1 | """Muse pack format — bundle of commits, snapshots, and blobs for wire transfer. |
| 2 | |
| 3 | A :class:`PackBundle` is the unit of exchange between the Muse CLI and a remote |
| 4 | (e.g. MuseHub). It carries everything needed to reconstruct a slice of commit |
| 5 | history locally: |
| 6 | |
| 7 | - :class:`CommitDict` records (full metadata) |
| 8 | - :class:`SnapshotDict` records (file manifests) |
| 9 | - :class:`ObjectPayload` entries (raw blob bytes) |
| 10 | - ``branch_heads`` mapping (branch name → commit ID, reflecting remote state) |
| 11 | |
| 12 | :func:`build_pack` collects all data reachable from a set of commit IDs. |
| 13 | :func:`apply_pack` writes a bundle into a local ``.muse/`` directory. |
| 14 | |
| 15 | MWP wire encoding |
| 16 | -------------------- |
| 17 | Object bytes are transmitted as raw ``bytes`` in :class:`ObjectPayload`. |
| 18 | The :class:`~muse.core.transport.HttpTransport` serialises the pack using |
| 19 | ``msgpack`` (``Content-Type: application/x-msgpack``) which handles binary |
| 20 | natively — no base64 inflation or encoding overhead. |
| 21 | """ |
| 22 | |
| 23 | from __future__ import annotations |
| 24 | |
| 25 | import collections |
| 26 | import logging |
| 27 | import pathlib |
| 28 | from typing import TypedDict |
| 29 | |
| 30 | from muse.core.object_store import read_object, write_object |
| 31 | from muse.core.validation import ( |
| 32 | MAX_OBJECT_WRITE_BYTES, |
| 33 | MAX_PACK_OBJECTS, |
| 34 | validate_object_id, |
| 35 | validate_workspace_path, |
| 36 | ) |
| 37 | from muse.core._types import BranchHeads |
| 38 | from muse.core.store import ( |
| 39 | CommitDict, |
| 40 | CommitRecord, |
| 41 | SnapshotDict, |
| 42 | SnapshotRecord, |
| 43 | TagDict, |
| 44 | TagRecord, |
| 45 | get_all_tags, |
| 46 | get_tags_for_commit, |
| 47 | read_commit, |
| 48 | read_snapshot, |
| 49 | write_commit, |
| 50 | write_snapshot, |
| 51 | write_tag, |
| 52 | ) |
| 53 | |
| 54 | logger = logging.getLogger(__name__) |
| 55 | |
| 56 | |
| 57 | # --------------------------------------------------------------------------- |
| 58 | # Wire-format TypedDicts |
| 59 | # --------------------------------------------------------------------------- |
| 60 | |
| 61 | |
| 62 | class _ObjectPayloadBase(TypedDict): |
| 63 | """Required fields for every object payload transmitted over MWP.""" |
| 64 | |
| 65 | object_id: str |
| 66 | content: bytes |
| 67 | |
| 68 | |
| 69 | class ObjectPayload(_ObjectPayloadBase, total=False): |
| 70 | """A single content-addressed blob with encoding metadata for MWP transport. |
| 71 | |
| 72 | Required fields (always present): |
| 73 | object_id: Content-addressed SHA-256 identifier (``sha256:<hex>``). |
| 74 | content: Raw or encoded bytes — see *encoding*. |
| 75 | |
| 76 | Optional fields (omit for ``"raw"`` with no base): |
| 77 | path: Repository path of this object; used by the server to look |
| 78 | up delta base candidates for the next push. |
| 79 | encoding: ``"raw"`` (default) | ``"zlib"`` | ``"delta+zlib"``. |
| 80 | base_id: Base object ID for ``"delta+zlib"`` encoding. |
| 81 | """ |
| 82 | |
| 83 | path: str |
| 84 | encoding: str |
| 85 | base_id: str |
| 86 | |
| 87 | |
| 88 | class WireTag(TypedDict): |
| 89 | """A tag record serialised for wire transfer inside a :class:`PackBundle`.""" |
| 90 | |
| 91 | tag_id: str |
| 92 | repo_id: str |
| 93 | commit_id: str |
| 94 | tag: str |
| 95 | created_at: str |
| 96 | |
| 97 | |
| 98 | class PackBundle(TypedDict, total=False): |
| 99 | """The unit of exchange between the Muse CLI and a remote. |
| 100 | |
| 101 | All fields are optional so that partial bundles (fetch-only, objects-only) |
| 102 | are valid wire messages. Callers check for presence before consuming. |
| 103 | """ |
| 104 | |
| 105 | commits: list[CommitDict] |
| 106 | snapshots: list[SnapshotDict] |
| 107 | objects: list[ObjectPayload] |
| 108 | #: Tags attached to any commit included in this bundle. |
| 109 | tags: list[WireTag] |
| 110 | #: Remote branch heads at the time the bundle was produced. |
| 111 | branch_heads: BranchHeads |
| 112 | |
| 113 | |
| 114 | class RemoteInfo(TypedDict, total=False): |
| 115 | """Repository metadata returned by ``GET {url}/refs``.""" |
| 116 | |
| 117 | repo_id: str # always present |
| 118 | domain: str # always present |
| 119 | #: Maps branch name → commit ID for every branch on the remote. |
| 120 | branch_heads: BranchHeads # always present |
| 121 | default_branch: str # always present |
| 122 | #: Optional Worker URL advertised by the server. When present, the CLI |
| 123 | #: routes POST /push/object-pack calls here instead of the primary URL. |
| 124 | pack_origin: str |
| 125 | |
| 126 | |
| 127 | class PushResult(TypedDict): |
| 128 | """Server response after a push attempt.""" |
| 129 | |
| 130 | ok: bool |
| 131 | message: str |
| 132 | #: Updated branch heads on the remote after the push (if successful). |
| 133 | branch_heads: BranchHeads |
| 134 | |
| 135 | |
| 136 | class FetchRequest(TypedDict, total=False): |
| 137 | """Body of ``POST {url}/fetch`` — negotiates which commits to transfer. |
| 138 | |
| 139 | ``want`` lists commit IDs the client wants to receive. |
| 140 | ``have`` lists commit IDs already present locally, allowing the server |
| 141 | to send only the commits the client lacks (delta negotiation). |
| 142 | """ |
| 143 | |
| 144 | want: list[str] |
| 145 | have: list[str] |
| 146 | |
| 147 | |
| 148 | class ApplyResult(TypedDict): |
| 149 | """Counts returned by :func:`apply_pack` describing what was written. |
| 150 | |
| 151 | ``objects_skipped`` counts blobs already present in the store (not |
| 152 | rewritten, idempotent). All other counts reflect *new* writes only. |
| 153 | """ |
| 154 | |
| 155 | commits_written: int |
| 156 | snapshots_written: int |
| 157 | objects_written: int |
| 158 | objects_skipped: int |
| 159 | |
| 160 | |
| 161 | class ObjectsChunkResponse(TypedDict): |
| 162 | """Response from ``POST {url}/push/objects`` — one chunk of a chunked push. |
| 163 | |
| 164 | Returned by both :class:`~muse.core.transport.HttpTransport` and |
| 165 | :class:`~muse.core.transport.LocalFileTransport` after pre-uploading a |
| 166 | batch of content-addressed objects. |
| 167 | |
| 168 | ``stored`` — objects written to storage in this call. |
| 169 | ``skipped`` — objects already present on the remote (idempotent no-ops). |
| 170 | """ |
| 171 | |
| 172 | stored: int |
| 173 | skipped: int |
| 174 | |
| 175 | |
| 176 | # --------------------------------------------------------------------------- |
| 177 | # Pack building |
| 178 | # --------------------------------------------------------------------------- |
| 179 | |
| 180 | |
| 181 | class _WalkResult(TypedDict): |
| 182 | """Cached result of a BFS commit-graph walk. |
| 183 | |
| 184 | Produced once by :func:`walk_commits` and consumed by both |
| 185 | :func:`collect_object_ids` (Phase 1 — send IDs to filter-objects) and |
| 186 | :func:`build_pack` (Phase 2 — load blobs for missing objects). |
| 187 | |
| 188 | Sharing this avoids two identical BFS traversals per push: the first to |
| 189 | gather object IDs for ``POST /filter-objects``, and the second to actually |
| 190 | load blobs and assemble the pack bundle. |
| 191 | |
| 192 | ``missing_snapshots`` is populated by :func:`walk_commits` with the |
| 193 | snapshot_ids of any reachable commit whose snapshot file is absent from |
| 194 | the local store. Callers should surface this to the user before pushing — |
| 195 | a pack that contains a commit but not its snapshot creates a dangling |
| 196 | reference on the remote. |
| 197 | """ |
| 198 | |
| 199 | commits: list[CommitRecord] |
| 200 | snapshot_ids: set[str] |
| 201 | all_object_ids: list[str] # sorted, deduplicated |
| 202 | oid_to_path: dict[str, str] # object_id → repository path (from snapshot manifests) |
| 203 | missing_snapshots: set[str] # snapshot_ids present in commits but absent on disk |
| 204 | |
| 205 | |
| 206 | def walk_commits( |
| 207 | repo_root: pathlib.Path, |
| 208 | commit_ids: list[str], |
| 209 | *, |
| 210 | have: list[str] | None = None, |
| 211 | ) -> _WalkResult: |
| 212 | """BFS-walk the commit graph from *commit_ids*, stopping at *have*. |
| 213 | |
| 214 | Returns a :class:`_WalkResult` that can be passed to both |
| 215 | :func:`collect_object_ids_from_walk` and :func:`build_pack_from_walk` |
| 216 | to avoid repeating the traversal. |
| 217 | |
| 218 | This is the **single source of truth** for what goes into a push bundle. |
| 219 | Callers that need both the object ID list (for ``POST /filter-objects``) |
| 220 | and the full pack (for uploading blobs) should call this once and pass |
| 221 | the result to both downstream functions. |
| 222 | """ |
| 223 | have_set: set[str] = set(have or []) |
| 224 | commits_to_send: list[CommitRecord] = [] |
| 225 | seen: set[str] = set(have_set) |
| 226 | queue: collections.deque[str] = collections.deque( |
| 227 | cid for cid in commit_ids if cid not in seen |
| 228 | ) |
| 229 | |
| 230 | while queue: |
| 231 | cid = queue.popleft() |
| 232 | if cid in seen: |
| 233 | continue |
| 234 | seen.add(cid) |
| 235 | commit = read_commit(repo_root, cid) |
| 236 | if commit is None: |
| 237 | logger.warning("⚠️ walk_commits: commit %s not found — skipping", cid[:8]) |
| 238 | continue |
| 239 | commits_to_send.append(commit) |
| 240 | if commit.parent_commit_id and commit.parent_commit_id not in seen: |
| 241 | queue.append(commit.parent_commit_id) |
| 242 | if commit.parent2_commit_id and commit.parent2_commit_id not in seen: |
| 243 | queue.append(commit.parent2_commit_id) |
| 244 | |
| 245 | snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_send} |
| 246 | missing_snapshots: set[str] = set() |
| 247 | all_object_ids: set[str] = set() |
| 248 | oid_to_path: dict[str, str] = {} |
| 249 | for sid in snapshot_ids: |
| 250 | snap = read_snapshot(repo_root, sid) |
| 251 | if snap is not None: |
| 252 | all_object_ids.update(snap.manifest.values()) |
| 253 | # Build oid→path from each snapshot manifest (path → oid). |
| 254 | # Later snapshots win on collision — any path is fine for delta lookup. |
| 255 | for path, oid in snap.manifest.items(): |
| 256 | oid_to_path[oid] = path |
| 257 | else: |
| 258 | missing_snapshots.add(sid) |
| 259 | |
| 260 | if missing_snapshots: |
| 261 | for sid in sorted(missing_snapshots): |
| 262 | logger.warning( |
| 263 | "⚠️ walk_commits: snapshot %s is missing from the local store — " |
| 264 | "the commit(s) referencing it will be excluded from the pack. " |
| 265 | "Run `muse verify` to audit store integrity.", |
| 266 | sid[:8], |
| 267 | ) |
| 268 | |
| 269 | return _WalkResult( |
| 270 | commits=commits_to_send, |
| 271 | snapshot_ids=snapshot_ids, |
| 272 | all_object_ids=sorted(all_object_ids), |
| 273 | oid_to_path=oid_to_path, |
| 274 | missing_snapshots=missing_snapshots, |
| 275 | ) |
| 276 | |
| 277 | |
| 278 | def stream_object_chunks( |
| 279 | repo_root: pathlib.Path, |
| 280 | object_ids: list[str], |
| 281 | chunk_size: int, |
| 282 | ) -> "collections.abc.Iterator[list[ObjectPayload]]": |
| 283 | """Yield blobs in chunks of *chunk_size* as they are read from disk. |
| 284 | |
| 285 | This is the hot path for ``muse push`` after ``POST /filter-objects``. |
| 286 | Instead of reading all blobs into RAM before uploading (peak RAM = all |
| 287 | missing objects), we read one chunk at a time and yield it immediately. |
| 288 | The caller can start the first upload while the second chunk is still |
| 289 | being assembled — reducing both peak memory and time-to-first-upload. |
| 290 | |
| 291 | Missing blobs are logged and skipped, consistent with :func:`build_pack`. |
| 292 | """ |
| 293 | import collections.abc # local to avoid circular at module level |
| 294 | |
| 295 | chunk: list[ObjectPayload] = [] |
| 296 | for oid in object_ids: |
| 297 | raw = read_object(repo_root, oid) |
| 298 | if raw is None: |
| 299 | logger.warning("⚠️ stream_object_chunks: blob %s absent — skipping", oid[:8]) |
| 300 | continue |
| 301 | chunk.append(ObjectPayload(object_id=oid, content=raw)) |
| 302 | if len(chunk) >= chunk_size: |
| 303 | yield chunk |
| 304 | chunk = [] |
| 305 | if chunk: |
| 306 | yield chunk |
| 307 | |
| 308 | |
| 309 | def collect_object_ids_from_walk(walk: _WalkResult) -> list[str]: |
| 310 | """Return the sorted object ID list from a pre-computed :func:`walk_commits` result. |
| 311 | |
| 312 | Zero disk I/O — the walk already read all snapshots. |
| 313 | """ |
| 314 | return walk["all_object_ids"] |
| 315 | |
| 316 | |
| 317 | def build_pack_from_walk( |
| 318 | repo_root: pathlib.Path, |
| 319 | walk: _WalkResult, |
| 320 | *, |
| 321 | only_objects: set[str] | None = None, |
| 322 | repo_id: str = "", |
| 323 | ) -> PackBundle: |
| 324 | """Assemble a :class:`PackBundle` from a pre-computed :func:`walk_commits` result. |
| 325 | |
| 326 | Avoids the second BFS traversal that :func:`build_pack` would otherwise |
| 327 | perform. Only reads blob bytes for objects in *only_objects* (or all |
| 328 | objects when *only_objects* is ``None``). |
| 329 | |
| 330 | This is the hot path for ``muse push`` after ``POST /filter-objects`` |
| 331 | narrows the upload set — we already have the commit+snapshot lists in |
| 332 | memory from the earlier :func:`walk_commits` call. |
| 333 | """ |
| 334 | missing_snapshots: set[str] = walk.get("missing_snapshots") or set() |
| 335 | all_object_ids_set: set[str] = set(walk["all_object_ids"]) |
| 336 | |
| 337 | # Hard failure on any missing snapshot — silently skipping would push |
| 338 | # commits without their snapshots, creating dangling references on the |
| 339 | # remote that can never be healed without rewriting history. |
| 340 | if missing_snapshots: |
| 341 | sample = sorted(missing_snapshots)[:3] |
| 342 | sample_str = ", ".join(s[:8] for s in sample) |
| 343 | raise ValueError( |
| 344 | f"Push aborted: {len(missing_snapshots)} snapshot(s) are missing from " |
| 345 | f"the local store but are required by commits being sent " |
| 346 | f"({sample_str}{'…' if len(missing_snapshots) > 3 else ''}). " |
| 347 | f"Run 'muse verify' to audit store integrity." |
| 348 | ) |
| 349 | |
| 350 | snapshot_dicts: list[SnapshotDict] = [] |
| 351 | for sid in sorted(walk["snapshot_ids"]): |
| 352 | snap = read_snapshot(repo_root, sid) |
| 353 | if snap is None: |
| 354 | # This should not happen — missing_snapshots guard above catches it. |
| 355 | raise ValueError( |
| 356 | f"Push aborted: snapshot {sid[:8]} is missing from the local store. " |
| 357 | f"Run 'muse verify' to audit store integrity." |
| 358 | ) |
| 359 | snapshot_dicts.append(snap.to_dict()) |
| 360 | |
| 361 | commits_to_send = list(walk["commits"]) |
| 362 | |
| 363 | candidate_ids = ( |
| 364 | all_object_ids_set & only_objects if only_objects is not None else all_object_ids_set |
| 365 | ) |
| 366 | |
| 367 | object_payloads: list[ObjectPayload] = [] |
| 368 | for oid in sorted(candidate_ids): |
| 369 | raw = read_object(repo_root, oid) |
| 370 | if raw is None: |
| 371 | logger.warning("⚠️ build_pack_from_walk: blob %s absent — skipping", oid[:8]) |
| 372 | continue |
| 373 | object_payloads.append(ObjectPayload(object_id=oid, content=raw)) |
| 374 | |
| 375 | sent_commit_ids = [c.commit_id for c in commits_to_send] |
| 376 | wire_tags = _tags_for_commits(repo_root, sent_commit_ids, repo_id) if repo_id else [] |
| 377 | |
| 378 | bundle: PackBundle = { |
| 379 | "commits": [c.to_dict() for c in commits_to_send], |
| 380 | "snapshots": snapshot_dicts, |
| 381 | "objects": object_payloads, |
| 382 | } |
| 383 | if wire_tags: |
| 384 | bundle["tags"] = wire_tags |
| 385 | |
| 386 | logger.info( |
| 387 | "✅ Built pack (from walk): %d commits, %d snapshots, %d objects, %d tags", |
| 388 | len(commits_to_send), |
| 389 | len(snapshot_dicts), |
| 390 | len(object_payloads), |
| 391 | len(wire_tags), |
| 392 | ) |
| 393 | return bundle |
| 394 | |
| 395 | |
| 396 | def _tags_for_commits( |
| 397 | repo_root: pathlib.Path, commit_ids: list[str], repo_id: str |
| 398 | ) -> list[WireTag]: |
| 399 | """Return all tags attached to *commit_ids* as serialisable :class:`WireTag` dicts.""" |
| 400 | seen_tag_ids: set[str] = set() |
| 401 | wire_tags: list[WireTag] = [] |
| 402 | for cid in commit_ids: |
| 403 | for tag in get_tags_for_commit(repo_root, repo_id, cid): |
| 404 | if tag.tag_id not in seen_tag_ids: |
| 405 | seen_tag_ids.add(tag.tag_id) |
| 406 | wire_tags.append(WireTag( |
| 407 | tag_id=tag.tag_id, |
| 408 | repo_id=tag.repo_id, |
| 409 | commit_id=tag.commit_id, |
| 410 | tag=tag.tag, |
| 411 | created_at=tag.created_at.isoformat(), |
| 412 | )) |
| 413 | return wire_tags |
| 414 | |
| 415 | |
| 416 | def build_pack( |
| 417 | repo_root: pathlib.Path, |
| 418 | commit_ids: list[str], |
| 419 | *, |
| 420 | have: list[str] | None = None, |
| 421 | only_objects: set[str] | None = None, |
| 422 | repo_id: str = "", |
| 423 | ) -> PackBundle: |
| 424 | """Assemble a :class:`PackBundle` from *commit_ids*, excluding commits in *have*. |
| 425 | |
| 426 | Performs a BFS walk of the commit graph from every ID in *commit_ids*, |
| 427 | stopping at any commit already in *have*. Collects all snapshot manifests |
| 428 | and object blobs reachable from the selected commits. |
| 429 | |
| 430 | Missing objects or snapshots are logged and skipped — the caller decides |
| 431 | whether that constitutes an error. |
| 432 | |
| 433 | Args: |
| 434 | repo_root: Root of the Muse repository. |
| 435 | commit_ids: Tip commit IDs to include (e.g. current branch HEAD). |
| 436 | have: Commit IDs already known to the receiver. The BFS stops |
| 437 | at these, reducing bundle size. Pass ``None`` or ``[]`` |
| 438 | to send the full history. |
| 439 | only_objects: When set, only include objects whose IDs are in this set. |
| 440 | Used after a ``POST /filter-objects`` negotiation so the |
| 441 | client only uploads objects the remote is missing. |
| 442 | repo_id: Repository UUID used to look up tags. When omitted, tags |
| 443 | are not included in the bundle. |
| 444 | |
| 445 | Returns: |
| 446 | A :class:`PackBundle` ready for serialisation and transfer. |
| 447 | """ |
| 448 | have_set: set[str] = set(have or []) |
| 449 | |
| 450 | # BFS walk from every tip, treating have_set as already-visited. |
| 451 | commits_to_send: list[CommitRecord] = [] |
| 452 | seen: set[str] = set(have_set) |
| 453 | queue: collections.deque[str] = collections.deque( |
| 454 | cid for cid in commit_ids if cid not in seen |
| 455 | ) |
| 456 | |
| 457 | while queue: |
| 458 | cid = queue.popleft() |
| 459 | if cid in seen: |
| 460 | continue |
| 461 | seen.add(cid) |
| 462 | commit = read_commit(repo_root, cid) |
| 463 | if commit is None: |
| 464 | logger.warning("⚠️ build_pack: commit %s not found — skipping", cid[:8]) |
| 465 | continue |
| 466 | commits_to_send.append(commit) |
| 467 | if commit.parent_commit_id and commit.parent_commit_id not in seen: |
| 468 | queue.append(commit.parent_commit_id) |
| 469 | if commit.parent2_commit_id and commit.parent2_commit_id not in seen: |
| 470 | queue.append(commit.parent2_commit_id) |
| 471 | |
| 472 | # Unique snapshot IDs referenced by selected commits. |
| 473 | snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_send} |
| 474 | |
| 475 | snapshot_dicts: list[SnapshotDict] = [] |
| 476 | all_object_ids: set[str] = set() |
| 477 | for sid in sorted(snapshot_ids): |
| 478 | snap = read_snapshot(repo_root, sid) |
| 479 | if snap is None: |
| 480 | # Hard failure — a commit whose snapshot is absent locally cannot |
| 481 | # be pushed safely. Silently skipping would create a dangling |
| 482 | # commit reference on the remote: the commit lands, the snapshot |
| 483 | # never does, and every subsequent pull restores the commit but |
| 484 | # not its snapshot. Fail loudly so the user knows the store is |
| 485 | # corrupt and can take corrective action before data is lost. |
| 486 | raise ValueError( |
| 487 | f"Push aborted: snapshot {sid[:8]} is missing from the local store " |
| 488 | f"but is required by a commit being sent. " |
| 489 | f"Run 'muse verify' to audit store integrity." |
| 490 | ) |
| 491 | snapshot_dicts.append(snap.to_dict()) |
| 492 | all_object_ids.update(snap.manifest.values()) |
| 493 | |
| 494 | # When only_objects is provided (post filter-objects negotiation) skip |
| 495 | # any object the remote already has — only transmit the missing delta. |
| 496 | candidate_ids = ( |
| 497 | all_object_ids & only_objects if only_objects is not None else all_object_ids |
| 498 | ) |
| 499 | |
| 500 | object_payloads: list[ObjectPayload] = [] |
| 501 | for oid in sorted(candidate_ids): |
| 502 | raw = read_object(repo_root, oid) |
| 503 | if raw is None: |
| 504 | logger.warning("⚠️ build_pack: blob %s absent from store — skipping", oid[:8]) |
| 505 | continue |
| 506 | object_payloads.append(ObjectPayload(object_id=oid, content=raw)) |
| 507 | |
| 508 | sent_commit_ids = [c.commit_id for c in commits_to_send] |
| 509 | wire_tags = _tags_for_commits(repo_root, sent_commit_ids, repo_id) if repo_id else [] |
| 510 | |
| 511 | bundle: PackBundle = { |
| 512 | "commits": [c.to_dict() for c in commits_to_send], |
| 513 | "snapshots": snapshot_dicts, |
| 514 | "objects": object_payloads, |
| 515 | } |
| 516 | if wire_tags: |
| 517 | bundle["tags"] = wire_tags |
| 518 | |
| 519 | logger.info( |
| 520 | "✅ Built pack: %d commits, %d snapshots, %d objects, %d tags", |
| 521 | len(commits_to_send), |
| 522 | len(snapshot_dicts), |
| 523 | len(object_payloads), |
| 524 | len(wire_tags), |
| 525 | ) |
| 526 | return bundle |
| 527 | |
| 528 | |
| 529 | # --------------------------------------------------------------------------- |
| 530 | # Object ID collection — for pre-push deduplication negotiation |
| 531 | # --------------------------------------------------------------------------- |
| 532 | |
| 533 | |
| 534 | def collect_object_ids( |
| 535 | repo_root: pathlib.Path, |
| 536 | commit_ids: list[str], |
| 537 | *, |
| 538 | have: list[str] | None = None, |
| 539 | ) -> list[str]: |
| 540 | """Return all object IDs reachable from *commit_ids*, excluding *have*. |
| 541 | |
| 542 | Identical BFS walk to :func:`build_pack` but without reading object bytes. |
| 543 | Used by ``muse push`` to call ``POST /filter-objects`` before building the |
| 544 | full pack — the client discovers which objects are missing on the remote |
| 545 | and then calls :func:`build_pack` with ``only_objects`` set to that subset. |
| 546 | This avoids loading any blob content until we know it is actually needed. |
| 547 | |
| 548 | Args: |
| 549 | repo_root: Root of the Muse repository. |
| 550 | commit_ids: Tip commit IDs to examine. |
| 551 | have: Commit IDs already known to the receiver (BFS stops here). |
| 552 | |
| 553 | Returns: |
| 554 | Sorted list of object IDs reachable from the delta. |
| 555 | """ |
| 556 | have_set: set[str] = set(have or []) |
| 557 | commits_to_examine: list[CommitRecord] = [] |
| 558 | seen: set[str] = set(have_set) |
| 559 | queue: collections.deque[str] = collections.deque( |
| 560 | cid for cid in commit_ids if cid not in seen |
| 561 | ) |
| 562 | while queue: |
| 563 | cid = queue.popleft() |
| 564 | if cid in seen: |
| 565 | continue |
| 566 | seen.add(cid) |
| 567 | commit = read_commit(repo_root, cid) |
| 568 | if commit is None: |
| 569 | continue |
| 570 | commits_to_examine.append(commit) |
| 571 | if commit.parent_commit_id and commit.parent_commit_id not in seen: |
| 572 | queue.append(commit.parent_commit_id) |
| 573 | if commit.parent2_commit_id and commit.parent2_commit_id not in seen: |
| 574 | queue.append(commit.parent2_commit_id) |
| 575 | |
| 576 | snapshot_ids: set[str] = {c.snapshot_id for c in commits_to_examine} |
| 577 | all_object_ids: set[str] = set() |
| 578 | for sid in snapshot_ids: |
| 579 | snap = read_snapshot(repo_root, sid) |
| 580 | if snap is not None: |
| 581 | all_object_ids.update(snap.manifest.values()) |
| 582 | |
| 583 | return sorted(all_object_ids) |
| 584 | |
| 585 | |
| 586 | # --------------------------------------------------------------------------- |
| 587 | # Pack applying |
| 588 | # --------------------------------------------------------------------------- |
| 589 | |
| 590 | |
| 591 | def apply_pack(repo_root: pathlib.Path, bundle: PackBundle) -> ApplyResult: |
| 592 | """Write the contents of *bundle* into a local ``.muse/`` directory. |
| 593 | |
| 594 | Writes in dependency order: objects first (blobs), then snapshots (which |
| 595 | reference object IDs), then commits (which reference snapshot IDs). All |
| 596 | writes are idempotent — already-present items are silently skipped. |
| 597 | |
| 598 | Args: |
| 599 | repo_root: Root of the Muse repository to write into. |
| 600 | bundle: :class:`PackBundle` received from the remote. |
| 601 | |
| 602 | Returns: |
| 603 | :class:`ApplyResult` with counts of newly written and skipped items. |
| 604 | """ |
| 605 | objects_written = 0 |
| 606 | objects_skipped = 0 |
| 607 | snapshots_written = 0 |
| 608 | commits_written = 0 |
| 609 | |
| 610 | raw_objects = bundle.get("objects") or [] |
| 611 | raw_snapshots = bundle.get("snapshots") or [] |
| 612 | raw_commits = bundle.get("commits") or [] |
| 613 | |
| 614 | # Pack-bomb guard: cap the total number of items accepted per call. |
| 615 | # A legitimate push carries at most tens of thousands of objects per chunk; |
| 616 | # a pack claiming millions is an adversarial input. |
| 617 | total_items = len(raw_objects) + len(raw_snapshots) + len(raw_commits) |
| 618 | if total_items > MAX_PACK_OBJECTS: |
| 619 | raise ValueError( |
| 620 | f"Pack rejected: {total_items:,} total items exceeds the " |
| 621 | f"{MAX_PACK_OBJECTS:,} item limit per apply_pack call. " |
| 622 | "Split the pack into smaller chunks." |
| 623 | ) |
| 624 | |
| 625 | # Deduplicate object IDs before the write loop. A malicious or buggy |
| 626 | # sender may repeat the same object_id N times, forcing N sha256 hashes |
| 627 | # before the "already exists" short-circuit returns False. |
| 628 | seen_object_ids: set[str] = set() |
| 629 | |
| 630 | for obj in raw_objects: |
| 631 | oid = obj.get("object_id", "") |
| 632 | raw = obj.get("content", b"") |
| 633 | if not oid or not isinstance(raw, bytes): |
| 634 | logger.warning("⚠️ apply_pack: blob entry missing fields — skipped") |
| 635 | continue |
| 636 | if oid in seen_object_ids: |
| 637 | logger.debug("⚠️ apply_pack: duplicate object_id %s — skipped", oid[:8]) |
| 638 | objects_skipped += 1 |
| 639 | continue |
| 640 | seen_object_ids.add(oid) |
| 641 | # Per-object size cap: check before calling write_object to avoid |
| 642 | # hashing a huge payload that will be rejected anyway. |
| 643 | if len(raw) > MAX_OBJECT_WRITE_BYTES: |
| 644 | logger.warning( |
| 645 | "⚠️ apply_pack: object %s is %d bytes, exceeding %d MiB limit — skipped", |
| 646 | oid[:8], len(raw), MAX_OBJECT_WRITE_BYTES // (1024 * 1024), |
| 647 | ) |
| 648 | continue |
| 649 | try: |
| 650 | if write_object(repo_root, oid, raw): |
| 651 | objects_written += 1 |
| 652 | else: |
| 653 | objects_skipped += 1 |
| 654 | except ValueError as exc: |
| 655 | # Malicious object IDs (non-hex, path traversal) or content/ID |
| 656 | # mismatch — log and skip rather than aborting the entire pack. |
| 657 | logger.warning("⚠️ apply_pack: malformed object entry — skipped: %s", exc) |
| 658 | |
| 659 | for snap_dict in raw_snapshots: |
| 660 | try: |
| 661 | snap = SnapshotRecord.from_dict(snap_dict) |
| 662 | # Guard against zip-slip: manifest keys are stored as-is and later |
| 663 | # used to construct checkout paths. A malicious bundle could inject |
| 664 | # keys like "../../etc/cron.d/evil". We validate all keys here — |
| 665 | # before writing — so no traversal path ever enters the store. |
| 666 | for key in snap.manifest: |
| 667 | validate_workspace_path(key) |
| 668 | # Manifest values are object IDs — validate they are safe hex strings. |
| 669 | for oid in snap.manifest.values(): |
| 670 | validate_object_id(oid) |
| 671 | is_new = read_snapshot(repo_root, snap.snapshot_id) is None |
| 672 | write_snapshot(repo_root, snap) |
| 673 | if is_new: |
| 674 | snapshots_written += 1 |
| 675 | except (KeyError, ValueError) as exc: |
| 676 | logger.warning("⚠️ apply_pack: malformed snapshot — skipped: %s", exc) |
| 677 | |
| 678 | for commit_dict in raw_commits: |
| 679 | try: |
| 680 | commit = CommitRecord.from_dict(commit_dict) |
| 681 | # from_dict is designed for trusted, typed callers and does not |
| 682 | # validate essential fields — guard them here before the commit |
| 683 | # reaches write_commit (which constructs paths from commit_id). |
| 684 | if not commit.commit_id: |
| 685 | logger.warning("⚠️ apply_pack: commit missing commit_id — skipped") |
| 686 | continue |
| 687 | if not commit.snapshot_id: |
| 688 | logger.warning( |
| 689 | "⚠️ apply_pack: commit %s missing snapshot_id — skipped", |
| 690 | commit.commit_id[:8], |
| 691 | ) |
| 692 | continue |
| 693 | is_new = read_commit(repo_root, commit.commit_id) is None |
| 694 | write_commit(repo_root, commit) |
| 695 | if is_new: |
| 696 | commits_written += 1 |
| 697 | except OSError as exc: |
| 698 | # write_commit raises OSError("Store integrity violation") when the |
| 699 | # existing commit file contains a DIFFERENT commit_id than the one |
| 700 | # being written — indicating the local store has been tampered with. |
| 701 | # Log CRITICAL and skip rather than crashing the entire apply_pack. |
| 702 | logger.critical( |
| 703 | "❌ apply_pack: store integrity violation for commit — skipped: %s", exc |
| 704 | ) |
| 705 | except (KeyError, ValueError, TypeError) as exc: |
| 706 | logger.warning("⚠️ apply_pack: malformed commit — skipped: %s", exc) |
| 707 | |
| 708 | for wire_tag in bundle.get("tags") or []: |
| 709 | try: |
| 710 | tag_record = TagRecord.from_dict(TagDict( |
| 711 | tag_id=wire_tag["tag_id"], |
| 712 | repo_id=wire_tag["repo_id"], |
| 713 | commit_id=wire_tag["commit_id"], |
| 714 | tag=wire_tag["tag"], |
| 715 | created_at=wire_tag["created_at"], |
| 716 | )) |
| 717 | write_tag(repo_root, tag_record) |
| 718 | except (KeyError, ValueError) as exc: |
| 719 | logger.warning("⚠️ apply_pack: malformed tag — skipped: %s", exc) |
| 720 | |
| 721 | logger.info( |
| 722 | "✅ Applied pack: %d new blobs, %d new snapshots, %d new commits (%d blobs skipped)", |
| 723 | objects_written, |
| 724 | snapshots_written, |
| 725 | commits_written, |
| 726 | objects_skipped, |
| 727 | ) |
| 728 | return ApplyResult( |
| 729 | commits_written=commits_written, |
| 730 | snapshots_written=snapshots_written, |
| 731 | objects_written=objects_written, |
| 732 | objects_skipped=objects_skipped, |
| 733 | ) |
File History
2 commits
sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6
chore(tests): add docstring to tests/__init__.py so rc14 tr…
Human
33 days ago
sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62
baseline: rc14 re-baseline after rc3 store corruption recovery
Human
patch
33 days ago