fix(stall): bound the verify retrieve separately from the create (X1D re-ship loop)

CLAUDE.md

@@ -230,9 +230,23 @@ Slurm's own wall-clock (up to 48h) would ever stop it. Long, output-streaming ca
 (create, round-trip retrieve) go through `_run_with_stall_watch()` (`ship_object`) or
 its self-contained inline twin `run_watched()` (`send_to_fortress`, invariant #1 —
 imports inside, no reference to the module-level copy), which kill the process and
-raise if `HTAR_STALL_SECONDS` (2h) passes with no new output — silence, not elapsed
+raise if the call's silence budget passes with no new output — silence, not elapsed
 time, is the signal, since legitimate transfers range from minutes to 22+ hours (see
-Known issues below). Short, no-progress calls (`hsi mkdir`/`hsi ls`) just need
+Known issues below).
+
+The budget is **not one number**: a *create* streams per-file progress, so silence there
+really is a hang and it keeps `HTAR_STALL_SECONDS` (2h); the post-create *round-trip
+verify retrieve* is silent **by design** while HPSS queues and stages the tape, so
+`ship_object` bounds it with the config's **`retrieve_stall_seconds`** instead (default
+in `stall_defaults.json` — data, not code; a per-asset config may override it; there is
+deliberately no code-side fallback, so a caller that forgets to thread it down raises).
+Conflating the two is a real, expensive bug, not a theoretical one: on 2026-08-03 a
+single 6.62 GB `X1D_3_metabolomics_rawspectra` shard had its healthy retrieve killed at
+exactly 2h twice per job (its stall log reads `activity resumed after 7200.1s silence`),
+so verify never passed, no success record was written, and every scheduled run re-shipped
+the whole shard — 48 identical tars, ~318 GB of duplicate tape, over 2.5 weeks.
+
+Short, no-progress calls (`hsi mkdir`/`hsi ls`) just need
 `timeout=HSI_SHORT_CALL_TIMEOUT_SECONDS` (5 min) on a plain `subprocess.run()`. A new
 call site that skips this reintroduces the exact hang this fixes.
 

archive.py

@@ -49,6 +49,51 @@
 # that --local can run in a plain stdlib Python with no Globus install at all.
 
 
+# Engine-wide stall-watch defaults live in DATA, not code: stall_defaults.json beside
+# this module (override the path with FORTRESS_STALL_DEFAULTS). The file is REQUIRED —
+# load_stall_defaults raises on a missing/malformed/invalid file rather than
+# substituting a value buried in a function signature, so the operative timeout for
+# every run is always readable in one inspectable place. A per-asset config may still
+# override any key by name (load_config's setdefault below).
+STALL_DEFAULTS_PATH = os.environ.get(
+    "FORTRESS_STALL_DEFAULTS",
+    os.path.join(os.path.dirname(os.path.realpath(__file__)), "stall_defaults.json"))
+
+# Keys load_stall_defaults requires, each a positive integer number of seconds.
+STALL_DEFAULT_KEYS = ("retrieve_stall_seconds",)
+
+
+def load_stall_defaults(path=None):
+    """Load stall_defaults.json (see STALL_DEFAULTS_PATH). Returns a dict of
+    {key: positive int seconds} for every key in STALL_DEFAULT_KEYS. Raises
+    ValueError if the file is missing, unparseable, or any required key is absent
+    or not a positive integer — fail loud, never a silent code-side fallback."""
+    path = path or STALL_DEFAULTS_PATH
+    try:
+        with open(path) as fh:
+            data = json.load(fh)
+    except OSError as exc:
+        raise ValueError(
+            f"stall defaults file is required but could not be read: {path} ({exc}). "
+            f"Restore it from the fortress-archive checkout, or point "
+            f"FORTRESS_STALL_DEFAULTS at a valid one.")
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"stall defaults file {path} is not valid JSON: {exc}")
+    if not isinstance(data, dict):
+        raise ValueError(f"stall defaults file {path} must contain a JSON object")
+    out = {}
+    for key in STALL_DEFAULT_KEYS:
+        if key not in data:
+            raise ValueError(f"stall defaults file {path} is missing required key {key!r}")
+        val = data[key]
+        if not isinstance(val, int) or isinstance(val, bool) or val <= 0:
+            raise ValueError(
+                f"stall defaults file {path}: {key!r} must be a positive integer "
+                f"(seconds), got: {val!r}")
+        out[key] = val
+    return out
+
+
 def load_config(config_path):
     """Load and validate the JSON config file."""
     fh = open(config_path, 'r')
@@ -182,6 +227,24 @@ def load_config(config_path):
             "(size-routing) and Phase 3 (leveled incremental) are mutually exclusive "
             "schemes for a target (docs/RFC_incremental_v2.md §3). Pick one.")
 
+    # retrieve_stall_seconds: how long ship_object's round-trip verify retrieve may go
+    # SILENT before it is killed as a hung HPSS connection. Default comes from
+    # stall_defaults.json (fail-loud if that file is gone); a per-asset config may
+    # override it for a target whose tape-stage latency is legitimately longer. Kept
+    # separate from the create-side HTAR_STALL_SECONDS on purpose — see that constant
+    # and stall_defaults.json for the X1D shard incident that split the two.
+    # Read LAZILY (not via setdefault's eagerly-evaluated argument): a config that
+    # pins the value itself must not need the defaults file at all.
+    if "retrieve_stall_seconds" not in config:
+        config["retrieve_stall_seconds"] = \
+            load_stall_defaults()["retrieve_stall_seconds"]
+    if not isinstance(config["retrieve_stall_seconds"], int) \
+            or isinstance(config["retrieve_stall_seconds"], bool) \
+            or config["retrieve_stall_seconds"] <= 0:
+        raise ValueError(
+            f"config 'retrieve_stall_seconds' must be a positive integer (seconds), "
+            f"got: {config['retrieve_stall_seconds']!r}")
+
     # Durability guard (RFC §2.4 / §3.2): EITHER routing scheme keeps a per-target
     # manifest at {log_dir}/_vault/ that MUST outlive runs and be shared, so reclaim
     # can find it. Refuse a log_dir under $HOME / scratch / the home default — home
@@ -888,9 +951,21 @@ def decode_blonde(blonde):
 # it's silence: a healthy htar/htar_large/hsi call streams per-file progress
 # continuously, a stalled one stops producing output entirely.
 #
-# HTAR_STALL_SECONDS bounds how long a create/retrieve call may go without any new
-# stdout/stderr before it's killed as a hung HPSS connection (see
-# _run_with_stall_watch). Generous relative to any per-file gap observed in practice.
+# HTAR_STALL_SECONDS bounds how long a CREATE call (and any other stall-watched call
+# that has no bound of its own) may go without any new stdout/stderr before it's killed
+# as a hung HPSS connection (see _run_with_stall_watch). Generous relative to any
+# per-file gap observed in practice for a create, which streams per-file progress
+# continuously.
+#
+# The post-create round-trip VERIFY RETRIEVE is bounded separately, by the config's
+# `retrieve_stall_seconds` (default in stall_defaults.json, threaded into ship_object)
+# — 2 hours is wrong there. A retrieve is silent BY DESIGN while HPSS queues and
+# stages the tape, so on a large routed shard this bound was killing a healthy
+# retrieve: 2026-08-03, repository_X1D_3_metabolomics_rawspectra (one 6,624,961,024-byte
+# shard) was killed at exactly the 2h bound on both attempts, and its own stall log
+# shows "activity resumed after 7200.1s silence" — 0.1s past the kill line. Verify
+# never passed, so no success record was written and every scheduled run re-shipped
+# the whole shard: 48 identical 6.62 GB tars (~318 GB of tape) over 2026-07-17..08-03.
 HTAR_STALL_SECONDS = 2 * 60 * 60  # 2 hours
 
 # mkdir / ls existence checks return in seconds to low minutes when healthy and carry
@@ -3760,7 +3835,8 @@ def _verify_members(heartbeat):
 
 
 def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir,
-                tmp_dir, log_dir, emails, cleanup_on_success=False):
+                tmp_dir, log_dir, emails, cleanup_on_success=False,
+                retrieve_stall_seconds=None):
     """
     Ship ONE routed object (Phase 2, RFC §2.5) — LOCAL only. Tars the object's members
     DIRECTLY from source (no zip), choosing the transport via choose_transport: real
@@ -3802,6 +3878,15 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir
       htar_large:  cd <src>; htar_large -cvf <fortress>/<stem>_<ts>.tar <relpath>
       verify htar: cd <scratch>; htar -xvf <tar> ; md5 each member vs source
       verify _lg:  hsi get <tar> -> <scratch>; tar xf ; md5
+
+    retrieve_stall_seconds is REQUIRED (no code-side default): the silence budget for
+    the round-trip verify retrieve only, threaded down from the config
+    (`retrieve_stall_seconds`, whose own default lives in stall_defaults.json). Passing
+    None raises rather than quietly reinstating the create-side HTAR_STALL_SECONDS,
+    which is far too tight for a large shard's tape stage — the exact regression that
+    burned ~318 GB of duplicate tape on repository_X1D_3_metabolomics_rawspectra (see
+    HTAR_STALL_SECONDS). The create call keeps HTAR_STALL_SECONDS: a create streams
+    per-file progress, so silence there really is a hang.
     """
     import os
     import re
@@ -3821,6 +3906,13 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir
     arcnames = list(arcnames)
     if not arcnames:
         raise RuntimeError(f"ship_object: empty member set for {object_stem}")
+    if not isinstance(retrieve_stall_seconds, int) \
+            or isinstance(retrieve_stall_seconds, bool) or retrieve_stall_seconds <= 0:
+        raise RuntimeError(
+            f"ship_object: retrieve_stall_seconds must be a positive integer "
+            f"(seconds), got {retrieve_stall_seconds!r} — thread it down from the "
+            f"config (default in stall_defaults.json); there is deliberately no "
+            f"code-side fallback.")
 
     transport = choose_transport(arcnames, catalog)
     timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
@@ -3957,7 +4049,9 @@ def _hsi_ls_ok(path):
             print(f"  WARNING: {create_warning}")
 
         # Round-trip verify: pull the members back and re-check per-file MD5.
-        # See RETRIEVE_STALL_MAX_ATTEMPTS above for why this step gets one retry.
+        # See RETRIEVE_STALL_MAX_ATTEMPTS above for why this step gets one retry, and
+        # HTAR_STALL_SECONDS for why the silence budget here is the config's
+        # retrieve_stall_seconds rather than the (create-sized) module default.
         retrieve_dir = os.path.join(tmp_dir, f"{run_timestamp}.verify")
         retrieve_stall_log = os.path.join(log_dir, f"{run_timestamp}.retrieve.stall.log")
 
@@ -3967,16 +4061,19 @@ def _attempt_retrieve():
             if transport == "htar":
                 vr = _run_with_stall_watch(["htar", "-xvf", fortress_tar],
                                             cwd=retrieve_dir,
+                                            stall_seconds=retrieve_stall_seconds,
                                             activity_log=retrieve_stall_log)
             else:
                 local_tar = os.path.join(retrieve_dir, "obj.tar")
                 g = _run_with_stall_watch(
                     f'hsi -q "get {local_tar} : {fortress_tar}"',
+                    stall_seconds=retrieve_stall_seconds,
                     activity_log=retrieve_stall_log)
                 if g.returncode != 0:
                     raise RuntimeError("hsi get failed:\n"
                                        + g.stdout.decode("utf-8", "replace")[-4000:])
                 vr = _run_with_stall_watch(["tar", "xf", local_tar], cwd=retrieve_dir,
+                                            stall_seconds=retrieve_stall_seconds,
                                             activity_log=retrieve_stall_log)
             if vr.returncode != 0:
                 raise RuntimeError(
@@ -4123,6 +4220,9 @@ def _verify_md5s(heartbeat):
     shard_count_cfg = config["shard_count"]
     shard_target    = config["shard_target"]
     incremental     = config["incremental"]
+    # Silence budget for ship_object's round-trip verify retrieve (size-routing path).
+    # load_config guarantees this key: its default lives in stall_defaults.json.
+    retrieve_stall_seconds = config["retrieve_stall_seconds"]
 
     # Size-routing runs LOCAL only (RFC §2.8): the per-target manifest lives on a
     # shared durable log_dir that __main__ reads/writes, and reclaim later reads. Refuse
@@ -4296,8 +4396,11 @@ def run(fn, *fn_args):
         def ship_one(obj, stem):
             # Tar the object's members directly from source (no zip); ship_object picks
             # htar (indexed) vs htar_large by max member size, verifies, and logs.
+            # retrieve_stall_seconds is passed POSITIONALLY: `run` is the local/globus
+            # dispatcher (def run(fn, *fn_args)) and forwards positional args only.
             return run(ship_object, source_folder, obj["arcnames"], stem, survivors,
-                       fortress_base, tmp_dir, log_dir, emails, cleanup_zip_on_success)
+                       fortress_base, tmp_dir, log_dir, emails, cleanup_zip_on_success,
+                       retrieve_stall_seconds)
 
         try:
             summary = route_and_ship(

config.example.json

@@ -35,6 +35,9 @@
     "comment_exclude_optional": "Optional (default []). Per-asset opt-out list of Tier-2 (OPTIONAL) rule ids from the exclude_spec (e.g. omics provenance/QC: aux_info, fastp, checksums). Honored only if this asset's file_pattern also ships the raw inputs those artifacts are regenerable from. Omit to back up all OPTIONAL artifacts (safe default).",
     "exclude_optional": [],
 
+    "comment_retrieve_stall_seconds": "Optional override (default: stall_defaults.json's retrieve_stall_seconds, currently 21600 = 6h). How long the post-create round-trip VERIFY RETRIEVE may go without producing any output before it is killed as a hung HPSS connection. Much larger than the create-side bound on purpose: a retrieve is silent while HPSS queues and stages the tape, and a too-tight bound kills a healthy retrieve — which, because verify never passes, makes the next run re-ship the whole object (the 2026-08 X1D shard loop; see stall_defaults.json). Raise it for a target whose tape-stage latency is legitimately longer.",
+    "retrieve_stall_seconds": 21600,
+
     "comment_size_routing": "Optional (default false). Phase-2 size-routing (docs/RFC_incremental_v2.md §2): instead of one whole-target tar, ship each file >= t_small as its own 'solo' object and bundle smaller files into content-addressed 'shards' (shard = hash(relpath) mod K). Unchanged objects skip, so a GROWING target ships only its new files (the append win). LOCAL mode only — refused under --globus. When true, log_dir MUST be shared+durable (a Depot logs dir), NOT home or scratch: the per-target manifest lives at {log_dir}/_vault/ and reclaim must find it. Omit for the default whole-target behavior.",
     "size_routing": false,
 

stall_defaults.json

@@ -0,0 +1,6 @@
+{
+    "comment": "Engine-wide stall-watch defaults. Read by archive.load_config() for EVERY config (per-asset configs may override a key by name). This file is REQUIRED: a missing/malformed/invalid file raises rather than falling back to a value hidden in code, so the operative timeout is always inspectable here.",
+
+    "comment_retrieve_stall_seconds": "How long ship_object's post-create round-trip verify retrieve (htar -xvf, or hsi get + tar xf) may go without producing ANY output before it is killed as a hung HPSS connection. Deliberately much larger than the create-side HTAR_STALL_SECONDS (2h): a create streams per-file progress continuously, so silence there really is a hang, whereas a retrieve of a large routed shard is silent BY DESIGN while HPSS queues and stages the tape. 2026-08-03 (repository_X1D_3_metabolomics_rawspectra, a single 6,624,961,024-byte shard): the retrieve went quiet, was killed at exactly the 2h bound on both attempts, and its own stall log shows activity resuming 0.1s past the kill line ('activity resumed after 7200.1s silence') — legitimate tape-stage latency, misread as a hang. Verify never passed, so no success record was written and the next scheduled run re-shipped the whole shard from scratch: 48 identical 6.62 GB tars (~318 GB of duplicate tape) over 2026-07-17..08-03. 6h gives ~3x margin over the one observed gap while staying well inside the 48h Slurm walltime even if both retrieve attempts burn their full budget.",
+    "retrieve_stall_seconds": 21600
+}