feat(incremental): level-decision + manifest engine (PR 2)

archive.py

@@ -109,6 +109,19 @@ def load_config(config_path):
             f"got: {config['cleanup_zip_on_success']!r}"
         )
 
+    # incremental: opt-in leveled backup (L0 full + L1/L2… deltas). Default false
+    # keeps the historical whole-snapshot behaviour byte-for-byte. When true, a
+    # run consults the target's manifest and ships only changed/added files as a
+    # delta. The decision/manifest engine lives in this module (decide_level,
+    # update_manifest, …); the __main__ wiring that acts on it lands in a
+    # follow-up. See docs/RFC_incremental_v2.md. Leveling is LOCAL-mode only.
+    config.setdefault("incremental", False)
+    if not isinstance(config["incremental"], bool):
+        raise ValueError(
+            f"config 'incremental' must be true or false, "
+            f"got: {config['incremental']!r}"
+        )
+
     return config
 
 
@@ -423,8 +436,235 @@ def decode_blonde(blonde):
     }
 
 
+# ---------------------------------------------------------------------------
+# Level decision + manifest engine (see docs/RFC_incremental_v2.md §2–3, PR 2).
+#
+# All LOCAL: these read/write log_dir, which exists only on the launching host
+# (invariant #1). Leveling is local-mode only; the __main__ wiring (PR 2b)
+# refuses it under --globus. classify_delta is split out pure for testing.
+# The manifest lives at {log_dir}/_vault/{badge}.manifest.json — the _vault/
+# subdir is invisible to reclaim's non-recursive "*.json" glob (no reclaim
+# edit forced). Per-level (mtime,size) catalogs live in _vault/ too, as
+# {stem}.catalog.json — also hidden from reclaim's glob; effective_catalog folds
+# them back. NOTE: the pure helpers here reference module-level names and are
+# NOT dill-safe — LOCAL-only, never pass them to run()/gce.submit() (invariant #1).
+# ---------------------------------------------------------------------------
+
+_VAULT_SUBDIR = "_vault"
+_MAX_DELTAS = 35   # detail is one base36 char ('1'..'z'); past this, re-anchor a new L0
+
+
+def _vault_dir(log_dir):
+    import os
+    return os.path.join(log_dir, _VAULT_SUBDIR)
+
+
+def manifest_path(log_dir, badge):
+    import os
+    return os.path.join(_vault_dir(log_dir), f"{badge}.manifest.json")
+
+
+def load_manifest(log_dir, badge):
+    """Return the target's manifest dict, or None if absent/unreadable."""
+    import json
+    try:
+        with open(manifest_path(log_dir, badge)) as fh:
+            return json.load(fh)
+    except (OSError, ValueError):
+        return None
+
+
+def write_level_catalog(log_dir, log_stem, catalog):
+    """Persist a level's {arcname:(mtime_ns,size)} catalog as {stem}.catalog.json
+    INSIDE _vault/ (so it's hidden from reclaim's non-recursive "*.json" glob,
+    exactly like the manifest). Atomic (tmp + fsync + os.replace). Returns the
+    catalog_ref (basename) stored in the manifest node."""
+    import os
+    import json
+    vd = _vault_dir(log_dir)
+    os.makedirs(vd, exist_ok=True)
+    ref = f"{log_stem}.catalog.json"
+    path = os.path.join(vd, ref)
+    tmp = path + ".tmp"
+    with open(tmp, "w") as fh:
+        json.dump({a: [mt, sz] for a, (mt, sz) in catalog.items()}, fh)
+        fh.flush()
+        os.fsync(fh.fileno())
+    os.replace(tmp, path)
+    return ref
+
+
+def read_level_catalog(log_dir, catalog_ref):
+    """Load a _vault/{catalog_ref} catalog JSON back to {arcname:(mtime_ns,size)}."""
+    import os
+    import json
+    with open(os.path.join(_vault_dir(log_dir), catalog_ref)) as fh:
+        raw = json.load(fh)
+    return {a: (v[0], v[1]) for a, v in raw.items()}
+
+
+def _ordered_levels(manifest):
+    """The current L0 followed by its deltas in apply order (run_timestamp).
+
+    Star topology: deltas reference their L0 (parent_blonde), not each other, so
+    run_timestamp is the only sequencer. Levels belonging to a superseded L0 are
+    ignored — only the current chain is reconstructed."""
+    l0 = manifest.get("current_l0_blonde")
+    levels = [n for n in manifest.get("levels", [])
+              if n.get("level") == 0 and n.get("blonde") == l0]
+    deltas = [n for n in manifest.get("levels", [])
+              if n.get("level", 0) > 0 and n.get("parent_blonde") == l0]
+    deltas.sort(key=lambda n: n.get("run_timestamp", ""))
+    return levels + deltas
+
+
+def effective_catalog(log_dir, manifest):
+    """Reconstruct the current effective {arcname:(mtime_ns,size)} from the L0
+    and its deltas — apply each level's members, then its deletions, in order.
+    Returns (catalog, total_bytes)."""
+    catalog = {}
+    for node in _ordered_levels(manifest):
+        ref = node.get("catalog_ref")
+        if ref:
+            catalog.update(read_level_catalog(log_dir, ref))
+        for arcname in node.get("deleted", []):
+            catalog.pop(arcname, None)
+    total = sum(sz for (_mt, sz) in catalog.values())
+    return catalog, total
+
+
+def _n_deltas(manifest):
+    """Number of deltas hanging off the current L0."""
+    l0 = manifest.get("current_l0_blonde")
+    return sum(1 for n in manifest.get("levels", [])
+               if n.get("level", 0) > 0 and n.get("parent_blonde") == l0)
+
+
+def next_detail(manifest):
+    """Single base36 detail char for the next delta off the current L0
+    (L0 is '0'; first delta '1', then '2'…). Caller re-anchors past _MAX_DELTAS."""
+    return _B36_DIGITS[_n_deltas(manifest) + 1]
+
+
+def classify_delta(live_catalog, prior_catalog):
+    """PURE. Compare the live catalog against the reconstructed prior catalog
+    (both {arcname:(mtime_ns,size)}). Returns dict(kind, members, deleted):
+
+      'delta'          — members (added ∪ modified) to ship, plus any deletions
+      'deletions_only' — only removals; ships no tar
+      'noop'           — nothing detectably changed
+
+    A file is 'modified' when its size differs OR its mtime (floored to whole
+    seconds via floor_s — sub-second jitter is not a change) differs. ACCEPTED
+    BLIND SPOT (docs/RFC_incremental_v2.md §1): an in-place rewrite preserving
+    BOTH size AND mtime is invisible here. A total-byte-sum check would add
+    nothing at this per-file granularity — a size change is already caught as a
+    per-file 'modified'. The operational mitigation is a periodic --full
+    re-baseline."""
+    added    = [a for a in live_catalog if a not in prior_catalog]
+    removed  = [a for a in prior_catalog if a not in live_catalog]
+    modified = [a for a in live_catalog if a in prior_catalog and
+                (floor_s(live_catalog[a][0]) != floor_s(prior_catalog[a][0])
+                 or live_catalog[a][1] != prior_catalog[a][1])]
+    if not (added or modified or removed):
+        return {"kind": "noop", "members": [], "deleted": []}
+    members = sorted(set(added) | set(modified))
+    if not members:
+        return {"kind": "deletions_only", "members": [], "deleted": sorted(removed)}
+    return {"kind": "delta", "members": members, "deleted": sorted(removed)}
+
+
+def decide_level(log_dir, badge, live_catalog, manifest, force_full=False):
+    """Decide this run's level. Returns dict(kind, detail, parent, members, deleted):
+
+      kind   'full' | 'delta' | 'deletions_only' | 'noop'
+      detail level char ('0' full; '1'..'z' delta; None for noop)
+      parent parent L0 blonde (None for a full)
+      members arcnames to archive (all matched for full; the change set for delta)
+      deleted arcnames removed since the prior level (deltas only)
+
+    A full L0 is chosen when force_full, when no current L0 exists yet, or when
+    the chain has hit _MAX_DELTAS (re-anchor). The manifest is NOT mutated here —
+    the caller assembles + writes the level node after the transfer verifies."""
+    has_l0 = bool(
+        manifest and manifest.get("current_l0_blonde")
+        and any(n.get("level") == 0 and n.get("blonde") == manifest["current_l0_blonde"]
+                for n in manifest.get("levels", []))
+    )
+    if force_full or not has_l0:
+        return {"kind": "full", "detail": "0", "parent": None,
+                "members": sorted(live_catalog), "deleted": []}
+
+    try:
+        prior_catalog, _prior_bytes = effective_catalog(log_dir, manifest)
+    except (OSError, ValueError):
+        # Prior chain unreadable (a missing/corrupt sibling catalog) — re-baseline
+        # to a safe full L0 rather than refuse to back up. A backup engine must
+        # never decline to back up because old state is unreadable.
+        return {"kind": "full", "detail": "0", "parent": None,
+                "members": sorted(live_catalog), "deleted": []}
+    cls = classify_delta(live_catalog, prior_catalog)
+    if cls["kind"] == "noop":
+        return {"kind": "noop", "detail": None,
+                "parent": manifest["current_l0_blonde"], "members": [], "deleted": []}
+
+    # Re-anchor once the chain is too deep to encode in one detail char.
+    if _n_deltas(manifest) + 1 > _MAX_DELTAS:
+        return {"kind": "full", "detail": "0", "parent": None,
+                "members": sorted(live_catalog), "deleted": []}
+
+    return {"kind": cls["kind"], "detail": next_detail(manifest),
+            "parent": manifest["current_l0_blonde"],
+            "members": cls["members"], "deleted": cls["deleted"]}
+
+
+def update_manifest(log_dir, badge, node):
+    """Append a level node to the target's manifest, atomically (tmp + os.replace).
+    Sets current_l0_blonde when node is an L0. Creates the manifest if absent;
+    replaces any existing node with the same blonde (idempotent re-runs). Returns
+    the manifest path. Best-effort by contract — the caller must treat a failure
+    as a warning, never failing an archive already verified on tape."""
+    import os
+    import json
+    vd = _vault_dir(log_dir)
+    os.makedirs(vd, exist_ok=True)
+    path = manifest_path(log_dir, badge)
+    if os.path.exists(path):
+        try:
+            with open(path) as fh:
+                m = json.load(fh)
+        except (OSError, ValueError):
+            # Present but unreadable: NEVER silently overwrite the only on-disk
+            # record of the level chain with a fresh one (that would truncate the
+            # chain and lie). Preserve the bad file aside and refuse — the
+            # best-effort caller logs the warning; the archive is already on tape.
+            aside = path + ".corrupt"
+            try:
+                os.replace(path, aside)
+            except OSError:
+                pass
+            raise RuntimeError(
+                f"manifest {path} is corrupt (moved to {aside}); refusing to "
+                f"overwrite the level chain. Rebuild it from the run logs."
+            )
+    else:
+        m = {"schema_version": 1, "badge": badge, "levels": []}
+    m["levels"] = [n for n in m.get("levels", []) if n.get("blonde") != node.get("blonde")]
+    m["levels"].append(node)
+    if node.get("level") == 0:
+        m["current_l0_blonde"] = node["blonde"]
+    tmp = path + ".tmp"
+    with open(tmp, "w") as fh:
+        json.dump(m, fh, indent=2)
+        fh.flush()
+        os.fsync(fh.fileno())
+    os.replace(tmp, path)
+    return path
+
+
 def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
-                   compression="deflate", allow_empty_files=False):
+                   compression="deflate", allow_empty_files=False,
+                   only_arcnames=None):
     """
     On the RCAC compute system: find files in source_folder matching
     file_pattern (regex), zip them preserving relative paths, verify
@@ -443,6 +683,14 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
     0-byte member. Set True to archive legitimately-empty files as-is (e.g.
     .gitkeep placeholders, empty __init__.py markers); they still get a valid
     (empty-string) MD5 and are counted, so reclaim's drift check stays exact.
+
+    only_arcnames: when not None, restrict the archive to this set/list of
+    arcnames AFTER pattern matching — used to build an incremental delta zip
+    containing only the changed/added members chosen by decide_level. The
+    arcnames must be the same os.path.relpath keys make_zip_files itself
+    produces (invariant #2). None (default) = archive every matched file (the
+    full / L0 behaviour). Only the selected files are zipped and MD5'd, so
+    unchanged bulk is never re-read.
     """
     import os
     import re
@@ -454,12 +702,15 @@ def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
     pattern = re.compile(file_pattern)
     source_folder = os.path.normpath(source_folder)
 
+    only = set(only_arcnames) if only_arcnames is not None else None
     for dirpath, dirnames, filenames in os.walk(source_folder):
         dirnames.sort()
         for fname in sorted(filenames):
             full_path = os.path.join(dirpath, fname)
             arcname = os.path.relpath(full_path, source_folder)
             if pattern.search(arcname):
+                if only is not None and arcname not in only:
+                    continue        # delta: skip members not in the change set
                 matched_files.append((full_path, arcname))
 
     if not matched_files: