feat(skip-unchanged): don't re-archive an unchanged target (incremental archiving Phase 1a)

archive.py

@@ -109,6 +109,20 @@ def load_config(config_path):
             f"got: {config['cleanup_zip_on_success']!r}"
         )
 
+    # skip_unchanged: when true, a target whose live source is the same set as its
+    # last SUCCESSFUL archive — same file count, nothing modified since, same total
+    # bytes (exactly reclaim.verify's no-drift conditions) — is SKIPPED: no new
+    # tar, no new run log, so the prior success log stays the canonical record
+    # reclaim/coverage read. Default false: the unchanged/changed decision is still
+    # computed and logged (free measurement), but the archive runs anyway. The gate
+    # is bypassed when --fresh / --force-resume is given (an explicit re-archive).
+    config.setdefault("skip_unchanged", False)
+    if not isinstance(config["skip_unchanged"], bool):
+        raise ValueError(
+            f"config 'skip_unchanged' must be true or false, "
+            f"got: {config['skip_unchanged']!r}"
+        )
+
     return config
 
 
@@ -301,6 +315,98 @@ def enumerate_source(source_folder, file_pattern):
     return count, newest_mtime
 
 
+def skippable_prior_log(log_dir, project_name):
+    """The prior successful archive log to skip against — or None if skipping is
+    unsafe.
+
+    Returns the latest SUCCESSFUL run-log for project_name ONLY IF it is also the
+    newest run overall. Two deliberate properties:
+
+      * "Latest" is by parsed run_timestamp, NOT by log-file mtime — so a log
+        whose mtime was rewritten out of order (FS migration, cp without -p,
+        snapshot rehydrate) can't pick the wrong canonical archive, and the
+        choice agrees with coverage.classify (which also keys on run_timestamp).
+      * If the NEWEST run is not that success — i.e. a later run is `running` or
+        `failed` (a crash/retry is in flight) — return None so the gate does NOT
+        skip; the in-progress recovery must be allowed to re-establish a success.
+
+    Matches the log's exact 'project' field (not just the filename prefix) so a
+    sibling target sharing this prefix — e.g. '<proj>_volatile' vs '<proj>' —
+    never masquerades as this project's log.
+    """
+    import reclaim
+    newest, newest_dt = None, None
+    success, success_dt = None, None
+    for path in reclaim.find_logs(log_dir, project=project_name):
+        try:
+            log = reclaim.load_json(path)
+        except (ValueError, OSError):
+            continue
+        if log.get("project") != project_name:
+            continue
+        dt = reclaim.run_datetime(log.get("run_timestamp"))
+        if dt is None:
+            continue
+        if newest_dt is None or dt > newest_dt:
+            newest, newest_dt = log, dt
+        if log.get("status") == "success" and (success_dt is None or dt > success_dt):
+            success, success_dt = log, dt
+    # Skip only against a success that is the most recent run (no newer
+    # running/failed run shadowing it).
+    if success is not None and success is newest:
+        return success
+    return None
+
+
+def source_matches_log(log, source_folder, file_pattern):
+    """True if the live source is UNCHANGED versus this archive log.
+
+    Uses exactly the conditions reclaim.verify() uses to rule out DRIFT — same
+    file count, nothing modified after the archive ran, same total bytes — so the
+    skip decision can never disagree with reclaim's notion of "still archived".
+    One metadata-only walk (reclaim.scan_source); reads no file contents. Returns
+    False if the source is gone or the archive time can't be determined (treat as
+    changed → re-archive, the safe default).
+
+    KNOWN LIMITATION (by design): this stat-only triple cannot see an in-place
+    rewrite that preserves both byte length AND mtime — i.e. an equal-length
+    content swap with a preserved/backdated mtime. Gating a *write* on a
+    *detection* heuristic means such a change, if it ever occurred, would never be
+    re-archived while skip_unchanged is on. That is an accepted trade: detecting
+    it would require re-reading every byte each run (defeating the point), and
+    bit-level integrity of already-archived data is the separate job of
+    fs-checksum-sweep (#60), which re-hashes from bytes on a cadence. Research
+    data here is write-once, so the case is pathological; fs-checksum-sweep is the
+    backstop if it ever isn't.
+    """
+    import reclaim
+    # Only a log for the SAME target definition is comparable. If the archive's
+    # scope changed — a different file_pattern (selects a different set) or a
+    # re-pointed source_folder — the prior count/bytes describe a different
+    # selection, so re-archive under the new definition rather than skip.
+    if (log.get("file_pattern") or "") != (file_pattern or ""):
+        return False
+    log_src = log.get("source_folder")
+    if log_src and os.path.normpath(log_src) != os.path.normpath(source_folder):
+        return False
+    cutoff = reclaim.run_datetime(log.get("run_timestamp"))
+    if cutoff is None:
+        return False
+    scan = reclaim.scan_source(source_folder, file_pattern, cutoff)
+    if scan is None:
+        return False
+    live_count, newer_count, _samples, live_bytes = scan
+    archived_count = len(log.get("source_files") or {})
+    archived_bytes = log.get("source_bytes")
+    if live_count != archived_count:
+        return False
+    if newer_count > 0:
+        return False
+    if archived_bytes is not None and live_bytes != archived_bytes:
+        return False
+    return True
+
+
 def make_zip_files(source_folder, file_pattern, tmp_dir, project_name,
                    compression="deflate", allow_empty_files=False):
     """
@@ -862,6 +968,34 @@ def send_email(subject, body):
     allow_empty_files = config["allow_empty_files"]
     cleanup_zip_on_success = config["cleanup_zip_on_success"]
 
+    # ── Skip-unchanged gate ───────────────────────────────────────────────────
+    # If this target's live source is the same set as its last SUCCESSFUL archive
+    # (reclaim's no-drift conditions: same count, nothing newer, same bytes),
+    # there is nothing new to send to tape — re-archiving would only write a
+    # duplicate tar + a new run log. The decision is ALWAYS computed and logged
+    # (free measurement of redundancy); it is ACTED ON (skip, writing NO tar and
+    # NO new run log so the prior success log stays canonical for reclaim/coverage)
+    # only when skip_unchanged is enabled and the user did not force a re-archive
+    # with --fresh / --force-resume. Runs before any preflight so an unchanged
+    # target costs only one metadata walk.
+    _forced = (not resume_enabled) or force_resume
+    _prior = skippable_prior_log(log_dir, project_name)
+    if _prior is not None and source_matches_log(_prior, source_folder, file_pattern):
+        _since = _prior.get("run_timestamp")
+        if config["skip_unchanged"] and not _forced:
+            print(f"\n[skip-unchanged] {project_name}: unchanged since {_since} — "
+                  f"skipping (no tar, no new log; prior archive remains canonical).")
+            sys.exit(0)
+        elif config["skip_unchanged"] and _forced:
+            print(f"\n[skip-unchanged] {project_name}: unchanged since {_since}, but "
+                  f"--fresh/--force-resume forces a re-archive.")
+        else:
+            print(f"\n[skip-unchanged] {project_name}: WOULD skip (unchanged since "
+                  f"{_since}); skip_unchanged disabled — archiving anyway.")
+    elif _prior is not None:
+        print(f"\n[skip-unchanged] {project_name}: changed since "
+              f"{_prior.get('run_timestamp')} — archiving.")
+
     # `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()
     # and are identical in both modes.

tests/test_skip_unchanged.py

@@ -0,0 +1,207 @@
+"""archive.py — the skip-unchanged gate.
+
+The gate lets the daily repository backup skip a target whose live source is the
+same set as its last SUCCESSFUL archive, and — crucially — write NO new run log
+when it skips, so reclaim/coverage keep reading the prior success log unchanged.
+
+Covers the two decision helpers (latest_success_log, source_matches_log) and the
+end-to-end "skip writes nothing" property via a real subprocess run.
+"""
+import hashlib
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO))
+import archive  # noqa: E402
+
+# Fixed file mtime far from any run_timestamp used below, so the local-time
+# interpretation of run_datetime() vs the UTC-epoch mtime can't flip the
+# newer-than comparison regardless of the test host's timezone.
+FILE_MTIME = 1780272000  # 2026-06-01 00:00:00 UTC
+
+
+def _seed_source(d):
+    """Two files with a fixed past mtime. Returns (source_files dict, total bytes)."""
+    files = {"a.txt": b"hello", "b.txt": b"world!!"}
+    src_files, total = {}, 0
+    for rel, data in files.items():
+        p = os.path.join(d, rel)
+        with open(p, "wb") as fh:
+            fh.write(data)
+        os.utime(p, (FILE_MTIME, FILE_MTIME))
+        src_files[rel] = hashlib.md5(data).hexdigest()
+        total += len(data)
+    return src_files, total
+
+
+def _write_log(log_dir, project, ts, **over):
+    rec = {
+        "project": over.get("project", project),
+        "run_timestamp": f"{project}_{ts}",
+        "source_folder": over.get("source_folder", "/some/src"),
+        "fortress_tar": "/group/fake.tar",
+        "source_files": over.get("source_files", {"a.txt": "x"}),
+        "source_bytes": over.get("source_bytes", 1),
+        "file_pattern": over.get("file_pattern", ".*"),
+        "status": over.get("status", "success"),
+        "error": None,
+    }
+    path = os.path.join(log_dir, f"{over.get('fname_project', project)}_{ts}.json")
+    with open(path, "w") as fh:
+        json.dump(rec, fh)
+    return path
+
+
+class TestSkippablePriorLog(unittest.TestCase):
+    def test_returns_success_when_it_is_newest(self):
+        with tempfile.TemporaryDirectory() as d:
+            _write_log(d, "P", "20260101_000000", status="success")
+            _write_log(d, "P", "20260201_000000", status="success")  # newest success
+            got = archive.skippable_prior_log(d, "P")
+            self.assertIsNotNone(got)
+            self.assertEqual(got["run_timestamp"], "P_20260201_000000")
+
+    def test_newer_failure_blocks_skip(self):
+        # a later run that failed = a retry/recovery is in flight → do NOT skip
+        with tempfile.TemporaryDirectory() as d:
+            _write_log(d, "P", "20260201_000000", status="success")
+            _write_log(d, "P", "20260301_000000", status="failed")
+            self.assertIsNone(archive.skippable_prior_log(d, "P"))
+
+    def test_newer_running_blocks_skip(self):
+        with tempfile.TemporaryDirectory() as d:
+            _write_log(d, "P", "20260201_000000", status="success")
+            _write_log(d, "P", "20260301_000000", status="running")
+            self.assertIsNone(archive.skippable_prior_log(d, "P"))
+
+    def test_selects_by_run_timestamp_not_file_mtime(self):
+        # find_logs sorts by file mtime; the selector must order by run_timestamp
+        # instead, so a log whose mtime was rewritten out of order can't win.
+        with tempfile.TemporaryDirectory() as d:
+            newer_run = _write_log(d, "P", "20260201_000000", status="success")
+            older_run = _write_log(d, "P", "20260101_000000", status="success")
+            os.utime(newer_run, (1000, 1000))                    # older mtime
+            os.utime(older_run, (2_000_000_000, 2_000_000_000))  # newer mtime (the trap)
+            got = archive.skippable_prior_log(d, "P")
+            self.assertEqual(got["run_timestamp"], "P_20260201_000000")
+
+    def test_none_when_no_success(self):
+        with tempfile.TemporaryDirectory() as d:
+            _write_log(d, "P", "20260101_000000", status="failed")
+            self.assertIsNone(archive.skippable_prior_log(d, "P"))
+
+    def test_prefix_sibling_does_not_masquerade(self):
+        # find_logs globs "P_*.json", which also matches "P_volatile_*.json";
+        # the exact 'project' field check must exclude the sibling target.
+        with tempfile.TemporaryDirectory() as d:
+            _write_log(d, "P_volatile", "20260301_000000", status="success")
+            self.assertIsNone(archive.skippable_prior_log(d, "P"))
+            _write_log(d, "P", "20260101_000000", status="success")
+            got = archive.skippable_prior_log(d, "P")
+            self.assertEqual(got["project"], "P")
+
+
+class TestSourceMatchesLog(unittest.TestCase):
+    def _log(self, src_files, total, ts="20260701_000000", file_pattern=".*",
+             source_folder=None):
+        log = {"run_timestamp": f"P_{ts}", "source_files": src_files,
+               "source_bytes": total, "file_pattern": file_pattern}
+        if source_folder is not None:
+            log["source_folder"] = source_folder
+        return log
+
+    def test_unchanged_matches(self):
+        with tempfile.TemporaryDirectory() as d:
+            sf, total = _seed_source(d)
+            self.assertTrue(archive.source_matches_log(self._log(sf, total), d, ".*"))
+
+    def test_count_change_is_not_match(self):
+        with tempfile.TemporaryDirectory() as d:
+            sf, total = _seed_source(d)
+            with open(os.path.join(d, "c.txt"), "wb") as fh:
+                fh.write(b"new")
+            os.utime(os.path.join(d, "c.txt"), (FILE_MTIME, FILE_MTIME))
+            self.assertFalse(archive.source_matches_log(self._log(sf, total), d, ".*"))
+
+    def test_byte_change_is_not_match(self):
+        with tempfile.TemporaryDirectory() as d:
+            sf, total = _seed_source(d)
+            self.assertFalse(
+                archive.source_matches_log(self._log(sf, total + 1), d, ".*"))
+
+    def test_a_file_newer_than_archive_is_not_match(self):
+        with tempfile.TemporaryDirectory() as d:
+            sf, total = _seed_source(d)
+            # run_timestamp 5 months BEFORE the files → they are "newer"
+            log = self._log(sf, total, ts="20260101_000000")
+            self.assertFalse(archive.source_matches_log(log, d, ".*"))
+
+    def test_pattern_change_is_not_match(self):
+        # a different file_pattern selects a different set — not comparable → re-archive
+        with tempfile.TemporaryDirectory() as d:
+            sf, total = _seed_source(d)
+            self.assertFalse(
+                archive.source_matches_log(self._log(sf, total, file_pattern=r"\.bin$"),
+                                           d, ".*"))
+
+    def test_source_folder_change_is_not_match(self):
+        # a re-pointed source_folder describes a different selection → re-archive
+        with tempfile.TemporaryDirectory() as d:
+            sf, total = _seed_source(d)
+            self.assertFalse(
+                archive.source_matches_log(
+                    self._log(sf, total, source_folder="/some/other/path"), d, ".*"))
+
+    def test_missing_source_is_not_match(self):
+        self.assertFalse(
+            archive.source_matches_log(self._log({"a": "x"}, 1), "/no/such/dir", ".*"))
+
+    def test_unparseable_timestamp_is_not_match(self):
+        with tempfile.TemporaryDirectory() as d:
+            sf, total = _seed_source(d)
+            self.assertFalse(
+                archive.source_matches_log(
+                    {"run_timestamp": "no-stamp", "source_files": sf,
+                     "source_bytes": total}, d, ".*"))
+
+
+class TestSkipWritesNoLog(unittest.TestCase):
+    """The contract-critical property: a skip writes NO new run log, so the prior
+    success log stays the canonical record reclaim/coverage read."""
+
+    def test_skip_unchanged_exits_zero_and_writes_no_new_log(self):
+        with tempfile.TemporaryDirectory() as base:
+            src = os.path.join(base, "src"); os.makedirs(src)
+            logs = os.path.join(base, "logs"); os.makedirs(logs)
+            sf, total = _seed_source(src)
+            seeded = _write_log(logs, "P", "20260701_000000", status="success",
+                                source_folder=src, source_files=sf,
+                                source_bytes=total, file_pattern=".*")
+            cfg = os.path.join(base, "config.json")
+            with open(cfg, "w") as fh:
+                json.dump({
+                    "source_folder": src, "file_pattern": ".*",
+                    "fortress_base_dir": "/group/fake", "emails": ["x@example.edu"],
+                    "project_name": "P", "tmp_dir": os.path.join(base, "tmp"),
+                    "log_dir": logs, "skip_unchanged": True,
+                }, fh)
+
+            before = set(os.listdir(logs))
+            proc = subprocess.run(
+                [sys.executable, str(REPO / "archive.py"), "--local", cfg],
+                capture_output=True, text=True, timeout=60)
+            self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
+            self.assertIn("skipping", proc.stdout)
+            after = set(os.listdir(logs))
+            self.assertEqual(after, before, "skip must not write a new run log")
+            self.assertTrue(os.path.exists(seeded))  # prior log untouched
+
+
+if __name__ == "__main__":
+    unittest.main()