feat(routing): Phase-2 size-routing primitives (slice 1, behavior-neutral)

archive.py

@@ -548,6 +548,254 @@ def decode_blonde(blonde):
     }
 
 
+# ---------------------------------------------------------------------------
+# Size-routing primitives (Phase 2 — see docs/RFC_incremental_v2.md §2).
+#
+# Partition a target's survivors into independent Fortress OBJECTS:
+#   - solo:  one file >= T_small, its own object, skip-if-unchanged.
+#   - shard: files < T_small bucketed by shard = hash(relpath) mod K
+#            (K frozen per badge at baseline), whole-shard re-ship on change.
+# Content-addressed => a file's bucket is a pure function of its relpath, stable
+# every run, so adding/changing one file touches exactly one object (no repack
+# cascade) — this is what makes APPEND cheap.
+#
+# Pure, additive, deterministic helpers. NOTHING wires them into the archive or
+# reclaim flow yet — this slice is BEHAVIOR-NEUTRAL. Later slices add
+# route_and_ship (archive), verify_target (reclaim aggregate), the reship-monitor,
+# and restore.py. Routing is LOCAL-only (a later slice refuses --globus). Imports
+# stay inside the bodies to match the file's convention.
+# ---------------------------------------------------------------------------
+
+# Defaults the RFC (§2.2) cites; floor/budgets are enforced/emitted by callers.
+T_SMALL_DEFAULT      = 256 * 1024 * 1024      # 256 MiB — solo/shard cutoff
+T_SMALL_FLOOR        = 100 * 1000 * 1000      # 100 MB  — COS disk->tape line; hard floor
+SHARD_TARGET_DEFAULT = 256 * 1024 ** 3        # 256 GiB — target uncompressed shard size
+MAX_SOLO_DEFAULT     = 500
+MAX_OBJECTS_DEFAULT  = 1000
+
+
+def clamp_t_small(t_small):
+    """
+    Raise a configured T_small to the 100 MB floor if needed (RFC §2.2): below it a
+    solo lands in the Fortress COS-10 disk cache — tape-fronted, so not unsafe, just
+    inefficient. Returns an int >= T_SMALL_FLOOR.
+    """
+    return max(int(t_small), T_SMALL_FLOOR)
+
+
+def shard_of(arcname, shard_count):
+    """
+    Content-addressed bucket for a relpath: k = int(shake_128(arcname)) mod K, in
+    [0, shard_count). PURE + STABLE — same arcname and same K always give the same k,
+    every run, so adding/changing one file touches exactly one shard (no repack
+    cascade). K is frozen per badge at baseline (recorded in the manifest); never
+    recompute it from a changed shard_count without a --fresh re-baseline.
+    """
+    import hashlib
+
+    if shard_count < 1:
+        raise ValueError("shard_count must be >= 1")
+    h = hashlib.shake_128(arcname.encode("utf-8")).digest(8)
+    return int.from_bytes(h, "big") % shard_count
+
+
+def solo_slot(arcname):
+    """
+    Stable logical slot key for a solo object: 'solo:<hex>', hex = 64-bit shake_128
+    of the relpath. Stable per file path across runs (a solo whose bytes change
+    re-ships to a new tar under the SAME slot). 64 bits so hundreds of solos never
+    collide in practice (birthday ~ n^2 / 2^65).
+    """
+    import hashlib
+
+    return "solo:" + hashlib.shake_128(arcname.encode("utf-8")).hexdigest(8)
+
+
+def shard_slot(k):
+    """Slot key for shard k: 'shard:<k>'."""
+    return "shard:{}".format(int(k))
+
+
+def derive_shard_count(catalog, t_small=T_SMALL_DEFAULT,
+                       shard_target=SHARD_TARGET_DEFAULT):
+    """
+    Baseline K for a target: ceil(small-tier bytes / shard_target), floored at 1.
+    Computed ONCE at baseline from the live catalog, then FROZEN in the manifest
+    (immutable for the badge's life — a content-addressed map must be stable). The
+    catalog is {arcname: (mtime_ns, size)} from enumerate_source_catalog.
+    """
+    small_bytes = sum(size for (_m, size) in catalog.values() if size < t_small)
+    if small_bytes <= 0:
+        return 1
+    return max(1, -(-small_bytes // shard_target))   # ceil division, integer-only
+
+
+def route(catalog, t_small=T_SMALL_DEFAULT, shard_count=None,
+          shard_target=SHARD_TARGET_DEFAULT):
+    """
+    Partition a catalog into objects. Returns a list of dicts, deterministic order
+    (solos by slot, then shards by k):
+
+        {"slot": str, "kind": "solo"|"shard", "arcnames": [relpath, ...]}
+
+    Files >= t_small each become a solo object; smaller files bucket into shards by
+    shard_of(arcname, K). Empty buckets are omitted (nothing to ship); K itself is
+    fixed, so an empty bucket that later gains a file simply reappears. If shard_count
+    is None it is derived (derive_shard_count) — callers WITH an existing manifest MUST
+    pass the frozen K from it instead of re-deriving.
+    """
+    if shard_count is None:
+        shard_count = derive_shard_count(catalog, t_small, shard_target)
+
+    solos, buckets = [], {}
+    for arcname in sorted(catalog):
+        size = catalog[arcname][1]
+        if size >= t_small:
+            solos.append({"slot": solo_slot(arcname), "kind": "solo",
+                          "arcnames": [arcname]})
+        else:
+            buckets.setdefault(shard_of(arcname, shard_count), []).append(arcname)
+
+    # Guard the (astronomically unlikely) solo-slot collision rather than silently
+    # merge two distinct files into one slot.
+    slots = [o["slot"] for o in solos]
+    if len(set(slots)) != len(slots):
+        raise RuntimeError("solo slot collision — two relpaths hashed to one solo "
+                           "slot; should be impossible at 64 bits")
+
+    solos.sort(key=lambda o: o["slot"])
+    shards = [{"slot": shard_slot(k), "kind": "shard",
+               "arcnames": sorted(buckets[k])} for k in sorted(buckets)]
+    return solos + shards
+
+
+def object_fingerprint(kind, arcnames, catalog):
+    """
+    Compact fingerprint of an object's CURRENT content, from the live catalog.
+    Change detection = recompute this and compare against the manifest's stored
+    fingerprint for the slot; any difference re-ships the object.
+
+      - solo:  {"mtime_s", "size"} of the single file.
+      - shard: {"members", "byte_sum", "digest"} where digest = shake_128 over the
+               sorted (arcname, floor_s(mtime), size) triples — the authoritative
+               change signal (catches add/remove/modify/size, and a same-count/
+               same-bytes reshuffle a pure sum would miss). members/byte_sum are
+               readable cross-checks.
+
+    BLIND SPOT (documented, shared with reclaim): a same-size + same-mtime in-place
+    rewrite defeats both fingerprints — mitigated by periodic re-hash + --fresh
+    (RFC §2.3). Floors mtimes to whole seconds (floor_s) so sub-second jitter never
+    reads as a change.
+    """
+    import hashlib
+
+    if kind == "solo":
+        mtime_ns, size = catalog[arcnames[0]]
+        return {"mtime_s": floor_s(mtime_ns), "size": size}
+
+    items = sorted(arcnames)
+    parts, byte_sum = [], 0
+    for a in items:
+        mtime_ns, size = catalog[a]
+        parts.append("{}\0{}\0{}".format(a, floor_s(mtime_ns), size))
+        byte_sum += size
+    digest = hashlib.shake_128("\n".join(parts).encode("utf-8")).hexdigest(16)
+    return {"members": len(items), "byte_sum": byte_sum, "digest": digest}
+
+
+def object_unchanged(current_object, stored_object, catalog):
+    """
+    True iff a routed object can be SKIPPED this run — its live content matches what
+    the manifest recorded for that slot (the append win + resume, RFC §2.3/§2.8).
+
+    current_object is a fresh route() entry (so its arcnames reflect live membership,
+    including a small file just appended to a shard); stored_object is the manifest
+    entry for the same slot (or None at baseline). Compares recomputed vs stored
+    fingerprint. A brand-new slot (stored_object is None) is never unchanged.
+    """
+    if not stored_object:
+        return False
+    fp = object_fingerprint(current_object["kind"],
+                            current_object["arcnames"], catalog)
+    return fp == stored_object.get("fingerprint")
+
+
+def budget_warnings(objects, catalog, shard_target=SHARD_TARGET_DEFAULT,
+                    max_solo=MAX_SOLO_DEFAULT, max_objects=MAX_OBJECTS_DEFAULT):
+    """
+    Warn-only object-count / shard-size guards (RFC §2.2). Returns a list of
+    human-readable strings; NEVER raises and never blocks a run — routing proceeds
+    regardless (the operator raises shard_target / re-baselines in response).
+    """
+    warnings = []
+    n_solo = sum(1 for o in objects if o["kind"] == "solo")
+    n_total = len(objects)
+    if n_solo > max_solo:
+        warnings.append("solo objects {} exceed max_solo {}".format(n_solo, max_solo))
+    if n_total > max_objects:
+        warnings.append("total objects {} exceed max_objects_per_target {}"
+                        .format(n_total, max_objects))
+    limit = 2 * shard_target
+    for o in objects:
+        if o["kind"] != "shard":
+            continue
+        b = sum(catalog[a][1] for a in o["arcnames"] if a in catalog)
+        if b > limit:
+            warnings.append("shard {} is {} bytes (> 2x shard_target {}) — raise "
+                            "shard_target and re-baseline (--fresh)"
+                            .format(o["slot"], b, shard_target))
+    return warnings
+
+
+def vault_dir_for(log_dir):
+    """
+    The per-deployment manifest vault: {log_dir}/_vault. The '_vault' child is
+    invisible to reclaim.find_logs' non-recursive *.json glob, so manifests never
+    collide with run-log scanning (RFC §2.4).
+    """
+    import os
+
+    return os.path.join(log_dir, "_vault")
+
+
+def manifest_path(vault_dir, badge):
+    """Path to a target's manifest file inside the vault."""
+    import os
+
+    return os.path.join(vault_dir, "{}.manifest.json".format(badge))
+
+
+def load_manifest(vault_dir, badge):
+    """Load a target's manifest dict, or None if it has none yet (baseline)."""
+    import os
+    import json
+
+    path = manifest_path(vault_dir, badge)
+    if not os.path.exists(path):
+        return None
+    with open(path) as fh:
+        return json.load(fh)
+
+
+def write_manifest(vault_dir, badge, manifest):
+    """
+    Atomically write a target's manifest (tmp + os.replace). Callers treat it as
+    best-effort (a manifest write must never fail an archive already verified on
+    tape); the atomic replace guarantees a reader never sees a half-written file.
+    Returns the written path.
+    """
+    import os
+    import json
+
+    os.makedirs(vault_dir, exist_ok=True)
+    path = manifest_path(vault_dir, badge)
+    tmp = path + ".tmp"
+    with open(tmp, "w") as fh:
+        json.dump(manifest, fh, indent=2, sort_keys=True)
+    os.replace(tmp, path)
+    return path
+
+
 # ---------------------------------------------------------------------------
 # Exclusion primitives (see docs/EXCLUSION_SPEC.md).
 #
@@ -1186,7 +1434,7 @@ def evaluate_exclusions(matched_members, source_folder, spec,
 
 def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
                    compression="deflate", allow_empty_files=False,
-                   exclude_spec=None, exclude_optional=()):
+                   exclude_spec=None, exclude_optional=(), only_arcnames=None):
     """
     On the RCAC compute system: find files in source_folder matching
     file_pattern (regex), zip them preserving relative paths, verify
@@ -1213,6 +1461,12 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
     exclude_optional is the per-asset Tier-2 opt-out list. evaluate_exclusions
     NEVER raises: any problem yields "back up everything" with a recorded reason.
 
+    only_arcnames: optional iterable of relpaths (Phase 2 size-routing, RFC §2.5).
+    When given, the matched set is narrowed to exactly these members — one routed
+    object (a solo's single file, or a shard's bucket). The router applies exclusion
+    LOCALLY before routing, so a routed call passes exclude_spec=None. Default None =
+    unchanged whole-target behavior; no caller supplies it yet (behavior-neutral).
+
     Returns (zip_path, zip_checksum, file_checksums, members, source_bytes,
     exclusion) — the 6th element is the exclusion report (status OFF/DISABLED/ON
     + excluded/demoted/counts) for the run log, banner, and completion email.
@@ -1236,11 +1490,22 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
             if pattern.search(arcname):
                 matched_files.append((full_path, arcname))
 
+    # Size-routing (Phase 2, RFC §2.5): when only_arcnames is given, keep just that
+    # object's members (a solo's one file, or a shard's bucket). The set is already
+    # exclusion-applied by the LOCAL router, so the caller passes exclude_spec=None
+    # and the git gate is not re-run per object. No caller passes this yet — routing
+    # wiring lands in a later slice; here it is inert unless explicitly supplied.
+    if only_arcnames is not None:
+        keep = set(only_arcnames)
+        matched_files = [(fp, arc) for (fp, arc) in matched_files if arc in keep]
+
     if not matched_files:
         raise RuntimeError(
             f"No files matched pattern '{file_pattern}' in '{source_folder}' "
             f"(searched recursively — pattern is tested against relative paths "
             f"such as 'raw/IMG_001.raw')"
+            + ("" if only_arcnames is None else
+               " — after the only_arcnames routing filter (empty object)")
         )
 
     # Second, subtractive filter (docs/EXCLUSION_SPEC.md): drop git-COVERED