diff --git a/archive.py b/archive.py index 75de05c..b950c6d 100644 --- a/archive.py +++ b/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. '_volatile' vs '' — + 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. diff --git a/tests/test_skip_unchanged.py b/tests/test_skip_unchanged.py new file mode 100644 index 0000000..fbfb036 --- /dev/null +++ b/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()