diff --git a/workers/pg_queue_consumer/supervisor.py b/workers/pg_queue_consumer/supervisor.py index 2474983083..b852a3369d 100644 --- a/workers/pg_queue_consumer/supervisor.py +++ b/workers/pg_queue_consumer/supervisor.py @@ -23,6 +23,17 @@ 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, 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). 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 consumer. ``/health`` cannot say this — the heartbeats are seeded fresh at @@ -72,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(). @@ -107,6 +122,47 @@ 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. + + 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", False, _parse_bool): + return None + stale: float | None = consumer_env("HEALTH_STALE_SECONDS", None, float) + if stale is 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 " + 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. @@ -169,6 +225,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: @@ -264,6 +323,20 @@ 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 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) + 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) @@ -297,7 +370,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 @@ -311,6 +384,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( @@ -406,6 +482,64 @@ 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 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, 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`` + 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. + """ + 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) + 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) — " + "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, + limit, + ) + 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 + + +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: """Fork ``concurrency`` consumer children and supervise them until SIGTERM.""" logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO")) @@ -417,6 +551,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: @@ -452,8 +596,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) - _restart_due_children(fleet, stopping) + _monitor_tick(fleet, stopping, watchdog_after, killed) stopping.wait(_MONITOR_INTERVAL_SECONDS) # responsive to SIGTERM finally: stopping.set() @@ -558,6 +701,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 5b203bd5be..d2e755251f 100644 --- a/workers/queue_backend/pg_queue/README.md +++ b/workers/queue_backend/pg_queue/README.md @@ -36,6 +36,7 @@ integers. | `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` | +| `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_*`) @@ -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 +(`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 e4d88f458d..30a9c1dff6 100644 --- a/workers/tests/test_pg_consumer_supervisor.py +++ b/workers/tests/test_pg_consumer_supervisor.py @@ -18,15 +18,19 @@ _CRASH_LOOP_THRESHOLD, _DEFAULT_SHUTDOWN_GRACE_SECONDS, _MAX_CONCURRENCY, + _MIN_BOOTSTRAP_BUDGET_SECONDS, _MIN_HEALTHY_UPTIME_SECONDS, _Fleet, _child_after_fork, _join_children, + _kill_stale_children, + _monitor_tick, _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 +218,194 @@ 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: opt-in, because it turns HEALTH_STALE into a hard per-task cap — + which docker-compose (no healthcheck) has never enforced. + """ + + 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) + + @pytest.mark.parametrize("off", ["false", "0", "no", "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_enabled_without_health_stale_raises(self, monkeypatch): + # The 60s code default would kill any task longer than a minute. + monkeypatch.setenv(_WATCHDOG, "true") + 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") + 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.setenv(_WATCHDOG, "true") + 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_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_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 ( + 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_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} # 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]) + with patch(f"{_MOD}.os.kill"): + _kill_stale_children(f, stale_after=7260, killed=set()) + 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: def test_dead_child_reaped_and_rescheduled(self): f = _Fleet(1) @@ -354,6 +546,19 @@ 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 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. 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..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() @@ -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