diff --git a/archive.py b/archive.py index 75de05c..bddec65 100644 --- a/archive.py +++ b/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): """ diff --git a/tests/test_incremental_primitives.py b/tests/test_incremental_primitives.py new file mode 100644 index 0000000..6a4f201 --- /dev/null +++ b/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()