"""Garbage collection — prune unreachable objects from the object store. Muse uses a content-addressed object store: every file snapshot is stored as a SHA-256-addressed blob under ``.muse/objects/``. Over time, after branch deletions, rebases, and abandoned experiments, objects that are no longer reachable from any live commit accumulate. This module identifies and removes them. Reachability ------------ An object is *reachable* if it can be reached by following the graph from any live ref (branch HEAD, tag, or the current HEAD): branch HEAD → CommitRecord → SnapshotRecord → manifest → object SHA-256 Any object not in the reachable set is *loose garbage* and is safe to delete. ``--full`` mode --------------- By default ``muse gc`` only prunes file-content blobs (``.muse/objects/``). With ``--full`` it also prunes: - Orphaned commit records (``.muse/commits/*.msgpack``) — commits not reachable from any live branch ref or tag by following ``parent_commit_id`` links. - Orphaned snapshot manifests (``.muse/snapshots/*.msgpack``) — snapshots not referenced by any reachable commit. This mirrors ``git prune`` / ``git gc``, which removes all unreachable loose objects regardless of type. Safety ------ The GC walk is always performed **before** any deletion. The ``dry_run`` option shows what *would* be deleted without touching the store, making it safe to run frequently in CI or by agents to estimate bloat. A ``grace_period_seconds`` guard (default 30 s) prevents deleting objects that were written within the last N seconds. This closes the TOCTOU window where a concurrent ``muse commit`` has written a blob to the object store but not yet created the commit record — without the grace period, GC would see the blob as unreachable and delete it mid-commit, causing data loss. Symlink safety -------------- Every prefix directory and object file is checked with ``is_symlink()`` before being treated as a real object. A crafted symlink inside ``.muse/objects/`` cannot cause GC to read or delete files outside the repository. Return value ------------ ``GcResult`` is a typed dataclass with integer counts and the list of collected IDs. The CLI command renders it; the plumbing layer can expose it as JSON. """ from __future__ import annotations import json import logging import pathlib import time from collections import deque from dataclasses import dataclass, field import msgpack from muse.core.store import get_all_commits, read_snapshot logger = logging.getLogger(__name__) # Maximum size of stash.json we will read during the reachability walk. # This mirrors the cap in muse/cli/commands/stash.py. _MAX_STASH_BYTES: int = 64 * 1024 * 1024 # 64 MiB # Default number of seconds to protect recently-written objects from GC. # See the "Safety" section of the module docstring. _DEFAULT_GRACE_PERIOD_SECONDS: int = 30 # --------------------------------------------------------------------------- # Result type # --------------------------------------------------------------------------- @dataclass class GcResult: """Statistics from one garbage-collection pass.""" # Blob store (.muse/objects/) reachable_count: int = 0 collected_count: int = 0 collected_bytes: int = 0 # Commit store (.muse/commits/) — populated only with --full commits_reachable: int = 0 commits_collected: int = 0 commits_collected_bytes: int = 0 # Snapshot store (.muse/snapshots/) — populated only with --full snapshots_reachable: int = 0 snapshots_collected: int = 0 snapshots_collected_bytes: int = 0 elapsed_seconds: float = 0.0 grace_period_seconds: int = _DEFAULT_GRACE_PERIOD_SECONDS collected_ids: list[str] = field(default_factory=list) dry_run: bool = False full: bool = False # --------------------------------------------------------------------------- # Reachability walk — blobs (conservative: all commits in store) # --------------------------------------------------------------------------- def _collect_reachable_objects( repo_root: pathlib.Path, ) -> set[str]: """Return the set of all object SHA-256 IDs reachable from any live ref. Uses the conservative walk (all commits in the store) so that orphaned commits don't cause their blobs to be deleted when ``--full`` is not set. Sources walked: - Every snapshot file stored in ``.muse/snapshots/`` — raw manifest values are read without hash verification so that objects are retained even when a commit's ``snapshot_id`` field is corrupt (bit-flip or earlier store bug). This is conservative: orphaned snapshots (not referenced by any commit) keep their objects alive until they themselves are pruned by ``muse gc --full``. - Every entry in ``.muse/stash.json`` (stashed work writes objects before the stash entry is committed, so they would otherwise be GCed). The stash file is protected by a symlink guard and a size cap to match the defences in ``muse stash`` itself. """ reachable: set[str] = set() # ── All snapshot files (raw, no hash verification) ──────────────────────── # Reading raw manifests (bypassing _verify_snapshot_id) is intentional here: # GC must never delete an object that is referenced in ANY snapshot on disk, # even a corrupt or orphaned one. If we relied on read_snapshot (which # verifies the hash), a commit with a corrupt snapshot_id field pointing to a # non-existent snapshot would cause its objects to be silently GCed. snapshots_dir = repo_root / ".muse" / "snapshots" if snapshots_dir.is_dir(): for snap_path in snapshots_dir.glob("*.msgpack"): # Symlink guard: skip symlinks to prevent traversal attacks. if snap_path.is_symlink(): logger.warning("⚠️ gc: skipping symlink snapshot file %s", snap_path.name) continue try: raw_data = msgpack.unpackb(snap_path.read_bytes(), raw=False) manifest = raw_data.get("manifest", {}) if isinstance(manifest, dict): for oid in manifest.values(): if isinstance(oid, str) and oid: reachable.add(oid) except Exception as exc: logger.debug("gc: could not read snapshot %s — skipped: %s", snap_path.name, exc) # ── Stash ───────────────────────────────────────────────────────────────── _collect_stash_objects(repo_root, reachable) return reachable def _collect_stash_objects(repo_root: pathlib.Path, reachable: set[str]) -> None: """Add object IDs from stash.json into *reachable* in-place.""" stash_path = repo_root / ".muse" / "stash.json" if not stash_path.exists(): return if stash_path.is_symlink(): logger.warning("⚠️ .muse/stash.json is a symlink — skipping during GC walk") return try: size = stash_path.stat().st_size if size > _MAX_STASH_BYTES: logger.warning( "⚠️ .muse/stash.json is %.1f MiB — exceeds cap; skipping stash walk", size / (1024 * 1024), ) return entries = json.loads(stash_path.read_text(encoding="utf-8")) if isinstance(entries, list): for entry in entries: if isinstance(entry, dict): manifest = entry.get("manifest") if isinstance(manifest, dict): for object_id in manifest.values(): if isinstance(object_id, str): reachable.add(object_id) except (OSError, json.JSONDecodeError) as exc: logger.warning("⚠️ Could not read stash.json during GC walk: %s", exc) # --------------------------------------------------------------------------- # Reachability walk — commits + snapshots (for --full) # --------------------------------------------------------------------------- def _collect_reachable_commits(repo_root: pathlib.Path) -> set[str]: """Return all commit IDs reachable from any live branch ref or tag. Walks ``parent_commit_id`` and ``parent2_commit_id`` links in each commit record (read as raw msgpack to survive any schema evolution). Corrupt or missing files are skipped with a warning — they are not added to the reachable set but are also not deleted (the grace period covers the window). """ reachable: set[str] = set() commits_dir = repo_root / ".muse" / "commits" if not commits_dir.exists(): return reachable # Collect all live tips: branch refs + tags tips: list[str] = [] refs_dir = repo_root / ".muse" / "refs" / "heads" if refs_dir.exists(): for ref_path in refs_dir.rglob("*"): if ref_path.is_file() and not ref_path.is_symlink(): tip = ref_path.read_text().strip() if tip: tips.append(tip) tags_dir = repo_root / ".muse" / "tags" if tags_dir.exists(): for tag_path in tags_dir.glob("*.msgpack"): if tag_path.is_symlink() or not tag_path.is_file(): continue try: tag_data = msgpack.unpackb(tag_path.read_bytes(), raw=False) cid = tag_data.get("commit_id") or tag_data.get("target") if cid: tips.append(cid) except Exception: pass # BFS through parent links queue: deque[str] = deque(tips) while queue: cid = queue.popleft() if cid in reachable: continue reachable.add(cid) commit_path = commits_dir / f"{cid}.msgpack" if not commit_path.exists() or commit_path.is_symlink(): continue try: data = msgpack.unpackb(commit_path.read_bytes(), raw=False) except Exception as exc: logger.warning("⚠️ Could not read commit %s during GC walk: %s", cid[:12], exc) continue p1 = data.get("parent_commit_id") p2 = data.get("parent2_commit_id") if p1: queue.append(p1) if p2: queue.append(p2) return reachable def _collect_reachable_snapshots( repo_root: pathlib.Path, reachable_commits: set[str], ) -> tuple[set[str], set[str]]: """Return ``(reachable_snapshot_ids, reachable_object_ids)`` from reachable commits. Also folds in any snapshot IDs referenced by stash entries. """ reachable_snaps: set[str] = set() reachable_objs: set[str] = set() commits_dir = repo_root / ".muse" / "commits" for cid in reachable_commits: commit_path = commits_dir / f"{cid}.msgpack" if not commit_path.exists() or commit_path.is_symlink(): continue try: data = msgpack.unpackb(commit_path.read_bytes(), raw=False) except Exception: continue snap_id = data.get("snapshot_id") if not snap_id: continue reachable_snaps.add(snap_id) snap = read_snapshot(repo_root, snap_id) if snap is not None: for object_id in snap.manifest.values(): reachable_objs.add(object_id) else: # read_snapshot returned None — the snapshot file may be corrupt # (hash verification failed) but still contains valid object IDs. # Read the raw manifest without verification to retain its objects: # it is always safer to keep objects than to silently lose them. snap_path = repo_root / ".muse" / "snapshots" / f"{snap_id}.msgpack" if snap_path.is_file() and not snap_path.is_symlink(): try: raw_snap = msgpack.unpackb(snap_path.read_bytes(), raw=False) manifest = raw_snap.get("manifest", {}) if isinstance(manifest, dict): for oid in manifest.values(): if isinstance(oid, str) and oid: reachable_objs.add(oid) logger.critical( "❌ gc: snapshot %s failed hash verification — " "its objects were retained conservatively. " "Run `muse verify-pack` to audit the store.", snap_id[:8], ) except Exception as exc: logger.critical( "❌ gc: could not read corrupt snapshot %s — " "some objects may have been lost: %s", snap_id[:8], exc, ) # Stash snapshot IDs _collect_stash_objects(repo_root, reachable_objs) return reachable_snaps, reachable_objs # --------------------------------------------------------------------------- # Store enumeration helpers # --------------------------------------------------------------------------- _HEX_CHARS: frozenset[str] = frozenset("0123456789abcdef") def _is_hex(s: str) -> bool: """Return True iff every character of *s* is a lowercase hex digit.""" return bool(s) and all(c in _HEX_CHARS for c in s) def _list_stored_objects( repo_root: pathlib.Path, *, grace_period_seconds: int = _DEFAULT_GRACE_PERIOD_SECONDS, ) -> list[tuple[str, pathlib.Path]]: """Return ``(object_id, path)`` for every valid SHA-256 object in the store. Only **real files** (not symlinks) whose name + parent prefix form a 64-char lowercase hex string are included. Stray files (editor temporaries, macOS ``.DS_Store``, etc.) are silently skipped. Symlink guard ~~~~~~~~~~~~~ Both prefix directories and object files are rejected if they are symlinks. Without this guard a crafted symlink inside ``.muse/objects/`` could cause GC to ``unlink()`` a file anywhere on the filesystem — a silent data destruction vector. Grace period ~~~~~~~~~~~~ Objects whose ``mtime`` is within *grace_period_seconds* seconds of now are excluded from the returned list even if they are not currently reachable. This closes the TOCTOU window where a concurrent ``muse commit`` has written a blob but not yet created the commit record. Args: repo_root: Repository root (contains ``.muse/``). grace_period_seconds: Number of seconds before an object becomes eligible for collection. Default: 30 s. """ objects_dir = repo_root / ".muse" / "objects" if not objects_dir.exists(): return [] cutoff = time.time() - grace_period_seconds pairs: list[tuple[str, pathlib.Path]] = [] for prefix_dir in objects_dir.iterdir(): prefix = prefix_dir.name # Reject symlinks — a symlink-to-dir would pass is_dir() but must not. if prefix_dir.is_symlink() or not prefix_dir.is_dir(): continue if len(prefix) != 2 or not _is_hex(prefix): continue for obj_file in prefix_dir.iterdir(): rest = obj_file.name # Reject symlinks — unlink()ing a symlink-to-file outside the repo # would silently delete an unrelated file. if obj_file.is_symlink() or not obj_file.is_file(): continue if len(rest) != 62 or not _is_hex(rest): continue # Grace period: protect objects written in the last N seconds. try: if obj_file.stat().st_mtime > cutoff: continue except OSError: # File disappeared between iterdir() and stat() — skip. continue object_id = prefix + rest pairs.append((object_id, obj_file)) return pairs def _list_stored_msgpack( directory: pathlib.Path, *, grace_period_seconds: int = _DEFAULT_GRACE_PERIOD_SECONDS, ) -> list[tuple[str, pathlib.Path]]: """Return ``(stem, path)`` for every non-symlink ``.msgpack`` file in *directory*. Applies the same grace-period and symlink guards as ``_list_stored_objects``. """ if not directory.exists(): return [] cutoff = time.time() - grace_period_seconds pairs: list[tuple[str, pathlib.Path]] = [] for p in directory.glob("*.msgpack"): if p.is_symlink() or not p.is_file(): continue try: if p.stat().st_mtime > cutoff: continue except OSError: continue pairs.append((p.stem, p)) return pairs # --------------------------------------------------------------------------- # Public API # --------------------------------------------------------------------------- def run_gc( repo_root: pathlib.Path, *, dry_run: bool = False, grace_period_seconds: int = _DEFAULT_GRACE_PERIOD_SECONDS, full: bool = False, ) -> GcResult: """Prune unreachable objects from the Muse object store. Args: repo_root: Root of the Muse repository (``.muse/`` lives here). dry_run: When ``True``, report what *would* be deleted without actually removing anything. grace_period_seconds: Objects written within the last N seconds are never deleted, even if currently unreachable. Protects concurrent ``muse commit`` operations. Default: 30 s. full: When ``True``, also prune orphaned commit records (``.muse/commits/``) and snapshot manifests (``.muse/snapshots/``), mirroring ``git prune``. Returns: A ``GcResult`` with counts and the list of collected object IDs. """ t0 = time.monotonic() result = GcResult(dry_run=dry_run, grace_period_seconds=grace_period_seconds, full=full) if full: # --full: tight reachability — only objects reachable from live refs. reachable_commits = _collect_reachable_commits(repo_root) reachable_snaps, reachable_objs = _collect_reachable_snapshots( repo_root, reachable_commits ) result.commits_reachable = len(reachable_commits) result.snapshots_reachable = len(reachable_snaps) result.reachable_count = len(reachable_objs) # Prune orphaned commits commits_dir = repo_root / ".muse" / "commits" for cid, commit_path in _list_stored_msgpack( commits_dir, grace_period_seconds=grace_period_seconds ): if cid not in reachable_commits: try: size = commit_path.stat().st_size except OSError: size = 0 result.commits_collected += 1 result.commits_collected_bytes += size if not dry_run: try: commit_path.unlink() except OSError as exc: logger.warning( "⚠️ Could not remove commit %s: %s", cid[:12], exc ) # Prune orphaned snapshots snapshots_dir = repo_root / ".muse" / "snapshots" for snap_id, snap_path in _list_stored_msgpack( snapshots_dir, grace_period_seconds=grace_period_seconds ): if snap_id not in reachable_snaps: try: size = snap_path.stat().st_size except OSError: size = 0 result.snapshots_collected += 1 result.snapshots_collected_bytes += size if not dry_run: try: snap_path.unlink() except OSError as exc: logger.warning( "⚠️ Could not remove snapshot %s: %s", snap_id[:12], exc ) # Prune unreachable blobs (tighter set: only from reachable commits) stored = _list_stored_objects(repo_root, grace_period_seconds=grace_period_seconds) for object_id, obj_path in stored: if object_id not in reachable_objs: try: size = obj_path.stat().st_size except OSError: size = 0 result.collected_ids.append(object_id) result.collected_bytes += size result.collected_count += 1 if not dry_run: try: obj_path.unlink() try: if not any(obj_path.parent.iterdir()): obj_path.parent.rmdir() except OSError: pass except OSError as exc: logger.warning( "⚠️ Could not remove object %s: %s", object_id[:12], exc ) else: # Default: conservative blob-only GC using all commits in the store. reachable = _collect_reachable_objects(repo_root) stored = _list_stored_objects(repo_root, grace_period_seconds=grace_period_seconds) result.reachable_count = len(reachable) for object_id, obj_path in stored: if object_id not in reachable: try: size = obj_path.stat().st_size except OSError: size = 0 result.collected_ids.append(object_id) result.collected_bytes += size result.collected_count += 1 if not dry_run: try: obj_path.unlink() try: if not any(obj_path.parent.iterdir()): obj_path.parent.rmdir() except OSError: pass except OSError as exc: logger.warning( "⚠️ Could not remove object %s: %s", object_id[:12], exc ) result.elapsed_seconds = time.monotonic() - t0 logger.info( "gc: %d reachable blobs, %d %s, full=%s, grace=%ds, %.3fs elapsed", result.reachable_count, result.collected_count, "would be removed" if dry_run else "removed", full, grace_period_seconds, result.elapsed_seconds, ) return result