From cd518cfb120aa8a65e80418013c8b2ee27d18150 Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Thu, 9 Jul 2026 11:07:10 -0400 Subject: [PATCH] =?UTF-8?q?feat(routing):=20Phase-2=20aggregate=20reclaim?= =?UTF-8?q?=20=E2=80=94=20verify=5Ftarget=20(slice=203)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit reclaim.py must verify a size-routed target as an AGGREGATE or every routed target perpetually false-DRIFTs (its newest log is one object of few files). Adds: - verify_target(manifest, holds, …): reconstruct the whole target from its manifest — expected set = ⋃ objects' arcnames, archived = Σ n_files, archived_bytes = Σ source_bytes; require EVERY object tar present (hsi_exists per object); newer-than-run is PER-OBJECT (a live member drifts iff newer than the run of the object that owns it). Same verdict vocabulary + INV-E exclusion re-gate as verify() (the manifest carries the same `exclusion` block, so recheck_exclusions works on it directly). Returns the standard result dict + all_tars. - find_manifests(log_dir): load {log_dir}/_vault/*.manifest.json (the aggregate indexes). - main(): --scan skips routed OBJECT logs (routed_object marker) and verifies each manifest via verify_target; --config with size_routing resolves its badge → manifest → verify_target; --match filters manifests too. Legacy/non-routed logs flow through verify() unchanged. - delete_source: guarded re-confirm of ALL object tars (all_tars) before rm, not just one. Tests: tests/test_routing_reclaim.py (13) — SAFE / count-drift / newer-drift / bytes-drift / missing-tar FAILED / no-objects FAILED / GONE / HOLD; INV-E no-longer-covered → DRIFTED and still-covered subtraction → SAFE; find_manifests; multi-tar delete ABORT. Suite 196 → 209. Based on main (NOT stacked). See docs/RFC_incremental_v2.md §2.6. Co-Authored-By: Claude Opus 4.8 --- reclaim.py | 212 ++++++++++++++++++++++++++++++++-- tests/test_routing_reclaim.py | 200 ++++++++++++++++++++++++++++++++ 2 files changed, 404 insertions(+), 8 deletions(-) create mode 100644 tests/test_routing_reclaim.py diff --git a/reclaim.py b/reclaim.py index c6c4088..aa9dd65 100644 --- a/reclaim.py +++ b/reclaim.py @@ -545,6 +545,164 @@ def verify(log, source_folder, holds, file_pattern=None, canonical_roots=(), } +def find_manifests(log_dir): + """ + Load every Phase-2 size-routing manifest ({log_dir}/_vault/*.manifest.json). + These are the per-target aggregate indexes verify_target() checks; they live in + the _vault/ child precisely so find_logs' non-recursive glob never sees them. + Unreadable/parse-broken manifests are skipped. + """ + out = [] + for mp in sorted(glob.glob(os.path.join(log_dir, "_vault", "*.manifest.json"))): + try: + out.append(load_json(mp)) + except (ValueError, OSError): + pass + return out + + +def verify_target(manifest, holds, canonical_roots=(), allowlist=None, opt_in=False): + """ + Per-target AGGREGATE structural double-check for a size-routed target (Phase 2, + RFC §2.6). A routed target ships as MANY objects (solos + shards); reclaim's + per-log verify() would see only one object's few files and perpetually false-DRIFT. + verify_target instead reconstructs the whole target from its manifest: + + expected member set = ⋃ objects' arcnames + archived count = Σ objects' n_files + archived bytes = Σ objects' source_bytes + every object tar must be present on tape (hsi_exists per object) + newer-than-run is per-object (a live member is drift iff newer than the run + of the object that currently owns it) + + Verdict vocabulary + INV-E exclusion re-gate match verify() exactly. Returns the + same result-dict shape (plus `all_tars`), so the report and guarded --delete paths + work unchanged. The manifest carries `source_folder`/`file_pattern`/`exclusion`, so + no config or original log is needed. + """ + project = manifest.get("project") + source_folder = manifest.get("source_folder") + pattern = manifest.get("file_pattern") + objects = manifest.get("objects") or [] + reasons = [] + + expected = set() + arc_run = {} # arcname -> owning object's run datetime + all_tars = [] + archived = 0 + archived_bytes = 0 + for o in objects: + rdt = run_datetime(o.get("run_timestamp")) + for a in (o.get("arcnames") or []): + expected.add(a) + arc_run[a] = rdt + archived += o.get("n_files") or 0 + archived_bytes += o.get("source_bytes") or 0 + if o.get("fortress_tar"): + all_tars.append(o["fortress_tar"]) + + held = any((h == project) or (source_folder and h in source_folder) for h in holds) + + # INV-E: re-prove the archive's recorded exclusions are still git-COVERED. The + # manifest carries the same `exclusion` block a log does, so recheck_exclusions + # works on it directly. + excl_check = recheck_exclusions(manifest, source_folder) + still_covered = excl_check["still_covered"] + + # Every object tar must be present on tape (bounded O(#objects) metadata calls). + missing_tars = [t for t in all_tars if not hsi_exists(t)] + tars_ok = bool(all_tars) and not missing_tars + + # One pattern-scoped live walk: count, bytes, and per-object newer-than-run. + live = None + live_bytes = None + newer_count = 0 + newer_samples = [] + if source_folder and os.path.isdir(source_folder): + rx = re.compile(pattern) if pattern else None + live = 0 + live_bytes = 0 + for root, _dirs, files in os.walk(source_folder): + for f in files: + p = os.path.join(root, f) + rel = os.path.relpath(p, source_folder) + if rx is not None and not rx.search(rel): + continue + if rel in still_covered: + continue # INV-E re-proven-covered exclusion — not on tape by design + live += 1 + try: + st = os.stat(p) + except OSError: + continue + live_bytes += st.st_size + rdt = arc_run.get(rel) + # A member newer than the run of the object that owns it changed since + # it was archived. A live file in NO object is new-since-archive; the + # count check (live != archived) catches it — leave the newer signal to + # genuine modifications of archived members. + if rdt is not None and st.st_mtime > rdt.timestamp(): + newer_count += 1 + if len(newer_samples) < 3: + newer_samples.append(rel) + + # Verdict — most-blocking first, mirroring verify(). + if is_canonical(source_folder, canonical_roots): + verdict = CANONICAL + reasons.append("under canonical root — never deletable (hard invariant)") + elif not objects or source_folder is None: + verdict = FAILED + reasons.append("manifest has no objects" if not objects + else "source_folder unknown") + elif not tars_ok: + verdict = FAILED + if not all_tars: + reasons.append("no object tars recorded in manifest") + else: + reasons.append(f"{len(missing_tars)} of {len(all_tars)} object tar(s) NOT " + f"found on Fortress (e.g. {missing_tars[0]})") + elif live is None: + verdict = GONE + reasons.append("source already removed from Depot") + elif excl_check.get("applies") and excl_check.get("no_longer_covered"): + verdict = DRIFTED + gone = excl_check["no_longer_covered"] + sample = ", ".join(f"{a} ({v})" for a, v in gone[:3]) + reasons.append(f"{len(gone)} previously-excluded file(s) no longer git-COVERED " + f"(e.g. {sample}) — current bytes are not on tape") + elif newer_count > 0: + verdict = DRIFTED + reasons.append(f"{newer_count} member(s) newer than the object run that " + f"archived them (e.g. {', '.join(newer_samples)})") + elif live != archived: + verdict = DRIFTED + reasons.append(f"live files ({live}) != archived files ({archived}) across " + f"{len(objects)} object(s)") + elif live_bytes != archived_bytes: + verdict = DRIFTED + reasons.append(f"live bytes ({live_bytes}) != archived bytes ({archived_bytes}) " + f"— same count, content changed") + elif held: + verdict = HOLD + reasons.append("on holds list") + elif opt_in and not on_allowlist(project, source_folder, allowlist): + verdict = KEEP + reasons.append("not on reclaim allowlist (opt-in mode) — kept") + else: + verdict = SAFE + reasons.append(f"{live} files across {len(objects)} object(s) match; nothing " + f"newer than run; all tars present") + + return { + "project": project, "source_folder": source_folder, + "fortress_tar": (all_tars[0] if all_tars else None), "all_tars": all_tars, + "archived": archived, "live": live, "newer": newer_count, + "archived_bytes": archived_bytes, "live_bytes": live_bytes, + "pattern": pattern, "run": manifest.get("badge"), + "verdict": verdict, "reasons": reasons, + } + + def safe_to_rm(folder, canonical_roots=()): """Refuse to rm anything that isn't a deep, existing Depot data path — or that lives under a declared canonical root (the hard invariant, re-checked @@ -603,7 +761,9 @@ def delete_source(result, assume_yes, allow_no_snapshot=False, canonical_roots=( never removed, even with --yes and --allow-no-snapshot. """ folder = result["source_folder"] - tar = result["fortress_tar"] + # A routed target has many object tars; a whole-target archive has one. Re-confirm + # EVERY tar (routed: all_tars) fresh, immediately before deleting. + tars = result.get("all_tars") or ([result["fortress_tar"]] if result.get("fortress_tar") else []) ok, why = safe_to_rm(folder, canonical_roots) if not ok: @@ -611,8 +771,13 @@ def delete_source(result, assume_yes, allow_no_snapshot=False, canonical_roots=( return False # Fresh re-verify immediately before deleting — never trust the earlier check. - if not hsi_exists(tar): - print(f" ABORT {folder}: Fortress tar not confirmed at delete time") + if not tars: + print(f" ABORT {folder}: no Fortress tar to confirm at delete time") + return False + missing = [t for t in tars if not hsi_exists(t)] + if missing: + print(f" ABORT {folder}: {len(missing)}/{len(tars)} Fortress tar(s) not " + f"confirmed at delete time (e.g. {missing[0]})") return False # Recoverability guard: only delete what a current Depot snapshot can restore. @@ -806,6 +971,7 @@ def main(): except (ValueError, OSError): pass + manifests = [] # Phase-2 size-routed targets, verified as aggregates (verify_target) if args.scan: seen_projects = set() for lpath in find_logs(args.log_dir, days=args.days): @@ -813,20 +979,36 @@ def main(): log = load_json(lpath) except (ValueError, OSError): continue + # Phase-2 routed OBJECT logs are covered by their target's manifest below; + # never feed them to the per-log verify() (they'd each false-DRIFT). + if log.get("routed_object"): + continue proj = log.get("project") # newest-first: only verify the most recent successful-or-not log per project if proj in seen_projects: continue seen_projects.add(proj) items.append((log, configs_by_project.get(proj))) + manifests = find_manifests(args.log_dir) elif args.config: config = load_json(args.config) proj = config["project_name"] - lpath = latest_log_for_project(args.log_dir, proj) - if not lpath: - print(f"No archive log found for project {proj!r} in {args.log_dir}") - sys.exit(1) - items.append((load_json(lpath), config)) + if config.get("size_routing"): + # Routed target: verify its manifest aggregate, not a per-object log. + badge = archive.badge_of(config["project_name"], config["source_folder"], + config["file_pattern"]) + mpath = os.path.join(config.get("log_dir") or args.log_dir, + "_vault", f"{badge}.manifest.json") + if not os.path.exists(mpath): + print(f"No size-routing manifest for {proj!r} at {mpath}") + sys.exit(1) + manifests = [load_json(mpath)] + else: + lpath = latest_log_for_project(args.log_dir, proj) + if not lpath: + print(f"No archive log found for project {proj!r} in {args.log_dir}") + sys.exit(1) + items.append((load_json(lpath), config)) else: ap.error("give a config file, or use --scan") @@ -841,6 +1023,12 @@ def matches(log, cfg): return args.match in hay items = [(log, cfg) for (log, cfg) in items if matches(log, cfg)] + def manifest_matches(m): + hay = " ".join([m.get("project") or "", m.get("source_folder") or ""] + + [o.get("fortress_tar") or "" for o in (m.get("objects") or [])]) + return args.match in hay + manifests = [m for m in manifests if manifest_matches(m)] + # Verify and report. Attach scratch-zip locators (for post-delete cleanup): # zip_name from the log + tmp_dir from the config. results = [] @@ -851,6 +1039,14 @@ def matches(log, cfg): r["zip_name"] = log.get("zip_name") r["tmp_dir"] = (cfg or {}).get("tmp_dir") results.append(r) + # Phase-2 routed targets: one aggregate verdict per manifest (verify_target). No + # staging zip to clean up (objects are tars of source), so zip_name/tmp_dir are None. + for m in manifests: + r = verify_target(m, holds, canonical_roots=canonical_roots, + allowlist=allowlist, opt_in=opt_in) + r["zip_name"] = None + r["tmp_dir"] = None + results.append(r) results.sort(key=lambda r: (r["verdict"] != SAFE, r["project"] or "")) display = [r for r in results if not (args.skip_gone and r["verdict"] == GONE)] diff --git a/tests/test_routing_reclaim.py b/tests/test_routing_reclaim.py new file mode 100644 index 0000000..ffe6998 --- /dev/null +++ b/tests/test_routing_reclaim.py @@ -0,0 +1,200 @@ +""" +Unit tests for Phase-2 aggregate reclaim (docs/RFC_incremental_v2.md §2.6, slice 3): +verify_target() reconstructs a size-routed target from its manifest (⋃ arcnames, +Σ n_files, Σ source_bytes, every object tar present, per-object newer-than-run) and +returns the same verdict vocabulary as verify(). Also covers find_manifests and the +multi-tar guarded delete. Tape is mocked (reclaim.hsi_exists); no Slurm/Globus/hsi. +""" +import json +import os +import sys +import tempfile +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import reclaim # noqa: E402 + + +FUTURE = "20990101_000000" # run after any live file mtime -> nothing newer +PAST = "20200101_000000" # run before live mtimes -> everything newer + + +def _write(path, body): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb") as fh: + fh.write(body) + return len(body) + + +def build_target(d, run=FUTURE): + """Create a source tree + a matching manifest. Returns (source, manifest).""" + src = os.path.join(d, "src") + sizes = {} + sizes["big.bin"] = _write(os.path.join(src, "big.bin"), b"B" * 4096) + for i in (1, 2, 3): + sizes["sub/s%d.bin" % i] = _write(os.path.join(src, "sub", "s%d.bin" % i), + b"s" * (100 + i)) + solo_arcs = ["big.bin"] + shard_arcs = ["sub/s1.bin", "sub/s2.bin", "sub/s3.bin"] + manifest = { + "schema_version": 1, "badge": "tstbadge", "project": "proj", + "source_folder": src, "file_pattern": ".*", + "exclusion": {"status": "OFF"}, + "objects": [ + {"slot": "solo:big", "kind": "solo", "arcnames": solo_arcs, + "fortress_tar": "/tape/proj__solo_%s.tar" % run, + "run_timestamp": "proj__solo_%s" % run, + "n_files": 1, "source_bytes": sizes["big.bin"]}, + {"slot": "shard:0", "kind": "shard", "arcnames": shard_arcs, + "fortress_tar": "/tape/proj__shard-0of1_%s.tar" % run, + "run_timestamp": "proj__shard-0of1_%s" % run, + "n_files": 3, "source_bytes": sum(sizes[a] for a in shard_arcs)}, + ], + } + return src, manifest + + +class VerifyTargetBase(unittest.TestCase): + def setUp(self): + self._orig_hsi = reclaim.hsi_exists + reclaim.hsi_exists = lambda tar: True # all tars present by default + + def tearDown(self): + reclaim.hsi_exists = self._orig_hsi + + def vt(self, manifest, holds=()): + return reclaim.verify_target(manifest, holds) + + +class TestVerifyTargetVerdicts(VerifyTargetBase): + def test_safe(self): + with tempfile.TemporaryDirectory() as d: + _, m = build_target(d) + r = self.vt(m) + self.assertEqual(r["verdict"], reclaim.SAFE) + self.assertEqual(r["live"], 4) + self.assertEqual(r["archived"], 4) + self.assertEqual(len(r["all_tars"]), 2) + + def test_count_drift_extra_file(self): + with tempfile.TemporaryDirectory() as d: + src, m = build_target(d) + _write(os.path.join(src, "sub", "s4_new.bin"), b"new") # not in any object + r = self.vt(m) + self.assertEqual(r["verdict"], reclaim.DRIFTED) + self.assertTrue(any("live files" in x for x in r["reasons"])) + + def test_newer_drift(self): + with tempfile.TemporaryDirectory() as d: + _, m = build_target(d, run=PAST) # live mtimes now > 2020 run + r = self.vt(m) + self.assertEqual(r["verdict"], reclaim.DRIFTED) + self.assertTrue(any("newer than the object run" in x for x in r["reasons"])) + + def test_bytes_drift(self): + with tempfile.TemporaryDirectory() as d: + _, m = build_target(d) + m["objects"][0]["source_bytes"] += 100 # count + mtime unchanged, bytes off + r = self.vt(m) + self.assertEqual(r["verdict"], reclaim.DRIFTED) + self.assertTrue(any("live bytes" in x for x in r["reasons"])) + + def test_missing_tar_failed(self): + with tempfile.TemporaryDirectory() as d: + _, m = build_target(d) + missing = m["objects"][1]["fortress_tar"] + reclaim.hsi_exists = lambda tar: tar != missing + r = self.vt(m) + self.assertEqual(r["verdict"], reclaim.FAILED) + self.assertTrue(any("object tar(s) NOT found" in x for x in r["reasons"])) + + def test_no_objects_failed(self): + with tempfile.TemporaryDirectory() as d: + _, m = build_target(d) + m["objects"] = [] + self.assertEqual(self.vt(m)["verdict"], reclaim.FAILED) + + def test_gone(self): + with tempfile.TemporaryDirectory() as d: + _, m = build_target(d) + m["source_folder"] = os.path.join(d, "does_not_exist") + self.assertEqual(self.vt(m)["verdict"], reclaim.GONE) + + def test_hold(self): + with tempfile.TemporaryDirectory() as d: + _, m = build_target(d) + self.assertEqual(self.vt(m, holds=["proj"])["verdict"], reclaim.HOLD) + + +class TestVerifyTargetExclusion(VerifyTargetBase): + def test_inv_e_no_longer_covered_drifts(self): + with tempfile.TemporaryDirectory() as d: + _, m = build_target(d) + orig = reclaim.recheck_exclusions + reclaim.recheck_exclusions = lambda mani, sf: { + "applies": True, "still_covered": set(), + "no_longer_covered": [("meta/x.json", "DIRTY")], "reason": None} + try: + r = self.vt(m) + finally: + reclaim.recheck_exclusions = orig + self.assertEqual(r["verdict"], reclaim.DRIFTED) + self.assertTrue(any("no longer git-COVERED" in x for x in r["reasons"])) + + def test_inv_e_still_covered_subtracted(self): + with tempfile.TemporaryDirectory() as d: + src, m = build_target(d) + # an excluded-but-still-covered file lives on disk but is NOT in any object; + # it must be subtracted from the walk so live still matches archived (SAFE). + _write(os.path.join(src, "meta", "cov.json"), b"{}") + orig = reclaim.recheck_exclusions + reclaim.recheck_exclusions = lambda mani, sf: { + "applies": True, "still_covered": {"meta/cov.json"}, + "no_longer_covered": [], "reason": None} + try: + r = self.vt(m) + finally: + reclaim.recheck_exclusions = orig + self.assertEqual(r["verdict"], reclaim.SAFE) # cov.json subtracted -> 4==4 + + +class TestFindManifests(unittest.TestCase): + def test_loads_vault_manifests(self): + with tempfile.TemporaryDirectory() as d: + vault = os.path.join(d, "_vault") + os.makedirs(vault) + for b in ("aaa", "bbb"): + with open(os.path.join(vault, "%s.manifest.json" % b), "w") as fh: + json.dump({"badge": b, "objects": []}, fh) + got = reclaim.find_manifests(d) + self.assertEqual(sorted(m["badge"] for m in got), ["aaa", "bbb"]) + + def test_empty_when_no_vault(self): + with tempfile.TemporaryDirectory() as d: + self.assertEqual(reclaim.find_manifests(d), []) + + +class TestDeleteMultiTar(unittest.TestCase): + def setUp(self): + self._hsi = reclaim.hsi_exists + self._safe = reclaim.safe_to_rm + reclaim.safe_to_rm = lambda folder, roots=(): (True, "") + + def tearDown(self): + reclaim.hsi_exists = self._hsi + reclaim.safe_to_rm = self._safe + + def test_aborts_when_any_tar_missing(self): + with tempfile.TemporaryDirectory() as d: + folder = os.path.join(d, "src") + os.makedirs(folder) + tars = ["/tape/a.tar", "/tape/b.tar"] + reclaim.hsi_exists = lambda tar: tar != "/tape/b.tar" # b missing + result = {"source_folder": folder, "fortress_tar": tars[0], "all_tars": tars} + deleted = reclaim.delete_source(result, assume_yes=True, allow_no_snapshot=True) + self.assertFalse(deleted) + self.assertTrue(os.path.isdir(folder)) # NOT deleted + + +if __name__ == "__main__": + unittest.main()