From 6adc6294dfc748dbf15d12853b32910b18cbfbda Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Sat, 18 Jul 2026 12:55:56 -0400 Subject: [PATCH] fix(archive): kill stalled hsi/htar subprocess calls instead of hanging Every subprocess.run() call to hsi/htar/htar_large/tar had no timeout, so a hung HPSS connection would block until Slurm's own wall-clock limit (up to 48h) - confirmed on Negishi job 40793013, which ran 32h+ before the htar_large create call's connection stalled with zero progress output. A flat overall timeout is wrong here since real transfers in the same batch legitimately ranged from minutes to 22+ hours; the fix instead watches for silence (no new stdout/stderr) via a Popen + non-blocking read/poll loop, killing and raising only when a create/retrieve call goes HTAR_STALL_SECONDS (2h) without output. Short mkdir/ls existence checks get a flat HSI_SHORT_CALL_TIMEOUT_SECONDS (5 min) instead, since they carry no progress signal to watch. ship_object() calls the new module-level _run_with_stall_watch() directly; send_to_fortress() carries a self-contained inline twin (run_watched) since it can run remotely via Globus Compute / dill (invariant #1). tests/test_ship_object_htar_rc.py is updated to drive real fake htar/hsi/tar executables on PATH (exercising the real Popen/select plumbing) instead of mocking subprocess.run directly, which the create/verify calls no longer use. tests/test_stall_detection.py adds direct coverage of both stall-watch modes (PIPE-captured and file-growth) and the short-call timeout path. Co-Authored-By: Claude Sonnet 5 --- CLAUDE.md | 26 +++ archive.py | 254 +++++++++++++++++++++++++++--- tests/test_ship_object_htar_rc.py | 249 ++++++++++++++++++----------- tests/test_stall_detection.py | 209 ++++++++++++++++++++++++ 4 files changed, 625 insertions(+), 113 deletions(-) create mode 100644 tests/test_stall_detection.py diff --git a/CLAUDE.md b/CLAUDE.md index 6234944..7941531 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -224,6 +224,18 @@ we do NOT use `htar_large -xvf` (it streams the archive through stdout to a `tar pipe, which breaks in subprocess). Instead `hsi get` writes the tar to a local file, then `tar xvf` extracts. Both steps log stdout+stderr. +**7. Every `hsi`/`htar`/`htar_large`/`tar` subprocess call must be stall-guarded, not +bare `subprocess.run()`** — a bare call has no bound on a hung HPSS connection; only +Slurm's own wall-clock (up to 48h) would ever stop it. Long, output-streaming calls +(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 +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 +`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. + --- ## Verification chain (in order) @@ -437,6 +449,20 @@ archives byte-for-byte as before. Authoritative design: `docs/RFC_incremental_v2 ## Known issues / history +- **Unbounded subprocess hang (32h+ on Negishi):** `repository_X1D_3_metabolomics_rawspectra` + (Slurm job 40793013, 2026-07-17/18) ran 32h+ before manual cancellation — its + `.out` file showed the destination `hsi -q mkdir -p ...` line and then nothing; + the following `htar_large` create call never produced a single per-file progress + line, i.e. the HPSS connection hung before transferring a byte. No `subprocess.run()` + call to `hsi`/`htar`/`htar_large`/`tar` had a `timeout=`, so only Slurm's own + wall-clock (up to 48h) would ever have stopped it — wasting the whole allocation and + delaying every archive queued behind it. `sacct` confirmed every other job in the + same batch completed fine (some legitimately taking 7–22h), ruling out a broader + HPSS outage and ruling out a flat overall timeout as the fix (it would have killed + those healthy long transfers too). Fixed by watching for *silence* instead: long + create/retrieve calls are killed if `HTAR_STALL_SECONDS` (2h) passes with no new + stdout/stderr; short mkdir/ls calls get a flat `HSI_SHORT_CALL_TIMEOUT_SECONDS` + (5 min). See invariant #7. - **htar 64 GB limit:** `htar` fails if any single file exceeds 64 GB. Use `htar_large` (no limit, but no `.idx` index file). - **htar 250-char path limit (silent, not fail-fast):** indexed `htar -L` drops diff --git a/archive.py b/archive.py index ab9ea8e..d7e27c9 100644 --- a/archive.py +++ b/archive.py @@ -776,6 +776,127 @@ def decode_blonde(blonde): # member gets anywhere near the drop threshold. HTAR_ARCNAME_LEN_LIMIT = 240 +# Stall detection (2026-07-18 incident: repository_X1D_3_metabolomics_rawspectra, +# Slurm job 40793013 on Negishi, ran 32h+ before manual cancellation). Its .out file +# showed the destination `hsi -q mkdir -p ...` line and then nothing — the following +# htar_large create call never produced a single per-file progress line, i.e. the +# underlying HPSS connection hung before transferring a byte. Only Slurm's own +# wall-clock limit (up to 48h) would ever have stopped it. +# +# A flat overall timeout is wrong here: real transfers in the SAME batch legitimately +# ranged from minutes to 22+ hours (sacct confirmed every other job that batch +# completed fine — this was an isolated stall, not a broader HPSS outage). The signal +# that distinguishes a hung connection from a merely-long transfer isn't elapsed time, +# 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 = 2 * 60 * 60 # 2 hours + +# mkdir / ls existence checks return in seconds to low minutes when healthy and carry +# no progressive output to distinguish "slow but working" from "stuck" — a flat +# timeout suffices for these rather than the full stall-watch treatment above. +HSI_SHORT_CALL_TIMEOUT_SECONDS = 5 * 60 # 5 minutes + + +def _run_with_stall_watch(cmd, *, cwd=None, watch_paths=None, + stall_seconds=HTAR_STALL_SECONDS, poll_interval=30): + """ + Run cmd via Popen, killing it and raising RuntimeError if it goes stall_seconds + without producing new output. See HTAR_STALL_SECONDS above for why this watches + for silence rather than enforcing a flat overall timeout. + + Two activity-detection modes, picked by whether the caller wants output captured + or the shell command redirects its own output to files: + + - watch_paths=None (default): cmd is run with stdout+stderr merged into a pipe; + activity = new bytes read. `cmd` may be a list (shell=False) or a string + (shell=True), matching plain subprocess.run's convention. Returns a + subprocess.CompletedProcess-like object with .returncode and .stdout (bytes) + — a drop-in replacement for `subprocess.run(cmd, stdout=PIPE, + stderr=STDOUT)`. + + - watch_paths=[...]: cmd is a shell string that redirects its own stdout/stderr + to those paths (send_to_fortress uses this — invariant #6: some retrieval + steps must not have their output piped through Python); activity = growth of + any watched file's size. Returns a CompletedProcess-like object with only + .returncode set (the output already landed in the watched files, as before). + + send_to_fortress carries a self-contained inline twin of this function (imports + inside, no reference to this module-level copy) since it can run remotely via + Globus Compute / dill (invariant #1) — kept in sync by inspection, same as the + notify_bell twin. + """ + import subprocess + import time + import os + + if watch_paths is None: + import select + + proc = subprocess.Popen(cmd, cwd=cwd, shell=isinstance(cmd, str), + stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + chunks = [] + last_activity = time.monotonic() + fd = proc.stdout.fileno() + try: + while True: + readable, _, _ = select.select([fd], [], [], + min(poll_interval, stall_seconds)) + if readable: + data = os.read(fd, 65536) + if data: + chunks.append(data) + last_activity = time.monotonic() + continue + proc.wait() + break + if proc.poll() is not None: + break + if time.monotonic() - last_activity > stall_seconds: + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=10) + raise RuntimeError( + f"command stalled — no output for " + f"{stall_seconds / 3600:.1f}h (HPSS connection likely " + f"hung), pid={proc.pid} killed. cmd={cmd!r}") + finally: + proc.stdout.close() + return subprocess.CompletedProcess(cmd, proc.returncode, b"".join(chunks)) + + proc = subprocess.Popen(cmd, cwd=cwd, shell=True) + last_activity = time.monotonic() + last_sizes = {} + while True: + for p in watch_paths: + try: + sz = os.path.getsize(p) + except OSError: + continue + if last_sizes.get(p) != sz: + last_sizes[p] = sz + last_activity = time.monotonic() + rc = proc.poll() + if rc is not None: + return subprocess.CompletedProcess(cmd, rc) + if time.monotonic() - last_activity > stall_seconds: + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=10) + raise RuntimeError( + f"command stalled — no output for {stall_seconds / 3600:.1f}h " + f"(HPSS connection likely hung), pid={proc.pid} killed. cmd={cmd!r}") + time.sleep(min(poll_interval, stall_seconds)) + def clamp_t_small(t_small): """ @@ -2281,6 +2402,52 @@ def notify_bell(subject, body): write_log(f"Teams bell not delivered ({type(e).__name__}); email still sent.", level="WARNING") + def run_watched(cmd, watch_paths, stall_seconds=7200, poll_interval=30): + """ + Inline, Compute-safe twin of the module-level _run_with_stall_watch()'s + file-growth mode: kills `cmd` (a shell string that redirects its own + stdout/stderr to watch_paths — invariant #6, some steps here must not pipe + output through Python) if none of watch_paths grow for stall_seconds, raising + RuntimeError. A healthy htar_large/hsi call streams per-file progress + continuously into those files; a stalled HPSS connection goes silent instead + of merely running long (see HTAR_STALL_SECONDS for the incident this fixes). + This one can run on a compute node under --globus, so it references no + module/outer names and imports inside (invariant #1). stall_seconds=7200 (2h) + mirrors the module-level HTAR_STALL_SECONDS constant — kept in sync by + inspection, same as the notify_bell twin. + """ + import subprocess + import time + import os + + proc = subprocess.Popen(cmd, shell=True) + last_activity = time.monotonic() + last_sizes = {} + while True: + for p in watch_paths: + try: + sz = os.path.getsize(p) + except OSError: + continue + if last_sizes.get(p) != sz: + last_sizes[p] = sz + last_activity = time.monotonic() + rc = proc.poll() + if rc is not None: + return subprocess.CompletedProcess(cmd, rc) + if time.monotonic() - last_activity > stall_seconds: + proc.terminate() + try: + proc.wait(timeout=10) + except subprocess.TimeoutExpired: + proc.kill() + proc.wait(timeout=10) + raise RuntimeError( + f"command stalled — no output for {stall_seconds / 3600:.1f}h " + f"(HPSS connection likely hung), pid={proc.pid} killed. " + f"cmd={cmd!r}") + time.sleep(min(poll_interval, stall_seconds)) + write_log(f"Starting archive workflow") write_log(f"Project: {project_name}") write_log(f"Source folder (Depot): {source_folder}") @@ -2351,15 +2518,34 @@ def notify_bell(subject, body): f"Log (JSON): {json_path}" ) - subprocess.run(f'hsi "mkdir -p {fortress_base_dir}"', - shell=True, capture_output=True) + # No progressive output to watch — a flat timeout bounds a hung connection here; + # a failed/timed-out mkdir was already silently ignored before this change (the + # create call below will surface a real Fortress-side problem either way). + try: + subprocess.run(f'hsi "mkdir -p {fortress_base_dir}"', + shell=True, capture_output=True, timeout=300) + except subprocess.TimeoutExpired: + pass # Use htar_large for write — handles zip files > 64 GB (htar's single-file limit). # htar_large does not generate an index file. # stdout redirect (>) is safe here — capturing progress text only. cmd = (f'cd {src_dir} && htar_large -cvf {fortress_tar} {filename} ' f'>{stdout_log} 2>{stderr_log}') - result = subprocess.run(cmd, shell=True) + try: + result = run_watched(cmd, [stdout_log, stderr_log]) + except RuntimeError as e: + write_log(f"htar_large transfer stalled: {e}", level="ERROR") + finalize_log("failed", error=f"htar_large transfer stalled: {e}") + send_email( + f"[FAILED] Fortress archive: {filename}", + f"Project: {project_name}\n\n" + f"The htar_large transfer to Fortress stalled (no output) for {filename}.\n\n" + f"Error:\n{e}\n\n" + f"Log (TXT): {txt_path}\n" + f"Log (JSON): {json_path}" + ) + raise if result.returncode != 0: fh = open(stderr_log) @@ -2405,7 +2591,20 @@ def notify_bell(subject, body): # Step A: hsi get — retrieve tar from Fortress to local disk hsi_get_cmd = (f'hsi "get {retrieved_tar} : {fortress_tar}" ' f'>{hsi_get_out} 2>{hsi_get_err}') - hsi_result = subprocess.run(hsi_get_cmd, shell=True) + try: + hsi_result = run_watched(hsi_get_cmd, [hsi_get_out, hsi_get_err]) + except RuntimeError as e: + write_log(f"hsi get stalled: {e}", level="ERROR") + finalize_log("failed", error=f"hsi get stalled: {e}") + send_email( + f"[FAILED] Fortress retrieval: {filename}", + f"Project: {project_name}\n\n" + f"hsi get stalled (no output) for {tar_basename}.\n\n" + f"Error:\n{e}\n\n" + f"Log (TXT): {txt_path}\n" + f"Log (JSON): {json_path}" + ) + raise if hsi_result.returncode != 0: fh = open(hsi_get_err) @@ -2443,7 +2642,20 @@ def notify_bell(subject, body): tar_extract_out = os.path.join(src_dir, f"tar_extract_{zip_ts}.out") tar_extract_err = os.path.join(src_dir, f"tar_extract_{zip_ts}.err") tar_cmd = f'tar xvf {retrieved_tar} -C {retrieve_dir} >{tar_extract_out} 2>{tar_extract_err}' - tar_result = subprocess.run(tar_cmd, shell=True) + try: + tar_result = run_watched(tar_cmd, [tar_extract_out, tar_extract_err]) + except RuntimeError as e: + write_log(f"tar extraction stalled: {e}", level="ERROR") + finalize_log("failed", error=f"tar extraction stalled: {e}") + send_email( + f"[FAILED] Fortress tar extraction: {filename}", + f"Project: {project_name}\n\n" + f"Local tar extraction stalled (no output) for {tar_basename}.\n\n" + f"Error:\n{e}\n\n" + f"Log (TXT): {txt_path}\n" + f"Log (JSON): {json_path}" + ) + raise if tar_result.returncode != 0: fh = open(tar_extract_err) @@ -2705,7 +2917,15 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir os.makedirs(tmp_dir, exist_ok=True) os.makedirs(log_dir, exist_ok=True) # Ensure the Fortress destination dir exists (htar_large has no -P; htar uses -P too). - subprocess.run(f'hsi -q "mkdir -p {fortress_base_dir}"', shell=True) + # No progressive output to watch (invariant: HSI_SHORT_CALL_TIMEOUT_SECONDS is for + # exactly these no-progress calls) — a flat timeout bounds a hung connection here; + # a failed/timed-out mkdir was already silently ignored before this change (the + # actual create call below will surface a real Fortress-side problem either way). + try: + subprocess.run(f'hsi -q "mkdir -p {fortress_base_dir}"', shell=True, + timeout=HSI_SHORT_CALL_TIMEOUT_SECONDS) + except subprocess.TimeoutExpired: + pass # A member list file (relpaths) drives htar -L; htar_large takes the paths directly. listfile = os.path.join(tmp_dir, f"{run_timestamp}.members") @@ -2719,8 +2939,7 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir idx = fortress_tar + ".idx" else: # htar_large — a single >=64 GiB member; pass the path directly cmd = ["htar_large", "-cvf", fortress_tar] + arcnames - create = subprocess.run(cmd, cwd=source_folder, - stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + create = _run_with_stall_watch(cmd, cwd=source_folder) create_stdout = create.stdout.decode("utf-8", "replace") # Fail fast on an explicit member-omission message, BEFORE the rc!=0 @@ -2777,13 +2996,15 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir 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. + # Fail-safe: any error (or a hung connection) checking presence is + # treated as "not landed", so an hsi/PATH problem — or a stall — can't + # mask a genuine create failure. try: r = subprocess.run(["hsi", "-q", "ls", path], - stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, + timeout=HSI_SHORT_CALL_TIMEOUT_SECONDS) return r.returncode == 0 - except OSError: + except (OSError, subprocess.TimeoutExpired): return False landed = _hsi_ls_ok(fortress_tar) and (idx is None or _hsi_ls_ok(idx)) @@ -2810,17 +3031,14 @@ def _hsi_ls_ok(path): shutil.rmtree(retrieve_dir, ignore_errors=True) os.makedirs(retrieve_dir) if transport == "htar": - vr = subprocess.run(["htar", "-xvf", fortress_tar], cwd=retrieve_dir, - stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + vr = _run_with_stall_watch(["htar", "-xvf", fortress_tar], cwd=retrieve_dir) else: local_tar = os.path.join(retrieve_dir, "obj.tar") - g = subprocess.run(f'hsi -q "get {local_tar} : {fortress_tar}"', shell=True, - stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + g = _run_with_stall_watch(f'hsi -q "get {local_tar} : {fortress_tar}"') if g.returncode != 0: raise RuntimeError("hsi get failed:\n" + g.stdout.decode("utf-8", "replace")[-4000:]) - vr = subprocess.run(["tar", "xf", local_tar], cwd=retrieve_dir, - stdout=subprocess.PIPE, stderr=subprocess.STDOUT) + vr = _run_with_stall_watch(["tar", "xf", local_tar], cwd=retrieve_dir) if vr.returncode != 0: raise RuntimeError( f"round-trip extract failed for {run_timestamp}:\n" diff --git a/tests/test_ship_object_htar_rc.py b/tests/test_ship_object_htar_rc.py index 20b02bc..4094c94 100644 --- a/tests/test_ship_object_htar_rc.py +++ b/tests/test_ship_object_htar_rc.py @@ -7,54 +7,113 @@ 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. +No real hsi/htar/tape: ship_object's create/verify calls now go through +_run_with_stall_watch(), which drives real subprocess.Popen objects (to exercise the +actual select()/os.read() stall-watch plumbing, not a mock of it) — so instead of +faking subprocess.run, this fakes the `htar`/`htar_large`/`hsi`/`tar` executables +themselves as tiny shell-script stand-ins on PATH. Each invocation is appended to a +call-log file so tests can still assert on what ran and in what order, exactly as +the old subprocess.run-mock-based `calls` list did. The round-trip "htar -xvf" step +copies 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 stat 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 _write_fake_bin(bin_dir, name, body): + path = os.path.join(bin_dir, name) + with open(path, "w") as fh: + fh.write("#!/bin/sh\n" + body) + st = os.stat(path) + os.chmod(path, st.st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH) + return path -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 ` outcome for both - the tar and its .idx, so tests can simulate "landed" vs "genuinely missing". +class FakeHsiHtarEnv: + """ + Installs fake `htar`, `htar_large`, `hsi`, `tar` executables on PATH for the + duration of a `with` block, so ship_object's real subprocess.Popen calls exercise + real (fast, local) processes instead of the actual RCAC tools. Every invocation + is appended to `self.call_log` (list of "prog arg1 arg2 ..." strings) so tests can + assert on what ran, mirroring the old mocked-subprocess.run `calls` list. """ - 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 + + def __init__(self, create_rc=0, create_stdout=b"", ls_rc=0, + source_folder=None, arcnames=None): + self.create_rc = create_rc + self.create_stdout = create_stdout + self.ls_rc = ls_rc + self.source_folder = source_folder + self.arcnames = arcnames or [] + self._d = None + self._old_path = None + self.call_log_path = None + + def __enter__(self): + self._d = tempfile.mkdtemp() + bin_dir = os.path.join(self._d, "bin") + os.makedirs(bin_dir) + self.call_log_path = os.path.join(self._d, "calls.log") + + stdout_file = os.path.join(self._d, "create_stdout.txt") + with open(stdout_file, "wb") as fh: + fh.write(self.create_stdout) + + arcnames_file = os.path.join(self._d, "arcnames.txt") + with open(arcnames_file, "w") as fh: + fh.write("\n".join(self.arcnames) + "\n") + + _write_fake_bin(bin_dir, "htar", f''' +printf 'htar %s\\n' "$*" >> "{self.call_log_path}" +if [ "$1" = "-xvf" ]; then + while IFS= read -r arc; do + [ -z "$arc" ] && continue + mkdir -p "$(dirname "$arc")" + cp "{self.source_folder}/$arc" "$arc" + done < "{arcnames_file}" + exit 0 +fi +cat "{stdout_file}" +exit {self.create_rc} +''') + _write_fake_bin(bin_dir, "htar_large", f''' +printf 'htar_large %s\\n' "$*" >> "{self.call_log_path}" +cat "{stdout_file}" +exit {self.create_rc} +''') + _write_fake_bin(bin_dir, "hsi", f''' +printf 'hsi %s\\n' "$*" >> "{self.call_log_path}" +case "$2" in + ls) exit {self.ls_rc} ;; + *) exit 0 ;; +esac +''') + _write_fake_bin(bin_dir, "tar", f''' +printf 'tar %s\\n' "$*" >> "{self.call_log_path}" +exit 0 +''') + + self._old_path = os.environ.get("PATH", "") + os.environ["PATH"] = bin_dir + os.pathsep + self._old_path + return self + + def __exit__(self, *exc): + os.environ["PATH"] = self._old_path + shutil.rmtree(self._d, ignore_errors=True) + + def calls(self): + if not os.path.isfile(self.call_log_path): + return [] + with open(self.call_log_path) as fh: + return [line.rstrip("\n") for line in fh] class TestShipObjectHtarRc(unittest.TestCase): @@ -73,20 +132,21 @@ def setUp(self): 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( + def _ship(self, env): + with env: + result = 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"]) + calls = env.calls() + return result, calls 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) + env = FakeHsiHtarEnv(create_rc=70, create_stdout=stdout, ls_rc=0, + source_folder=self.source_folder, arcnames=self.arcnames) - result = self._ship(fake_run) + result, calls = self._ship(env) self.assertIsNotNone(result["create_warning"]) self.assertIn("rc=70", result["create_warning"]) @@ -111,19 +171,17 @@ def test_nonzero_rc_but_tar_present_recovers(self): # 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) + ls_calls = [c for c in calls if c.startswith("hsi -q ls")] + self.assertIn(f"hsi -q ls {result['fortress_tar']}", ls_calls) + self.assertIn(f"hsi -q ls {result['idx']}", ls_calls) 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) + env = FakeHsiHtarEnv(create_rc=1, create_stdout=b"HTAR: ERROR: something went wrong\n", + ls_rc=1, source_folder=self.source_folder, arcnames=self.arcnames) with self.assertRaises(RuntimeError) as ctx: - self._ship(fake_run) + self._ship(env) self.assertIn("rc=1", str(ctx.exception)) self.assertIn("could not confirm", str(ctx.exception)) @@ -133,11 +191,10 @@ def test_nonzero_rc_and_tar_missing_raises(self): 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) + env = FakeHsiHtarEnv(create_rc=0, create_stdout=b"HTAR: HTAR SUCCESSFUL\n", ls_rc=0, + source_folder=self.source_folder, arcnames=self.arcnames) - result = self._ship(fake_run) + result, _calls = self._ship(env) self.assertIsNone(result["create_warning"]) debug_log = os.path.join(self.log_dir, f"{result['run_timestamp']}.htar_create.log") @@ -174,12 +231,26 @@ def setUp(self): 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( + def _ship(self, env): + with env: + result = 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"]) + calls = env.calls() + return result, calls + + def _ship_raises(self, env): + # env.__exit__ deletes its tempdir (incl. the call log), so both the + # exception and the call log must be captured before the `with` closes. + with env: + with self.assertRaises(RuntimeError) as ctx: + 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"]) + calls = env.calls() + return ctx.exception, calls def test_omission_warning_raises_immediately_and_skips_verify(self): omitted_relpath = "sub/" + ("x" * 260) + ".txt" @@ -189,20 +260,17 @@ def test_omission_warning_raises_immediately_and_skips_verify(self): b" [" + omitted_relpath.encode() + b"]\n" b"HTAR: 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) + env = FakeHsiHtarEnv(create_rc=70, create_stdout=stdout, ls_rc=0, + source_folder=self.source_folder, arcnames=self.arcnames) - with self.assertRaises(RuntimeError) as ctx: - self._ship(fake_run) + exc, calls = self._ship_raises(env) - self.assertIn(omitted_relpath, str(ctx.exception)) - self.assertIn("too long", str(ctx.exception)) + self.assertIn(omitted_relpath, str(exc)) + self.assertIn("too long", str(exc)) # Fails at the create step — hsi ls / round-trip extract must never run. - self.assertFalse(any(c[0][0] == "hsi" and c[0][1:3] == ["-q", "ls"] - for c in calls)) - self.assertFalse(any(c[0][0] == "htar" and c[0][1] == "-xvf" for c in calls)) + self.assertFalse(any(c.startswith("hsi -q ls") for c in calls)) + self.assertFalse(any(c.startswith("htar -xvf") for c in calls)) # Full stdout + the omitted-member list are captured durably even though # the run fails fast. @@ -226,12 +294,11 @@ def test_multiple_omissions_all_named(self): b"WARNING: name too long for tar archive- file omitted\n" b" [" + omitted_b.encode() + b"]\n" ) - fake_run, _calls = make_fake_run( - self.source_folder, self.arcnames, create_rc=70, create_stdout=stdout, - ls_ok=lambda path: True) + env = FakeHsiHtarEnv(create_rc=70, create_stdout=stdout, ls_rc=0, + source_folder=self.source_folder, arcnames=self.arcnames) with self.assertRaises(RuntimeError) as ctx: - self._ship(fake_run) + self._ship(env) self.assertIn(omitted_a, str(ctx.exception)) self.assertIn(omitted_b, str(ctx.exception)) @@ -244,14 +311,12 @@ def test_omission_warning_with_rc_zero_still_raises(self): b" [" + omitted_relpath.encode() + b"]\n" b"HTAR: HTAR SUCCESSFUL\n" ) - fake_run, calls = make_fake_run( - self.source_folder, self.arcnames, create_rc=0, create_stdout=stdout, - ls_ok=lambda path: True) + env = FakeHsiHtarEnv(create_rc=0, create_stdout=stdout, ls_rc=0, + source_folder=self.source_folder, arcnames=self.arcnames) - with self.assertRaises(RuntimeError) as ctx: - self._ship(fake_run) - self.assertIn(omitted_relpath, str(ctx.exception)) - self.assertFalse(any(c[0][0] == "htar" and c[0][1] == "-xvf" for c in calls)) + exc, calls = self._ship_raises(env) + self.assertIn(omitted_relpath, str(exc)) + self.assertFalse(any(c.startswith("htar -xvf") for c in calls)) def test_not_readable_omission_raises(self): # htar (indexed) skips unreadable source files with an INFO line rather @@ -263,17 +328,14 @@ def test_not_readable_omission_raises(self): b"HTAR: a a.txt\n" b"HTAR: HTAR SUCCESSFUL\n" ) - fake_run, calls = make_fake_run( - self.source_folder, self.arcnames, create_rc=72, create_stdout=stdout, - ls_ok=lambda path: True) + env = FakeHsiHtarEnv(create_rc=72, create_stdout=stdout, ls_rc=0, + source_folder=self.source_folder, arcnames=self.arcnames) - with self.assertRaises(RuntimeError) as ctx: - self._ship(fake_run) + exc, calls = self._ship_raises(env) - self.assertIn("sub/b.txt", str(ctx.exception)) - self.assertIn("not readable", str(ctx.exception)) - self.assertFalse(any(c[0][0] == "hsi" and c[0][1:3] == ["-q", "ls"] - for c in calls)) + self.assertIn("sub/b.txt", str(exc)) + self.assertIn("not readable", str(exc)) + self.assertFalse(any(c.startswith("hsi -q ls") for c in calls)) def test_htar_large_tar_cannot_open_raises_even_with_rc_zero(self): # htar_large wraps `tar cvf - | hsi put`; when tar hits a per-file @@ -286,17 +348,14 @@ def test_htar_large_tar_cannot_open_raises_even_with_rc_zero(self): b"tar: sub/b.txt: Cannot open: Permission denied\n" b"tar: Exiting with failure status due to previous errors\n" ) - fake_run, calls = make_fake_run( - self.source_folder, self.arcnames, create_rc=0, create_stdout=stdout, - ls_ok=lambda path: True) + env = FakeHsiHtarEnv(create_rc=0, create_stdout=stdout, ls_rc=0, + source_folder=self.source_folder, arcnames=self.arcnames) - with self.assertRaises(RuntimeError) as ctx: - self._ship(fake_run) + exc, calls = self._ship_raises(env) - self.assertIn("sub/b.txt", str(ctx.exception)) - self.assertIn("Permission denied", str(ctx.exception)) - self.assertFalse(any(c[0][0] == "hsi" and c[0][1:3] == ["-q", "ls"] - for c in calls)) + self.assertIn("sub/b.txt", str(exc)) + self.assertIn("Permission denied", str(exc)) + self.assertFalse(any(c.startswith("hsi -q ls") for c in calls)) if __name__ == "__main__": diff --git a/tests/test_stall_detection.py b/tests/test_stall_detection.py new file mode 100644 index 0000000..979742f --- /dev/null +++ b/tests/test_stall_detection.py @@ -0,0 +1,209 @@ +""" +Unit tests for archive._run_with_stall_watch() (2026-07-18 incident: +repository_X1D_3_metabolomics_rawspectra, Slurm job 40793013 on Negishi, hung 32h+ +because the htar_large create call's HPSS connection stalled — no per-file progress +line ever followed the destination mkdir, and nothing but Slurm's own 48h wall-clock +limit would ever have stopped it). + +Every subprocess.run() call to hsi/htar/htar_large/tar previously had no timeout at +all. The fix distinguishes "hung" from "merely long" by watching for NEW OUTPUT +rather than enforcing a flat overall timeout — real transfers in the same batch +legitimately ranged from minutes to 22+ hours, so a flat timeout would kill healthy +transfers too. This exercises the real mechanism directly: real subprocesses (plain +`sh -c ...`), real Popen/select/os.read plumbing, no mocking of subprocess itself — +only stall_seconds/poll_interval are shrunk so the tests run in well under a second. + +Covers both activity-detection modes _run_with_stall_watch supports: + - PIPE mode (watch_paths=None): used by ship_object's create/round-trip calls. + - file-growth mode (watch_paths=[...]): used by send_to_fortress's inline + `run_watched` twin, whose shell commands redirect their own output to files + (invariant #6) rather than piping through Python. + +Also covers the short, no-progress-output calls (hsi mkdir / hsi ls) which use a +flat subprocess.run(timeout=...) instead of the full stall-watch treatment, wired +through ship_object's _hsi_ls_ok and mkdir calls. +""" +import os +import re +import shutil +import sys +import tempfile +import time +import unittest +from unittest import mock + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import archive # noqa: E402 + + +def _extract_pid(message): + m = re.search(r"pid=(\d+)", message) + assert m, f"no pid= in stall message: {message!r}" + return int(m.group(1)) + + +def _assert_process_dead(testcase, pid): + # _run_with_stall_watch calls proc.wait() as part of the kill sequence before + # raising, so the child should already be reaped by the time we get here. + with testcase.assertRaises(ProcessLookupError): + os.kill(pid, 0) + + +class TestStallWatchPipeMode(unittest.TestCase): + """watch_paths=None — PIPE-captured mode, used by ship_object.""" + + def test_steady_output_under_stall_window_completes_normally(self): + # Three ticks with 0.05s gaps, well under a 0.4s stall window — must run to + # completion untouched, not get killed for merely taking a little while. + cmd = ["sh", "-c", + "for i in 1 2 3; do echo tick; sleep 0.05; done; exit 0"] + result = archive._run_with_stall_watch(cmd, stall_seconds=0.4, + poll_interval=0.05) + self.assertEqual(result.returncode, 0) + self.assertEqual(result.stdout.count(b"tick"), 3) + + def test_silent_process_is_killed_and_raises(self): + cmd = ["sh", "-c", "sleep 5"] + start = time.monotonic() + with self.assertRaises(RuntimeError) as ctx: + archive._run_with_stall_watch(cmd, stall_seconds=0.3, poll_interval=0.05) + elapsed = time.monotonic() - start + self.assertIn("stalled", str(ctx.exception)) + self.assertIn("no output", str(ctx.exception)) + # Killed well before the process's own 5s sleep would have finished — + # proves it was actually detected as silent, not just naturally waited out. + self.assertLess(elapsed, 2.0) + _assert_process_dead(self, _extract_pid(str(ctx.exception))) + + def test_normal_nonzero_exit_unaffected(self): + # Existing "benign transient rc" handling (ship_object's own docstring) + # must be untouched: a process that produces output then exits nonzero is + # not treated as a stall at all. + cmd = ["sh", "-c", "echo HTAR SUCCESSFUL; exit 70"] + result = archive._run_with_stall_watch(cmd, stall_seconds=0.4, + poll_interval=0.05) + self.assertEqual(result.returncode, 70) + self.assertIn(b"HTAR SUCCESSFUL", result.stdout) + + def test_shell_string_command_runs_with_shell_true(self): + # ship_object's htar_large-verify `hsi get` call passes a quoted shell + # string rather than an argv list — confirm that shape still works. + result = archive._run_with_stall_watch('echo from-shell-string', + stall_seconds=0.4, poll_interval=0.05) + self.assertEqual(result.returncode, 0) + self.assertIn(b"from-shell-string", result.stdout) + + +class TestStallWatchFileGrowthMode(unittest.TestCase): + """watch_paths=[...] — file-growth mode, used by send_to_fortress's twin.""" + + def setUp(self): + self.d = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.d, ignore_errors=True) + self.out = os.path.join(self.d, "out.log") + + def test_steady_growth_under_stall_window_completes_normally(self): + cmd = (f"for i in 1 2 3; do echo tick >> {self.out}; sleep 0.05; done; " + f"exit 0") + result = archive._run_with_stall_watch(cmd, watch_paths=[self.out], + stall_seconds=0.4, poll_interval=0.05) + self.assertEqual(result.returncode, 0) + with open(self.out) as fh: + self.assertEqual(fh.read().count("tick"), 3) + + def test_silent_file_is_killed_and_raises(self): + # The file is created by the shell redirect but never written to — + # exactly the incident signature (an .out file frozen right after mkdir). + cmd = f"sleep 5 > {self.out} 2>&1" + start = time.monotonic() + with self.assertRaises(RuntimeError) as ctx: + archive._run_with_stall_watch(cmd, watch_paths=[self.out], + stall_seconds=0.3, poll_interval=0.05) + elapsed = time.monotonic() - start + self.assertIn("stalled", str(ctx.exception)) + self.assertLess(elapsed, 2.0) + _assert_process_dead(self, _extract_pid(str(ctx.exception))) + + def test_normal_nonzero_exit_unaffected(self): + cmd = f"echo done > {self.out} 2>&1; exit 7" + result = archive._run_with_stall_watch(cmd, watch_paths=[self.out], + stall_seconds=0.4, poll_interval=0.05) + self.assertEqual(result.returncode, 7) + with open(self.out) as fh: + self.assertIn("done", fh.read()) + + def test_watch_path_not_yet_created_is_not_activity(self): + # A watched path that doesn't exist yet (e.g. hasn't been created by the + # shell redirect for some reason) must not be misread as activity — absence + # is simply skipped, not treated as a size change. + missing = os.path.join(self.d, "does_not_exist.log") + cmd = "sleep 5" + with self.assertRaises(RuntimeError): + archive._run_with_stall_watch(cmd, watch_paths=[missing], + stall_seconds=0.2, poll_interval=0.05) + + +class TestShipObjectShortCallTimeout(unittest.TestCase): + """ + mkdir/ls existence checks stream no progress output, so they use a flat + subprocess.run(timeout=...) rather than the full stall-watch treatment. A + timeout there must be treated exactly like any other failure to confirm + presence — fail-safe, not an uncaught TimeoutExpired. + """ + + 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(self.source_folder) + with open(os.path.join(self.source_folder, "a.txt"), "wb") as fh: + fh.write(b"hello world") + self.arcnames = ["a.txt"] + self.catalog = {"a.txt": (0, 11)} + 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 test_hsi_ls_timeout_treated_as_not_landed(self): + bin_dir = os.path.join(self.d, "bin") + os.makedirs(bin_dir) + htar_path = os.path.join(bin_dir, "htar") + with open(htar_path, "w") as fh: + fh.write("#!/bin/sh\n" + "if [ \"$1\" = \"-xvf\" ]; then exit 0; fi\n" + "echo 'HTAR: ERROR'\n" + "exit 1\n") + os.chmod(htar_path, 0o755) + hsi_path = os.path.join(bin_dir, "hsi") + with open(hsi_path, "w") as fh: + # `ls` never returns within the (patched, tiny) short-call timeout — + # simulates a hung hsi existence check. + fh.write("#!/bin/sh\n" + "case \"$2\" in\n" + " ls) sleep 5 ;;\n" + " *) exit 0 ;;\n" + "esac\n") + os.chmod(hsi_path, 0o755) + + old_path = os.environ.get("PATH", "") + os.environ["PATH"] = bin_dir + os.pathsep + old_path + try: + with mock.patch.object(archive, "HSI_SHORT_CALL_TIMEOUT_SECONDS", 0.2): + start = time.monotonic() + with self.assertRaises(RuntimeError) as ctx: + 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"]) + elapsed = time.monotonic() - start + finally: + os.environ["PATH"] = old_path + + # Treated as "not landed" (fail-safe) rather than hanging the full 5s or + # raising a bare, uncaught TimeoutExpired. + self.assertIn("could not confirm", str(ctx.exception)) + self.assertLess(elapsed, 3.0) + + +if __name__ == "__main__": + unittest.main()