From 18ae189e85ba26128f3c410596cb89da7bc7f196 Mon Sep 17 00:00:00 2001 From: ali Date: Mon, 5 Oct 2026 12:15:27 +0530 Subject: [PATCH 1/4] UN-4223 Kill and re-fork only the hung child in the PG consumer supervisor One child whose task hangs stops publishing heartbeats, which made the fleet's /health 503 and a liveness kill drain every sibling for up to the grace period. The supervisor now SIGKILLs just that child once it is silent past HEALTH_STALE_SECONDS, so its lease lapses, the message redelivers, and the slot is re-forked. Off when HEALTH_STALE_SECONDS is unset; CHILD_WATCHDOG=false disables it. Co-Authored-By: Claude Opus 5.5 --- workers/pg_queue_consumer/supervisor.py | 94 +++++++++++++++ workers/queue_backend/pg_queue/README.md | 8 +- workers/tests/test_pg_consumer_supervisor.py | 114 +++++++++++++++++++ 3 files changed, 213 insertions(+), 3 deletions(-) diff --git a/workers/pg_queue_consumer/supervisor.py b/workers/pg_queue_consumer/supervisor.py index 2474983083..4c4cedd261 100644 --- a/workers/pg_queue_consumer/supervisor.py +++ b/workers/pg_queue_consumer/supervisor.py @@ -23,6 +23,14 @@ a row, never reaching a real poll) forces the probe to 503 so k8s restarts the pod rather than the supervisor masking a wedged fleet with fresh-looking re-forks. +**Child watchdog** (UN-4223): a child whose heartbeat goes stale past +``HEALTH_STALE_SECONDS`` is SIGKILLed and re-forked on its own. Its lease-renewal +thread dies with it, so the claim lapses and the reaper redelivers the message +(the poison cap bounds a task that hangs on every attempt). Without this, one hung +task made the fleet's ``/health`` 503 and a liveness kill drained every sibling; +now only the stuck slot is lost, for one stale window. SIGKILL rather than SIGTERM: +a graceful stop waits for the in-flight task, which is the thing that is hung. + **Readiness** (UN-4136): the same port serves ``/ready``, which answers 200 only once EVERY child has finished its ``import worker`` bootstrap and built its consumer. ``/health`` cannot say this — the heartbeats are seeded fresh at @@ -107,6 +115,40 @@ def concurrency_from_env() -> int: return n +def _parse_bool(raw: str) -> bool: + value = raw.strip().lower() + if value in ("1", "true", "yes", "on"): + return True + if value in ("0", "false", "no", "off"): + return False + raise ValueError(f"expected a boolean, got {raw!r}") + + +def child_watchdog_from_env() -> float | None: + """Heartbeat age (seconds) past which the supervisor SIGKILLs a single child, or + ``None`` when the watchdog is off. + + Reuses ``WORKER_PG_QUEUE_CONSUMER_HEALTH_STALE_SECONDS`` — the threshold at which + the fleet probe already declares a child stale, which every deployment sizes at + or above the VT so a legitimately long task never crosses it. Off when that knob + is unset (its 60s code default would kill any task longer than a minute) or when + ``WORKER_PG_QUEUE_CONSUMER_CHILD_WATCHDOG`` is false. + """ + from queue_backend.pg_queue.consumer import consumer_env + + if not consumer_env("CHILD_WATCHDOG", True, _parse_bool): + return None + stale: float | None = consumer_env("HEALTH_STALE_SECONDS", None, float) + if stale is None: + return None + if not math.isfinite(stale) or stale <= 0: + raise ValueError( + "WORKER_PG_QUEUE_CONSUMER_HEALTH_STALE_SECONDS must be a finite number " + f"> 0, got {stale!r}" + ) + return stale + + def shutdown_grace_from_env() -> float: """Graceful-drain budget (seconds) on shutdown, shared across all children, before SIGKILL. @@ -264,6 +306,15 @@ def all_loaded(self) -> bool: """Readiness verdict source: True once every slot's child has loaded.""" return self.loaded_count() == self._n + def is_loaded(self, slot: int) -> bool: + self._validate(slot) + return bool(self._loaded[slot]) + + def slot_age(self, slot: int) -> float: + """Seconds since ``slot``'s child last polled, per its published heartbeat.""" + self._validate(slot) + return time.time() - self._heartbeats[slot] + def oldest_age(self) -> float: now = time.time() return max((now - hb for hb in self._heartbeats), default=0.0) @@ -406,6 +457,36 @@ def _restart_due_children(fleet: _Fleet, stopping: threading.Event) -> None: _try_fork_child(fleet, slot) +def _kill_stale_children(fleet: _Fleet, stale_after: float, killed: set[int]) -> None: + """SIGKILL every loaded child whose heartbeat is older than ``stale_after``. + + Only a *loaded* child is judged: a re-forked slot keeps its predecessor's old + heartbeat until the new child bootstraps and publishes (record_fork does not + reseed it), so judging it earlier would kill every replacement during import. + ``killed`` holds pids already signalled and not yet reaped, so a child is + killed and logged once; the next ``_reap_dead`` reaps it and schedules the + re-fork like any other exit. + """ + for slot, pid in fleet.alive_items(): + if pid in killed or not fleet.is_loaded(slot): + continue + age = fleet.slot_age(slot) + if age <= stale_after: + continue + logger.error( + "PG-queue consumer: child slot=%s pid=%s has not polled for %.0fs " + "(> %.0fs) — presumed hung; SIGKILL so its message redelivers and the " + "slot is re-forked", + slot, + pid, + age, + stale_after, + ) + with contextlib.suppress(ProcessLookupError): + os.kill(pid, signal.SIGKILL) + killed.add(pid) + + def run_supervised(concurrency: int) -> None: """Fork ``concurrency`` consumer children and supervise them until SIGTERM.""" logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO")) @@ -417,6 +498,16 @@ def run_supervised(concurrency: int) -> None: "children)", grace_seconds, ) + watchdog_after = child_watchdog_from_env() + if watchdog_after is None: + logger.info("PG-queue consumer supervisor: child watchdog off") + else: + logger.info( + "PG-queue consumer supervisor: child watchdog kills a child silent for " + "> %.0fs", + watchdog_after, + ) + killed: set[int] = set() stopping = threading.Event() def _signal_children(sig: int) -> None: @@ -453,7 +544,10 @@ def _on_term(signum: int, _frame: object) -> None: try: while not stopping.is_set(): _reap_dead(fleet, stopping) + killed.intersection_update(pid for _slot, pid in fleet.alive_items()) _restart_due_children(fleet, stopping) + if watchdog_after is not None and not stopping.is_set(): + _kill_stale_children(fleet, watchdog_after, killed) stopping.wait(_MONITOR_INTERVAL_SECONDS) # responsive to SIGTERM finally: stopping.set() diff --git a/workers/queue_backend/pg_queue/README.md b/workers/queue_backend/pg_queue/README.md index 5b203bd5be..c03928bedd 100644 --- a/workers/queue_backend/pg_queue/README.md +++ b/workers/queue_backend/pg_queue/README.md @@ -35,7 +35,8 @@ integers. | `BACKOFF_MAX` | Max empty-queue poll backoff | `2.0` | | `MAX_ATTEMPTS` | Max deliveries before a message is dropped as **poison** | `5` | | `HEALTH_PORT` | Liveness HTTP port (unset → probe disabled) | unset | -| `HEALTH_STALE_SECONDS` | A poll loop idle beyond this is reported unhealthy | `60` | +| `HEALTH_STALE_SECONDS` | A poll loop idle beyond this is reported unhealthy; with `CONCURRENCY > 1` the supervisor also SIGKILLs and re-forks a child silent this long | `60` (watchdog off when unset) | +| `CHILD_WATCHDOG` | Kill switch for that per-child watchdog | `true` | | `WORKER_TYPE` | Which source worker's tasks this consumer registers (bootstrap) | — | ### Reaper (`WORKER_PG_REAPER_*`) @@ -76,8 +77,9 @@ keeps its claim; a **dead** worker's renewal stops, so its `vt` expires in ~`LEA redelivery in **minutes, not hours**. `VT_SECONDS` is retained as the *drain / max-runtime bound* (grace, health-stale), and the lease is clamped to it. Note the lease is **not** a hard cap on runtime — a live-but-hung task keeps renewing forever; -the backstop for that is the liveness probe restarting the pod (process death stops -renewal). The renewal owns its own DB connection (closed on exit) and is best-effort: +the backstop is process death, which stops renewal: the supervisor's child watchdog +kills a child silent past `HEALTH_STALE_SECONDS`, and a single-process consumer +relies on its liveness probe restarting the pod. The renewal owns its own DB connection (closed on exit) and is best-effort: a connection death retries within the `~2×` slack the `LEASE/3` interval leaves before expiry, and escalates to an ERROR log once it keeps failing past `LEASE` (the lease is then genuinely lost and the message may double-run). Because renewal covers only the diff --git a/workers/tests/test_pg_consumer_supervisor.py b/workers/tests/test_pg_consumer_supervisor.py index e4d88f458d..3576857006 100644 --- a/workers/tests/test_pg_consumer_supervisor.py +++ b/workers/tests/test_pg_consumer_supervisor.py @@ -22,11 +22,13 @@ _Fleet, _child_after_fork, _join_children, + _kill_stale_children, _reap_dead, _restart_due_children, _run_child, _try_fork_child, _wait_for_exit, + child_watchdog_from_env, concurrency_from_env, shutdown_grace_from_env, ) @@ -214,6 +216,118 @@ def test_reap_clears_the_slot_loaded_flag(self): assert f.all_loaded() is False +_STALE = "WORKER_PG_QUEUE_CONSUMER_HEALTH_STALE_SECONDS" +_WATCHDOG = "WORKER_PG_QUEUE_CONSUMER_CHILD_WATCHDOG" + + +class TestChildWatchdogFromEnv: + """UN-4223: the watchdog kills at the stale threshold the fleet probe uses, and + only when that threshold was deliberately configured. + """ + + def test_tracks_health_stale_when_set(self, monkeypatch): + monkeypatch.delenv(_WATCHDOG, raising=False) + monkeypatch.setenv(_STALE, "7260") + assert child_watchdog_from_env() == pytest.approx(7260.0) + + def test_off_when_health_stale_unset(self, monkeypatch): + # The 60s code default would kill any task longer than a minute. + monkeypatch.delenv(_WATCHDOG, raising=False) + monkeypatch.delenv(_STALE, raising=False) + assert child_watchdog_from_env() is None + + @pytest.mark.parametrize("off", ["false", "0", "no", "OFF"]) + def test_kill_switch(self, monkeypatch, off): + monkeypatch.setenv(_STALE, "7260") + monkeypatch.setenv(_WATCHDOG, off) + assert child_watchdog_from_env() is None + + def test_explicit_true_is_on(self, monkeypatch): + monkeypatch.setenv(_STALE, "600") + monkeypatch.setenv(_WATCHDOG, "true") + assert child_watchdog_from_env() == pytest.approx(600.0) + + def test_malformed_switch_raises(self, monkeypatch): + monkeypatch.setenv(_STALE, "600") + monkeypatch.setenv(_WATCHDOG, "maybe") + with pytest.raises(ValueError, match=_WATCHDOG): + child_watchdog_from_env() + + @pytest.mark.parametrize("bad", ["0", "-5", "inf", "nan"]) + def test_non_positive_or_non_finite_stale_raises(self, monkeypatch, bad): + monkeypatch.delenv(_WATCHDOG, raising=False) + monkeypatch.setenv(_STALE, bad) + with pytest.raises(ValueError, match=_STALE): + child_watchdog_from_env() + + +class TestKillStaleChildren: + """UN-4223: one hung child is killed on its own instead of a liveness kill + draining every sibling with it. + """ + + @staticmethod + def _fleet(ages: list[float], loaded: list[int]) -> _Fleet: + f = _Fleet(len(ages)) + now = time.time() + for slot, age in enumerate(ages): + f._heartbeats[slot] = now - age + f.loaded[slot] = loaded[slot] + f.record_fork(slot, 100 + slot) + return f + + def test_only_the_stale_child_is_sigkilled(self): + import signal as _signal + + f = self._fleet([5, 9000, 5], loaded=[1, 1, 1]) + killed: set[int] = set() + with patch(f"{_MOD}.os.kill") as kill: + _kill_stale_children(f, stale_after=7260, killed=killed) + kill.assert_called_once_with(101, _signal.SIGKILL) + assert killed == {101} + + def test_child_still_bootstrapping_is_spared(self): + # A re-forked slot keeps its predecessor's old heartbeat until it loads. + f = self._fleet([9000], loaded=[0]) + with patch(f"{_MOD}.os.kill") as kill: + _kill_stale_children(f, stale_after=7260, killed=set()) + kill.assert_not_called() + + def test_age_at_threshold_is_spared(self): + f = self._fleet([60], loaded=[1]) + with ( + patch(f"{_MOD}.time.time", return_value=f._heartbeats[0] + 60), + patch(f"{_MOD}.os.kill") as kill, + ): + _kill_stale_children(f, stale_after=60, killed=set()) + kill.assert_not_called() + + def test_already_signalled_child_is_not_killed_again(self): + f = self._fleet([9000], loaded=[1]) + with patch(f"{_MOD}.os.kill") as kill: + _kill_stale_children(f, stale_after=7260, killed={100}) + kill.assert_not_called() + + def test_vanished_child_is_tolerated(self): + f = self._fleet([9000], loaded=[1]) + killed: set[int] = set() + with patch(f"{_MOD}.os.kill", side_effect=ProcessLookupError): + _kill_stale_children(f, stale_after=7260, killed=killed) + assert killed == {100} + + def test_killed_child_is_reforked_without_counting_a_crash(self): + # A watchdog kill ends a long-running child, so the slot re-forks with no + # crash-loop penalty — exactly like any other healthy exit. + f = self._fleet([9000], loaded=[1]) + f._last_fork[0] -= _MIN_HEALTHY_UPTIME_SECONDS + 100 + with patch(f"{_MOD}.os.kill"): + _kill_stale_children(f, stale_after=7260, killed=set()) + with patch(f"{_MOD}.os.waitpid", return_value=(100, 9)): + _reap_dead(f, threading.Event()) + assert f._consecutive_crashes[0] == 0 + assert f.is_loaded(0) is False # replacement must bootstrap before judged + + class TestReapDead: def test_dead_child_reaped_and_rescheduled(self): f = _Fleet(1) From 5d46ad9e8a524b809bc377aa5f826ab6e1b4dec0 Mon Sep 17 00:00:00 2001 From: ali Date: Mon, 5 Oct 2026 12:34:43 +0530 Subject: [PATCH 2/4] UN-4223 Make the child watchdog opt-in and close the gaps the review found - CHILD_WATCHDOG now defaults off: docker-compose has no healthcheck, so HEALTH_STALE was never enforced there and may sit below a long batch. The chart enables it where a liveness probe already killed at that threshold. - poll_once stamps the heartbeat before each queue read, so a multi-queue consumer's age covers one task, which is what HEALTH_STALE is sized for. - A child that never finishes loading is judged by time since fork. - The child publishes its heartbeat before marking itself loaded. - pg_consumer_child_watchdog_kills_total on /metrics. - The loop iteration is _monitor_tick, now under test. Co-Authored-By: Claude Opus 5.5 --- workers/pg_queue_consumer/supervisor.py | 103 ++++++++++----- workers/queue_backend/pg_queue/README.md | 8 +- workers/queue_backend/pg_queue/consumer.py | 18 +-- workers/queue_backend/pg_queue/metrics.py | 31 +++++ workers/tests/test_pg_consumer_supervisor.py | 125 +++++++++++++++---- workers/tests/test_pg_metrics.py | 12 ++ workers/tests/test_pg_queue_consumer.py | 18 +++ 7 files changed, 248 insertions(+), 67 deletions(-) diff --git a/workers/pg_queue_consumer/supervisor.py b/workers/pg_queue_consumer/supervisor.py index 4c4cedd261..4ceca208c5 100644 --- a/workers/pg_queue_consumer/supervisor.py +++ b/workers/pg_queue_consumer/supervisor.py @@ -23,13 +23,16 @@ a row, never reaching a real poll) forces the probe to 503 so k8s restarts the pod rather than the supervisor masking a wedged fleet with fresh-looking re-forks. -**Child watchdog** (UN-4223): a child whose heartbeat goes stale past -``HEALTH_STALE_SECONDS`` is SIGKILLed and re-forked on its own. Its lease-renewal +**Child watchdog** (UN-4223, opt-in via ``CHILD_WATCHDOG``): a loaded child whose +heartbeat goes stale past ``HEALTH_STALE_SECONDS``, or a child still not loaded +that long after its fork, is SIGKILLed and re-forked on its own. Its lease-renewal thread dies with it, so the claim lapses and the reaper redelivers the message -(the poison cap bounds a task that hangs on every attempt). Without this, one hung -task made the fleet's ``/health`` 503 and a liveness kill drained every sibling; -now only the stuck slot is lost, for one stale window. SIGKILL rather than SIGTERM: -a graceful stop waits for the in-flight task, which is the thing that is hung. +(the poison cap bounds a task that hangs on every attempt). The siblings keep +running, so a pod without a liveness probe on ``/health`` loses one slot for one +stale window instead of the whole fleet. ``/health`` itself still reports 503 from +the stale crossing until the replacement loads (~20s), because ``oldest_age`` +counts the killed slot's heartbeat. SIGKILL rather than SIGTERM: a graceful stop +waits for the in-flight task, which is the thing that is hung. **Readiness** (UN-4136): the same port serves ``/ready``, which answers 200 only once EVERY child has finished its ``import worker`` bootstrap and built its @@ -128,19 +131,26 @@ def child_watchdog_from_env() -> float | None: """Heartbeat age (seconds) past which the supervisor SIGKILLs a single child, or ``None`` when the watchdog is off. - Reuses ``WORKER_PG_QUEUE_CONSUMER_HEALTH_STALE_SECONDS`` — the threshold at which - the fleet probe already declares a child stale, which every deployment sizes at - or above the VT so a legitimately long task never crosses it. Off when that knob - is unset (its 60s code default would kill any task longer than a minute) or when - ``WORKER_PG_QUEUE_CONSUMER_CHILD_WATCHDOG`` is false. + Off unless ``WORKER_PG_QUEUE_CONSUMER_CHILD_WATCHDOG`` is true. Opt-in because + it turns ``HEALTH_STALE_SECONDS`` into a hard per-task wall-clock cap: on k8s a + liveness probe already killed at that threshold (the whole pod), but a + docker-compose install has no healthcheck, so there it has never been enforced + and may sit below a legitimately long batch. Enable it only where + ``HEALTH_STALE_SECONDS`` exceeds the longest legitimate task. Enabling it + without that knob is a misconfiguration (its 60s code default would kill any + task longer than a minute), so it raises. """ from queue_backend.pg_queue.consumer import consumer_env - if not consumer_env("CHILD_WATCHDOG", True, _parse_bool): + if not consumer_env("CHILD_WATCHDOG", False, _parse_bool): return None stale: float | None = consumer_env("HEALTH_STALE_SECONDS", None, float) if stale is None: - return None + raise ValueError( + "WORKER_PG_QUEUE_CONSUMER_CHILD_WATCHDOG is enabled but " + "WORKER_PG_QUEUE_CONSUMER_HEALTH_STALE_SECONDS is unset — set it above " + "the longest legitimate task" + ) if not math.isfinite(stale) or stale <= 0: raise ValueError( "WORKER_PG_QUEUE_CONSUMER_HEALTH_STALE_SECONDS must be a finite number " @@ -211,6 +221,9 @@ def __init__(self, concurrency: int) -> None: self._last_fork: dict[int, float] = {} self._consecutive_crashes: dict[int, int] = {} self._restart_due: dict[int, float] = {} # slot -> monotonic not-before + # Written by the main thread, read by the liveness thread for /metrics; an + # int read is atomic under the GIL. + self.watchdog_kills = 0 @property def concurrency(self) -> int: @@ -310,6 +323,11 @@ def is_loaded(self, slot: int) -> bool: self._validate(slot) return bool(self._loaded[slot]) + def fork_age(self, slot: int) -> float: + """Seconds since ``slot``'s current child was forked.""" + self._validate(slot) + return time.monotonic() - self._last_fork.get(slot, time.monotonic()) + def slot_age(self, slot: int) -> float: """Seconds since ``slot``'s child last polled, per its published heartbeat.""" self._validate(slot) @@ -362,6 +380,9 @@ def _publish_heartbeat() -> None: ) time.sleep(_REPORT_INTERVAL_SECONDS) + # Publish once before ``loaded`` so the watchdog never pairs a loaded slot with + # the previous child's stale heartbeat. + heartbeats[slot] = time.time() - consumer.seconds_since_last_poll() threading.Thread(target=_publish_heartbeat, daemon=True, name=f"pg-hb-{slot}").start() loaded[slot] = 1 logger.info( @@ -458,33 +479,56 @@ def _restart_due_children(fleet: _Fleet, stopping: threading.Event) -> None: def _kill_stale_children(fleet: _Fleet, stale_after: float, killed: set[int]) -> None: - """SIGKILL every loaded child whose heartbeat is older than ``stale_after``. - - Only a *loaded* child is judged: a re-forked slot keeps its predecessor's old - heartbeat until the new child bootstraps and publishes (record_fork does not - reseed it), so judging it earlier would kill every replacement during import. - ``killed`` holds pids already signalled and not yet reaped, so a child is - killed and logged once; the next ``_reap_dead`` reaps it and schedules the - re-fork like any other exit. + """SIGKILL every child silent for longer than ``stale_after``. + + A *loaded* child is judged by its heartbeat. A child not yet loaded is judged + by time since its fork instead: a re-forked slot keeps its predecessor's old + heartbeat until the new child bootstraps (record_fork does not reseed it), so + the heartbeat would kill every replacement during import, while a child that + hangs in ``import worker`` would otherwise never be judged at all. ``killed`` + holds pids already signalled and not yet reaped, so a child is killed and + logged once; the next ``_reap_dead`` reaps it and schedules the re-fork like + any other exit. """ for slot, pid in fleet.alive_items(): - if pid in killed or not fleet.is_loaded(slot): + if pid in killed: continue - age = fleet.slot_age(slot) + loaded = fleet.is_loaded(slot) + age = fleet.slot_age(slot) if loaded else fleet.fork_age(slot) if age <= stale_after: continue logger.error( - "PG-queue consumer: child slot=%s pid=%s has not polled for %.0fs " - "(> %.0fs) — presumed hung; SIGKILL so its message redelivers and the " - "slot is re-forked", + "PG-queue consumer: child slot=%s pid=%s %s for %.0fs (> %.0fs) — " + "presumed hung; SIGKILL so its message redelivers and the slot is " + "re-forked", slot, pid, + "has not polled" if loaded else "has not finished loading", age, stale_after, ) with contextlib.suppress(ProcessLookupError): os.kill(pid, signal.SIGKILL) killed.add(pid) + fleet.watchdog_kills += 1 + + +def _monitor_tick( + fleet: _Fleet, + stopping: threading.Event, + watchdog_after: float | None, + killed: set[int], +) -> None: + """One supervisor iteration: reap exits, re-fork due slots, then the watchdog. + + ``killed`` is trimmed to live pids right after the reap, so a recycled pid is + never mistaken for one already signalled. + """ + _reap_dead(fleet, stopping) + killed.intersection_update(pid for _slot, pid in fleet.alive_items()) + _restart_due_children(fleet, stopping) + if watchdog_after is not None and not stopping.is_set(): + _kill_stale_children(fleet, watchdog_after, killed) def run_supervised(concurrency: int) -> None: @@ -543,11 +587,7 @@ def _on_term(signum: int, _frame: object) -> None: health = _maybe_start_supervisor_health(fleet) try: while not stopping.is_set(): - _reap_dead(fleet, stopping) - killed.intersection_update(pid for _slot, pid in fleet.alive_items()) - _restart_due_children(fleet, stopping) - if watchdog_after is not None and not stopping.is_set(): - _kill_stale_children(fleet, watchdog_after, killed) + _monitor_tick(fleet, stopping, watchdog_after, killed) stopping.wait(_MONITOR_INTERVAL_SECONDS) # responsive to SIGTERM finally: stopping.set() @@ -652,6 +692,7 @@ def _extra_status() -> dict[str, object]: freshness_fn=fleet.freshness, alive_children_fn=lambda: float(fleet.alive_count()), concurrency_fn=lambda: float(fleet.concurrency), + watchdog_kills_fn=lambda: float(fleet.watchdog_kills), ) server = LivenessServer( freshness_fn=fleet.freshness, diff --git a/workers/queue_backend/pg_queue/README.md b/workers/queue_backend/pg_queue/README.md index c03928bedd..d2e755251f 100644 --- a/workers/queue_backend/pg_queue/README.md +++ b/workers/queue_backend/pg_queue/README.md @@ -35,8 +35,8 @@ integers. | `BACKOFF_MAX` | Max empty-queue poll backoff | `2.0` | | `MAX_ATTEMPTS` | Max deliveries before a message is dropped as **poison** | `5` | | `HEALTH_PORT` | Liveness HTTP port (unset → probe disabled) | unset | -| `HEALTH_STALE_SECONDS` | A poll loop idle beyond this is reported unhealthy; with `CONCURRENCY > 1` the supervisor also SIGKILLs and re-forks a child silent this long | `60` (watchdog off when unset) | -| `CHILD_WATCHDOG` | Kill switch for that per-child watchdog | `true` | +| `HEALTH_STALE_SECONDS` | A poll loop idle beyond this is reported unhealthy | `60` | +| `CHILD_WATCHDOG` | With `CONCURRENCY > 1`, SIGKILL and re-fork a child silent past `HEALTH_STALE_SECONDS` (requires it set above the longest legitimate task) | `false` | | `WORKER_TYPE` | Which source worker's tasks this consumer registers (bootstrap) | — | ### Reaper (`WORKER_PG_REAPER_*`) @@ -78,8 +78,8 @@ redelivery in **minutes, not hours**. `VT_SECONDS` is retained as the *drain / max-runtime bound* (grace, health-stale), and the lease is clamped to it. Note the lease is **not** a hard cap on runtime — a live-but-hung task keeps renewing forever; the backstop is process death, which stops renewal: the supervisor's child watchdog -kills a child silent past `HEALTH_STALE_SECONDS`, and a single-process consumer -relies on its liveness probe restarting the pod. The renewal owns its own DB connection (closed on exit) and is best-effort: +(`CHILD_WATCHDOG`) kills a child silent past `HEALTH_STALE_SECONDS`; otherwise a +liveness probe restarting the pod, where one exists. The renewal owns its own DB connection (closed on exit) and is best-effort: a connection death retries within the `~2×` slack the `LEASE/3` interval leaves before expiry, and escalates to an ERROR log once it keeps failing past `LEASE` (the lease is then genuinely lost and the message may double-run). Because renewal covers only the diff --git a/workers/queue_backend/pg_queue/consumer.py b/workers/queue_backend/pg_queue/consumer.py index 057c8d85e8..c31b427957 100644 --- a/workers/queue_backend/pg_queue/consumer.py +++ b/workers/queue_backend/pg_queue/consumer.py @@ -96,10 +96,11 @@ # response, and the row is TTL'd. _TASK_STATUS_RETENTION_SECONDS = 86400 # Liveness: a poll loop that hasn't cycled in this many seconds is reported -# unhealthy. The heartbeat is stamped at the top of each poll_once and frozen -# during task execution, so this threshold doubles as an UPPER BOUND on a single -# task's wall-clock: a task running longer than it trips the probe → pod restart -# → the in-flight task is killed and (at-least-once) redelivered. 60s suits the +# unhealthy. The heartbeat is stamped before each queue read and frozen during +# task execution, so this threshold doubles as an UPPER BOUND on a single task's +# wall-clock: a task running longer than it trips the probe → pod restart (or, +# under the prefork supervisor with CHILD_WATCHDOG on, a SIGKILL of just that +# child) → the in-flight task is killed and (at-least-once) redelivered. 60s suits the # current sub-second leaf (send_webhook_notification); for longer-running tasks, # raise WORKER_PG_QUEUE_CONSUMER_HEALTH_STALE_SECONDS above # max(batch_size x worst_case_task_seconds, backoff_max). @@ -365,8 +366,8 @@ def __init__( self._result_backend: PgResultBackend | None = None # Heartbeat for the liveness probe: monotonic timestamp of the most # recent poll attempt. Seeded at construction so a just-started consumer - # reads healthy. Updated at the TOP of poll_once, so a loop wedged on a - # long-running task (poll_once not returning) goes stale and is caught — + # reads healthy. Updated before EACH queue read in poll_once, so a loop + # wedged on a long-running task goes stale and is caught — # something pgrep-based --status and the launch-time check cannot see. self._last_poll_monotonic = time.monotonic() @@ -379,9 +380,12 @@ def poll_once(self) -> int: cycle still counts (so run() doesn't take the empty-queue backoff path after a partial failure). """ - self._last_poll_monotonic = time.monotonic() total = 0 for queue_name in self.queue_names: + # Per queue, not per cycle: a cycle runs one task from EACH queue, and + # HEALTH_STALE is sized for one task, so a per-cycle stamp would let + # two legitimate long tasks add up to a false "stale". + self._last_poll_monotonic = time.monotonic() try: messages = self._client.read( queue_name, vt_seconds=self.lease_seconds, qty=self.batch_size diff --git a/workers/queue_backend/pg_queue/metrics.py b/workers/queue_backend/pg_queue/metrics.py index d34c7d885d..e3a6e58488 100644 --- a/workers/queue_backend/pg_queue/metrics.py +++ b/workers/queue_backend/pg_queue/metrics.py @@ -87,6 +87,7 @@ def __init__( freshness_fn: Callable[[], float], alive_children_fn: Callable[[], float] | None = None, concurrency_fn: Callable[[], float] | None = None, + watchdog_kills_fn: Callable[[], float] | None = None, ) -> None: super().__init__() self._function_gauge( @@ -107,6 +108,36 @@ def __init__( "Configured child-process concurrency of the supervisor fleet", concurrency_fn, ) + if watchdog_kills_fn is not None: + self.registry.register( + _FunctionCounter( + "pg_consumer_child_watchdog_kills", + "Children the supervisor SIGKILLed for going silent past " + "HEALTH_STALE_SECONDS", + watchdog_kills_fn, + ) + ) + + +class _FunctionCounter: + """A counter read from a callable at scrape time (``Counter`` has no + ``set_function``). Exposed as ``_total``. + """ + + def __init__(self, name: str, doc: str, fn: Callable[[], float]) -> None: + self._name = name + self._doc = doc + self._fn = fn + + def describe(self) -> Iterable[Metric]: + from prometheus_client.core import CounterMetricFamily + + return [CounterMetricFamily(self._name, self._doc)] + + def collect(self) -> Iterable[Metric]: + from prometheus_client.core import CounterMetricFamily + + yield CounterMetricFamily(self._name, self._doc, value=self._fn()) @dataclass(frozen=True) diff --git a/workers/tests/test_pg_consumer_supervisor.py b/workers/tests/test_pg_consumer_supervisor.py index 3576857006..20ed73f829 100644 --- a/workers/tests/test_pg_consumer_supervisor.py +++ b/workers/tests/test_pg_consumer_supervisor.py @@ -23,6 +23,7 @@ _child_after_fork, _join_children, _kill_stale_children, + _monitor_tick, _reap_dead, _restart_due_children, _run_child, @@ -221,31 +222,33 @@ def test_reap_clears_the_slot_loaded_flag(self): class TestChildWatchdogFromEnv: - """UN-4223: the watchdog kills at the stale threshold the fleet probe uses, and - only when that threshold was deliberately configured. + """UN-4223: opt-in, because it turns HEALTH_STALE into a hard per-task cap — + which docker-compose (no healthcheck) has never enforced. """ - def test_tracks_health_stale_when_set(self, monkeypatch): + def test_off_by_default_even_with_health_stale(self, monkeypatch): monkeypatch.delenv(_WATCHDOG, raising=False) + monkeypatch.setenv(_STALE, "3720") + assert child_watchdog_from_env() is None + + @pytest.mark.parametrize("on", ["true", "1", "yes", "ON"]) + def test_enabled_tracks_health_stale(self, monkeypatch, on): + monkeypatch.setenv(_WATCHDOG, on) monkeypatch.setenv(_STALE, "7260") assert child_watchdog_from_env() == pytest.approx(7260.0) - def test_off_when_health_stale_unset(self, monkeypatch): - # The 60s code default would kill any task longer than a minute. - monkeypatch.delenv(_WATCHDOG, raising=False) - monkeypatch.delenv(_STALE, raising=False) - assert child_watchdog_from_env() is None - @pytest.mark.parametrize("off", ["false", "0", "no", "OFF"]) - def test_kill_switch(self, monkeypatch, off): + def test_explicitly_off(self, monkeypatch, off): monkeypatch.setenv(_STALE, "7260") monkeypatch.setenv(_WATCHDOG, off) assert child_watchdog_from_env() is None - def test_explicit_true_is_on(self, monkeypatch): - monkeypatch.setenv(_STALE, "600") + def test_enabled_without_health_stale_raises(self, monkeypatch): + # The 60s code default would kill any task longer than a minute. monkeypatch.setenv(_WATCHDOG, "true") - assert child_watchdog_from_env() == pytest.approx(600.0) + monkeypatch.delenv(_STALE, raising=False) + with pytest.raises(ValueError, match=_STALE): + child_watchdog_from_env() def test_malformed_switch_raises(self, monkeypatch): monkeypatch.setenv(_STALE, "600") @@ -255,7 +258,7 @@ def test_malformed_switch_raises(self, monkeypatch): @pytest.mark.parametrize("bad", ["0", "-5", "inf", "nan"]) def test_non_positive_or_non_finite_stale_raises(self, monkeypatch, bad): - monkeypatch.delenv(_WATCHDOG, raising=False) + monkeypatch.setenv(_WATCHDOG, "true") monkeypatch.setenv(_STALE, bad) with pytest.raises(ValueError, match=_STALE): child_watchdog_from_env() @@ -286,13 +289,23 @@ def test_only_the_stale_child_is_sigkilled(self): kill.assert_called_once_with(101, _signal.SIGKILL) assert killed == {101} - def test_child_still_bootstrapping_is_spared(self): - # A re-forked slot keeps its predecessor's old heartbeat until it loads. + def test_bootstrapping_child_is_judged_by_fork_age_not_heartbeat(self): + # A re-forked slot keeps its predecessor's old heartbeat until it loads, + # so a recent fork is spared however stale that heartbeat looks. f = self._fleet([9000], loaded=[0]) with patch(f"{_MOD}.os.kill") as kill: _kill_stale_children(f, stale_after=7260, killed=set()) kill.assert_not_called() + def test_child_hung_in_bootstrap_is_killed(self): + import signal as _signal + + f = self._fleet([0], loaded=[0]) + f._last_fork[0] -= 7300 # forked long ago, never loaded + with patch(f"{_MOD}.os.kill") as kill: + _kill_stale_children(f, stale_after=7260, killed=set()) + kill.assert_called_once_with(100, _signal.SIGKILL) + def test_age_at_threshold_is_spared(self): f = self._fleet([60], loaded=[1]) with ( @@ -315,17 +328,67 @@ def test_vanished_child_is_tolerated(self): _kill_stale_children(f, stale_after=7260, killed=killed) assert killed == {100} - def test_killed_child_is_reforked_without_counting_a_crash(self): - # A watchdog kill ends a long-running child, so the slot re-forks with no - # crash-loop penalty — exactly like any other healthy exit. - f = self._fleet([9000], loaded=[1]) - f._last_fork[0] -= _MIN_HEALTHY_UPTIME_SECONDS + 100 + def test_kill_is_counted_for_metrics(self): + f = self._fleet([9000, 9000], loaded=[1, 1]) with patch(f"{_MOD}.os.kill"): _kill_stale_children(f, stale_after=7260, killed=set()) - with patch(f"{_MOD}.os.waitpid", return_value=(100, 9)): - _reap_dead(f, threading.Event()) - assert f._consecutive_crashes[0] == 0 - assert f.is_loaded(0) is False # replacement must bootstrap before judged + assert f.watchdog_kills == 2 + + +class TestMonitorTick: + """The loop wiring: reap, trim ``killed``, re-fork, then the watchdog.""" + + @staticmethod + def _stale_fleet() -> _Fleet: + f = _Fleet(1) + f._heartbeats[0] = time.time() - 9000 + f.loaded[0] = 1 + f.record_fork(0, 100) + return f + + def test_watchdog_runs_when_enabled(self): + f = self._stale_fleet() + with ( + patch(f"{_MOD}.os.waitpid", return_value=(0, 0)), + patch(f"{_MOD}.os.kill") as kill, + ): + _monitor_tick(f, threading.Event(), 7260, set()) + kill.assert_called_once() + + def test_watchdog_skipped_when_disabled(self): + f = self._stale_fleet() + with ( + patch(f"{_MOD}.os.waitpid", return_value=(0, 0)), + patch(f"{_MOD}.os.kill") as kill, + ): + _monitor_tick(f, threading.Event(), None, set()) + kill.assert_not_called() + + def test_watchdog_skipped_while_stopping(self): + # Draining children must not be SIGKILLed during shutdown. + f = self._stale_fleet() + stopping = threading.Event() + stopping.set() + with ( + patch(f"{_MOD}.os.waitpid", return_value=(0, 0)), + patch(f"{_MOD}.os.kill") as kill, + ): + _monitor_tick(f, stopping, 7260, set()) + kill.assert_not_called() + + def test_reaped_pid_leaves_killed_and_slot_is_scheduled(self): + f = self._stale_fleet() + f._last_fork[0] -= _MIN_HEALTHY_UPTIME_SECONDS + 100 + killed = {100} + with ( + patch(f"{_MOD}.os.waitpid", return_value=(100, 9)), + patch(f"{_MOD}.os.kill"), + patch(f"{_MOD}._try_fork_child"), + ): + _monitor_tick(f, threading.Event(), 7260, killed) + assert killed == set() # a recycled pid can't inherit the mark + assert 0 in f._restart_due # re-fork scheduled + assert f._consecutive_crashes[0] == 0 # a long-lived child, not a crash class TestReapDead: @@ -468,6 +531,18 @@ def test_marks_its_slot_loaded_before_polling(self): self._run_slot_1(f, lambda: consumer) assert seen == [[0, 1]] # own slot only, and set before run() + def test_heartbeat_is_fresh_by_the_time_the_slot_is_loaded(self): + # UN-4223: the publisher thread is patched out here, so only the + # synchronous publish can refresh the predecessor's stale heartbeat. + f = _Fleet(2) + f._heartbeats[1] = time.time() - 9000 + consumer = MagicMock() + consumer.seconds_since_last_poll.return_value = 0.0 + ages: list[float] = [] + consumer.run.side_effect = lambda: ages.append(f.slot_age(1)) + self._run_slot_1(f, lambda: consumer) + assert ages and ages[0] < 5 + def test_not_marked_loaded_when_the_build_fails(self): # A child that cannot finish its bootstrap must never count as loaded. f = _Fleet(2) diff --git a/workers/tests/test_pg_metrics.py b/workers/tests/test_pg_metrics.py index e21bc2b67f..9ca7e6314c 100644 --- a/workers/tests/test_pg_metrics.py +++ b/workers/tests/test_pg_metrics.py @@ -598,3 +598,15 @@ def test_supervisor_fleet_metrics_served(self, monkeypatch): assert b"pg_consumer_configured_concurrency 2.0" in body finally: server.stop() + + +def test_consumer_metrics_exports_watchdog_kills_as_a_counter(): + # UN-4223: a slot killed and re-forked leaves alive_children unchanged, so + # the kill count is the only scrapeable trace of the watchdog acting. + from queue_backend.pg_queue.metrics import ConsumerMetrics + + body = ConsumerMetrics( + freshness_fn=lambda: 0.0, watchdog_kills_fn=lambda: 3.0 + ).render() + assert b"# TYPE pg_consumer_child_watchdog_kills_total counter" in body + assert b"pg_consumer_child_watchdog_kills_total 3.0" in body diff --git a/workers/tests/test_pg_queue_consumer.py b/workers/tests/test_pg_queue_consumer.py index f4105cc1ef..2daac385c8 100644 --- a/workers/tests/test_pg_queue_consumer.py +++ b/workers/tests/test_pg_queue_consumer.py @@ -681,6 +681,24 @@ def test_heartbeat_stamped_before_read(self): consumer.poll_once() assert seen["during"] > before # refreshed BEFORE read ran, not after + def test_heartbeat_restamped_before_each_queue_read(self): + # UN-4223: a cycle runs one task per queue and HEALTH_STALE is sized for + # one task, so each queue's read must start from a fresh stamp — a single + # per-cycle stamp would add two long tasks into a false "stale". + client = MagicMock() + consumer = PgQueueConsumer(["q1", "q2"], client=client) + seen: list[float] = [] + + def _read(*_a, **_k): + seen.append(consumer._last_poll_monotonic) + consumer._last_poll_monotonic -= 5000 # the first queue's long task + return [] + + client.read.side_effect = _read + consumer.poll_once() + assert len(seen) == 2 + assert seen[1] > seen[0] - 1 # re-stamped, not inherited from queue 1 + def test_health_server_disabled_without_port(self): # No port configured → no server bound (opt-in). from queue_backend.pg_queue.consumer import _maybe_start_health_server From 7ba4d0e47963b61e2593138347fdff70da9b5d8f Mon Sep 17 00:00:00 2001 From: ali Date: Mon, 5 Oct 2026 12:46:51 +0530 Subject: [PATCH 3/4] UN-4223 Correct two comments that still describe a per-cycle heartbeat stamp Co-Authored-By: Claude Opus 5.5 --- workers/pg_queue_consumer/supervisor.py | 2 +- workers/tests/test_pg_queue_consumer.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/workers/pg_queue_consumer/supervisor.py b/workers/pg_queue_consumer/supervisor.py index 4ceca208c5..036e2b9e68 100644 --- a/workers/pg_queue_consumer/supervisor.py +++ b/workers/pg_queue_consumer/supervisor.py @@ -366,7 +366,7 @@ def _run_child(slot: int, heartbeats, loaded) -> None: # noqa: ANN001 (ctypes a def _publish_heartbeat() -> None: # last-poll wall-time = now − (seconds since last poll). Frozen while a - # task runs (the consumer stamps its heartbeat at the top of poll_once), + # task runs (the consumer stamps its heartbeat before each queue read), # so a child stuck on a too-long task goes stale exactly as the single # consumer does. Guarded so a transient error (e.g. teardown during # shutdown) logs loudly and the loop continues instead of dying silently diff --git a/workers/tests/test_pg_queue_consumer.py b/workers/tests/test_pg_queue_consumer.py index 2daac385c8..80555b3a80 100644 --- a/workers/tests/test_pg_queue_consumer.py +++ b/workers/tests/test_pg_queue_consumer.py @@ -666,8 +666,8 @@ def test_poll_once_refreshes_heartbeat(self): assert consumer.seconds_since_last_poll() < 1.0 def test_heartbeat_stamped_before_read(self): - # Pins the headline design: the stamp lands at the TOP of poll_once - # (before read), so a task running longer than the threshold still trips + # Pins the headline design: the stamp lands before each queue read (not + # after it), so a task running longer than the threshold still trips # the probe. A bottom-of-poll stamp would pass test_poll_once_refreshes # but fail here. client = MagicMock() From b611adfa05ff76b7f1d6fc6bc8cc661fa74d186c Mon Sep 17 00:00:00 2001 From: ali Date: Mon, 5 Oct 2026 12:55:17 +0530 Subject: [PATCH 4/4] UN-4223 Give a slow bootstrap its own floor and count only real watchdog kills Review comments on #2311: - An unloaded child is judged against max(HEALTH_STALE, 600s): HEALTH_STALE is sized for task runtime and can be 180s, while an import under a CPU cap can take minutes. - A child that exits before the SIGKILL lands no longer counts as a kill. - Split a composite test assertion (Sonar S9073). Co-Authored-By: Claude Opus 5.5 --- workers/pg_queue_consumer/supervisor.py | 19 ++++++++++++----- workers/tests/test_pg_consumer_supervisor.py | 22 +++++++++++++++++--- 2 files changed, 33 insertions(+), 8 deletions(-) diff --git a/workers/pg_queue_consumer/supervisor.py b/workers/pg_queue_consumer/supervisor.py index 036e2b9e68..b852a3369d 100644 --- a/workers/pg_queue_consumer/supervisor.py +++ b/workers/pg_queue_consumer/supervisor.py @@ -83,6 +83,10 @@ _MIN_HEALTHY_UPTIME_SECONDS = 10.0 # Consecutive immediate crashes after which the fleet probe is forced unhealthy. _CRASH_LOOP_THRESHOLD = 3 +# Least time a child gets to finish ``import worker`` before the watchdog treats it +# as hung. HEALTH_STALE is sized for task runtime and can be short (ide-callback: +# 180s), while a bootstrap under a CPU cap can take minutes. +_MIN_BOOTSTRAP_BUDGET_SECONDS = 600.0 # Fallback graceful-drain budget (s, shared across all children) on shutdown, used # only when neither an explicit override nor the consumer VT is set — see # shutdown_grace_from_env(). @@ -482,7 +486,8 @@ def _kill_stale_children(fleet: _Fleet, stale_after: float, killed: set[int]) -> """SIGKILL every child silent for longer than ``stale_after``. A *loaded* child is judged by its heartbeat. A child not yet loaded is judged - by time since its fork instead: a re-forked slot keeps its predecessor's old + by time since its fork, against the larger of ``stale_after`` and + ``_MIN_BOOTSTRAP_BUDGET_SECONDS``: a re-forked slot keeps its predecessor's old heartbeat until the new child bootstraps (record_fork does not reseed it), so the heartbeat would kill every replacement during import, while a child that hangs in ``import worker`` would otherwise never be judged at all. ``killed`` @@ -490,12 +495,14 @@ def _kill_stale_children(fleet: _Fleet, stale_after: float, killed: set[int]) -> logged once; the next ``_reap_dead`` reaps it and schedules the re-fork like any other exit. """ + bootstrap_budget = max(stale_after, _MIN_BOOTSTRAP_BUDGET_SECONDS) for slot, pid in fleet.alive_items(): if pid in killed: continue loaded = fleet.is_loaded(slot) age = fleet.slot_age(slot) if loaded else fleet.fork_age(slot) - if age <= stale_after: + limit = stale_after if loaded else bootstrap_budget + if age <= limit: continue logger.error( "PG-queue consumer: child slot=%s pid=%s %s for %.0fs (> %.0fs) — " @@ -505,11 +512,13 @@ def _kill_stale_children(fleet: _Fleet, stale_after: float, killed: set[int]) -> pid, "has not polled" if loaded else "has not finished loading", age, - stale_after, + limit, ) - with contextlib.suppress(ProcessLookupError): - os.kill(pid, signal.SIGKILL) killed.add(pid) + try: + os.kill(pid, signal.SIGKILL) + except ProcessLookupError: + continue # exited on its own first; the reap handles it, not a kill fleet.watchdog_kills += 1 diff --git a/workers/tests/test_pg_consumer_supervisor.py b/workers/tests/test_pg_consumer_supervisor.py index 20ed73f829..30a9c1dff6 100644 --- a/workers/tests/test_pg_consumer_supervisor.py +++ b/workers/tests/test_pg_consumer_supervisor.py @@ -18,6 +18,7 @@ _CRASH_LOOP_THRESHOLD, _DEFAULT_SHUTDOWN_GRACE_SECONDS, _MAX_CONCURRENCY, + _MIN_BOOTSTRAP_BUDGET_SECONDS, _MIN_HEALTHY_UPTIME_SECONDS, _Fleet, _child_after_fork, @@ -306,6 +307,19 @@ def test_child_hung_in_bootstrap_is_killed(self): _kill_stale_children(f, stale_after=7260, killed=set()) kill.assert_called_once_with(100, _signal.SIGKILL) + def test_slow_bootstrap_gets_at_least_the_bootstrap_floor(self): + # A short task threshold (ide-callback: 180s) must not cut off an import + # that is merely slow, e.g. under a CPU cap. + f = self._fleet([0], loaded=[0]) + f._last_fork[0] -= _MIN_BOOTSTRAP_BUDGET_SECONDS - 60 + with patch(f"{_MOD}.os.kill") as kill: + _kill_stale_children(f, stale_after=180, killed=set()) + kill.assert_not_called() + f._last_fork[0] -= 120 # now past the floor + with patch(f"{_MOD}.os.kill") as kill: + _kill_stale_children(f, stale_after=180, killed=set()) + kill.assert_called_once() + def test_age_at_threshold_is_spared(self): f = self._fleet([60], loaded=[1]) with ( @@ -321,12 +335,13 @@ def test_already_signalled_child_is_not_killed_again(self): _kill_stale_children(f, stale_after=7260, killed={100}) kill.assert_not_called() - def test_vanished_child_is_tolerated(self): + def test_vanished_child_is_tolerated_and_not_counted(self): f = self._fleet([9000], loaded=[1]) killed: set[int] = set() with patch(f"{_MOD}.os.kill", side_effect=ProcessLookupError): _kill_stale_children(f, stale_after=7260, killed=killed) - assert killed == {100} + assert killed == {100} # not signalled again before the reap + assert f.watchdog_kills == 0 # the metric counts kills that happened def test_kill_is_counted_for_metrics(self): f = self._fleet([9000, 9000], loaded=[1, 1]) @@ -541,7 +556,8 @@ def test_heartbeat_is_fresh_by_the_time_the_slot_is_loaded(self): ages: list[float] = [] consumer.run.side_effect = lambda: ages.append(f.slot_age(1)) self._run_slot_1(f, lambda: consumer) - assert ages and ages[0] < 5 + assert len(ages) == 1 + assert ages[0] < 5 def test_not_marked_loaded_when_the_build_fails(self): # A child that cannot finish its bootstrap must never count as loaded.