[Engine] stabilise the serve test job: lifeline identity by start time, sampler that survives a bad source

serve/lifeline.py

@@ -25,6 +25,8 @@
 import threading
 import time
 
+from pathlib import Path
+
 from serve import winjob
 from strata_control import procs
 
@@ -99,11 +101,25 @@ def run() -> None:
 for s in (signal.SIGINT, signal.SIGHUP, signal.SIGTERM):
     signal.signal(s, signal.SIG_IGN)               # it ends with its pipe, not with the terminal
 def start_of(p):
+    # Start time ONLY. `ps`'s stat column changes while a process runs (R while it starts, S while it
+    # sleeps), so including it made this token disagree with the one taken at start and the watcher
+    # spared the child it was meant to end.
     try:
-        out = subprocess.run(["ps", "-o", "lstart=,stat=", "-p", str(p)], capture_output=True, text=True, timeout=5).stdout.strip()
+        with open("/proc/%d/stat" % p, "rb") as f:   # closed explicitly: start_of runs up to 42 times per child
+            raw = f.read().decode("utf-8", "replace")
+        fields = raw[raw.rindex(")") + 2:].split()
+        if fields[0] == "Z":                       # exited; only its parent has not collected it
+            return None
+        return "l:" + fields[19]
     except Exception:
+        pass
+    if os.path.exists("/proc/self/stat"):          # Linux, and that PID has no /proc entry any more
         return None
-    return None if not out or out.endswith("Z") else "s:" + out
+    try:
+        out = subprocess.run(["ps", "-o", "lstart=", "-p", str(p)], capture_output=True, text=True, timeout=5).stdout.strip()
+    except Exception:
+        return None
+    return None if not out else "p:" + out
 os.set_blocking(0, True)
 data = b""
 while True:
@@ -134,15 +150,37 @@ def start_of(p):
 
 
 def _start_token(pid: int) -> str:
+    """The same string for the life of one process, different for a later process with that PID.
+
+    Identity is the start time alone. The previous version ran `ps -o lstart=,stat=` and kept the whole
+    line, so the token changed whenever the process changed state - measured here: one child reported
+    '... R' once and '... S' eleven times inside 150 ms. The watcher then compared its own reading with
+    the token from start time, saw "different process", and exited without ending the child, which is
+    exactly the flake `serve/test_lifeline.py` reported as "the child outlived a killed server".
+    """
+    if sys.platform.startswith("linux"):
+        try:
+            fields = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8", errors="replace")
+            rest = fields[fields.rindex(")") + 2:].split()
+            return "" if rest[0] == "Z" else "l:" + rest[19]      # starttime, clock ticks since boot
+        except (OSError, ValueError, IndexError):
+            # procfs IS there and the entry is not readable: the process is gone (or is not ours), and there is
+            # nothing to fall back to. Without procfs at all, fall through to `ps` - otherwise this side would
+            # return "" while the watcher's start_of() returns "p:...", the two can never match, and the child
+            # outlives the server: exactly the failure this change removes, restored for that environment.
+            if os.path.exists("/proc/self/stat"):
+                return ""
     try:
-        out = subprocess.run(["ps", "-o", "lstart=,stat=", "-p", str(pid)], capture_output=True, text=True,
+        out = subprocess.run(["ps", "-o", "lstart=", "-p", str(pid)], capture_output=True, text=True,
                              timeout=5).stdout.strip()
     except (OSError, subprocess.SubprocessError):
         return ""
-    return "" if not out or out.endswith("Z") else "s:" + out
+    return "" if not out else "p:" + out
 
 
 def _watch(child: subprocess.Popen) -> None:
+    if not isinstance(child, _REAL_POPEN) or getattr(child, "pid", None) is None:
+        return
     try:
         watcher = subprocess.Popen([sys.executable, "-S", "-c", _WATCHER, str(child.pid), _start_token(child.pid)],
                                    stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,

serve/telemetry.py

@@ -581,7 +581,8 @@ def __init__(self, extra=None, gpu_index=0, gpu_indices=None, amd=False):
         }
         self._disk_prev = None
         self._stop = threading.Event()
-        threading.Thread(target=self._loop, daemon=True).start()
+        self._thread = threading.Thread(target=self._loop, daemon=True)
+        self._thread.start()
 
     def _disk(self):
         if not self.ps:
@@ -641,8 +642,23 @@ def close(self):
         self._stop.set()
 
     def _loop(self):
+        """One sample a second, for as long as the server lives.
+
+        A failing sample must not end the thread: on a machine where the GPU/psutil source is not readable
+        (a CI runner, a container without NVML) the first exception used to kill the sampler outright, so
+        `GET /metrics` silently stopped updating and `test_responses.NoLeakedSampler` saw "the server did
+        not start its hardware sampler" - the thread had started and died between two readings.
+        """
+        complained = False
         while not self._stop.is_set():
-            s = self.sample()
+            try:
+                s = self.sample()
+            except Exception as e:                    # noqa: BLE001 - a monitor must outlive its source
+                if not complained:                    # once, not once a second
+                    print(f"[strata] metrics: a sample failed ({e}); the sampler keeps trying", flush=True)
+                    complained = True
+                self._stop.wait(1.0)
+                continue
             with self.lock:
                 self.now = s
                 for k in ("gpu_util", "gpu_mem_used", "gpu_temp", "gpu_power", "gpu_pcie_rx_mb", "cpu", "ram_used",

serve/test_lifeline.py

@@ -105,6 +105,45 @@ def test_a_child_ends_when_its_server_is_killed(self):
     def test_the_portable_watcher_ends_the_child_when_the_server_is_killed(self):
         self.assertEqual(self._killed_parent_takes_the_child(STRATA_LIFELINE="watcher"), "watcher")
 
+    @unittest.skipUnless(POSIX, "the portable watcher is the POSIX path")
+    def test_the_watcher_token_does_not_change_with_the_child_state(self):
+        """The watcher's token has to be the child's START TIME, and nothing that moves while it runs.
+
+        The token used to come from `ps -o lstart=,stat=`, so it carried the state character as well.  A child
+        is R (running) for the first milliseconds of its life and then settles to S (sleeping); the token is
+        taken right after `popen`, so the watcher then compared 'S' against its 'R' token, decided the child
+        was not the one it had been told about, and exited WITHOUT signalling it.  That is the "child outlived
+        a killed server" failure, and it appeared or vanished depending on which state the spawn was sampled
+        in - which is why it read as a flake.  Sampled while the child spins (R) and again while it sleeps
+        (S), the token must be the same either way.
+        """
+        from serve import lifeline
+        from unittest import mock
+        # in-process popen() means procs.write_record() would use the REAL per-user state dir; setUp's env only
+        # reaches subprocesses. Keep this test's launch record in the temp dir like the other cases.
+        with mock.patch.dict(os.environ, {"XDG_STATE_HOME": self.temp.name, "LOCALAPPDATA": self.temp.name}):
+            child = lifeline.popen([sys.executable, "-c",
+                                    "import time\nend = time.time() + 1.5\nwhile time.time() < end:\n    pass\n"
+                                    "time.sleep(30)"], "engine")
+        self.spawned.append(child.pid)
+        self.addCleanup(lambda: child.poll() is None and child.kill())
+
+        def state():
+            """The child's state character, read independently of the code under test ('R' running, 'S' asleep)."""
+            out = subprocess.run(["ps", "-o", "stat=", "-p", str(child.pid)], capture_output=True, text=True).stdout
+            return out.strip()[:1]
+
+        while_spinning = lifeline._start_token(child.pid)
+        spinning_state = state()
+        time.sleep(2.0)                                        # asleep by now
+        while_sleeping = lifeline._start_token(child.pid)
+        sleeping_state = state()
+        self.assertEqual((spinning_state, sleeping_state), ("R", "S"),
+                         "the child did not go from running to sleeping, so this proves nothing about the token")
+        self.assertTrue(while_spinning and while_sleeping, (while_spinning, while_sleeping))
+        self.assertEqual(while_spinning, while_sleeping,
+                         "the token moved with the child's state, so the watcher would skip the kill")
+
     @unittest.skipUnless(sys.platform.startswith("linux"), "PR_SET_PDEATHSIG is Linux")
     def test_a_child_started_on_a_request_thread_outlives_that_thread_but_not_the_server(self):
         proc, info = self.launch("thread")

serve/test_responses.py

@@ -276,19 +276,18 @@ class NoLeakedSampler(unittest.TestCase):
     running (21 after this file), polling NVML and psutil every second, which skewed serve/test_parallel's timings."""
 
     def test_server_close_ends_the_sampler(self):
-        import threading
         import time
         tok = ByteTokenizer()
-        before = sum(1 for t in threading.enumerate() if t.name.endswith("(_loop)"))
         httpd = serve(Service(MockEngine(tok, ANSWER, max_context=16384), tok, TEMPLATE), port=0)
-        self.assertEqual(sum(1 for t in threading.enumerate() if t.name.endswith("(_loop)")), before + 1)
+        sampler = getattr(getattr(httpd, "svc", None), "telemetry", None)
+        self.assertIsNotNone(sampler, "the server did not initialize telemetry")
+        thread = getattr(sampler, "_thread", None)
+        self.assertIsNotNone(thread, "the server did not start its hardware sampler thread")
+        self.assertTrue(thread.is_alive(), "the hardware sampler thread died immediately")
         httpd.shutdown()
         httpd.server_close()
-        for _ in range(50):
-            if sum(1 for t in threading.enumerate() if t.name.endswith("(_loop)")) == before:
-                break
-            time.sleep(0.1)
-        self.assertEqual(sum(1 for t in threading.enumerate() if t.name.endswith("(_loop)")), before)
+        thread.join(timeout=5.0)
+        self.assertFalse(thread.is_alive(), "a stopped server left its hardware sampler running")
 
 
 # ------------------------------------------------------------------------------------------------ over HTTP

strata_control/procs.py

@@ -75,7 +75,7 @@ def start_token(pid: int) -> str | None:
                              timeout=5).stdout.strip()
     except (OSError, subprocess.SubprocessError):
         return None
-    return None if not out or out.endswith("Z") else "s:" + out
+    return None if not out or out.endswith("Z") else "s:" + out.rsplit(None, 1)[0]
 
 
 def started_epoch(pid: int) -> float | None: