diff --git a/archive.py b/archive.py index c168d6c..312b7ab 100644 --- a/archive.py +++ b/archive.py @@ -129,6 +129,45 @@ def load_config(config_path): f"got: {config['exclude_optional']!r}" ) + # Phase-2 size-routing (docs/RFC_incremental_v2.md §2). Opt-in per config and + # default-off, so an unset config is byte-for-byte the whole-target behavior. + config.setdefault("size_routing", False) + if not isinstance(config["size_routing"], bool): + raise ValueError( + f"config 'size_routing' must be true or false, got: {config['size_routing']!r}") + config.setdefault("t_small", T_SMALL_DEFAULT) + if not isinstance(config["t_small"], int) or isinstance(config["t_small"], bool) \ + or config["t_small"] <= 0: + raise ValueError( + f"config 't_small' must be a positive integer (bytes), got: {config['t_small']!r}") + config.setdefault("shard_count", None) # None -> derive K at baseline + if config["shard_count"] is not None and ( + not isinstance(config["shard_count"], int) + or isinstance(config["shard_count"], bool) or config["shard_count"] < 1): + raise ValueError( + f"config 'shard_count' must be null or an integer >= 1, " + f"got: {config['shard_count']!r}") + config.setdefault("shard_target", SHARD_TARGET_DEFAULT) + if not isinstance(config["shard_target"], int) or isinstance(config["shard_target"], bool) \ + or config["shard_target"] <= 0: + raise ValueError( + f"config 'shard_target' must be a positive integer (bytes), " + f"got: {config['shard_target']!r}") + + # Durability guard (RFC §2.4): a routed target keeps a per-target manifest at + # {log_dir}/_vault/ that MUST outlive runs and be shared, so reclaim can find it. + # Refuse a log_dir under $HOME / scratch / the home default — home is per-user and + # scratch is purged (a lost manifest makes every routed target false-DRIFT). + if config["size_routing"]: + ok, why = routing_log_dir_ok(config["log_dir"]) + if not ok: + raise ValueError( + f"size_routing needs log_dir on SHARED, DURABLE storage, but it points at " + f"{why} (log_dir={config['log_dir']!r}). The per-target manifest lives at " + f"{{log_dir}}/_vault/ and must outlive runs so reclaim can rebuild the " + f"aggregate. Point log_dir at the deployment's shared Depot logs dir " + f"(e.g. /depot//etc/logs_fortress/...). tmp_dir may stay in scratch.") + return config @@ -473,6 +512,62 @@ def enumerate_source_catalog(source_folder, file_pattern): return catalog +def enumerate_source_catalog_excluded(source_folder, file_pattern, + exclude_spec=None, exclude_optional=()): + """ + Routing data source (Phase 2, RFC §2.5): ONE walk returning the SURVIVOR catalog + (matched − git-COVERED exclusions) plus the exclusion report to record: + + ({arcname: (mtime_ns, size)}, exclusion_report) + + Combines enumerate_source_catalog's per-file (mtime,size) with the same exclusion + filter make_zip_files / enumerate_source_excluded apply — so the router partitions + exactly the bytes that will ship, and the report is recorded once for reclaim + (INV-E). evaluate_exclusions never raises: any problem yields status DISABLED and + an empty excluded set (route everything). Imports inside the body (dill, inv #1). + """ + import os + import re + import json + + pattern = re.compile(file_pattern) + source_folder = os.path.normpath(source_folder) + matched = [] # (full_path, arcname) — evaluate_exclusions' input shape + stats = {} # arcname -> (mtime_ns, size) + for dirpath, dirnames, filenames in os.walk(source_folder): + for fname in filenames: + full_path = os.path.join(dirpath, fname) + arcname = os.path.relpath(full_path, source_folder) + if pattern.search(arcname): + try: + st = os.stat(full_path) + except OSError: + continue + matched.append((full_path, arcname)) + stats[arcname] = (st.st_mtime_ns, st.st_size) + + report = {"status": "OFF", "spec_path": exclude_spec} + if exclude_spec: + try: + with open(exclude_spec) as fh: + spec = json.load(fh) + except Exception as e: + report = {"status": "DISABLED", "spec_path": exclude_spec, + "disabled_reason": f"could not read/parse exclude_spec " + f"{exclude_spec!r}: {e}", + "excluded": [], "excluded_arcnames": [], "demoted": [], + "contested_data": [], "companion_violations": [], + "counts": {}, "excluded_bytes": 0, "error": str(e)} + spec = None + if spec is not None: + report = evaluate_exclusions(matched, source_folder, spec, exclude_optional) + report["spec_path"] = exclude_spec + + drop = set(report.get("excluded_arcnames") or []) + survivors = {a: v for a, v in stats.items() if a not in drop} + return survivors, report + + def floor_s(mtime_ns): """ Floor an integer-nanosecond mtime to whole epoch seconds. @@ -796,6 +891,155 @@ def write_manifest(vault_dir, badge, manifest): return path +def routing_log_dir_ok(log_dir): + """ + Durability gate for a routed target's manifest location (RFC §2.4). Returns + (ok: bool, reason: str). The manifest lives at {log_dir}/_vault/ and must be + SHARED + DURABLE, so a routed log_dir must NOT be: + - the ~/globus_archive_tmp home default (the sneaky one), + - anywhere under $HOME (per-user, not shared), + - a scratch path ($RCAC_SCRATCH/$SCRATCH, or a 'scratch' path component — + purged on RCAC's retention timer; a lost manifest false-DRIFTs every target). + Point log_dir at the deployment's shared Depot logs dir instead. + """ + import os + + rp = os.path.realpath(os.path.expanduser(log_dir)) + home = os.path.realpath(os.path.expanduser("~")) + default = os.path.join(home, "globus_archive_tmp") + + def _under(root): + return rp == root or rp.startswith(root + os.sep) + + if _under(default): + return False, "the ~/globus_archive_tmp home default" + for env in ("RCAC_SCRATCH", "SCRATCH"): + s = os.environ.get(env) + if s and _under(os.path.realpath(s)): + return False, f"a scratch path (${env}) — purged on RCAC's retention timer" + if "scratch" in rp.split(os.sep): + return False, "a scratch path — purged on RCAC's retention timer" + if _under(home): + return False, "your home directory ($HOME) — per-user, not shared" + return True, "" + + +def object_project_name(project_name, obj, shard_count): + """ + Per-object log/zip/tar stem (invariant #4 — each object gets its own log pair): + solo: -> {project}__solo- + shard: -> {project}__shard-of + Slots carry a ':' which is unsafe in filenames; this yields the RFC §2.4 stems + (e.g. repository_X0E_4_omics-analysis__shard-3of8). make_zip_files appends + _{timestamp}.zip, so each object run writes {stem}_{ts}.zip/.txt/.json. + """ + kind = obj["kind"] + ident = obj["slot"].split(":", 1)[1] + if kind == "solo": + return "{}__solo-{}".format(project_name, ident) + return "{}__shard-{}of{}".format(project_name, ident, shard_count) + + +def route_and_ship(project_name, source_folder, file_pattern, survivors, exclusion, + vault_dir, t_small, shard_count_cfg, shard_target, now_epoch, + ship_one, fresh=False, max_solo=MAX_SOLO_DEFAULT, + max_objects=MAX_OBJECTS_DEFAULT): + """ + LOCAL orchestration for size-routing (RFC §2.5). Partitions the survivor catalog + into objects, ships only the changed/new ones (skip-unchanged = the append win + + resume), and maintains the per-target manifest. Returns a summary dict. + + Args: + survivors: {arcname: (mtime_ns, size)} post-exclusion (from + enumerate_source_catalog_excluded). + exclusion: the recorded exclusion report (stored in the manifest for reclaim). + vault_dir: {log_dir}/_vault — where the manifest lives (durable+shared). + t_small/shard_count_cfg/shard_target: routing knobs; K is frozen from the + manifest once a baseline exists (a content-addressed map must be + stable), else derived (or taken from shard_count_cfg) at baseline. + now_epoch: run epoch seconds (BLONDE quaver). + ship_one(obj, obj_stem) -> dict(fortress_tar, zip_checksum, n_files, + source_bytes, run_timestamp[, tar_bytes]) — INJECTED so tests can + mock the tape path; production wires it to make_zip_files + + send_to_fortress. Only called for objects that actually ship. + fresh: True (--fresh) ignores the manifest, re-derives K, re-ships everything. + + The manifest is written atomically AFTER EACH shipped object (best-effort), so a + crash leaves already-shipped objects recorded and the next run skips them. + """ + badge = badge_of(project_name, source_folder, file_pattern) + t_small = clamp_t_small(t_small) + prior = None if fresh else load_manifest(vault_dir, badge) + + prior_k = (prior or {}).get("routing", {}).get("shard_count") + if prior_k: + shard_count = int(prior_k) # frozen for the badge's life + elif shard_count_cfg: + shard_count = int(shard_count_cfg) + else: + shard_count = derive_shard_count(survivors, t_small, shard_target) + + objects = route(survivors, t_small, shard_count) + warnings = budget_warnings(objects, survivors, shard_target, max_solo, max_objects) + stored_by_slot = {o["slot"]: o for o in (prior or {}).get("objects", [])} + + manifest = { + "schema_version": 1, "badge": badge, "project": project_name, + "source_folder": source_folder, "file_pattern": file_pattern, + "routing": {"t_small": t_small, "shard_count": shard_count, + "shard_target": shard_target}, + "exclusion": exclusion, + "objects": [] if fresh else list((prior or {}).get("objects", [])), + } + index = {o["slot"]: i for i, o in enumerate(manifest["objects"])} + + shipped, skipped = [], [] + current_slots = {o["slot"] for o in objects} + for obj in objects: + stored = stored_by_slot.get(obj["slot"]) + if not fresh and object_unchanged(obj, stored, survivors): + skipped.append(obj["slot"]) # stored entry stays in manifest + continue + + stem = object_project_name(project_name, obj, shard_count) + r = ship_one(obj, stem) + detail = "s" if obj["kind"] == "solo" else "h" + entry = { + "slot": obj["slot"], "kind": obj["kind"], "arcnames": list(obj["arcnames"]), + "blonde": make_blonde(badge, now_epoch, detail), + "fortress_tar": r["fortress_tar"], "run_timestamp": r["run_timestamp"], + "zip_checksum": r["zip_checksum"], + "fingerprint": object_fingerprint(obj["kind"], obj["arcnames"], survivors), + "n_files": r["n_files"], "source_bytes": r["source_bytes"], + "tar_bytes": r.get("tar_bytes"), + } + if obj["slot"] in index: + manifest["objects"][index[obj["slot"]]] = entry + else: + index[obj["slot"]] = len(manifest["objects"]) + manifest["objects"].append(entry) + shipped.append(obj["slot"]) + write_manifest(vault_dir, badge, manifest) # resume-safe: persist per object + + # Prune slots no longer present (deleted big files / emptied shards). + dropped = [s for s in list(index) if s not in current_slots] + if dropped: + manifest["objects"] = [o for o in manifest["objects"] + if o["slot"] in current_slots] + write_manifest(vault_dir, badge, manifest) + elif not manifest["objects"] and not prior: + # Nothing shipped and no prior manifest (e.g. everything skipped is impossible + # at baseline) — still write so reclaim can see the (empty) target. + write_manifest(vault_dir, badge, manifest) + + return { + "badge": badge, "shard_count": shard_count, "t_small": t_small, + "n_objects": len(objects), "shipped": shipped, "skipped": skipped, + "dropped": dropped, "warnings": warnings, "exclusion": exclusion, + "manifest": manifest, + } + + # --------------------------------------------------------------------------- # Exclusion primitives (see docs/EXCLUSION_SPEC.md). # @@ -2135,6 +2379,24 @@ def send_email(subject, body): cleanup_zip_on_success = config["cleanup_zip_on_success"] exclude_spec = config["exclude_spec"] exclude_optional = config["exclude_optional"] + size_routing = config["size_routing"] + t_small = config["t_small"] + shard_count_cfg = config["shard_count"] + shard_target = config["shard_target"] + + # Size-routing runs LOCAL only (RFC §2.8): the per-target manifest lives on a + # shared durable log_dir that __main__ reads/writes, and reclaim later reads. Refuse + # --globus up front (before any endpoint auth) rather than diverge manifest and logs + # across hosts. (Phase-1 exclusion is stateless per-run and still runs under --globus.) + if size_routing and use_globus: + msg = ("size_routing is enabled in this config, but --globus was passed.\n" + "Routing keeps a per-target manifest on shared durable storage and runs\n" + "LOCAL only (docs/RFC_incremental_v2.md §2.8). Re-run without --globus\n" + "(e.g. via archive_local.sbatch), or disable size_routing for a --globus run.") + print(f"\n ERROR: {msg}") + send_alert_email(emails, + f"[FAILED] Fortress archive: size_routing + --globus ({project_name})", msg) + sys.exit(1) # `run` abstracts how a worker function is executed: in-process by default, or # submitted to the endpoint and waited on with --globus. Step 1/Step 2 call run() @@ -2231,6 +2493,92 @@ def run(fn, *fn_args): def run(fn, *fn_args): return gce.submit(fn, *fn_args).result() + # ------------------------------------------------------------------ # + # Size-routing path (Phase 2, RFC §2.5) — LOCAL only (--globus refused above). + # Partition survivors into solo/shard OBJECTS and ship only the changed/new ones + # (skip-unchanged = the append win). Each object goes through the SAME + # make_zip_files -> send_to_fortress chain (verify chain byte-for-byte). Terminal: + # this branch replaces the whole-target Step 1/Step 2 below. + # ------------------------------------------------------------------ # + if size_routing: + import time + + print("#" * 64) + print("# SIZE-ROUTING (Phase 2) — per-object append; whole-target ship disabled") + print(f"# T_small: {clamp_t_small(t_small)} bytes shard_target: {shard_target} bytes") + print("#" * 64) + + vault_dir = vault_dir_for(log_dir) + now_epoch = int(time.time()) + + # One walk: survivor catalog (post-exclusion) + the exclusion report to record. + try: + survivors, exclusion = run(enumerate_source_catalog_excluded, + source_folder, file_pattern, + exclude_spec, exclude_optional) + except Exception as e: + msg = (f"Size-routing enumerate failed for {project_name}.\n\n" + f"Source: {source_folder}\nPattern: {file_pattern}\n\nError:\n{e}") + print(f"\n ERROR: {msg}") + send_alert_email(emails, f"[FAILED] Fortress routing enumerate: {project_name}", msg) + sys.exit(1) + + if not survivors: + msg = (f"Size-routing found no files to archive for {project_name} " + f"(pattern {file_pattern!r} under {source_folder}, after exclusion).") + print(f"\n ERROR: {msg}") + send_alert_email(emails, f"[FAILED] Fortress routing: nothing to archive ({project_name})", msg) + sys.exit(1) + + def ship_one(obj, stem): + # exclusion already applied to survivors -> pass exclude_spec=None (no re-gate). + zp, zc, fc, mem, sb, _ex = run( + make_zip_files, source_folder, file_pattern, tmp_dir, stem, + compression, allow_empty_files, None, (), obj["arcnames"]) + ftar, txt_path, json_path = run( + send_to_fortress, zp, zc, fc, mem, sb, source_folder, file_pattern, + fortress_base, emails, stem, log_dir, cleanup_zip_on_success, None) + return {"fortress_tar": ftar, "zip_checksum": zc, "n_files": len(mem), + "source_bytes": sb, "txt_path": txt_path, "json_path": json_path, + "run_timestamp": os.path.splitext(os.path.basename(txt_path))[0]} + + try: + summary = route_and_ship( + project_name, source_folder, file_pattern, survivors, exclusion, + vault_dir, t_small, shard_count_cfg, shard_target, now_epoch, + ship_one, fresh=(not resume_enabled)) + except Exception as e: + msg = (f"Size-routing ship failed for {project_name}.\n\n" + f"Some objects may already be on tape and recorded in the manifest;\n" + f"re-running resumes (unchanged objects skip).\n\nError:\n{e}") + print(f"\n ERROR: {msg}") + send_alert_email(emails, f"[FAILED] Fortress routing ship: {project_name}", msg) + sys.exit(1) + + n_ship, n_skip = len(summary["shipped"]), len(summary["skipped"]) + print() + print("=" * 60) + print("SIZE-ROUTING COMPLETE") + print("=" * 60) + print(f" Badge: {summary['badge']} K={summary['shard_count']}") + print(f" Objects: {summary['n_objects']} total — {n_ship} shipped, " + f"{n_skip} skipped (unchanged), {len(summary['dropped'])} dropped") + for w in summary["warnings"]: + print(f" WARNING: {w}") + print(f" Manifest: {manifest_path(vault_dir, summary['badge'])}") + + body = (f"Size-routing archive complete for {project_name}.\n\n" + f"Badge {summary['badge']} (K={summary['shard_count']}, " + f"T_small={summary['t_small']} bytes)\n" + f"Objects: {summary['n_objects']} total — {n_ship} shipped, " + f"{n_skip} skipped (unchanged), {len(summary['dropped'])} dropped.\n" + + ("\nBudget warnings:\n " + "\n ".join(summary["warnings"]) + if summary["warnings"] else "") + + f"\n\nShipped slots: {', '.join(summary['shipped']) or '(none)'}\n" + f"Manifest: {manifest_path(vault_dir, summary['badge'])}\n") + send_alert_email(emails, f"[SUCCESS] Fortress size-routing: {project_name}", body) + sys.exit(0) + # Step 1: Find files, zip, and verify on the compute system print() print("=" * 60) diff --git a/config.example.json b/config.example.json index 38231c2..740815d 100644 --- a/config.example.json +++ b/config.example.json @@ -33,5 +33,17 @@ "exclude_spec": null, "comment_exclude_optional": "Optional (default []). Per-asset opt-out list of Tier-2 (OPTIONAL) rule ids from the exclude_spec (e.g. omics provenance/QC: aux_info, fastp, checksums). Honored only if this asset's file_pattern also ships the raw inputs those artifacts are regenerable from. Omit to back up all OPTIONAL artifacts (safe default).", - "exclude_optional": [] + "exclude_optional": [], + + "comment_size_routing": "Optional (default false). Phase-2 size-routing (docs/RFC_incremental_v2.md §2): instead of one whole-target tar, ship each file >= t_small as its own 'solo' object and bundle smaller files into content-addressed 'shards' (shard = hash(relpath) mod K). Unchanged objects skip, so a GROWING target ships only its new files (the append win). LOCAL mode only — refused under --globus. When true, log_dir MUST be shared+durable (a Depot logs dir), NOT home or scratch: the per-target manifest lives at {log_dir}/_vault/ and reclaim must find it. Omit for the default whole-target behavior.", + "size_routing": false, + + "comment_t_small": "Optional (default 268435456 = 256 MiB; floored at 100 MB). Files >= this ship as solo objects (skip-if-unchanged); smaller files bundle into shards. Only used when size_routing is true.", + "t_small": 268435456, + + "comment_shard_count": "Optional (default null). Number of content-addressed shards K. null => derived at baseline as ceil(small-tier bytes / shard_target). K is FROZEN in the manifest once a baseline exists (a content-addressed map must be stable); changing it requires --fresh. Only used when size_routing is true.", + "shard_count": null, + + "comment_shard_target": "Optional (default 274877906944 = 256 GiB). Target uncompressed shard size, used to derive K at baseline and to warn (not fail) when a live shard exceeds ~2x this. Only used when size_routing is true.", + "shard_target": 274877906944 } diff --git a/tests/test_routing_wiring.py b/tests/test_routing_wiring.py new file mode 100644 index 0000000..a4bac42 --- /dev/null +++ b/tests/test_routing_wiring.py @@ -0,0 +1,272 @@ +""" +Unit tests for Phase-2 size-routing archive wiring (docs/RFC_incremental_v2.md §2, +build-order slice 2): the durability guard, config-key validation, per-object stem, +the survivor-catalog enumerator, and route_and_ship (baseline / append / change / +delete / K-frozen / --fresh / resume). ship_one is injected so no tape/Slurm/Globus +is touched; the manifest is exercised against a real temp vault dir. +""" +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 archive # noqa: E402 + + +MiB = 1024 * 1024 +BIG = 300 * MiB # >= 256 MiB default T_small -> solo +SMALL = 4 * MiB # < T_small -> shard + + +def cat(**entries): + """catalog from name=(mtime_s, size); mtime given in whole seconds.""" + return {a: (mt * 1_000_000_000, sz) for a, (mt, sz) in entries.items()} + + +class Recorder: + """Injectable ship_one: records shipped slots, returns plausible metadata.""" + def __init__(self, catalog, raise_on_call=None): + self.catalog = catalog + self.calls = [] + self.raise_on_call = raise_on_call + + def __call__(self, obj, stem): + self.calls.append(obj["slot"]) + if self.raise_on_call is not None and len(self.calls) == self.raise_on_call: + raise RuntimeError("simulated tape failure") + sb = sum(self.catalog[a][1] for a in obj["arcnames"]) + return {"fortress_tar": "/tape/%s.tar" % stem, + "zip_checksum": "md5:%s" % obj["slot"], + "n_files": len(obj["arcnames"]), "source_bytes": sb, + "run_timestamp": "%s_20260101_000000" % stem} + + +def ship(catalog, vault, recorder, fresh=False, shard_count_cfg=4, + shard_target=archive.SHARD_TARGET_DEFAULT, exclusion=None): + return archive.route_and_ship( + "proj", "/src", ".*", catalog, exclusion or {"status": "OFF"}, + vault, archive.T_SMALL_DEFAULT, shard_count_cfg, shard_target, + 1_700_000_000, recorder, fresh=fresh) + + +class TestRoutingLogDirOk(unittest.TestCase): + def test_rejects_home_default(self): + ok, why = archive.routing_log_dir_ok("~/globus_archive_tmp") + self.assertFalse(ok) + self.assertIn("default", why) + + def test_rejects_under_home(self): + ok, why = archive.routing_log_dir_ok(os.path.expanduser("~/some/logs")) + self.assertFalse(ok) + self.assertIn("home", why) + + def test_rejects_scratch_component(self): + ok, why = archive.routing_log_dir_ok("/scratch/negishi/jdoucett/logs") + self.assertFalse(ok) + self.assertIn("scratch", why) + + def test_accepts_depot_shared_path(self): + ok, why = archive.routing_log_dir_ok("/depot/florasense/etc/logs_fortress/archive_log") + self.assertTrue(ok) + self.assertEqual(why, "") + + +class TestObjectProjectName(unittest.TestCase): + def test_solo_and_shard_stems(self): + solo = {"slot": "solo:9f2c1ab7deadbeef", "kind": "solo", "arcnames": ["x"]} + shard = {"slot": "shard:3", "kind": "shard", "arcnames": ["y"]} + self.assertEqual(archive.object_project_name("P", solo, 8), "P__solo-9f2c1ab7deadbeef") + self.assertEqual(archive.object_project_name("P", shard, 8), "P__shard-3of8") + + +class TestConfigValidation(unittest.TestCase): + def _write_cfg(self, d, **over): + cfg = {"project_name": "p", "source_folder": "/s", "file_pattern": ".*", + "fortress_base_dir": "/f", "emails": ["a@b.c"]} + cfg.update(over) + path = os.path.join(d, "c.json") + with open(path, "w") as fh: + json.dump(cfg, fh) + return path + + def test_defaults_off(self): + with tempfile.TemporaryDirectory() as d: + c = archive.load_config(self._write_cfg(d)) + self.assertFalse(c["size_routing"]) + self.assertEqual(c["t_small"], archive.T_SMALL_DEFAULT) + self.assertIsNone(c["shard_count"]) + + def test_durability_guard_rejects_home_logdir(self): + with tempfile.TemporaryDirectory() as d: + p = self._write_cfg(d, size_routing=True, + log_dir=os.path.expanduser("~/globus_archive_tmp")) + with self.assertRaises(ValueError): + archive.load_config(p) + + def test_durability_guard_accepts_depot_logdir(self): + with tempfile.TemporaryDirectory() as d: + p = self._write_cfg(d, size_routing=True, + log_dir="/depot/florasense/etc/logs_fortress") + c = archive.load_config(p) + self.assertTrue(c["size_routing"]) + + def test_bad_shard_count_rejected(self): + with tempfile.TemporaryDirectory() as d: + p = self._write_cfg(d, size_routing=True, + log_dir="/depot/x/logs", shard_count=0) + with self.assertRaises(ValueError): + archive.load_config(p) + + +class TestEnumerateSurvivors(unittest.TestCase): + def _make(self, d): + for rel, body in [("a.txt", b"aaa"), ("b.txt", b"bbbb")]: + with open(os.path.join(d, rel), "wb") as fh: + fh.write(body) + + def test_no_spec_full_catalog(self): + with tempfile.TemporaryDirectory() as src: + self._make(src) + survivors, report = archive.enumerate_source_catalog_excluded(src, ".*") + self.assertEqual(set(survivors), {"a.txt", "b.txt"}) + self.assertEqual(report["status"], "OFF") + for a in survivors: + self.assertEqual(len(survivors[a]), 2) # (mtime_ns, size) + + def test_exclusion_drops_from_survivors(self): + with tempfile.TemporaryDirectory() as src, tempfile.TemporaryDirectory() as sd: + self._make(src) + spec_path = os.path.join(sd, "spec.json") + with open(spec_path, "w") as fh: + fh.write("{}") + orig = archive.evaluate_exclusions + archive.evaluate_exclusions = lambda m, s, sp, o: { + "status": "ON", "excluded_arcnames": ["b.txt"], "excluded": [], "counts": {}} + try: + survivors, report = archive.enumerate_source_catalog_excluded( + src, ".*", exclude_spec=spec_path) + finally: + archive.evaluate_exclusions = orig + self.assertEqual(set(survivors), {"a.txt"}) + self.assertEqual(report["status"], "ON") + + +class TestRouteAndShip(unittest.TestCase): + def test_baseline_ships_all_and_writes_manifest(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c = cat(big1=(1, BIG), a=(1, SMALL), b=(1, SMALL)) + rec = Recorder(c) + summ = ship(c, vault, rec) + expected = {o["slot"] for o in archive.route(c, shard_count=4)} + self.assertEqual(set(summ["shipped"]), expected) + self.assertEqual(summ["skipped"], []) + m = archive.load_manifest(vault, summ["badge"]) + self.assertEqual({o["slot"] for o in m["objects"]}, expected) + self.assertEqual(m["routing"]["shard_count"], 4) + + def test_append_ships_only_new_solo(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c1 = cat(big1=(1, BIG), a=(1, SMALL)) + ship(c1, vault, Recorder(c1)) + c2 = cat(big1=(1, BIG), a=(1, SMALL), big2=(2, BIG)) # append a big file + rec2 = Recorder(c2) + summ = ship(c2, vault, rec2) + self.assertEqual(summ["shipped"], [archive.solo_slot("big2")]) + self.assertIn(archive.solo_slot("big1"), summ["skipped"]) + m = archive.load_manifest(vault, summ["badge"]) + self.assertEqual(len(m["objects"]), + len(archive.route(c2, shard_count=4))) + + def test_solo_change_reships_only_that_solo(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c1 = cat(big1=(1, BIG), big2=(1, BIG)) + ship(c1, vault, Recorder(c1)) + c2 = cat(big1=(5, BIG), big2=(1, BIG)) # big1 mtime bumped + summ = ship(c2, vault, Recorder(c2)) + self.assertEqual(summ["shipped"], [archive.solo_slot("big1")]) + self.assertIn(archive.solo_slot("big2"), summ["skipped"]) + + def test_shard_member_change_reships_whole_shard(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c1 = cat(a=(1, SMALL), b=(1, SMALL)) + ship(c1, vault, Recorder(c1), shard_count_cfg=4) + c2 = cat(a=(1, SMALL + 1), b=(1, SMALL)) # a grew + summ = ship(c2, vault, Recorder(c2), shard_count_cfg=4) + changed_shard = archive.shard_slot(archive.shard_of("a", 4)) + self.assertIn(changed_shard, summ["shipped"]) + # b's shard (if different) is untouched + b_shard = archive.shard_slot(archive.shard_of("b", 4)) + if b_shard != changed_shard: + self.assertIn(b_shard, summ["skipped"]) + + def test_deleted_big_file_drops_slot(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c1 = cat(big1=(1, BIG), big2=(1, BIG)) + ship(c1, vault, Recorder(c1)) + c2 = cat(big2=(1, BIG)) # big1 deleted + rec2 = Recorder(c2) + summ = ship(c2, vault, rec2) + self.assertIn(archive.solo_slot("big1"), summ["dropped"]) + self.assertEqual(rec2.calls, []) # nothing shipped (big2 unchanged) + m = archive.load_manifest(vault, summ["badge"]) + slots = {o["slot"] for o in m["objects"]} + self.assertNotIn(archive.solo_slot("big1"), slots) + self.assertIn(archive.solo_slot("big2"), slots) + + def test_k_frozen_across_runs(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c = cat(big1=(1, BIG), a=(1, SMALL)) + s1 = ship(c, vault, Recorder(c), shard_count_cfg=4) + self.assertEqual(s1["shard_count"], 4) + # a later run with a different shard_count must NOT change K + s2 = ship(c, vault, Recorder(c), shard_count_cfg=16) + self.assertEqual(s2["shard_count"], 4) + + def test_fresh_reships_everything(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c = cat(big1=(1, BIG), a=(1, SMALL)) + ship(c, vault, Recorder(c)) + rec2 = Recorder(c) + summ = ship(c, vault, rec2, fresh=True) + self.assertEqual(set(summ["shipped"]), + {o["slot"] for o in archive.route(c, shard_count=4)}) + self.assertEqual(summ["skipped"], []) + + def test_records_exclusion_in_manifest(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c = cat(big1=(1, BIG)) + excl = {"status": "ON", "excluded_arcnames": ["x.json"], "excluded": []} + summ = ship(c, vault, Recorder(c), exclusion=excl) + m = archive.load_manifest(vault, summ["badge"]) + self.assertEqual(m["exclusion"]["status"], "ON") + + def test_resume_after_crash_persists_shipped_objects(self): + with tempfile.TemporaryDirectory() as d: + vault = archive.vault_dir_for(d) + c = cat(big1=(1, BIG), big2=(1, BIG), big3=(1, BIG)) + # crash on the 2nd ship: object #1 must already be recorded. + rec = Recorder(c, raise_on_call=2) + with self.assertRaises(RuntimeError): + ship(c, vault, rec) + badge = archive.badge_of("proj", "/src", ".*") + m = archive.load_manifest(vault, badge) + self.assertEqual(len(m["objects"]), 1) # only the first survived + # clean re-run: the persisted object skips, the rest ship. + rec2 = Recorder(c) + summ = ship(c, vault, rec2) + self.assertEqual(len(summ["skipped"]), 1) + self.assertEqual(len(summ["shipped"]), 2) + + +if __name__ == "__main__": + unittest.main()