fix(archive): re-zip fresh on an unusable resume candidate instead of hard-failing

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

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()