diff --git a/archive.py b/archive.py index 75de05c..1beb696 100644 --- a/archive.py +++ b/archive.py @@ -194,8 +194,13 @@ def find_existing_zip(tmp_dir, project_name): import os import glob pattern = os.path.join(tmp_dir, f"{project_name}_*.zip") - matches = sorted(glob.glob(pattern)) - return matches[-1] if matches else None + # Skip candidates that no longer exist or are zero-byte: /scratch is purged + # periodically, and a crashed zip run can leave a 0-byte stub. Neither is a + # usable resume source, so ignore it and let the caller re-zip from the + # (intact) source rather than "resuming" into a FileNotFoundError. + usable = [m for m in sorted(glob.glob(pattern)) + if os.path.exists(m) and os.path.getsize(m) > 0] + return usable[-1] if usable else None def reconstruct_from_zip(zip_path): @@ -258,6 +263,34 @@ def reconstruct_from_zip(zip_path): return zip_path, zip_checksum, file_checksums, members, source_bytes +def try_reconstruct(existing_zip, runner): + """ + Attempt to rebuild the archive tuple from an existing resume candidate via + `runner` (which dispatches reconstruct_from_zip — on a compute node under + --globus). Returns the tuple on success, or None if the candidate is + unusable (missing/scratch-purged/corrupt) and the caller should re-zip fresh. + + A purged or partially-written zip is FULLY RECOVERABLE — the source on + /depot is intact, so re-zipping from it is correct and complete. Treating + that as a fatal error (hard exit + alert) instead burns the resume + watchdog's retry budget on a self-healing condition and strands the + experiment with nothing on Fortress (this is what happened to X0H_Soy). + """ + import os + try: + return runner(reconstruct_from_zip, existing_zip) + except Exception as e: + print(f" [RESUME] Candidate unusable ({type(e).__name__}: {e}); " + f"re-zipping fresh from source.") + try: + if os.path.exists(existing_zip): + os.remove(existing_zip) + print(f" [RESUME] Removed unusable zip: {existing_zip}") + except OSError as rm_e: + print(f" [RESUME] (could not remove {existing_zip}: {rm_e})") + return None + + def enumerate_source(source_folder, file_pattern): """ Enumerate the live source EXACTLY as make_zip_files would, but without @@ -973,24 +1006,17 @@ def run(fn, *fn_args): # make_zip_files entirely and recompute checksums on a compute node. Lets us # recover from any failure in later steps (upload, verify) without redoing the zip. existing_zip = find_existing_zip(tmp_dir, project_name) if resume_enabled else None + resume_result = None if existing_zip: print(f" [RESUME] Existing zip found: {existing_zip}") print(f" [RESUME] Validating + recomputing checksums on compute node (may take several minutes for large zips)...") - try: - zip_path, zip_checksum, file_checksums, members, source_bytes = run(reconstruct_from_zip, existing_zip) - except Exception as e: - msg = ( - f"Resume from existing zip failed.\n\n" - f"Project: {project_name}\n" - f"Existing zip: {existing_zip}\n\n" - f"If the zip is corrupt or partially written, delete it and rerun to re-zip\n" - f"(or pass --fresh to ignore it):\n" - f" rm {existing_zip}\n\n" - f"Error:\n{e}" - ) - print(f"\n ERROR: {msg}") - send_alert_email(emails, f"[FAILED] Fortress archive resume: {project_name}", msg) - sys.exit(1) + # A missing/purged/corrupt candidate returns None here (not a fatal + # error) so we fall through and re-zip fresh from the intact source. + resume_result = try_reconstruct(existing_zip, run) + if resume_result is None: + existing_zip = None + if resume_result is not None: + zip_path, zip_checksum, file_checksums, members, source_bytes = resume_result # Staleness guard: an existing zip is only safe to reuse if it still reflects # the live source. Reusing a zip from an earlier run after the source has diff --git a/tests/test_resume_fallback.py b/tests/test_resume_fallback.py new file mode 100644 index 0000000..1eb28b3 --- /dev/null +++ b/tests/test_resume_fallback.py @@ -0,0 +1,88 @@ +""" +Resume-candidate robustness: a missing (scratch-purged), zero-byte (partial), +or corrupt zip must NOT crash the archive run — find_existing_zip ignores +unusable candidates, and try_reconstruct returns None so main() re-zips fresh +from the intact source instead of hard-failing and burning the watchdog's +retries (the X0H_Soy stranding). + +Stdlib unittest; run from the repo root: + python3 -m unittest discover -s tests -v +""" +import os +import sys +import tempfile +import shutil +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +import archive + + +def _write(path, data): + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "wb") as fh: + fh.write(data) + + +# run() passes (fn, *args) straight through; this mimics it without Globus. +def _local_runner(fn, *args): + return fn(*args) + + +class TestFindExistingZip(unittest.TestCase): + def setUp(self): + self.d = tempfile.mkdtemp() + + def tearDown(self): + shutil.rmtree(self.d, ignore_errors=True) + + def test_none_when_no_candidate(self): + self.assertIsNone(archive.find_existing_zip(self.d, "T")) + + def test_zero_byte_stub_is_skipped(self): + # A crashed zip run can leave a 0-byte stub — not a usable resume source. + _write(os.path.join(self.d, "T_20260102_000000.zip"), b"") + self.assertIsNone(archive.find_existing_zip(self.d, "T")) + + def test_returns_valid_and_skips_newer_zero_byte(self): + valid = os.path.join(self.d, "T_20260101_000000.zip") + _write(valid, b"PK\x05\x06" + b"\x00" * 18) # non-empty + _write(os.path.join(self.d, "T_20260102_000000.zip"), b"") # newer, 0-byte + # The newer candidate is a partial stub; fall back to the older valid one. + self.assertEqual(archive.find_existing_zip(self.d, "T"), valid) + + +class TestTryReconstruct(unittest.TestCase): + def setUp(self): + self.d = tempfile.mkdtemp() + self.src = os.path.join(self.d, "src") + _write(os.path.join(self.src, "a.txt"), b"hello") + _write(os.path.join(self.src, "sub", "b.bin"), b"\x00" * 10) + + def tearDown(self): + shutil.rmtree(self.d, ignore_errors=True) + + def test_valid_zip_reconstructs(self): + zip_path, *_ = archive.make_zip_files( + self.src, ".*", os.path.join(self.d, "tmp"), "T", + compression="store", allow_empty_files=True) + res = archive.try_reconstruct(zip_path, _local_runner) + self.assertIsNotNone(res) + self.assertEqual(res, archive.reconstruct_from_zip(zip_path)) + + def test_missing_zip_returns_none(self): + # The scratch-purged case: candidate path no longer exists. + gone = os.path.join(self.d, "tmp", "T_20260101_000000.zip") + self.assertIsNone(archive.try_reconstruct(gone, _local_runner)) + + def test_corrupt_zip_returns_none_and_is_removed(self): + bad = os.path.join(self.d, "T_20260101_000000.zip") + _write(bad, b"not a real zip file") + self.assertIsNone(archive.try_reconstruct(bad, _local_runner)) + # Unusable candidate is cleared so the next run re-zips cleanly. + self.assertFalse(os.path.exists(bad)) + + +if __name__ == "__main__": + unittest.main()