backup.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | #!/usr/bin/env python3 |
| 2 | """Belt-and-suspenders backups for canonical muse/musehub repos (#185 Phase 5). |
| 3 | |
| 4 | Two independent mechanisms, on purpose — not either/or: |
| 5 | |
| 6 | - **Suspenders** (fast, frequent, local): APFS copy-on-write snapshots of |
| 7 | just ``.muse/`` (not the working tree). Near-instant, cheap enough to take |
| 8 | before any risky operation. |
| 9 | - **Belt** (verified, portable, survives an APFS-level disaster): a |
| 10 | `muse bundle` archive of every branch, immediately verified after |
| 11 | creation. |
| 12 | |
| 13 | Restore refuses to overwrite a canonical repo that currently passes |
| 14 | `muse verify` unless ``force=True`` — don't clobber a good copy by accident. |
| 15 | """ |
| 16 | from __future__ import annotations |
| 17 | |
| 18 | import json |
| 19 | import shutil |
| 20 | import subprocess |
| 21 | import sys |
| 22 | from datetime import datetime, timezone |
| 23 | from pathlib import Path |
| 24 | |
| 25 | DEFAULT_CANONICAL_ROOTS = { |
| 26 | "muse": Path.home() / "ecosystem" / "muse", |
| 27 | "musehub": Path.home() / "ecosystem" / "musehub", |
| 28 | } |
| 29 | DEFAULT_BACKUP_BASE = Path.home() / "dev" / "backups" |
| 30 | DEFAULT_SNAPSHOT_KEEP = 20 |
| 31 | DEFAULT_BUNDLE_KEEP = 20 |
| 32 | |
| 33 | |
| 34 | class CanonicalHealthyError(RuntimeError): |
| 35 | """Raised when a restore is refused because canonical already passes `muse verify`.""" |
| 36 | |
| 37 | |
| 38 | def _resolve_canonical(repo: str, canonical_root: Path | None) -> Path: |
| 39 | if canonical_root is not None: |
| 40 | return canonical_root |
| 41 | if repo not in DEFAULT_CANONICAL_ROOTS: |
| 42 | raise ValueError(f"Unknown repo {repo!r} — no default canonical root registered for it.") |
| 43 | return DEFAULT_CANONICAL_ROOTS[repo] |
| 44 | |
| 45 | |
| 46 | def _timestamp() -> str: |
| 47 | return datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S.%f") |
| 48 | |
| 49 | |
| 50 | def _canonical_is_healthy(canonical_root: Path) -> bool: |
| 51 | proc = subprocess.run( |
| 52 | ["muse", "verify", "--json"], cwd=canonical_root, capture_output=True, text=True, |
| 53 | ) |
| 54 | if proc.returncode not in (0, 1): |
| 55 | raise RuntimeError(f"`muse verify` failed unexpectedly: {proc.stderr.strip()}") |
| 56 | try: |
| 57 | return bool(json.loads(proc.stdout).get("all_ok", False)) |
| 58 | except json.JSONDecodeError: |
| 59 | return False |
| 60 | |
| 61 | |
| 62 | def _guard_restore(canonical_root: Path, force: bool) -> None: |
| 63 | if force: |
| 64 | return |
| 65 | if _canonical_is_healthy(canonical_root): |
| 66 | raise CanonicalHealthyError( |
| 67 | f"{canonical_root} currently passes `muse verify` — refusing to restore over " |
| 68 | f"a healthy repo. Pass force=True if you're certain." |
| 69 | ) |
| 70 | |
| 71 | |
| 72 | def _prune(dir_path: Path, *, keep: int, is_relevant) -> None: |
| 73 | entries = sorted((p for p in dir_path.iterdir() if is_relevant(p)), key=lambda p: p.name) |
| 74 | for stale in entries[:-keep] if keep > 0 else entries: |
| 75 | if stale.is_dir(): |
| 76 | shutil.rmtree(stale) |
| 77 | else: |
| 78 | stale.unlink() |
| 79 | |
| 80 | |
| 81 | # ── Snapshots ("suspenders") ────────────────────────────────────────────── |
| 82 | |
| 83 | def _snapshot_dir(repo: str, backup_base: Path) -> Path: |
| 84 | return backup_base / f"{repo}-snapshots" |
| 85 | |
| 86 | |
| 87 | def create_snapshot( |
| 88 | repo: str, *, canonical_root: Path | None = None, backup_base: Path | None = None, |
| 89 | keep: int = DEFAULT_SNAPSHOT_KEEP, |
| 90 | ) -> Path: |
| 91 | """APFS copy-on-write snapshot of canonical's ``.muse/`` only.""" |
| 92 | root = _resolve_canonical(repo, canonical_root) |
| 93 | base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE |
| 94 | snap_root = _snapshot_dir(repo, base) |
| 95 | snap_root.mkdir(parents=True, exist_ok=True) |
| 96 | |
| 97 | dest = snap_root / _timestamp() |
| 98 | if dest.exists(): |
| 99 | raise RuntimeError(f"Snapshot destination already exists (clock collision?): {dest}") |
| 100 | dest.mkdir(parents=True) |
| 101 | |
| 102 | subprocess.run( |
| 103 | ["cp", "-c", "-R", str(root / ".muse"), str(dest / ".muse")], |
| 104 | check=True, capture_output=True, text=True, |
| 105 | ) |
| 106 | |
| 107 | _prune(snap_root, keep=keep, is_relevant=lambda p: p.is_dir()) |
| 108 | return dest |
| 109 | |
| 110 | |
| 111 | def list_snapshots(repo: str, *, backup_base: Path | None = None) -> list[Path]: |
| 112 | base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE |
| 113 | snap_root = _snapshot_dir(repo, base) |
| 114 | if not snap_root.exists(): |
| 115 | return [] |
| 116 | return sorted((p for p in snap_root.iterdir() if p.is_dir()), key=lambda p: p.name) |
| 117 | |
| 118 | |
| 119 | def restore_from_snapshot( |
| 120 | repo: str, snapshot_name: str, *, canonical_root: Path | None = None, |
| 121 | backup_base: Path | None = None, force: bool = False, |
| 122 | ) -> None: |
| 123 | root = _resolve_canonical(repo, canonical_root) |
| 124 | base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE |
| 125 | snapshot_path = _snapshot_dir(repo, base) / snapshot_name / ".muse" |
| 126 | if not snapshot_path.exists(): |
| 127 | raise FileNotFoundError(f"No such snapshot: {snapshot_path}") |
| 128 | |
| 129 | _guard_restore(root, force) |
| 130 | |
| 131 | live_muse = root / ".muse" |
| 132 | if live_muse.exists(): |
| 133 | shutil.rmtree(live_muse) |
| 134 | subprocess.run( |
| 135 | ["cp", "-c", "-R", str(snapshot_path), str(live_muse)], |
| 136 | check=True, capture_output=True, text=True, |
| 137 | ) |
| 138 | |
| 139 | |
| 140 | # ── Bundles ("belt") ────────────────────────────────────────────────────── |
| 141 | |
| 142 | def _bundle_dir(repo: str, backup_base: Path) -> Path: |
| 143 | return backup_base / f"{repo}-bundles" |
| 144 | |
| 145 | |
| 146 | def create_bundle_backup( |
| 147 | repo: str, *, canonical_root: Path | None = None, backup_base: Path | None = None, |
| 148 | keep: int = DEFAULT_BUNDLE_KEEP, |
| 149 | ) -> Path: |
| 150 | """`muse bundle create` over every local branch, immediately verified.""" |
| 151 | root = _resolve_canonical(repo, canonical_root) |
| 152 | base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE |
| 153 | bundle_root = _bundle_dir(repo, base) |
| 154 | bundle_root.mkdir(parents=True, exist_ok=True) |
| 155 | |
| 156 | branches_proc = subprocess.run( |
| 157 | ["muse", "branch", "--json"], cwd=root, capture_output=True, text=True, check=True, |
| 158 | ) |
| 159 | branch_names = [b["name"] for b in json.loads(branches_proc.stdout)] |
| 160 | if not branch_names: |
| 161 | raise RuntimeError(f"No branches found in {root} — nothing to bundle.") |
| 162 | |
| 163 | dest = bundle_root / f"{_timestamp()}.muse" |
| 164 | subprocess.run( |
| 165 | ["muse", "bundle", "create", str(dest), *branch_names, "--json"], |
| 166 | cwd=root, check=True, capture_output=True, text=True, |
| 167 | ) |
| 168 | |
| 169 | verify_proc = subprocess.run( |
| 170 | ["muse", "bundle", "verify", str(dest), "--json"], |
| 171 | cwd=root, capture_output=True, text=True, |
| 172 | ) |
| 173 | if not json.loads(verify_proc.stdout).get("all_ok", False): |
| 174 | dest.unlink(missing_ok=True) |
| 175 | raise RuntimeError(f"Freshly created bundle failed verification: {verify_proc.stdout}") |
| 176 | |
| 177 | _prune(bundle_root, keep=keep, is_relevant=lambda p: p.is_file() and p.suffix == ".muse") |
| 178 | return dest |
| 179 | |
| 180 | |
| 181 | def list_bundles(repo: str, *, backup_base: Path | None = None) -> list[Path]: |
| 182 | base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE |
| 183 | bundle_root = _bundle_dir(repo, base) |
| 184 | if not bundle_root.exists(): |
| 185 | return [] |
| 186 | return sorted( |
| 187 | (p for p in bundle_root.iterdir() if p.is_file() and p.suffix == ".muse"), |
| 188 | key=lambda p: p.name, |
| 189 | ) |
| 190 | |
| 191 | |
| 192 | def restore_from_bundle( |
| 193 | repo: str, bundle_name: str, *, canonical_root: Path | None = None, |
| 194 | backup_base: Path | None = None, force: bool = False, |
| 195 | ) -> None: |
| 196 | root = _resolve_canonical(repo, canonical_root) |
| 197 | base = backup_base if backup_base is not None else DEFAULT_BACKUP_BASE |
| 198 | bundle_path = _bundle_dir(repo, base) / bundle_name |
| 199 | if not bundle_path.exists(): |
| 200 | raise FileNotFoundError(f"No such bundle: {bundle_path}") |
| 201 | |
| 202 | _guard_restore(root, force) |
| 203 | |
| 204 | # `muse bundle unbundle` treats an object as "already present" purely by |
| 205 | # path existing on disk — it never re-validates content, so unbundling |
| 206 | # on top of an existing (possibly corrupted) store can silently leave |
| 207 | # corruption in place. A restore must fully replace, the same way |
| 208 | # restore_from_snapshot does, not overlay. |
| 209 | live_muse = root / ".muse" |
| 210 | if live_muse.exists(): |
| 211 | shutil.rmtree(live_muse) |
| 212 | subprocess.run(["muse", "init", "--json"], cwd=root, check=True, capture_output=True, text=True) |
| 213 | |
| 214 | subprocess.run( |
| 215 | ["muse", "bundle", "unbundle", str(bundle_path), "--verify", "--json"], |
| 216 | cwd=root, check=True, capture_output=True, text=True, |
| 217 | ) |
| 218 | |
| 219 | |
| 220 | def main(argv: list[str] | None = None) -> int: |
| 221 | import argparse |
| 222 | |
| 223 | parser = argparse.ArgumentParser(description=__doc__) |
| 224 | subparsers = parser.add_subparsers(dest="action", required=True) |
| 225 | |
| 226 | for name, fn in (("snapshot", create_snapshot), ("bundle", create_bundle_backup)): |
| 227 | p = subparsers.add_parser(name, help=f"Create a {name} backup.") |
| 228 | p.add_argument("repo") |
| 229 | p.add_argument("--keep", type=int, default=DEFAULT_SNAPSHOT_KEEP if name == "snapshot" else DEFAULT_BUNDLE_KEEP) |
| 230 | p.add_argument("--backup-base", default=None, help=argparse.SUPPRESS) |
| 231 | p.add_argument("--json", action="store_true") |
| 232 | p.set_defaults(_create=fn) |
| 233 | |
| 234 | restore_p = subparsers.add_parser("restore", help="Restore canonical from a snapshot or bundle.") |
| 235 | restore_p.add_argument("repo") |
| 236 | restore_p.add_argument("--from-snapshot", default=None) |
| 237 | restore_p.add_argument("--from-bundle", default=None) |
| 238 | restore_p.add_argument("--force", action="store_true") |
| 239 | restore_p.add_argument("--backup-base", default=None, help=argparse.SUPPRESS) |
| 240 | restore_p.add_argument("--json", action="store_true") |
| 241 | |
| 242 | args = parser.parse_args(argv) |
| 243 | backup_base = Path(args.backup_base) if args.backup_base else None |
| 244 | |
| 245 | try: |
| 246 | if args.action in ("snapshot", "bundle"): |
| 247 | path = args._create(args.repo, backup_base=backup_base, keep=args.keep) |
| 248 | print(json.dumps({"repo": args.repo, "path": str(path)}) if args.json |
| 249 | else f"✅ Created {args.action} backup: {path}") |
| 250 | return 0 |
| 251 | |
| 252 | if args.action == "restore": |
| 253 | if bool(args.from_snapshot) == bool(args.from_bundle): |
| 254 | print("❌ Exactly one of --from-snapshot or --from-bundle is required.", file=sys.stderr) |
| 255 | return 1 |
| 256 | if args.from_snapshot: |
| 257 | restore_from_snapshot(args.repo, args.from_snapshot, backup_base=backup_base, force=args.force) |
| 258 | else: |
| 259 | restore_from_bundle(args.repo, args.from_bundle, backup_base=backup_base, force=args.force) |
| 260 | print(json.dumps({"repo": args.repo, "status": "restored"}) if args.json |
| 261 | else f"✅ Restored {args.repo}.") |
| 262 | return 0 |
| 263 | except (ValueError, FileNotFoundError, RuntimeError, CanonicalHealthyError) as e: |
| 264 | print(f"❌ {e}", file=sys.stderr) |
| 265 | return 1 |
| 266 | |
| 267 | return 1 |
| 268 | |
| 269 | |
| 270 | if __name__ == "__main__": |
| 271 | sys.exit(main()) |