feat(routing): Phase-2 archive wiring — the append engine (slice 2)

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/<project>/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:<hex>  -> {project}__solo-<hex>
+      shard:<k>   -> {project}__shard-<k>of<K>
+    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)