fix(swarm-test): one watcher, and a wedge it can actually tell apart

The main merge left two step watchers: run.run_watched from main and the
newer watchdog.run_watched from here. Their disk and memory floors are
byte-identical (6144M, 6% of RAM, abort 75), so run.py now keeps only
run_step and takes the watcher from watchdog. mesh.py reaches it through
run_step and matrix.py already imported it directly.

That alone would have been a regression. The watchdog aborts on io60
alone, which its own docstring says cannot separate a wedged mount from
a busy disk, and the six mesh steps had no I/O kill before: a healthy
sleep on a host with 126G free and no task in D state died at rc=75
after 10s where the old watcher let it finish. The first sample also
lands 10s in, so 50 of the 60 seconds it averages belong to whatever ran
before -- an image pull could abort its successor before that step read
a byte.

The ceiling now takes three things together: a task in D state on the
same sample, WEDGE_SAMPLES breaches in a row, and a step old enough to
own a whole STALL_WINDOW_S of the average. Both refuted scenarios are
regression tests. A real wedge is caught about a minute later than
before; the disk floor that covers ENOSPC is an instantaneous read and
is untouched.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

tests/unit/python/utils/tests/swarm/test_watchdog.py

@@ -70,6 +70,13 @@ class TestSampleLine(unittest.TestCase):
         self.assertNotIn("\n", line)
 
 
+_WEDGE_SAMPLES = (
+    watchdog.STALL_WINDOW_S // watchdog.SAMPLE_INTERVAL_S + watchdog.WEDGE_SAMPLES
+)
+"""Samples before the earliest possible wedge abort: one whole stall window
+the step owns, then the consecutive breaches on top of it."""
+
+
 class TestRunWatched(unittest.TestCase):
     def _watch(
         self,
@@ -80,6 +87,7 @@ class TestRunWatched(unittest.TestCase):
         avail: int = 9324,
         total: int = 15947,
         free_mb: int = 38948,
+        dstate: int = 3,
     ) -> tuple[int, str, _FakeProc]:
         proc = _FakeProc(timeouts=timeouts, rc=rc)
         buffer = io.StringIO()
@@ -96,7 +104,9 @@ class TestRunWatched(unittest.TestCase):
             mock.patch.object(watchdog, "mem_stall_pct", return_value=0.0),
             mock.patch.object(watchdog, "io_stall_pct", return_value=io_stall),
             mock.patch.object(watchdog, "cpu_stall_pct", return_value=11.0),
-            mock.patch.object(watchdog, "uninterruptible_procs", return_value=3),
+            mock.patch.object(
+                watchdog, "uninterruptible_procs", return_value=dstate
+            ),
             contextlib.redirect_stdout(buffer),
         ):
             returned = watchdog.run_watched(["true"], env={})
@@ -120,19 +130,37 @@ class TestRunWatched(unittest.TestCase):
 
     def test_an_io_wedge_aborts_the_step_with_a_banner(self) -> None:
         returned, output, proc = self._watch(
-            timeouts=5, rc=0, io_stall=watchdog.IO_STALL_CEILING_PCT
+            timeouts=_WEDGE_SAMPLES, rc=0, io_stall=watchdog.IO_STALL_CEILING_PCT
         )
         self.assertEqual(returned, watchdog.ABORT_RC)
         self.assertTrue(proc.terminated)
         self.assertIn("I/O WEDGE", output)
         self.assertIn(f"{watchdog.IO_STALL_CEILING_PCT:.1f}% full stall60", output)
 
+    def test_a_busy_disk_with_nothing_blocked_is_not_a_wedge(self) -> None:
+        returned, output, proc = self._watch(
+            timeouts=_WEDGE_SAMPLES, rc=0, io_stall=99.0, dstate=0
+        )
+        self.assertEqual(returned, 0)
+        self.assertFalse(proc.terminated)
+        self.assertNotIn("I/O WEDGE", output)
+
+    def test_pressure_a_predecessor_left_behind_does_not_abort(self) -> None:
+        returned, output, proc = self._watch(
+            timeouts=watchdog.STALL_WINDOW_S // watchdog.SAMPLE_INTERVAL_S - 1,
+            rc=0,
+            io_stall=99.0,
+        )
+        self.assertEqual(returned, 0)
+        self.assertFalse(proc.terminated)
+        self.assertNotIn("I/O WEDGE", output)
+
     def test_the_wedge_sample_is_printed_before_the_abort(self) -> None:
-        _, output, _ = self._watch(timeouts=5, rc=0, io_stall=99.0)
+        _, output, _ = self._watch(timeouts=_WEDGE_SAMPLES, rc=0, io_stall=99.0)
         self.assertLess(output.index("host mem"), output.index("I/O WEDGE"))
 
     def test_the_abort_probes_the_pressure_file_not_the_mount_table(self) -> None:
-        self._watch(timeouts=5, rc=0, io_stall=99.0)
+        self._watch(timeouts=_WEDGE_SAMPLES, rc=0, io_stall=99.0)
         self.probe.assert_called_once_with(["cat", "/proc/pressure/io"], check=False)
 
     def test_io_below_the_ceiling_does_not_abort(self) -> None:

utils/tests/swarm/run.py

@@ -1,26 +1,10 @@
 from __future__ import annotations
 
-import shutil
-import subprocess
 import time
 
-from utils import PROJECT_ROOT
-from utils.env.runtime import mem_available_mb, mem_stall_pct, mem_total_mb
+from utils.tests.swarm.watchdog import run_watched
 
-DISK_FLOOR_MB = 6 * 2**10
-MEM_FLOOR_RATIO = 0.06
-
-
-def _abort(proc: subprocess.Popen[bytes], banner: str, probe: list[str]) -> int:
-    print(banner, flush=True)
-    subprocess.run(probe, check=False)
-    proc.terminate()
-    try:
-        proc.wait(timeout=30)
-    except subprocess.TimeoutExpired:
-        proc.kill()
-        proc.wait()
-    return 75
+__all__ = ["run_step", "run_watched"]
 
 
 def run_step(cmd: list[str], *, env: dict[str, str], label: str) -> int:
@@ -42,48 +26,3 @@ def run_step(cmd: list[str], *, env: dict[str, str], label: str) -> int:
         flush=True,
     )
     return rc
-
-
-def run_watched(cmd: list[str], *, env: dict[str, str]) -> int:
-    """Run a step, aborting visibly before the runner disk or RAM fills.
-
-    The Actions Worker dies silently on ENOSPC while writing its own logs, so
-    a full disk truncates the job without diagnostics; terminating the step at
-    DISK_FLOOR_MB keeps enough room for rescue artifacts and the log upload.
-
-    Args:
-        cmd: argv of the step.
-        env: environment the step runs with.
-
-    Returns:
-        The step's exit code, or the abort code when a floor was hit.
-    """
-    proc = subprocess.Popen(cmd, cwd=str(PROJECT_ROOT), env=env)
-    while True:
-        try:
-            return int(proc.wait(timeout=30))
-        except subprocess.TimeoutExpired:
-            free_mb = shutil.disk_usage("/").free // 2**20
-            if free_mb < DISK_FLOOR_MB:
-                return _abort(
-                    proc,
-                    "=== swarm-matrix: DISK EXHAUSTION IMMINENT "
-                    f"(<{DISK_FLOOR_MB}M free on /) - aborting step ===",
-                    ["df", "-h", "/"],
-                )
-            avail = mem_available_mb()
-            total = mem_total_mb()
-            stall = mem_stall_pct()
-            print(
-                f"=== swarm-matrix: host mem {avail}M/{total}M available, "
-                f"disk {free_mb}M free on /, stall60 {stall:.1f}% ===",
-                flush=True,
-            )
-            if total and avail < total * MEM_FLOOR_RATIO:
-                return _abort(
-                    proc,
-                    "=== swarm-matrix: MEMORY EXHAUSTION IMMINENT "
-                    f"({avail}M available, below {MEM_FLOOR_RATIO:.0%} of RAM)"
-                    " - aborting step ===",
-                    ["free", "-m"],
-                )

utils/tests/swarm/watchdog.py

@@ -31,6 +31,8 @@ DISK_FLOOR_MB = 6 * 2**10
 MEM_FLOOR_RATIO = 0.06
 IO_STALL_CEILING_PCT = 50.0
 SAMPLE_INTERVAL_S = 10
+STALL_WINDOW_S = 60
+WEDGE_SAMPLES = 2
 ABORT_RC = 75
 
 
@@ -86,7 +88,13 @@ def run_watched(cmd: list[str], *, env: dict[str, str]) -> int:
     a full disk truncates the job without diagnostics; terminating the step at
     DISK_FLOOR_MB keeps enough room for rescue artifacts and the log upload.
     A stuck mount produces the same silence without touching either floor, so
-    IO_STALL_CEILING_PCT ends the step while the runner can still upload.
+    IO_STALL_CEILING_PCT ends the step while the runner can still upload. It
+    takes three things together, because the pressure reading alone cannot
+    tell a wedge from a busy disk and is not even this step's: a task in D
+    state on the same sample, WEDGE_SAMPLES breaches in a row, and a step old
+    enough to own a whole STALL_WINDOW_S of the trailing average. Without the
+    last one an I/O-heavy predecessor aborts its successor before that step
+    has issued a single read.
 
     Args:
         cmd: argv of the step.
@@ -96,10 +104,13 @@ def run_watched(cmd: list[str], *, env: dict[str, str]) -> int:
         The step's exit code, or ABORT_RC when a floor was hit.
     """
     proc = subprocess.Popen(cmd, cwd=str(PROJECT_ROOT), env=env)
+    samples = 0
+    wedged = 0
     while True:
         try:
             return int(proc.wait(timeout=SAMPLE_INTERVAL_S))
         except subprocess.TimeoutExpired:
+            samples += 1
             free_mb = shutil.disk_usage("/").free // 2**20
             if free_mb < DISK_FLOOR_MB:
                 return _abort(
@@ -111,6 +122,7 @@ def run_watched(cmd: list[str], *, env: dict[str, str]) -> int:
             avail = mem_available_mb()
             total = mem_total_mb()
             io_stall = io_stall_pct()
+            dstate = uninterruptible_procs()
             print(
                 sample_line(
                     avail=avail,
@@ -119,7 +131,7 @@ def run_watched(cmd: list[str], *, env: dict[str, str]) -> int:
                     mem_stall=mem_stall_pct(),
                     io_stall=io_stall,
                     cpu_stall=cpu_stall_pct(),
-                    dstate=uninterruptible_procs(),
+                    dstate=dstate,
                 ),
                 flush=True,
             )
@@ -131,11 +143,15 @@ def run_watched(cmd: list[str], *, env: dict[str, str]) -> int:
                     " - aborting step ===",
                     ["free", "-m"],
                 )
-            if io_stall >= IO_STALL_CEILING_PCT:
+            owns_window = samples * SAMPLE_INTERVAL_S >= STALL_WINDOW_S
+            breached = io_stall >= IO_STALL_CEILING_PCT and dstate > 0
+            wedged = wedged + 1 if owns_window and breached else 0
+            if wedged >= WEDGE_SAMPLES:
                 return _abort(
                     proc,
                     "=== swarm-matrix: I/O WEDGE "
-                    f"({io_stall:.1f}% full stall60, at or above "
+                    f"({io_stall:.1f}% full stall60 with {dstate} task(s) in D "
+                    f"state, {wedged} samples at or above "
                     f"{IO_STALL_CEILING_PCT:.0f}%) - aborting step ===",
                     ["cat", "/proc/pressure/io"],
                 )