Skip to content

fix(archive): kill stalled hsi/htar subprocess calls instead of hanging #32

Merged
Jarrod S Doucette (jdoucett) merged 1 commit into from
Jul 18, 2026
+625 −113
Merged

fix(archive): kill stalled hsi/htar subprocess calls instead of hanging #32

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
26 changes: 26 additions & 0 deletions CLAUDE.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down Expand Up @@ -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
Expand Down
254 changes: 236 additions & 18 deletions archive.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):
"""
Expand Down Expand Up @@ -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}")
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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")
Expand All @@ -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
Expand Down Expand Up @@ -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))
Expand All @@ -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"
Expand Down
Loading