From a734082e816204d1f1be746fbad5ab0e6861bf49 Mon Sep 17 00:00:00 2001 From: "Doucette, Jarrod S" Date: Mon, 20 Jul 2026 14:03:44 -0400 Subject: [PATCH] fix(stall-watch): instrument, close MD5-verify gap, retry retrieve stalls 2026-07-20 investigation (job 40937735, repository_X1D_3_metabolomics_rawspectra): the 2h stall-watch reliably killed 4 prior identical htar -xvf retrieve hangs, but this one ran ~30min past that bound with no way to tell whether that was a real mechanism gap or a last-second activity reset short of a live strace/py-spy capture. - Add activity_log to _run_with_stall_watch (and send_to_fortress's inline twin): quiet until silence passes half of stall_seconds, then a timestamped trail of every activity/poll event, so a recurrence is diagnosable from disk alone. - Add _run_callable_with_stall_watch / VerifyStallError: a thread-based watchdog for the pure-Python MD5 verify loops (ship_object + send_to_fortress), which had no subprocess to terminate and thus no timeout at all - a confirmed-real gap. - ship_object's round-trip retrieve gets one automatic retry (with backoff) when the failure is specifically a stall-watch timeout, not a genuine extract failure - 4 of 5 recent stalls on this object were the create succeeding and only the retrieve hanging on a dead HPSS mover session. 409 tests passing (14 new), stable across repeated runs. Co-Authored-By: Claude Sonnet 5 --- archive.py | 482 +++++++++++++++++++---- tests/test_ship_object_retrieve_retry.py | 204 ++++++++++ tests/test_stall_activity_log.py | 183 +++++++++ 3 files changed, 795 insertions(+), 74 deletions(-) create mode 100644 tests/test_ship_object_retrieve_retry.py create mode 100644 tests/test_stall_activity_log.py diff --git a/archive.py b/archive.py index e32de4d..2a56a60 100644 --- a/archive.py +++ b/archive.py @@ -898,6 +898,18 @@ def decode_blonde(blonde): # timeout suffices for these rather than the full stall-watch treatment above. HSI_SHORT_CALL_TIMEOUT_SECONDS = 5 * 60 # 5 minutes +# ship_object's round-trip retrieve (htar -xvf / hsi get + tar xf) gets ONE +# automatic retry on a stall-watch timeout, after a short pause — 2026-07-20 stall +# investigation: 4 of 5 recent stalls on repository_X1D_3_metabolomics_rawspectra +# were the create step succeeding and ONLY the retrieve step hanging (a dead HPSS +# mover session, confirmed via CLOSE_WAIT sockets + a wedged client), so a cheap +# retry of just the retrieve clears a transient dead mover without re-running the +# whole multi-hour create. Only a stall-watch RuntimeError ("command stalled...") +# is retried — a genuine extract failure (nonzero rc, corrupt data) is not, since +# retrying that would just mask a real problem. A second stall still raises. +RETRIEVE_STALL_MAX_ATTEMPTS = 2 +RETRIEVE_STALL_RETRY_BACKOFF_SECONDS = 60 + # hsi/htar live in /opt/hsi/bin and are not always on a compute-node PATH; htar_large # is in /usr/local/bin. ship_object() prepends both defensively ahead of PATH so they # resolve even when a compute node's default shell hasn't sourced the profile snippet @@ -910,7 +922,8 @@ def decode_blonde(blonde): def _run_with_stall_watch(cmd, *, cwd=None, watch_paths=None, - stall_seconds=HTAR_STALL_SECONDS, poll_interval=30): + stall_seconds=HTAR_STALL_SECONDS, poll_interval=30, + activity_log=None): """ 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 @@ -932,6 +945,18 @@ def _run_with_stall_watch(cmd, *, cwd=None, watch_paths=None, any watched file's size. Returns a CompletedProcess-like object with only .returncode set (the output already landed in the watched files, as before). + activity_log: optional path. When set, appends one durable, timestamped line per + "new output" event AND — once silence has run past HALF of stall_seconds — one + line per poll (~every poll_interval) showing exactly how long it's been quiet. + Silent (no writes at all) the rest of the time, so a multi-hour healthy transfer + doesn't bloat this file — detail only shows up once things are already getting + concerning, which is exactly the forensic trail needed to answer, after the + fact, whether a stall that outlived stall_seconds without being killed was a + genuine mechanism gap or a last-second activity reset (2026-07-20: job 40937735 + ran ~30min past the 2.0h bound with no way to tell which, short of a live + strace/py-spy capture — this closes that gap for any recurrence). Never raises + on a logging failure (best-effort; a full /scratch must never break the run). + 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 @@ -941,6 +966,17 @@ def _run_with_stall_watch(cmd, *, cwd=None, watch_paths=None, import time import os + concern_seconds = stall_seconds / 2.0 + + def _log(msg): + if not activity_log: + return + try: + with open(activity_log, "a") as fh: + fh.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n") + except OSError: + pass + if watch_paths is None: import select @@ -956,20 +992,31 @@ def _run_with_stall_watch(cmd, *, cwd=None, watch_paths=None, if readable: data = os.read(fd, 65536) if data: + now = time.monotonic() + gap = now - last_activity + if gap > concern_seconds: + _log(f"activity resumed after {gap:.1f}s silence " + f"(+{len(data)} bytes)") + last_activity = now 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: + elapsed = time.monotonic() - last_activity + if elapsed > concern_seconds: + _log(f"poll: {elapsed:.1f}s since last activity " + f"(stall limit {stall_seconds:.0f}s)") + if elapsed > stall_seconds: proc.terminate() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=10) + _log(f"STALL DETECTED after {elapsed:.1f}s silence — " + f"killed pid={proc.pid}") raise RuntimeError( f"command stalled — no output for " f"{stall_seconds / 3600:.1f}h (HPSS connection likely " @@ -988,24 +1035,124 @@ def _run_with_stall_watch(cmd, *, cwd=None, watch_paths=None, except OSError: continue if last_sizes.get(p) != sz: + now = time.monotonic() + gap = now - last_activity + if gap > concern_seconds: + _log(f"activity resumed after {gap:.1f}s silence " + f"({p} grew to {sz} bytes)") last_sizes[p] = sz - last_activity = time.monotonic() + last_activity = now rc = proc.poll() if rc is not None: return subprocess.CompletedProcess(cmd, rc) - if time.monotonic() - last_activity > stall_seconds: + elapsed = time.monotonic() - last_activity + if elapsed > concern_seconds: + _log(f"poll: {elapsed:.1f}s since last activity " + f"(stall limit {stall_seconds:.0f}s)") + if elapsed > stall_seconds: proc.terminate() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=10) + _log(f"STALL DETECTED after {elapsed:.1f}s silence — killed pid={proc.pid}") 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)) +class VerifyStallError(RuntimeError): + """ + Raised by _run_callable_with_stall_watch when a pure-Python verify loop (no + subprocess to terminate) goes silent for too long — e.g. an MD5 read loop + stuck on a hung NFS/Depot mount. Distinct from a plain RuntimeError so callers + can tell "the watchdog gave up waiting" apart from a genuine verification + failure (missing file, checksum mismatch) raised by the wrapped callable + itself, which propagates unchanged and is NOT this type. + """ + + +def _run_callable_with_stall_watch(fn, *, stall_seconds=HTAR_STALL_SECONDS, + poll_interval=30, activity_log=None): + """ + Run fn(heartbeat) on a background daemon thread, raising VerifyStallError if + stall_seconds passes without fn calling heartbeat(). Mirrors + _run_with_stall_watch's silence-detection philosophy for pure-Python blocking + I/O that has no subprocess to terminate — e.g. the per-file MD5 comparison + loop that runs after a round-trip htar/hsi retrieve (confirmed real gap, + 2026-07-20 stall investigation: every htar/hsi subprocess call was already + stall-guarded, but that read loop had no timeout of any kind). + + fn must call heartbeat() periodically (e.g. once per chunk or once per file) + to prove progress. There is no safe way to forcibly cancel a blocking read + syscall from another Python thread, so on a real stall this raises and lets + the caller treat the run as failed — exactly like a killed htar/hsi + subprocess — while the wrapped thread (daemon=True) is simply abandoned and + dies with the process rather than being cleanly stopped. + + A genuine exception raised BY fn (e.g. a checksum mismatch) propagates + unchanged once the thread finishes — only a real stall raises + VerifyStallError, so callers can distinguish "the watchdog gave up" from + "the wrapped work itself failed" without inspecting message text. + + activity_log: same semantics as _run_with_stall_watch's — quiet until silence + passes half of stall_seconds, then one line per heartbeat/poll. + """ + import threading + import time + + concern_seconds = stall_seconds / 2.0 + + def _log(msg): + if not activity_log: + return + try: + with open(activity_log, "a") as fh: + fh.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n") + except OSError: + pass + + result = {} + state = {"last_activity": time.monotonic()} + + def heartbeat(): + now = time.monotonic() + gap = now - state["last_activity"] + if gap > concern_seconds: + _log(f"activity resumed after {gap:.1f}s silence") + state["last_activity"] = now + + def worker(): + try: + result["value"] = fn(heartbeat) + except BaseException as e: # re-raised on the main thread below + result["error"] = e + + thread = threading.Thread(target=worker, daemon=True) + thread.start() + while True: + thread.join(timeout=min(poll_interval, stall_seconds)) + if not thread.is_alive(): + break + elapsed = time.monotonic() - state["last_activity"] + if elapsed > concern_seconds: + _log(f"poll: {elapsed:.1f}s since last activity " + f"(stall limit {stall_seconds:.0f}s)") + if elapsed > stall_seconds: + _log(f"STALL DETECTED after {elapsed:.1f}s silence — " + f"read thread could not be forcibly stopped") + raise VerifyStallError( + f"MD5 verify read stalled — no progress for " + f"{stall_seconds / 3600:.1f}h (local/NFS filesystem read likely " + f"hung); the read thread could not be forcibly cancelled and may " + f"still be running in the background.") + if "error" in result: + raise result["error"] + return result["value"] + + def clamp_t_small(t_small): """ Raise a configured T_small to the 100 MB floor if needed (RFC §2.2): below it a @@ -3017,7 +3164,8 @@ 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): + def run_watched(cmd, watch_paths, stall_seconds=7200, poll_interval=30, + activity_log=None): """ 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 @@ -3030,11 +3178,31 @@ def run_watched(cmd, watch_paths, stall_seconds=7200, poll_interval=30): 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. + + activity_log: same semantics as the module-level twin's — quiet until + silence passes half of stall_seconds, then one durable, timestamped line per + growth event/poll, so a stall that outlives stall_seconds without being + killed leaves a forensic trail instead of requiring a live strace/py-spy + capture to explain (2026-07-20 stall investigation: job 40937735 ran ~30min + past the 2.0h bound with no way, after the fact, to tell whether that was a + last-second activity reset or a real mechanism gap). Kept in sync with the + module-level twin's activity_log behavior. """ import subprocess import time import os + concern_seconds = stall_seconds / 2.0 + + def _log(msg): + if not activity_log: + return + try: + with open(activity_log, "a") as fh: + fh.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n") + except OSError: + pass + proc = subprocess.Popen(cmd, shell=True) last_activity = time.monotonic() last_sizes = {} @@ -3045,24 +3213,101 @@ def run_watched(cmd, watch_paths, stall_seconds=7200, poll_interval=30): except OSError: continue if last_sizes.get(p) != sz: + now = time.monotonic() + gap = now - last_activity + if gap > concern_seconds: + _log(f"activity resumed after {gap:.1f}s silence " + f"({p} grew to {sz} bytes)") last_sizes[p] = sz - last_activity = time.monotonic() + last_activity = now rc = proc.poll() if rc is not None: return subprocess.CompletedProcess(cmd, rc) - if time.monotonic() - last_activity > stall_seconds: + elapsed = time.monotonic() - last_activity + if elapsed > concern_seconds: + _log(f"poll: {elapsed:.1f}s since last activity " + f"(stall limit {stall_seconds:.0f}s)") + if elapsed > stall_seconds: proc.terminate() try: proc.wait(timeout=10) except subprocess.TimeoutExpired: proc.kill() proc.wait(timeout=10) + _log(f"STALL DETECTED after {elapsed:.1f}s silence — " + f"killed pid={proc.pid}") 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)) + def run_callable_watched(fn, stall_seconds=7200, poll_interval=30, + activity_log=None): + """ + Inline, Compute-safe twin of the module-level + _run_callable_with_stall_watch(): runs fn(heartbeat) on a background daemon + thread, raising RuntimeError if stall_seconds passes without a heartbeat() + call. Used for the pure-Python MD5 verify loops below, which have no + subprocess to terminate on a hang (e.g. a stuck local/NFS read) — a + confirmed-real gap found in the 2026-07-20 stall investigation. This one can + run on a compute node under --globus, so it references no module/outer names + and imports inside (invariant #1); kept in sync with the module-level twin + (which raises the distinct VerifyStallError type there — not available here + since this function must not reference module-level names). + """ + import threading + import time + + concern_seconds = stall_seconds / 2.0 + + def _log(msg): + if not activity_log: + return + try: + with open(activity_log, "a") as fh: + fh.write(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {msg}\n") + except OSError: + pass + + result = {} + state = {"last_activity": time.monotonic()} + + def heartbeat(): + now = time.monotonic() + gap = now - state["last_activity"] + if gap > concern_seconds: + _log(f"activity resumed after {gap:.1f}s silence") + state["last_activity"] = now + + def worker(): + try: + result["value"] = fn(heartbeat) + except BaseException as e: + result["error"] = e + + thread = threading.Thread(target=worker, daemon=True) + thread.start() + while True: + thread.join(timeout=min(poll_interval, stall_seconds)) + if not thread.is_alive(): + break + elapsed = time.monotonic() - state["last_activity"] + if elapsed > concern_seconds: + _log(f"poll: {elapsed:.1f}s since last activity " + f"(stall limit {stall_seconds:.0f}s)") + if elapsed > stall_seconds: + _log(f"STALL DETECTED after {elapsed:.1f}s silence — " + f"read thread could not be forcibly stopped") + raise RuntimeError( + f"MD5 verify read stalled — no progress for " + f"{stall_seconds / 3600:.1f}h (local/NFS filesystem read likely " + f"hung); the read thread could not be forcibly cancelled and " + f"may still be running in the background.") + if "error" in result: + raise result["error"] + return result["value"] + write_log(f"Starting archive workflow") write_log(f"Project: {project_name}") write_log(f"Source folder (Depot): {source_folder}") @@ -3148,7 +3393,8 @@ def run_watched(cmd, watch_paths, stall_seconds=7200, poll_interval=30): cmd = (f'cd {src_dir} && htar_large -cvf {fortress_tar} {filename} ' f'>{stdout_log} 2>{stderr_log}') try: - result = run_watched(cmd, [stdout_log, stderr_log]) + result = run_watched(cmd, [stdout_log, stderr_log], + activity_log=os.path.join(log_dir, f"{zip_ts}.create.stall.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}") @@ -3207,7 +3453,8 @@ def run_watched(cmd, watch_paths, stall_seconds=7200, poll_interval=30): hsi_get_cmd = (f'hsi "get {retrieved_tar} : {fortress_tar}" ' f'>{hsi_get_out} 2>{hsi_get_err}') try: - hsi_result = run_watched(hsi_get_cmd, [hsi_get_out, hsi_get_err]) + hsi_result = run_watched(hsi_get_cmd, [hsi_get_out, hsi_get_err], + activity_log=os.path.join(log_dir, f"{zip_ts}.hsiget.stall.log")) except RuntimeError as e: write_log(f"hsi get stalled: {e}", level="ERROR") finalize_log("failed", error=f"hsi get stalled: {e}") @@ -3267,7 +3514,8 @@ def run_watched(cmd, watch_paths, stall_seconds=7200, poll_interval=30): 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}' try: - tar_result = run_watched(tar_cmd, [tar_extract_out, tar_extract_err]) + tar_result = run_watched(tar_cmd, [tar_extract_out, tar_extract_err], + activity_log=os.path.join(log_dir, f"{zip_ts}.tarextract.stall.log")) except RuntimeError as e: write_log(f"tar extraction stalled: {e}", level="ERROR") finalize_log("failed", error=f"tar extraction stalled: {e}") @@ -3314,12 +3562,34 @@ def run_watched(cmd, watch_paths, stall_seconds=7200, poll_interval=30): f"Log (JSON): {json_path}") raise RuntimeError(msg) - md5 = hashlib.md5() - fh = open(retrieved_zip, 'rb') - for chunk in iter(lambda: fh.read(8192), b''): - md5.update(chunk) - fh.close() - retrieved_checksum = md5.hexdigest() + # No subprocess to terminate for this read, so a hung local/NFS read gets the + # callable-based watchdog instead — confirmed-real gap from the 2026-07-20 + # stall investigation (every htar/hsi subprocess call was already + # stall-guarded, but this read loop had no timeout of any kind). + def _hash_zip(heartbeat): + m = hashlib.md5() + with open(retrieved_zip, 'rb') as fh: + for chunk in iter(lambda: fh.read(8192), b''): + m.update(chunk) + heartbeat() + return m.hexdigest() + + try: + retrieved_checksum = run_callable_watched( + _hash_zip, + activity_log=os.path.join(log_dir, f"{zip_ts}.zipmd5.stall.log")) + except RuntimeError as e: + write_log(f"Zip MD5 read stalled: {e}", level="ERROR") + finalize_log("failed", error=f"Zip MD5 read stalled: {e}") + send_email( + f"[FAILED] Fortress checksum read stalled: {filename}", + f"Project: {project_name}\n\n" + f"Reading the retrieved zip to verify its MD5 stalled (no progress) " + f"for {filename}.\n\nError:\n{e}\n\n" + f"Log (TXT): {txt_path}\n" + f"Log (JSON): {json_path}" + ) + raise if retrieved_checksum != zip_checksum: msg = ( @@ -3362,41 +3632,70 @@ def run_watched(cmd, watch_paths, stall_seconds=7200, poll_interval=30): ) raise RuntimeError(f"Retrieved zip extraction failed: {e}") - for member in members: - extracted_path = os.path.join(extract_dir, member) - if not os.path.exists(extracted_path): - msg = f"Missing file after extraction: {member}" - write_log(msg, level="ERROR") - finalize_log("failed", error=msg) - send_email(f"[FAILED] Fortress missing file: {filename}", - f"Project: {project_name}\n\n{msg}\n\n" - f"Log (TXT): {txt_path}\n" - f"Log (JSON): {json_path}") - raise RuntimeError(msg) + # No subprocess to terminate for these reads, so a hung local/NFS read gets + # the callable-based watchdog instead — confirmed-real gap from the + # 2026-07-20 stall investigation. Genuine verification failures (missing + # file, mismatch) raise from inside the worker exactly as before, with + # their own existing write_log/send_email — the watchdog only adds a NEW, + # distinctly-worded stall case, distinguished below by message prefix so + # a real failure's email is never sent twice. + def _verify_members(heartbeat): + for member in members: + extracted_path = os.path.join(extract_dir, member) + if not os.path.exists(extracted_path): + msg = f"Missing file after extraction: {member}" + write_log(msg, level="ERROR") + finalize_log("failed", error=msg) + send_email(f"[FAILED] Fortress missing file: {filename}", + f"Project: {project_name}\n\n{msg}\n\n" + f"Log (TXT): {txt_path}\n" + f"Log (JSON): {json_path}") + raise RuntimeError(msg) - md5 = hashlib.md5() - fh = open(extracted_path, 'rb') - for chunk in iter(lambda: fh.read(8192), b''): - md5.update(chunk) - fh.close() - extracted_checksum = md5.hexdigest() - expected = file_checksums.get(member) + md5 = hashlib.md5() + fh = open(extracted_path, 'rb') + for chunk in iter(lambda: fh.read(8192), b''): + md5.update(chunk) + heartbeat() + fh.close() + extracted_checksum = md5.hexdigest() + expected = file_checksums.get(member) - if expected and extracted_checksum != expected: - msg = ( - f"File MD5 mismatch after round-trip: {member}\n" - f" Expected: {expected}\n" - f" Got: {extracted_checksum}" - ) - write_log(msg, level="ERROR") - finalize_log("failed", error=msg) - send_email(f"[FAILED] Fortress file checksum mismatch: {filename}", - f"Project: {project_name}\n\n{msg}\n\n" - f"Log (TXT): {txt_path}\n" - f"Log (JSON): {json_path}") - raise RuntimeError(msg) + if expected and extracted_checksum != expected: + msg = ( + f"File MD5 mismatch after round-trip: {member}\n" + f" Expected: {expected}\n" + f" Got: {extracted_checksum}" + ) + write_log(msg, level="ERROR") + finalize_log("failed", error=msg) + send_email(f"[FAILED] Fortress file checksum mismatch: {filename}", + f"Project: {project_name}\n\n{msg}\n\n" + f"Log (TXT): {txt_path}\n" + f"Log (JSON): {json_path}") + raise RuntimeError(msg) + + write_log(f"File MD5 verified: {member} = {extracted_checksum}") + heartbeat() - write_log(f"File MD5 verified: {member} = {extracted_checksum}") + try: + run_callable_watched( + _verify_members, + activity_log=os.path.join(log_dir, f"{zip_ts}.mdverify.stall.log")) + except RuntimeError as e: + if str(e).startswith("MD5 verify read stalled"): + write_log(f"Per-file MD5 verify stalled: {e}", level="ERROR") + finalize_log("failed", error=f"Per-file MD5 verify stalled: {e}") + send_email( + f"[FAILED] Fortress verify stalled: {filename}", + f"Project: {project_name}\n\n" + f"Reading an extracted file to verify its MD5 stalled (no " + f"progress) during round-trip verification for {filename}.\n\n" + f"Error:\n{e}\n\n" + f"Log (TXT): {txt_path}\n" + f"Log (JSON): {json_path}" + ) + raise # Clean up round-trip temp files. The local staging zip is retained by # default (manual removal); cleanup_zip_on_success opts in to removing it @@ -3511,6 +3810,7 @@ def ship_object(source_folder, arcnames, object_stem, catalog, fortress_base_dir import datetime import json import shutil + import time # See HSI_PATH_DIRS above — prepend both defensively (mirrors send_to_fortress). for _bin in HSI_PATH_DIRS: @@ -3564,7 +3864,9 @@ 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 = _run_with_stall_watch(cmd, cwd=source_folder) + create = _run_with_stall_watch( + cmd, cwd=source_folder, + activity_log=os.path.join(log_dir, f"{run_timestamp}.create.stall.log")) create_stdout = create.stdout.decode("utf-8", "replace") # Fail fast on an explicit member-omission message, BEFORE the rc!=0 @@ -3655,32 +3957,64 @@ 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. retrieve_dir = os.path.join(tmp_dir, f"{run_timestamp}.verify") - shutil.rmtree(retrieve_dir, ignore_errors=True) - os.makedirs(retrieve_dir) - if transport == "htar": - vr = _run_with_stall_watch(["htar", "-xvf", fortress_tar], cwd=retrieve_dir) - else: - local_tar = os.path.join(retrieve_dir, "obj.tar") - 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 = _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" - + vr.stdout.decode("utf-8", "replace")[-4000:]) - - for arc, expected in file_checksums.items(): - got = hashlib.md5() - with open(os.path.join(retrieve_dir, arc), "rb") as fh: - for chunk in iter(lambda: fh.read(1024 * 1024), b""): - got.update(chunk) - if got.hexdigest() != expected: + retrieve_stall_log = os.path.join(log_dir, f"{run_timestamp}.retrieve.stall.log") + + def _attempt_retrieve(): + shutil.rmtree(retrieve_dir, ignore_errors=True) + os.makedirs(retrieve_dir) + if transport == "htar": + vr = _run_with_stall_watch(["htar", "-xvf", fortress_tar], + cwd=retrieve_dir, + 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}"', + 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, + activity_log=retrieve_stall_log) + if vr.returncode != 0: raise RuntimeError( - f"MD5 mismatch on round-trip for {arc} in {run_timestamp} " - f"(expected {expected}, got {got.hexdigest()})") + f"round-trip extract failed for {run_timestamp}:\n" + + vr.stdout.decode("utf-8", "replace")[-4000:]) + + for attempt in range(1, RETRIEVE_STALL_MAX_ATTEMPTS + 1): + try: + _attempt_retrieve() + break + except RuntimeError as e: + if ("command stalled" not in str(e) + or attempt >= RETRIEVE_STALL_MAX_ATTEMPTS): + raise + print(f" WARNING: retrieve stalled (attempt {attempt}/" + f"{RETRIEVE_STALL_MAX_ATTEMPTS}) for {run_timestamp}, " + f"retrying in {RETRIEVE_STALL_RETRY_BACKOFF_SECONDS}s: {e}") + time.sleep(RETRIEVE_STALL_RETRY_BACKOFF_SECONDS) + + # MD5 verify loop: no subprocess to terminate here, so a hung local/NFS + # read gets the callable-based watchdog (VerifyStallError) instead — + # confirmed-real gap from the 2026-07-20 stall investigation. + def _verify_md5s(heartbeat): + for arc, expected in file_checksums.items(): + got = hashlib.md5() + with open(os.path.join(retrieve_dir, arc), "rb") as fh: + for chunk in iter(lambda: fh.read(1024 * 1024), b""): + got.update(chunk) + heartbeat() + heartbeat() + if got.hexdigest() != expected: + raise RuntimeError( + f"MD5 mismatch on round-trip for {arc} in {run_timestamp} " + f"(expected {expected}, got {got.hexdigest()})") + + _run_callable_with_stall_watch( + _verify_md5s, + activity_log=os.path.join(log_dir, f"{run_timestamp}.mdverify.stall.log")) shutil.rmtree(retrieve_dir, ignore_errors=True) finally: try: diff --git a/tests/test_ship_object_retrieve_retry.py b/tests/test_ship_object_retrieve_retry.py new file mode 100644 index 0000000..daca43a --- /dev/null +++ b/tests/test_ship_object_retrieve_retry.py @@ -0,0 +1,204 @@ +""" +Unit tests for ship_object()'s retrieve-stall retry-with-backoff (2026-07-20 stall +investigation): 4 of 5 recent stalls on repository_X1D_3_metabolomics_rawspectra were +the create step succeeding and ONLY the round-trip retrieve step (htar -xvf) hanging +on a dead HPSS mover session, each caught cleanly by the stall-watch at exactly "no +output for 2.0h" — real, reproducible flakiness worth one cheap automatic retry +before failing the whole object (which would otherwise force a full multi-hour +re-create on the very next run). Only a stall-watch RuntimeError ("command +stalled...") is retried; a genuine extract failure (nonzero rc, corrupt data) is not, +since retrying that would just mask a real problem. + +ship_object always calls _run_with_stall_watch with its production defaults +(HTAR_STALL_SECONDS=2h, bound at function-definition time) — it never overrides +stall_seconds/poll_interval per call, unlike the direct tests in +test_stall_detection.py and test_stall_activity_log.py. To exercise the retry control +flow in well under a second, these tests wrap archive._run_with_stall_watch with a +thin adapter that forces small stall_seconds/poll_interval while still driving the +REAL function (real Popen/select/os.read plumbing, real fake htar/hsi binaries via +FakeHsiHtarEnv) — no mocking of subprocess itself, matching this suite's existing +philosophy. +""" +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 +from test_ship_object_htar_rc import FakeHsiHtarEnv, _write_fake_bin # noqa: E402 + + +class _FastStallWatch: + """Wraps the real archive._run_with_stall_watch, forcing a small + stall_seconds/poll_interval regardless of what the caller passes — ship_object + never overrides these, so without this the retry tests would need to actually + wait out the real 2-hour HTAR_STALL_SECONDS default.""" + + def __init__(self, stall_seconds=0.5, poll_interval=0.05): + self.real = archive._run_with_stall_watch + self.stall_seconds = stall_seconds + self.poll_interval = poll_interval + + def __call__(self, cmd, *, cwd=None, watch_paths=None, stall_seconds=None, + poll_interval=None, activity_log=None): + return self.real(cmd, cwd=cwd, watch_paths=watch_paths, + stall_seconds=self.stall_seconds, + poll_interval=self.poll_interval, + activity_log=activity_log) + + +class TestShipObjectRetrieveStallRetry(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(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 _write_stalling_htar(self, bin_dir, call_log_path, stdout_file, + arcnames_file, stall_marker, sleep_seconds, + create_rc=0, xvf_rc=0): + """A fake `htar` whose -xvf mode sleeps (simulating a hung HPSS mover + connection) exactly once — tracked via a marker file, so the retry attempt + behaves like a real recovered connection: fast, and (if xvf_rc == 0) copies + files into cwd (retrieve_dir) so the subsequent MD5 verify has real bytes + to check, exactly like the recovered real htar_process would.""" + _write_fake_bin(bin_dir, "htar", f''' +printf 'htar %s\\n' "$*" >> "{call_log_path}" +if [ "$1" = "-xvf" ]; then + if [ ! -f "{stall_marker}" ]; then + touch "{stall_marker}" + sleep {sleep_seconds} + fi + if [ {xvf_rc} -ne 0 ]; then + exit {xvf_rc} + fi + 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 {create_rc} +''') + + def _env_with_stalling_retrieve(self, sleep_seconds, xvf_rc=0): + """Builds a FakeHsiHtarEnv-like PATH setup, but with the -xvf handler + replaced by _write_stalling_htar. Mirrors FakeHsiHtarEnv's __enter__ so the + rest of ship_object's PATH-prepend guard behaves identically.""" + env = FakeHsiHtarEnv(create_rc=0, create_stdout=b"HTAR: HTAR SUCCESSFUL\n", + ls_rc=0, source_folder=self.source_folder, + arcnames=self.arcnames) + env.__enter__() + bin_dir = os.environ["PATH"].split(os.pathsep)[0] + stdout_file = os.path.join(os.path.dirname(bin_dir), "create_stdout.txt") + arcnames_file = os.path.join(os.path.dirname(bin_dir), "arcnames.txt") + stall_marker = os.path.join(os.path.dirname(bin_dir), "stall_marker") + self._write_stalling_htar(bin_dir, env.call_log_path, stdout_file, + arcnames_file, stall_marker, sleep_seconds, + create_rc=0, xvf_rc=xvf_rc) + return env + + def test_retrieve_stall_then_retry_succeeds(self): + env = self._env_with_stalling_retrieve(sleep_seconds=3) + try: + with mock.patch.object(archive, "_run_with_stall_watch", + _FastStallWatch()), \ + mock.patch.object(archive, "RETRIEVE_STALL_RETRY_BACKOFF_SECONDS", + 0.05): + 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() + finally: + env.__exit__(None, None, None) + + # Overall ship succeeded despite the first retrieve attempt stalling. + self.assertEqual(result["transport"], "htar") + xvf_calls = [c for c in calls if c.startswith("htar -xvf")] + self.assertEqual(len(xvf_calls), 2, "expected exactly one retry") + + # The stalled attempt's activity log recorded the detection. + stall_log = os.path.join(self.log_dir, + f"{result['run_timestamp']}.retrieve.stall.log") + self.assertTrue(os.path.isfile(stall_log)) + with open(stall_log) as fh: + contents = fh.read() + self.assertIn("STALL DETECTED", contents) + + def test_retrieve_stalls_twice_raises_after_max_attempts(self): + # sleep_seconds only applies once per the marker-file design above, so to + # simulate a PERSISTENT stall (every attempt hangs) the marker must never + # get created — point the marker at a path under a directory that will + # never exist, so `[ ! -f ... ]` is always true. + env = FakeHsiHtarEnv(create_rc=0, create_stdout=b"HTAR: HTAR SUCCESSFUL\n", + ls_rc=0, source_folder=self.source_folder, + arcnames=self.arcnames) + env.__enter__() + try: + bin_dir = os.environ["PATH"].split(os.pathsep)[0] + stdout_file = os.path.join(os.path.dirname(bin_dir), "create_stdout.txt") + arcnames_file = os.path.join(os.path.dirname(bin_dir), "arcnames.txt") + never_marker = os.path.join(os.path.dirname(bin_dir), + "no", "such", "dir", "marker") + self._write_stalling_htar(bin_dir, env.call_log_path, stdout_file, + arcnames_file, never_marker, sleep_seconds=3) + + with mock.patch.object(archive, "_run_with_stall_watch", + _FastStallWatch()), \ + mock.patch.object(archive, "RETRIEVE_STALL_RETRY_BACKOFF_SECONDS", + 0.05): + 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() + finally: + env.__exit__(None, None, None) + + self.assertIn("command stalled", str(ctx.exception)) + xvf_calls = [c for c in calls if c.startswith("htar -xvf")] + self.assertEqual(len(xvf_calls), archive.RETRIEVE_STALL_MAX_ATTEMPTS, + "must give up after exactly RETRIEVE_STALL_MAX_ATTEMPTS, " + "not retry forever") + + def test_nonstall_extract_failure_is_not_retried(self): + # A genuine extract failure (nonzero rc, no stall) must raise immediately — + # retrying that would just mask a real data problem, not a transient + # dead-mover session. + env = self._env_with_stalling_retrieve(sleep_seconds=0, xvf_rc=1) + try: + with mock.patch.object(archive, "_run_with_stall_watch", + _FastStallWatch()), \ + mock.patch.object(archive, "RETRIEVE_STALL_RETRY_BACKOFF_SECONDS", + 0.05): + 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() + finally: + env.__exit__(None, None, None) + + self.assertIn("round-trip extract failed", str(ctx.exception)) + self.assertNotIn("command stalled", str(ctx.exception)) + xvf_calls = [c for c in calls if c.startswith("htar -xvf")] + self.assertEqual(len(xvf_calls), 1, "a non-stall failure must not retry") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_stall_activity_log.py b/tests/test_stall_activity_log.py new file mode 100644 index 0000000..28e9509 --- /dev/null +++ b/tests/test_stall_activity_log.py @@ -0,0 +1,183 @@ +""" +Unit tests for the stall-watch instrumentation added by the 2026-07-20 investigation +(repository_X1D_3_metabolomics_rawspectra, job 40937735): a live retrieve stall ran +~30 minutes past the 2.0h HTAR_STALL_SECONDS bound without being killed, and there was +no way — short of a live strace/py-spy capture — to tell whether that was a genuine +mechanism gap or a last-second activity reset. `activity_log` closes that gap: a +durable, timestamped trail of every activity event once silence passes half of +stall_seconds, quiet the rest of the time so a multi-hour healthy transfer doesn't +bloat the file. + +Also covers _run_callable_with_stall_watch / VerifyStallError, the pure-Python +callable-based watchdog added for the confirmed-real gap in the per-file MD5 verify +loop (no subprocess to terminate on a hung local/NFS read). + +Real subprocesses / real threads throughout (matching test_stall_detection.py's +stated philosophy) — only stall_seconds/poll_interval are shrunk. +""" +import os +import shutil +import sys +import tempfile +import time +import unittest + +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) +import archive # noqa: E402 + + +def _read_log(path): + if not os.path.isfile(path): + return "" + with open(path) as fh: + return fh.read() + + +class TestActivityLogPipeMode(unittest.TestCase): + def setUp(self): + self.d = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.d, ignore_errors=True) + self.log_path = os.path.join(self.d, "activity.log") + + def test_healthy_run_leaves_no_log_file(self): + # Steady output well under the stall window must write NOTHING — the whole + # point is staying quiet unless something's actually concerning, so a + # multi-hour healthy transfer doesn't bloat log_dir with a file per call. + cmd = ["sh", "-c", "for i in 1 2 3; do echo tick; sleep 0.02; done; exit 0"] + result = archive._run_with_stall_watch(cmd, stall_seconds=1.0, + poll_interval=0.05, + activity_log=self.log_path) + self.assertEqual(result.returncode, 0) + self.assertFalse(os.path.exists(self.log_path)) + + def test_stalled_run_logs_poll_and_stall_lines(self): + cmd = ["sh", "-c", "sleep 5"] + with self.assertRaises(RuntimeError): + archive._run_with_stall_watch(cmd, stall_seconds=0.3, poll_interval=0.05, + activity_log=self.log_path) + contents = _read_log(self.log_path) + self.assertIn("poll:", contents) + self.assertIn("STALL DETECTED", contents) + self.assertIn("killed pid=", contents) + + def test_activity_resumed_after_concern_threshold_is_logged(self): + # Silent past HALF the stall window, then one more tick before the full + # window elapses — must survive (not killed) AND log the recovery, so a + # post-mortem can see exactly when/whether activity actually resumed. + cmd = ["sh", "-c", "sleep 0.3; echo late-tick; exit 0"] + result = archive._run_with_stall_watch(cmd, stall_seconds=0.5, + poll_interval=0.05, + activity_log=self.log_path) + self.assertEqual(result.returncode, 0) + contents = _read_log(self.log_path) + self.assertIn("activity resumed after", contents) + self.assertIn("late-tick", str(result.stdout)) + + def test_logging_failure_never_raises(self): + # activity_log pointed at a path under a directory that doesn't exist — + # _log's open() will fail; must be swallowed, never break the run. + bad_path = os.path.join(self.d, "no", "such", "dir", "activity.log") + cmd = ["sh", "-c", "sleep 5"] + with self.assertRaises(RuntimeError) as ctx: + archive._run_with_stall_watch(cmd, stall_seconds=0.3, poll_interval=0.05, + activity_log=bad_path) + self.assertIn("stalled", str(ctx.exception)) + + +class TestActivityLogFileGrowthMode(unittest.TestCase): + 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") + self.log_path = os.path.join(self.d, "activity.log") + + def test_healthy_growth_leaves_no_log_file(self): + cmd = f"for i in 1 2 3; do echo tick >> {self.out}; sleep 0.02; done; exit 0" + result = archive._run_with_stall_watch(cmd, watch_paths=[self.out], + stall_seconds=1.0, poll_interval=0.05, + activity_log=self.log_path) + self.assertEqual(result.returncode, 0) + self.assertFalse(os.path.exists(self.log_path)) + + def test_stalled_growth_logs_poll_and_stall_lines(self): + cmd = f"sleep 5 > {self.out} 2>&1" + with self.assertRaises(RuntimeError): + archive._run_with_stall_watch(cmd, watch_paths=[self.out], + stall_seconds=0.3, poll_interval=0.05, + activity_log=self.log_path) + contents = _read_log(self.log_path) + self.assertIn("poll:", contents) + self.assertIn("STALL DETECTED", contents) + + def test_activity_resumed_is_logged_with_path_and_size(self): + cmd = f"sleep 0.3; echo late >> {self.out}; exit 0" + result = archive._run_with_stall_watch(cmd, watch_paths=[self.out], + stall_seconds=0.5, poll_interval=0.05, + activity_log=self.log_path) + self.assertEqual(result.returncode, 0) + contents = _read_log(self.log_path) + self.assertIn("activity resumed after", contents) + self.assertIn(self.out, contents) + + +class TestCallableStallWatch(unittest.TestCase): + """_run_callable_with_stall_watch / VerifyStallError — the pure-Python watchdog + for the MD5 verify loops, which have no subprocess to terminate on a hang.""" + + def setUp(self): + self.d = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, self.d, ignore_errors=True) + self.log_path = os.path.join(self.d, "activity.log") + + def test_completes_normally_with_heartbeats(self): + def fn(heartbeat): + for i in range(3): + heartbeat() + time.sleep(0.02) + return "done" + + result = archive._run_callable_with_stall_watch( + fn, stall_seconds=1.0, poll_interval=0.05, activity_log=self.log_path) + self.assertEqual(result, "done") + self.assertFalse(os.path.exists(self.log_path)) + + def test_no_heartbeat_raises_verify_stall_error(self): + def fn(heartbeat): + time.sleep(5) + return "never" + + start = time.monotonic() + with self.assertRaises(archive.VerifyStallError) as ctx: + archive._run_callable_with_stall_watch( + fn, stall_seconds=0.3, poll_interval=0.05, activity_log=self.log_path) + elapsed = time.monotonic() - start + self.assertIn("MD5 verify read stalled", str(ctx.exception)) + self.assertIn("no progress", str(ctx.exception)) + # Detected well before fn's own 5s sleep would have finished. + self.assertLess(elapsed, 2.0) + contents = _read_log(self.log_path) + self.assertIn("STALL DETECTED", contents) + + def test_genuine_exception_from_fn_propagates_unchanged(self): + # A real verification failure (e.g. checksum mismatch) must come through + # exactly as raised — not wrapped, not swallowed, not turned into a + # VerifyStallError — so callers can still tell "the work itself failed" + # apart from "the watchdog gave up waiting". + def fn(heartbeat): + heartbeat() + raise RuntimeError("MD5 mismatch on round-trip for a.txt") + + with self.assertRaises(RuntimeError) as ctx: + archive._run_callable_with_stall_watch(fn, stall_seconds=1.0, + poll_interval=0.05) + self.assertNotIsInstance(ctx.exception, archive.VerifyStallError) + self.assertIn("MD5 mismatch", str(ctx.exception)) + + def test_verify_stall_error_is_a_runtime_error_subclass(self): + # Callers that only catch RuntimeError (e.g. ship_object's retrieve retry + # loop, or a bare except RuntimeError) must still catch this. + self.assertTrue(issubclass(archive.VerifyStallError, RuntimeError)) + + +if __name__ == "__main__": + unittest.main()