feat(incremental): catalog + BLONDE primitives (PR 1)

archive.py

@@ -301,6 +301,128 @@ def enumerate_source(source_folder, file_pattern):
     return count, newest_mtime
 
 
+# ---------------------------------------------------------------------------
+# Leveled-incremental primitives (see docs/RFC_incremental_v2.md, PR 1).
+#
+# Pure, additive helpers: the BLONDE id encode/decode + a per-file (mtime,size)
+# catalog enumerator. NOTHING in the archive flow calls these yet — this PR is
+# behavior-neutral; PR 2 wires them into the level decision. Kept here, beside
+# enumerate_source, because enumerate_source_catalog shares its exact walk/match
+# (and so its arcname keys — invariant #2).
+# ---------------------------------------------------------------------------
+
+def enumerate_source_catalog(source_folder, file_pattern):
+    """
+    Walk the live source EXACTLY as make_zip_files / enumerate_source do and
+    return a per-file catalog keyed by arcname:
+
+        {arcname: (mtime_ns, size)}
+
+    arcname is os.path.relpath(full_path, normpath(source_folder)) — the SAME
+    key make_zip_files uses for file_checksums and the round-trip verify
+    (invariant #2), so a delta member set computed from this catalog lines up
+    with the zip namelist and the per-file MD5 map.
+
+    mtime is st_mtime_ns (integer nanoseconds) — the single catalog precision;
+    callers floor to whole seconds (see floor_s) before comparing against a zip
+    timestamp. size is st_size (bytes). A file that vanishes / can't be stat'd
+    mid-walk is skipped. REMOTE-eligible (runs via run()); imports stay inside
+    the body for Globus Compute / dill (invariant #1).
+    """
+    import os
+    import re
+
+    pattern = re.compile(file_pattern)
+    source_folder = os.path.normpath(source_folder)
+    catalog = {}
+    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
+                catalog[arcname] = (st.st_mtime_ns, st.st_size)
+    return catalog
+
+
+def floor_s(mtime_ns):
+    """
+    Floor an integer-nanosecond mtime to whole epoch seconds.
+
+    The catalog keys on st_mtime_ns; the resume guard and run timestamps work in
+    whole seconds. Any comparison MUST floor BOTH sides to one precision
+    (seconds) or sub-second jitter produces phantom 'modified' files. Integer
+    floor division — never float.
+    """
+    return int(mtime_ns) // 1_000_000_000
+
+
+_B36_DIGITS = "0123456789abcdefghijklmnopqrstuvwxyz"
+
+
+def _b36encode(n):
+    """Lowercase, unpadded base36 of a non-negative integer (lossless)."""
+    if n < 0:
+        raise ValueError("base36 is for non-negative integers")
+    if n == 0:
+        return "0"
+    out = []
+    while n:
+        n, r = divmod(n, 36)
+        out.append(_B36_DIGITS[r])
+    return "".join(reversed(out))
+
+
+def badge_of(project_name, source_folder, file_pattern):
+    """
+    Stable 40-bit identity of a logical backup target, as 8 lowercase base32
+    chars: base32(shake_128(project \\0 source_folder \\0 file_pattern)[:5]).
+
+    Same target -> same badge across every run (it is the manifest key); a change
+    to any of the three fields yields a different target. NUL-joined so the
+    fields can't ambiguously run together. shake_128 / b32encode are stdlib.
+    """
+    import hashlib
+    import base64
+
+    curie = "\0".join([project_name or "", source_folder or "", file_pattern or ""])
+    digest = hashlib.shake_128(curie.encode("utf-8")).digest(5)   # 40 bits
+    return base64.b32encode(digest).decode("ascii").rstrip("=").lower()
+
+
+def make_blonde(badge, run_epoch_seconds, detail):
+    """
+    Compose a BLONDE id 'badge.quaver.detail.genus'
+    (see docs/RFC_incremental_v2.md §3):
+
+      quaver = base36(run_epoch_seconds) — lossless, orders same-level siblings
+               off one L0 by run time.
+      detail = 1-char level: '0' = full (L0), '1','2',... = deltas.
+      genus  = badge[:3] — display only, NEVER a key.
+
+    Dot is a safe separator: base32 (a-z2-7) and base36 (0-9a-z) never contain '.'.
+    """
+    quaver = _b36encode(int(run_epoch_seconds))
+    return "{}.{}.{}.{}".format(badge, quaver, detail, badge[:3])
+
+
+def decode_blonde(blonde):
+    """
+    Inverse of make_blonde. Returns dict(badge, when_epoch, detail, genus);
+    when_epoch is the exact run epoch seconds recovered from quaver (lossless).
+    """
+    badge, quaver, detail, genus = blonde.split(".")
+    return {
+        "badge": badge,
+        "when_epoch": int(quaver, 36),
+        "detail": detail,
+        "genus": genus,
+    }
+
+
 def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
                    compression="deflate", allow_empty_files=False):
     """

tests/test_incremental_primitives.py

@@ -0,0 +1,138 @@
+"""
+Unit tests for the leveled-incremental primitives added in PR 1
+(docs/RFC_incremental_v2.md): the BLONDE id encode/decode, the base36 helper,
+floor_s, and the (mtime,size) catalog enumerator.
+
+These are pure, additive helpers — nothing in the archive flow calls them yet,
+so this suite is fully self-contained (no tape, no Slurm, no Globus).
+"""
+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
+
+
+class TestBadge(unittest.TestCase):
+    def test_deterministic(self):
+        self.assertEqual(
+            archive.badge_of("X1D", "/depot/x", ".*"),
+            archive.badge_of("X1D", "/depot/x", ".*"),
+        )
+
+    def test_format_is_8_char_lowercase_base32(self):
+        badge = archive.badge_of("X1D", "/depot/x", ".*")
+        self.assertEqual(len(badge), 8)               # 40 bits -> 8 base32 chars
+        self.assertRegex(badge, r"^[a-z2-7]{8}$")     # lowercase base32, never a dot
+
+    def test_distinct_per_field(self):
+        base = archive.badge_of("X1D", "/depot/x", ".*")
+        self.assertNotEqual(base, archive.badge_of("X1E", "/depot/x", ".*"))
+        self.assertNotEqual(base, archive.badge_of("X1D", "/depot/y", ".*"))
+        self.assertNotEqual(base, archive.badge_of("X1D", "/depot/x", r"\.bin$"))
+
+    def test_nul_join_prevents_field_run_together(self):
+        # ("ab","c") must not collide with ("a","bc")
+        self.assertNotEqual(
+            archive.badge_of("ab", "c", ".*"),
+            archive.badge_of("a", "bc", ".*"),
+        )
+
+
+class TestBlonde(unittest.TestCase):
+    def test_roundtrip_lossless(self):
+        badge = archive.badge_of("X1D", "/depot/x", ".*")
+        epoch = 1781000000
+        b = archive.make_blonde(badge, epoch, "0")
+        self.assertEqual(b.count("."), 3)
+        d = archive.decode_blonde(b)
+        self.assertEqual(d["badge"], badge)
+        self.assertEqual(d["when_epoch"], epoch)        # exact, lossless
+        self.assertEqual(d["detail"], "0")
+        self.assertEqual(d["genus"], badge[:3])
+
+    def test_delta_detail_levels(self):
+        badge = archive.badge_of("p", "/s", ".*")
+        for lvl in ("0", "1", "2"):
+            self.assertEqual(
+                archive.decode_blonde(archive.make_blonde(badge, 1, lvl))["detail"], lvl
+            )
+
+    def test_quaver_orders_by_run_time(self):
+        badge = archive.badge_of("p", "/s", ".*")
+        early = archive.decode_blonde(archive.make_blonde(badge, 1781000000, "1"))["when_epoch"]
+        late = archive.decode_blonde(archive.make_blonde(badge, 1781000060, "1"))["when_epoch"]
+        self.assertLess(early, late)
+
+    def test_b36_zero(self):
+        self.assertEqual(archive._b36encode(0), "0")
+
+    def test_b36_roundtrip(self):
+        for n in (1, 35, 36, 37, 1781000000, 999999999999):
+            self.assertEqual(int(archive._b36encode(n), 36), n)
+
+    def test_b36_rejects_negative(self):
+        with self.assertRaises(ValueError):
+            archive._b36encode(-1)
+
+
+class TestFloorS(unittest.TestCase):
+    def test_floor_to_seconds(self):
+        self.assertEqual(archive.floor_s(1_500_000_000_123_456_789), 1_500_000_000)
+        self.assertEqual(archive.floor_s(999_999_999), 0)
+
+    def test_sub_second_jitter_collapses(self):
+        lo = 1781000000_000000001
+        hi = 1781000000_999999999
+        self.assertEqual(archive.floor_s(lo), archive.floor_s(hi))
+
+
+class TestCatalog(unittest.TestCase):
+    FILES = [("raw/a.bin", b"aaa"), ("raw/b.bin", b"bbbb"), ("top.txt", b"x")]
+
+    def _make_tree(self, d):
+        os.makedirs(os.path.join(d, "raw"))
+        for rel, data in self.FILES:
+            with open(os.path.join(d, rel), "wb") as fh:
+                fh.write(data)
+
+    def test_keys_are_arcnames_values_are_mtime_ns_size(self):
+        with tempfile.TemporaryDirectory() as d:
+            self._make_tree(d)
+            cat = archive.enumerate_source_catalog(d, ".*")
+            self.assertEqual(set(cat), {"raw/a.bin", "raw/b.bin", "top.txt"})
+            self.assertEqual(cat["raw/a.bin"][1], 3)        # size
+            self.assertEqual(cat["raw/b.bin"][1], 4)
+            self.assertEqual(cat["top.txt"][1], 1)
+            self.assertIsInstance(cat["top.txt"][0], int)   # mtime_ns is int
+
+    def test_arcname_parity_with_enumerate_source(self):
+        # Same walk/match as the resume guard's enumerate_source -> same file set.
+        with tempfile.TemporaryDirectory() as d:
+            self._make_tree(d)
+            cat = archive.enumerate_source_catalog(d, ".*")
+            count, _ = archive.enumerate_source(d, ".*")
+            self.assertEqual(len(cat), count)
+
+    def test_pattern_scopes_catalog(self):
+        with tempfile.TemporaryDirectory() as d:
+            self._make_tree(d)
+            cat = archive.enumerate_source_catalog(d, r"\.bin$")
+            self.assertEqual(set(cat), {"raw/a.bin", "raw/b.bin"})
+
+    def test_top_level_only_pattern(self):
+        with tempfile.TemporaryDirectory() as d:
+            self._make_tree(d)
+            cat = archive.enumerate_source_catalog(d, r"^[^/]+$")
+            self.assertEqual(set(cat), {"top.txt"})
+
+    def test_empty_when_nothing_matches(self):
+        with tempfile.TemporaryDirectory() as d:
+            self._make_tree(d)
+            self.assertEqual(archive.enumerate_source_catalog(d, r"\.nope$"), {})
+
+
+if __name__ == "__main__":
+    unittest.main()