#45 fix: liveness in-flight means the post hasn't returned, not that the thread is alive

src/processor/liveness.py

@@ -13,7 +13,9 @@
 in a daemon thread, awaited for at most ``timeout_s``:
 - not the default executor, which the handler's GCS calls use, and which
   ``asyncio.run`` waits for at exit;
-- a check-in still running at the next tick skips that tick, so threads never pile up;
+- a check-in still running at the next tick skips that tick, so threads never pile up.
+  "Running" means its ``post`` has not returned, not that its thread is alive: the
+  thread hands back its result a moment before it exits (#45);
 - one attempt per tick, never a retry loop;
 - a failure is one WARNING record, and the monitor's grace absorbs a missed tick.
 
@@ -65,7 +67,9 @@ def __init__(
         self._stale_after_s = stale_after_s
         self._timeout_s = timeout_s
         self._clock = clock
-        self._thread: threading.Thread | None = None
+        # Set before a check-in thread starts; cleared by that thread once its post
+        # returns, never by the timeout path (#39 trap 1, #45).
+        self._in_flight = False
         # Whether the last tick checked in: an ok is logged only when it was not.
         self._ok = False
 
@@ -95,7 +99,7 @@ async def tick(self) -> None:
                 extra={"last_progress_age_s": age, "stale_after_s": self._stale_after_s},
             )
             return
-        if self._thread is not None and self._thread.is_alive():
+        if self._in_flight:
             self._ok = False
             logger.warning("previous check-in still in flight; skipping this tick")
             return
@@ -125,7 +129,12 @@ def _failed(self, error: str) -> None:
         logger.warning("liveness check-in failed", extra={"error": error})
 
     def _in_thread(self, variables: dict[str, str]) -> asyncio.Future:
-        """``post(variables)`` in a daemon thread; its future settles on the loop."""
+        """``post(variables)`` in a daemon thread; its future settles on the loop.
+
+        Owns ``_in_flight``: set here before the thread starts, cleared by the thread
+        once ``post`` returns and before the loop can see the result (#45). A post
+        cut off by the timeout clears it only when it returns (#39 trap 1).
+        """
         loop = asyncio.get_running_loop()
         future = loop.create_future()
 
@@ -142,11 +151,16 @@ def target() -> None:
                 result, error = self._post(variables), None
             except BaseException as exc:
                 result, error = None, exc
+            self._in_flight = False  # before the loop can see the result (#45)
             try:
                 loop.call_soon_threadsafe(settle, result, error)
             except RuntimeError:
                 pass  # the loop closed while this thread waited on Status
 
-        self._thread = threading.Thread(target=target, name="liveness-checkin", daemon=True)
-        self._thread.start()
+        self._in_flight = True
+        try:
+            threading.Thread(target=target, name="liveness-checkin", daemon=True).start()
+        except BaseException:
+            self._in_flight = False  # no thread will clear it
+            raise
         return future

tests/test_consumer.py

@@ -941,7 +941,7 @@ async def test_a_hung_status_never_delays_a_command(admin, bus, stores, hung_sta
             await issue(admin, stores)
             while not await admin.xlen(CONTENT_DERIVED):
                 await asyncio.sleep(0.01)
-        assert beat._thread is not None and beat._thread.is_alive()  # still hung on Status
+        assert beat._in_flight  # still hung on Status
     finally:
         heartbeat.cancel()
         stop.set()

tests/test_liveness.py

@@ -10,6 +10,7 @@
 import json
 import logging
 import socket
+import threading
 import time
 from functools import partial
 from types import SimpleNamespace
@@ -147,6 +148,110 @@ async def test_a_hung_status_is_cut_off_and_the_next_tick_is_skipped(hung_status
     assert skipped.levelno == logging.WARNING
 
 
+async def test_a_cut_off_check_in_frees_the_next_tick_once_its_post_returns(
+    progress, clock, caplog
+):
+    # #39 trap 1's other half (#45 CR 1): a post cut off by the timeout keeps later
+    # ticks skipping only until it returns. Were the marker cleared on the loop, where
+    # settle drops a cut-off result, one hung Status would silence the heartbeat.
+    release = threading.Event()
+    threads = []
+
+    def post(variables):
+        threads.append(threading.current_thread())
+        if len(threads) == 1:
+            release.wait(5)
+        return 202
+
+    beat = Heartbeat(
+        progress, post, build="b", clock=clock, interval_s=300, stale_after_s=605, timeout_s=0.05
+    )
+    try:
+        with caplog.at_level(logging.INFO, logger="processor.liveness"):
+            await beat.tick()  # cut off: the post is still running
+            await beat.tick()  # skipped
+            release.set()
+            threads[0].join(5)
+            await beat.tick()
+    finally:
+        release.set()
+    assert len(threads) == 2
+    (failed,) = records(caplog, "liveness check-in failed")
+    assert failed.error == "timed out after 0.05 s"
+    assert len(records(caplog, "previous check-in still in flight; skipping this tick")) == 1
+    (ok,) = records(caplog, "liveness check-in")
+    assert ok.checkin == 202
+
+
+async def test_a_finished_check_in_whose_thread_lingers_does_not_skip_the_next_tick(
+    progress, clock, caplog, monkeypatch
+):
+    # #45: the result reaches the loop before its thread exits. A tick that awaits
+    # nothing (a stale one) let the next run while that thread was still alive, and
+    # it skipped. Here every check-in thread is held after its target returns.
+    linger = threading.Event()
+
+    class Lingering(threading.Thread):
+        def run(self) -> None:
+            super().run()
+            if self.name == "liveness-checkin":
+                linger.wait(5)
+
+    monkeypatch.setattr(threading, "Thread", Lingering)
+    posts = []
+    beat = Heartbeat(
+        progress,
+        lambda v: posts.append(v) or 202,
+        build="b",
+        clock=clock,
+        interval_s=300,
+        stale_after_s=605,
+        timeout_s=10,
+    )
+    try:
+        with caplog.at_level(logging.INFO, logger="processor.liveness"):
+            await beat.tick()
+            await beat.tick()
+    finally:
+        linger.set()
+    assert len(posts) == 2
+    assert records(caplog, "previous check-in still in flight; skipping this tick") == []
+
+
+async def test_a_thread_that_fails_to_start_does_not_wedge_the_heartbeat(
+    progress, clock, caplog, monkeypatch
+):
+    # The in-flight marker is set before start(): a start that raises must clear it,
+    # or every later tick would skip and the monitor would page for nothing.
+    real_start = threading.Thread.start
+    calls = 0
+
+    def start(self) -> None:
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            raise RuntimeError("can't start new thread")
+        real_start(self)
+
+    monkeypatch.setattr(threading.Thread, "start", start)
+    beat = Heartbeat(
+        progress,
+        lambda v: 202,
+        build="b",
+        clock=clock,
+        interval_s=300,
+        stale_after_s=605,
+        timeout_s=10,
+    )
+    with caplog.at_level(logging.INFO, logger="processor.liveness"):
+        await beat.tick()
+        await beat.tick()
+    (failed,) = records(caplog, "liveness check-in failed")
+    assert failed.error == "RuntimeError: can't start new thread"
+    (ok,) = records(caplog, "liveness check-in")
+    assert ok.checkin == 202
+
+
 async def test_ok_is_logged_on_the_first_and_after_a_failure_only(
     http_stub, progress, clock, caplog
 ):