diff --git a/src/processor/liveness.py b/src/processor/liveness.py index f6d3980..1920c32 100644 --- a/src/processor/liveness.py +++ b/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 diff --git a/tests/test_consumer.py b/tests/test_consumer.py index 5ef5794..f48c670 100644 --- a/tests/test_consumer.py +++ b/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() diff --git a/tests/test_liveness.py b/tests/test_liveness.py index 6432f85..1196109 100644 --- a/tests/test_liveness.py +++ b/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 ):