From a39bb0d91ba18782a2acc4e6868f4a9fd20cf2b6 Mon Sep 17 00:00:00 2001 From: gregoryfoster Date: Sat, 10 Oct 2026 19:36:37 +0000 Subject: [PATCH 1/5] #45 test: a check-in thread that outlives its result must not skip the next tick Every check-in thread is held after its target returns: the next tick skipped ("previous check-in still in flight"), the CI flake reproduced without sleeps. A thread that fails to start must not leave the heartbeat skipping for good. Co-Authored-By: Claude Opus 5.5 --- tests/test_liveness.py | 69 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) diff --git a/tests/test_liveness.py b/tests/test_liveness.py index 6432f85..fd12eae 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,74 @@ 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_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 + starts = [] + + def start(self) -> None: + starts.append(self.name) + if len(starts) == 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 ): From 2428fe8f32e67702c6a6040aa8fbcb8b5ad9947f Mon Sep 17 00:00:00 2001 From: gregoryfoster Date: Sat, 10 Oct 2026 19:36:37 +0000 Subject: [PATCH 2/5] #45 fix: in flight means the post hasn't returned, not that the thread is alive The thread clears an explicit marker before handing its result to the loop, so a tick that follows at once no longer sees a finishing thread. Only the thread clears it, never the timeout path: a post cut off by asyncio.timeout still counts as in flight until it returns (#39 trap 1). A failed start() clears it too. Co-Authored-By: Claude Opus 5.5 --- src/processor/liveness.py | 19 ++++++++++++++----- tests/test_consumer.py | 2 +- 2 files changed, 15 insertions(+), 6 deletions(-) diff --git a/src/processor/liveness.py b/src/processor/liveness.py index f6d3980..6001e28 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 @@ -142,11 +146,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() From d11693293fb5681e0a98d804db84fa851f1d36fc Mon Sep 17 00:00:00 2001 From: gregoryfoster Date: Sat, 10 Oct 2026 19:46:20 +0000 Subject: [PATCH 3/5] =?UTF-8?q?#45=20test:=20CR=201=20=E2=80=94=20a=20cut-?= =?UTF-8?q?off=20check-in=20frees=20the=20next=20tick=20once=20its=20post?= =?UTF-8?q?=20returns?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- tests/test_liveness.py | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/tests/test_liveness.py b/tests/test_liveness.py index fd12eae..1eea1e6 100644 --- a/tests/test_liveness.py +++ b/tests/test_liveness.py @@ -148,6 +148,41 @@ 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 ): From e68d1734bff4cb23f8f65f5c0df2ea707181563b Mon Sep 17 00:00:00 2001 From: gregoryfoster Date: Sat, 10 Oct 2026 19:46:21 +0000 Subject: [PATCH 4/5] =?UTF-8?q?#45=20docs:=20CR=202=20=E2=80=94=20=5Fin=5F?= =?UTF-8?q?thread's=20docstring=20names=20the=20in-flight=20marker=20it=20?= =?UTF-8?q?owns?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- src/processor/liveness.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/processor/liveness.py b/src/processor/liveness.py index 6001e28..1920c32 100644 --- a/src/processor/liveness.py +++ b/src/processor/liveness.py @@ -129,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() From 2088d71df86871246bd7e8ce12d3ea5140e178cb Mon Sep 17 00:00:00 2001 From: gregoryfoster Date: Sat, 10 Oct 2026 19:46:21 +0000 Subject: [PATCH 5/5] =?UTF-8?q?#45=20test:=20CR=203=20=E2=80=94=20the=20st?= =?UTF-8?q?art-failure=20test=20counts=20its=20calls?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Opus 5.5 --- tests/test_liveness.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/test_liveness.py b/tests/test_liveness.py index 1eea1e6..1196109 100644 --- a/tests/test_liveness.py +++ b/tests/test_liveness.py @@ -224,11 +224,12 @@ async def test_a_thread_that_fails_to_start_does_not_wedge_the_heartbeat( # 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 - starts = [] + calls = 0 def start(self) -> None: - starts.append(self.name) - if len(starts) == 1: + nonlocal calls + calls += 1 + if calls == 1: raise RuntimeError("can't start new thread") real_start(self)