feat(restore): Phase-2 full restore capstone (slice 5)

restore.py

@@ -0,0 +1,374 @@
+#!/usr/bin/env python3
+"""
+restore.py — full / selective restore of a Phase-2 size-routed target
+(docs/RFC_incremental_v2.md §2.9, the Phase-2 capstone).
+
+A size-routed target lives on Fortress as MANY objects (solos + shards), indexed
+by the per-target manifest at {log_dir}/_vault/{badge}.manifest.json. Because the
+objects are a DISJOINT partition of the current member set, restore is a plain
+union: extract every current object into an empty destination and reconcile —
+no overlay ordering, no deletion replay (that is the deferred Phase-3 leveled
+restore, RFC §3.4).
+
+Order of operations (each gate ABORTS the whole restore — never partial):
+  (a) completeness gate BEFORE any transfer — every selected object's tar must
+      still exist on Fortress (reclaim.hsi_exists), else "not fully restorable";
+  (b) destination MUST be empty (created if absent) — never restore over data;
+  (c) per object, by recorded transport:
+        "htar"       — `htar -xvf <tar> -L <memberlist>` random-access indexed
+                       extract straight into dest (reads only those members);
+        "htar_large" — `hsi get` the plain tar to scratch, then `tar xf`
+                       (htar_large tars have no .idx; invariant #6 — never
+                       stream htar_large/-xvf through stdout);
+  (d) every restored member's MD5 is verified against the object's own run-log
+      ({log_dir}/{run_timestamp}.json, key `source_files`, arcname-keyed —
+      invariant #2);
+  (e) on a FULL restore, a MANDATORY final reconciliation:
+      set(walk(dest)) == ⋃ objects' arcnames, else ABORT.
+
+Selective restore (`--slot solo:<hex>` / `--slot shard:<k>` / `--member relpath`)
+filters the object set — for `htar` objects this reads only the requested
+members' bytes off tape. Reconciliation (e) runs only on a full restore.
+
+LOCAL only (needs `hsi`/`htar` at /opt/hsi/bin + the filesystem) — run on the
+storage host (Negishi), like reclaim.py. Operator-invoked; NOT wired into
+`reclaim --delete` (a full restore pulls every tar off tape, TB-scale).
+
+Usage:
+  python3 restore.py config.json           --dest /scratch/.../restore_here
+  python3 restore.py --manifest /depot/.../logs/_vault/<badge>.manifest.json \
+                     --dest DIR [--log-dir DIR] [--tmp-dir DIR]
+  python3 restore.py config.json --dest DIR --slot shard:3 --slot solo:1a2b...
+  python3 restore.py config.json --dest DIR --member sub/path/file.bin
+"""
+import argparse
+import hashlib
+import json
+import os
+import subprocess
+import sys
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import archive   # noqa: E402  (badge_of, vault_dir_for, manifest_path, load_config)
+import reclaim   # noqa: E402  (hsi_exists — the same tape-presence check reclaim trusts)
+
+
+class RestoreError(RuntimeError):
+    """Any condition that makes the restored tree untrustworthy — always aborts."""
+
+
+def prepend_hpss_path():
+    """hsi/htar live in /opt/hsi/bin and are not on the non-interactive PATH;
+    htar_large is /usr/local/bin. Prepend both (mirrors archive.ship_object)."""
+    for _bin in ("/opt/hsi/bin", "/usr/local/bin"):
+        if os.path.isdir(_bin) and _bin not in os.environ.get("PATH", "").split(":"):
+            os.environ["PATH"] = _bin + ":" + os.environ.get("PATH", "")
+
+
+def check_dest(dest):
+    """Gate (b): the destination must not exist, or exist EMPTY — restoring over
+    live data is forbidden (RFC §2.9: 'always into an empty scratch dir')."""
+    if os.path.exists(dest):
+        if not os.path.isdir(dest):
+            raise RestoreError(f"restore dest {dest} exists and is not a directory")
+        if os.listdir(dest):
+            raise RestoreError(
+                f"restore dest {dest} is NOT empty — refusing to restore over "
+                f"existing data. Point --dest at a new/empty directory.")
+
+
+def select_objects(manifest, slots=None, members=None):
+    """
+    Resolve the object set to restore. Returns [(object, arcnames_to_restore)]
+    in manifest order. No filter = full restore (every object, every member).
+    `slots` picks whole objects by slot id; `members` picks single files (each
+    mapped to the one current object that owns it). Unknown slot/member = abort
+    (a typo must never silently restore nothing).
+    """
+    objects = manifest.get("objects") or []
+    if not objects:
+        raise RestoreError("manifest has no objects — nothing to restore")
+    if not slots and not members:
+        return [(o, list(o.get("arcnames") or [])) for o in objects]
+
+    by_slot = {o.get("slot"): o for o in objects}
+    picked = {}   # slot -> set(arcnames)
+    for s in (slots or ()):
+        if s not in by_slot:
+            known = ", ".join(sorted(by_slot)[:8])
+            raise RestoreError(f"unknown slot {s!r} — manifest has: {known}"
+                               + (" ..." if len(by_slot) > 8 else ""))
+        picked.setdefault(s, set()).update(by_slot[s].get("arcnames") or [])
+    if members:
+        owner = {}
+        for o in objects:
+            for a in (o.get("arcnames") or ()):
+                owner[a] = o.get("slot")
+        for m in members:
+            if m not in owner:
+                raise RestoreError(
+                    f"member {m!r} is not in any current object — check the relpath "
+                    f"(arcnames are relative to the archived source_folder)")
+            picked.setdefault(owner[m], set()).add(m)
+
+    return [(o, sorted(picked[o.get("slot")])) for o in objects
+            if o.get("slot") in picked]
+
+
+def extract_object(obj, arcnames, dest, tmp_dir):
+    """
+    Gate (c): pull ONE object's members off tape into dest, by the transport the
+    manifest recorded at ship time (archive.ship_object):
+      htar       — indexed random-access extract of exactly `arcnames` via a
+                   member-list file (only those members' bytes are read);
+      htar_large — plain tar (no .idx): hsi get to scratch, tar xf the members
+                   (invariant #6: never htar_large -xvf through a pipe).
+    htar/tar resolve members relative to CWD, so both run with cwd=dest.
+    """
+    prepend_hpss_path()
+    transport = obj.get("transport") or "htar"
+    tar = obj["fortress_tar"]
+    os.makedirs(tmp_dir, exist_ok=True)
+
+    if transport == "htar":
+        listfile = os.path.join(tmp_dir, f"{obj['run_timestamp']}.restore.members")
+        with open(listfile, "w") as fh:
+            fh.write("\n".join(arcnames) + "\n")
+        try:
+            r = subprocess.run(["htar", "-xvf", tar, "-L", listfile], cwd=dest,
+                               stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        finally:
+            try:
+                os.remove(listfile)
+            except OSError:
+                pass
+        if r.returncode != 0:
+            raise RestoreError(
+                f"htar extract failed (rc={r.returncode}) for {tar}:\n"
+                + r.stdout.decode("utf-8", "replace")[-4000:])
+    else:
+        local_tar = os.path.join(tmp_dir, f"{obj['run_timestamp']}.restore.tar")
+        try:
+            g = subprocess.run(f'hsi -q "get {local_tar} : {tar}"', shell=True,
+                               stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            if g.returncode != 0:
+                raise RestoreError(
+                    f"hsi get failed (rc={g.returncode}) for {tar}:\n"
+                    + g.stdout.decode("utf-8", "replace")[-4000:])
+            x = subprocess.run(["tar", "xf", local_tar] + list(arcnames), cwd=dest,
+                               stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+            if x.returncode != 0:
+                raise RestoreError(
+                    f"tar extract failed (rc={x.returncode}) for {local_tar}:\n"
+                    + x.stdout.decode("utf-8", "replace")[-4000:])
+        finally:
+            try:
+                os.remove(local_tar)
+            except OSError:
+                pass
+
+
+def object_runlog_checksums(log_dir, run_timestamp):
+    """The object's own run-log ({run_timestamp}.json, written by ship_object)
+    carries the arcname-keyed per-file MD5s (`source_files`, invariant #2)."""
+    path = os.path.join(log_dir, f"{run_timestamp}.json")
+    if not os.path.exists(path):
+        raise RestoreError(
+            f"object run-log not found: {path} — cannot MD5-verify this object "
+            f"(pass --log-dir if logs live elsewhere)")
+    with open(path) as fh:
+        log = json.load(fh)
+    checks = log.get("source_files")
+    if not isinstance(checks, dict) or not checks:
+        raise RestoreError(f"object run-log {path} has no source_files checksums")
+    return checks
+
+
+def md5_file(path):
+    h = hashlib.md5()
+    with open(path, "rb") as fh:
+        for chunk in iter(lambda: fh.read(1024 * 1024), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def verify_members(obj, arcnames, dest, log_dir):
+    """Gate (d): every restored member's MD5 must equal the MD5 recorded when the
+    object shipped. Any mismatch/missing file poisons the whole restore."""
+    checks = object_runlog_checksums(log_dir, obj["run_timestamp"])
+    for arc in arcnames:
+        expected = checks.get(arc)
+        if expected is None:
+            raise RestoreError(
+                f"no recorded MD5 for {arc!r} in run-log {obj['run_timestamp']}.json "
+                f"— checksum keys must match arcnames (invariant #2)")
+        full = os.path.join(dest, arc)
+        if not os.path.isfile(full):
+            raise RestoreError(
+                f"member {arc!r} missing from dest after extracting "
+                f"{obj.get('fortress_tar')}")
+        got = md5_file(full)
+        if got != expected:
+            raise RestoreError(
+                f"MD5 MISMATCH for {arc!r} from {obj.get('fortress_tar')} "
+                f"(expected {expected}, got {got})")
+
+
+def walk_restored(dest):
+    """Set of relpaths of every file now under dest (reconciliation walk)."""
+    out = set()
+    for root, _dirs, files in os.walk(dest):
+        for f in files:
+            out.add(os.path.relpath(os.path.join(root, f), dest))
+    return out
+
+
+def restore_target(manifest, dest, log_dir, tmp_dir, slots=None, members=None,
+                   extract=extract_object):
+    """
+    Restore a routed target (or a --slot/--member selection of it) into an empty
+    dest. Implements gates (a)-(e) from the module docstring in order; raises
+    RestoreError on the first violated gate. `extract` is injected so tests can
+    mock the tape path (mirrors route_and_ship's ship_one).
+
+    Returns a summary dict: badge, objects, files, restored (set of arcnames),
+    reconciled (True on a verified full restore, None when a selection skipped
+    the reconciliation).
+    """
+    selected = select_objects(manifest, slots, members)
+    full_restore = not slots and not members
+
+    check_dest(dest)                                     # (b) — cheap, no side effects
+
+    # (a) completeness gate BEFORE any transfer — a partially-restorable target is
+    # NOT restorable ("never 'restored minus one'", RFC §2.9).
+    missing = [o.get("fortress_tar") for o, _arcs in selected
+               if not reclaim.hsi_exists(o.get("fortress_tar"))]
+    if missing:
+        raise RestoreError(
+            f"target NOT fully restorable — {len(missing)} of {len(selected)} object "
+            f"tar(s) missing on Fortress (e.g. {missing[0]}); aborting before any "
+            f"extraction")
+
+    os.makedirs(dest, exist_ok=True)
+    restored = set()
+    for obj, arcs in selected:
+        print(f"  [RESTORE] {obj.get('slot')} ({obj.get('transport') or 'htar'}, "
+              f"{len(arcs)} member(s)) <- {obj.get('fortress_tar')}")
+        extract(obj, arcs, dest, tmp_dir)                # (c)
+        verify_members(obj, arcs, dest, log_dir)         # (d)
+        print(f"  [OK] {obj.get('slot')}: {len(arcs)} member MD5(s) verified")
+        restored.update(arcs)
+
+    reconciled = None
+    if full_restore:                                     # (e) MANDATORY, full only
+        expected = set()
+        for o in manifest.get("objects") or []:
+            expected.update(o.get("arcnames") or ())
+        actual = walk_restored(dest)
+        if actual != expected:
+            extra = sorted(actual - expected)
+            miss = sorted(expected - actual)
+            raise RestoreError(
+                "final reconciliation FAILED — restored tree != manifest aggregate "
+                f"({len(extra)} unexpected, e.g. {extra[:3]}; "
+                f"{len(miss)} missing, e.g. {miss[:3]}). "
+                "The restored tree must NOT be trusted.")
+        reconciled = True
+        print(f"  [OK] final reconciliation: {len(actual)} restored file(s) == "
+              f"manifest aggregate")
+
+    return {"badge": manifest.get("badge"), "objects": len(selected),
+            "files": len(restored), "restored": restored, "reconciled": reconciled}
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser(
+        description="Restore a Phase-2 size-routed target from Fortress "
+                    "(docs/RFC_incremental_v2.md §2.9). LOCAL only; empty dest only.")
+    ap.add_argument("config", nargs="?",
+                    help="The target's archive config JSON (resolves badge -> "
+                         "{log_dir}/_vault manifest). Alternative: --manifest.")
+    ap.add_argument("--manifest",
+                    help="Path to a {badge}.manifest.json (bypasses the config)")
+    ap.add_argument("--dest", required=True,
+                    help="Destination directory — MUST be empty (or absent)")
+    ap.add_argument("--log-dir", dest="log_dir",
+                    help="Where the object run-logs (*.json) live. Default: the "
+                         "config's log_dir, or the manifest's _vault parent.")
+    ap.add_argument("--tmp-dir", dest="tmp_dir",
+                    help="Scratch for member lists + htar_large staging tars. "
+                         "Default: the config's tmp_dir, else <dest>.restore-tmp")
+    ap.add_argument("--slot", action="append", dest="slots", metavar="SLOT",
+                    help="Restore only this object (solo:<hex> or shard:<k>); repeatable")
+    ap.add_argument("--member", action="append", dest="members", metavar="RELPATH",
+                    help="Restore only this file (relpath under source_folder); repeatable")
+    args = ap.parse_args(argv)
+
+    if bool(args.config) == bool(args.manifest):
+        ap.error("give exactly one of: a config JSON, or --manifest")
+
+    if args.manifest:
+        mpath = os.path.abspath(args.manifest)
+        try:
+            with open(mpath) as fh:
+                manifest = json.load(fh)
+        except (OSError, ValueError) as e:
+            print(f"  [ABORT] cannot read manifest {mpath}: {e}")
+            sys.exit(1)
+        log_dir = args.log_dir
+        if not log_dir:
+            vault = os.path.dirname(mpath)
+            if os.path.basename(vault) != "_vault":
+                ap.error("--manifest is not under a _vault/ dir; pass --log-dir")
+            log_dir = os.path.dirname(vault)
+        tmp_dir = args.tmp_dir
+    else:
+        try:
+            config = archive.load_config(args.config)
+        except (OSError, ValueError) as e:
+            print(f"  [ABORT] bad config {args.config}: {e}")
+            sys.exit(1)
+        badge = archive.badge_of(config["project_name"], config["source_folder"],
+                                 config["file_pattern"])
+        vault = archive.vault_dir_for(config["log_dir"])
+        mpath = archive.manifest_path(vault, badge)
+        manifest = archive.load_manifest(vault, badge)
+        if manifest is None:
+            print(f"  [ABORT] no size-routing manifest for "
+                  f"{config['project_name']!r} at {mpath} — was this target ever "
+                  f"archived with size_routing on?")
+            sys.exit(1)
+        log_dir = args.log_dir or config["log_dir"]
+        tmp_dir = args.tmp_dir or config["tmp_dir"]
+
+    dest = os.path.abspath(args.dest)
+    if not tmp_dir:
+        tmp_dir = dest.rstrip(os.sep) + ".restore-tmp"
+
+    print("#" * 64)
+    print("#  FORTRESS RESTORE — Phase-2 routed target (RFC §2.9), LOCAL")
+    print(f"#  badge:    {manifest.get('badge')}  project: {manifest.get('project')}")
+    print(f"#  manifest: {mpath}")
+    print(f"#  objects:  {len(manifest.get('objects') or [])}"
+          + (f"  (selection: {(args.slots or []) + (args.members or [])})"
+             if (args.slots or args.members) else "  (FULL restore)"))
+    print(f"#  dest:     {dest}")
+    print("#" * 64)
+
+    try:
+        summary = restore_target(manifest, dest, log_dir, tmp_dir,
+                                 slots=args.slots, members=args.members)
+    except RestoreError as e:
+        print(f"\n  [ABORT] {e}")
+        print("  Nothing restored is trustworthy until every gate passes — if dest "
+              "was partially written, remove it before re-running.")
+        sys.exit(1)
+
+    scope = "FULL restore, reconciled" if summary["reconciled"] else "selective restore"
+    print(f"\n  [SUCCESS] {scope}: {summary['files']} file(s) from "
+          f"{summary['objects']} object(s) -> {dest}")
+
+
+if __name__ == "__main__":
+    main()