fix(routing): tolerate transient nonzero htar/htar_large rc in ship_object

archive.py

@@ -2612,9 +2612,20 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir
     and writes the object's own {stem}_{ts}.txt/.json run-log pair (invariant #4).
 
     Returns dict(fortress_tar, transport, idx, run_timestamp, n_files, source_bytes,
-    tar_bytes, file_checksums, txt_path, json_path). Raises RuntimeError on any transfer
-    or verify failure (the caller emails + a re-run resumes: shipped objects are already
-    in the manifest). Imports stay inside the body (consistency with the other workers).
+    tar_bytes, file_checksums, txt_path, json_path, create_warning). Raises RuntimeError
+    on any transfer or verify failure (the caller emails + a re-run resumes: shipped
+    objects are already in the manifest). Imports stay inside the body (consistency with
+    the other workers).
+
+    A nonzero htar/htar_large create exit code is NOT immediately fatal: HPSS can return
+    a transient nonzero rc under concurrent load even though the tar landed correctly
+    (observed on Negishi, job 40527505 — rc=70, stdout ended "HTAR SUCCESSFUL", tar
+    confirmed fine). On rc != 0, the full create stdout is written to
+    {stem}_{ts}.htar_create.log, then `hsi ls` checks whether the tar (+ .idx for `htar`)
+    actually landed. If not, this raises exactly as before (genuine failure). If it did
+    land, this proceeds to the round-trip retrieve + per-file MD5 verify below — a
+    stronger proof than a size check — and only raises if THAT fails too; a warning
+    (`create_warning`) is recorded in the return value and the run-log instead.
 
     Command strings were verified live on Negishi (HPSS 10.3.0.p3, 2026-07-09):
       htar:        cd <src>; htar -P -cvf <fortress>/<stem>_<ts>.tar -L <memberlist>
@@ -2676,10 +2687,41 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir
             cmd = ["htar_large", "-cvf", fortress_tar] + arcnames
         create = subprocess.run(cmd, cwd=source_folder,
                                 stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
+        create_warning = None
         if create.returncode != 0:
-            raise RuntimeError(
-                f"{transport} create failed (rc={create.returncode}) for {run_timestamp}:\n"
-                + create.stdout.decode("utf-8", "replace")[-4000:])
+            create_stdout = create.stdout.decode("utf-8", "replace")
+            debug_log = os.path.join(log_dir, f"{run_timestamp}.htar_create.log")
+            with open(debug_log, "w") as fh:
+                fh.write(create_stdout)
+
+            def _hsi_ls_ok(path):
+                # Fail-safe: any error checking presence is treated as "not landed",
+                # so an hsi/PATH problem can't mask a genuine create failure.
+                try:
+                    r = subprocess.run(["hsi", "-q", "ls", path],
+                                       stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+                    return r.returncode == 0
+                except OSError:
+                    return False
+
+            landed = _hsi_ls_ok(fortress_tar) and (idx is None or _hsi_ls_ok(idx))
+            if not landed:
+                raise RuntimeError(
+                    f"{transport} create failed (rc={create.returncode}) for {run_timestamp}, "
+                    f"and hsi ls could not confirm the tar landed on Fortress. "
+                    f"Full output: {debug_log}\n" + create_stdout[-4000:])
+            # rc != 0 but hsi ls confirms the tar (+ .idx) is on Fortress — seen on
+            # Negishi 2026-07-11 (job 40527505, htar rc=70 under 14 concurrent archive
+            # jobs hitting Fortress; stdout's own last line was "HTAR SUCCESSFUL" and
+            # the tar was fine). Don't trust rc alone, but don't trust hsi ls alone
+            # either: fall through to the round-trip retrieve+verify below, which
+            # proves every member is byte-identical — a stronger check than a size
+            # comparison. Only raise above if the tar is genuinely missing.
+            create_warning = (
+                f"{transport} exited rc={create.returncode} but hsi ls confirmed the "
+                f"tar{' + idx' if idx else ''} is present on Fortress; proceeding to "
+                f"round-trip verification before trusting it. Full create output: {debug_log}")
+            print(f"  WARNING: {create_warning}")
 
         # Round-trip verify: pull the members back and re-check per-file MD5.
         retrieve_dir = os.path.join(tmp_dir, f"{run_timestamp}.verify")
@@ -2727,6 +2769,8 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir
         "source_files": file_checksums, "source_bytes": source_bytes,
         "n_files": len(arcnames),
     }
+    if create_warning:
+        record["create_warning"] = create_warning
     txt_path = os.path.join(log_dir, f"{run_timestamp}.txt")
     json_path = os.path.join(log_dir, f"{run_timestamp}.json")
     with open(json_path, "w") as fh:
@@ -2735,13 +2779,14 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir
         fh.write(f"Routed object {object_stem} ({transport})\n"
                  f"Fortress: {fortress_tar}\n"
                  f"Members: {len(arcnames)}  Bytes: {source_bytes}\n"
-                 f"Round-trip MD5 verified: OK\n")
+                 f"Round-trip MD5 verified: OK\n"
+                 + (f"WARNING: {create_warning}\n" if create_warning else ""))
 
     return {"fortress_tar": fortress_tar, "transport": transport, "idx": idx,
             "run_timestamp": run_timestamp, "n_files": len(arcnames),
             "source_bytes": source_bytes, "tar_bytes": None,
             "file_checksums": file_checksums, "txt_path": txt_path,
-            "json_path": json_path}
+            "json_path": json_path, "create_warning": create_warning}
 
 
 # ── Main ──────────────────────────────────────────────────────────────────────

tests/test_ship_object_htar_rc.py

@@ -0,0 +1,148 @@
+"""
+Unit tests for archive.ship_object()'s tolerance of a nonzero htar/htar_large create
+exit code (Negishi 2026-07-11, job 40527505: htar rc=70 under 14 concurrent archive
+jobs hitting Fortress, even though stdout ended "HTAR SUCCESSFUL" and the tar was
+confirmed fine via hsi ls). A nonzero rc must no longer be unconditionally fatal: it
+should only raise if `hsi ls` also can't confirm the tar (+ .idx) landed, or if the
+subsequent round-trip retrieve+verify fails. A genuinely missing tar must still raise
+exactly as before.
+
+No real hsi/htar/tape: subprocess.run is faked. The round-trip "htar -xvf" step is
+simulated by copying the real source files into the retrieve dir, so the existing
+per-file MD5 verification exercises real bytes, not a stub.
+"""
+import os
+import shutil
+import sys
+import tempfile
+import unittest
+from unittest import mock
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+import archive  # noqa: E402
+
+
+class FakeCompleted:
+    def __init__(self, returncode, stdout=b""):
+        self.returncode = returncode
+        self.stdout = stdout
+
+
+def make_fake_run(source_folder, arcnames, create_rc, create_stdout, ls_ok):
+    """subprocess.run stand-in dispatching on the command shape ship_object issues.
+
+    ls_ok: predicate(path) -> bool, controls the `hsi -q ls <path>` outcome for both
+    the tar and its .idx, so tests can simulate "landed" vs "genuinely missing".
+    """
+    calls = []
+
+    def _fake(cmd, *args, **kwargs):
+        calls.append((cmd, kwargs))
+        if isinstance(cmd, str):
+            return FakeCompleted(0, b"")
+        prog = cmd[0]
+        if prog in ("htar", "htar_large") and "-cvf" in cmd:
+            return FakeCompleted(create_rc, create_stdout)
+        if prog == "hsi" and cmd[1:3] == ["-q", "ls"]:
+            return FakeCompleted(0 if ls_ok(cmd[3]) else 1, b"")
+        if prog == "htar" and cmd[1] == "-xvf":
+            dest = kwargs["cwd"]
+            for arc in arcnames:
+                dst = os.path.join(dest, arc)
+                os.makedirs(os.path.dirname(dst), exist_ok=True)
+                shutil.copyfile(os.path.join(source_folder, arc), dst)
+            return FakeCompleted(0, b"")
+        return FakeCompleted(0, b"")
+
+    return _fake, calls
+
+
+class TestShipObjectHtarRc(unittest.TestCase):
+    def setUp(self):
+        self.d = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.d, ignore_errors=True)
+        self.source_folder = os.path.join(self.d, "src")
+        os.makedirs(os.path.join(self.source_folder, "sub"))
+        with open(os.path.join(self.source_folder, "a.txt"), "wb") as fh:
+            fh.write(b"hello world")
+        with open(os.path.join(self.source_folder, "sub", "b.txt"), "wb") as fh:
+            fh.write(b"more data here")
+        self.arcnames = ["a.txt", "sub/b.txt"]
+        self.catalog = {"a.txt": (0, 11), "sub/b.txt": (0, 14)}
+        self.tmp_dir = os.path.join(self.d, "tmp")
+        self.log_dir = os.path.join(self.d, "logs")
+        self.fortress_base_dir = "/tape/base"
+
+    def _ship(self, fake_run):
+        with mock.patch("subprocess.run", side_effect=fake_run):
+            return archive.ship_object(
+                self.source_folder, self.arcnames, "testproj__shard-0of1",
+                self.catalog, self.fortress_base_dir, self.tmp_dir, self.log_dir,
+                ["ops@example.com"])
+
+    def test_nonzero_rc_but_tar_present_recovers(self):
+        stdout = b"HTAR: a a.txt\nHTAR: a sub/b.txt\nHTAR: HTAR SUCCESSFUL\n"
+        fake_run, calls = make_fake_run(
+            self.source_folder, self.arcnames, create_rc=70, create_stdout=stdout,
+            ls_ok=lambda path: True)
+
+        result = self._ship(fake_run)
+
+        self.assertIsNotNone(result["create_warning"])
+        self.assertIn("rc=70", result["create_warning"])
+        self.assertIn("hsi ls confirmed", result["create_warning"])
+
+        # Full (untruncated) create stdout landed in its own debug log.
+        debug_log = os.path.join(self.log_dir, f"{result['run_timestamp']}.htar_create.log")
+        self.assertTrue(os.path.isfile(debug_log))
+        with open(debug_log, "rb") as fh:
+            self.assertEqual(fh.read(), stdout)
+
+        # The object's own run-log records the warning too.
+        with open(result["json_path"]) as fh:
+            import json
+            record = json.load(fh)
+        self.assertIn("create_warning", record)
+        self.assertEqual(record["status"], "success")
+
+        with open(result["txt_path"]) as fh:
+            txt = fh.read()
+        self.assertIn("WARNING", txt)
+
+        # Round-trip verify genuinely ran (not skipped) — both hsi ls checks fired
+        # (tar + .idx for the "htar" transport) before falling through.
+        ls_paths = [c[0][3] for c in calls if c[0][0] == "hsi" and c[0][1:3] == ["-q", "ls"]]
+        self.assertIn(result["fortress_tar"], ls_paths)
+        self.assertIn(result["idx"], ls_paths)
+        self.assertEqual(result["transport"], "htar")
+
+    def test_nonzero_rc_and_tar_missing_raises(self):
+        fake_run, _calls = make_fake_run(
+            self.source_folder, self.arcnames, create_rc=1,
+            create_stdout=b"HTAR: ERROR: something went wrong\n",
+            ls_ok=lambda path: False)
+
+        with self.assertRaises(RuntimeError) as ctx:
+            self._ship(fake_run)
+        self.assertIn("rc=1", str(ctx.exception))
+        self.assertIn("could not confirm", str(ctx.exception))
+
+        # No run-log written for a genuinely failed object.
+        self.assertEqual(os.listdir(self.log_dir) if os.path.isdir(self.log_dir) else [],
+                          [f for f in os.listdir(self.log_dir)
+                           if f.endswith(".htar_create.log")])
+
+    def test_zero_rc_success_path_unaffected(self):
+        fake_run, _calls = make_fake_run(
+            self.source_folder, self.arcnames, create_rc=0,
+            create_stdout=b"HTAR: HTAR SUCCESSFUL\n", ls_ok=lambda path: True)
+
+        result = self._ship(fake_run)
+
+        self.assertIsNone(result["create_warning"])
+        debug_log = os.path.join(self.log_dir, f"{result['run_timestamp']}.htar_create.log")
+        self.assertFalse(os.path.isfile(debug_log))
+
+
+if __name__ == "__main__":
+    unittest.main()