diff --git a/modelq/app/base.py b/modelq/app/base.py index 3f96ff1..b786361 100644 --- a/modelq/app/base.py +++ b/modelq/app/base.py @@ -1,5 +1,6 @@ import redis import json +import contextlib import functools import threading import time @@ -13,6 +14,28 @@ import requests # For sending error payloads to a webhook +def _tcp_keepalive_options() -> Dict[int, int]: + """TCP keepalive tuning, restricted to what the running platform supports. + + Linux exposes all three knobs. macOS spells the idle timer TCP_KEEPALIVE and + has no TCP_KEEPCNT, so we probe each name instead of assuming. An empty dict + is a valid result: keepalive still runs, just with the OS defaults (2h idle + on Linux), which is far too slow to be useful but never harmful. + """ + wanted = ( + ("TCP_KEEPIDLE", 60), # start probing after 60s idle + ("TCP_KEEPALIVE", 60), # macOS name for the same timer + ("TCP_KEEPINTVL", 10), # probe every 10s + ("TCP_KEEPCNT", 3), # give up after 3 failed probes (~90s total) + ) + options: Dict[int, int] = {} + for name, value in wanted: + opt = getattr(socket, name, None) + if opt is not None: + options[opt] = value + return options + + def get_system_info() -> Dict[str, Any]: """ Collect system information including CPU, RAM, and GPU details. @@ -105,6 +128,16 @@ class ModelQ: TASK_TTL = 86400 # 24 hours TTL for all tasks DEFAULT_STREAM_TIMEOUT = 300 # 5 minutes default stream timeout + # --- connection liveness ------------------------------------------------- + # BLPOP_TIMEOUT must stay comfortably below SOCKET_TIMEOUT: redis-py applies + # the socket read timeout to blocking commands too, so a BLPOP that blocks + # longer than the socket allows would raise on every idle poll. + BLPOP_TIMEOUT = 15 # seconds: bound each queue read so a dead socket surfaces + SOCKET_TIMEOUT = 60 # seconds: hard ceiling on any single read + SOCKET_CONNECT_TIMEOUT = 10 # seconds: fail fast when the host is unreachable + HEALTH_CHECK_INTERVAL = 30 # seconds: PING a pooled connection idle this long + BACKGROUND_LOOP_BACKOFF = 5 # seconds: pause before retrying a crashed loop body + def __init__( self, host: str = "localhost", @@ -252,6 +285,20 @@ def _connect_to_redis( # ssl=ssl, # ssl_cert_reqs=ssl_cert_reqs, max_connections=max_connections, + # --- half-open connection detection ------------------------------- + # Without these a silently dropped TCP connection is invisible to us. + # A reader parked in BLPOP transmits nothing, so the kernel never + # retransmits, never gives up, and never raises: the worker waits + # forever on a socket the server has already forgotten. Keepalive + # makes the kernel probe an idle connection; health_check_interval + # makes redis-py PING a pooled connection that has been idle before + # handing it out; socket_timeout bounds every read. + socket_keepalive=True, + socket_keepalive_options=_tcp_keepalive_options(), + socket_timeout=self.SOCKET_TIMEOUT, + socket_connect_timeout=self.SOCKET_CONNECT_TIMEOUT, + health_check_interval=self.HEALTH_CHECK_INTERVAL, + retry_on_timeout=True, ) return redis.Redis(connection_pool=pool) @@ -546,16 +593,17 @@ def requeue_delayed_tasks(self): then moves them into 'ml_tasks' for immediate processing. """ while True: - now = time.time() - ready_tasks = self.redis_client.zrangebyscore("delayed_tasks", 0, now) - for task_json in ready_tasks: - self.redis_client.zrem("delayed_tasks", task_json) - self.redis_client.lpush("ml_tasks", task_json) - try: - _td = json.loads(task_json) - self.redis_client.zadd("queued_requests", {_td["task_id"]: _td.get("queued_at", now)}) - except Exception: - pass + with self._guarded_iteration("requeue_delayed_tasks"): + now = time.time() + ready_tasks = self.redis_client.zrangebyscore("delayed_tasks", 0, now) + for task_json in ready_tasks: + self.redis_client.zrem("delayed_tasks", task_json) + self.redis_client.lpush("ml_tasks", task_json) + try: + _td = json.loads(task_json) + self.redis_client.zadd("queued_requests", {_td["task_id"]: _td.get("queued_at", now)}) + except Exception: + pass time.sleep(1) def requeue_inprogress_tasks(self): @@ -712,7 +760,14 @@ def worker_loop(worker_id): continue self.update_server_status(f"worker_{worker_id}: idle") - task_data = self.redis_client.blpop("ml_tasks") # blocks until a task is available + # Bounded block. An unbounded BLPOP is a pure read-wait: the + # worker transmits nothing, so a silently dropped connection + # is never retransmitted, never times out, and never raises + # — the thread parks forever on a socket Redis has already + # forgotten, while the queue behind it grows unattended. + # Timing out and looping forces the read to complete, which + # is what lets keepalive/health-check reap the dead socket. + task_data = self.redis_client.blpop("ml_tasks", timeout=self.BLPOP_TIMEOUT) if not task_data: continue @@ -822,12 +877,34 @@ def worker_loop(worker_id): f"Registered tasks: {task_names}" ) + @contextlib.contextmanager + def _guarded_iteration(self, loop_name: str): + """Swallow and log any exception raised by one background-loop iteration. + + A bare `while True:` in a thread is one unhandled exception away from + being gone for good — Python unwinds the thread and the process carries + on looking healthy, so no orchestrator ever restarts it. On 2026-08-15 a + transient Redis timeout took out the pruning thread on every replica of + several endpoints this way. Loops must survive a blip; only the process + exiting should end them. + """ + try: + yield + except Exception as exc: # noqa: BLE001 - a loop must outlive any body error + logger.error( + f"Background loop '{loop_name}' iteration failed " + f"({exc.__class__.__name__}: {exc}). Continuing.", + exc_info=True, + ) + time.sleep(self.BACKGROUND_LOOP_BACKOFF) + def _heartbeat_loop(self): """ Continuously updates the heartbeat for this server. """ while True: - self.heartbeat() + with self._guarded_iteration("heartbeat"): + self.heartbeat() time.sleep(self.HEARTBEAT_INTERVAL) def _pruning_loop(self): @@ -835,9 +912,10 @@ def _pruning_loop(self): Continuously prunes servers that have not updated their heartbeat in a while. """ while True: - self.prune_inactive_servers(timeout_seconds=self.PRUNE_TIMEOUT) - self.requeue_stuck_processing_tasks(threshold=180) - self.prune_old_task_results(older_than_seconds=self.TASK_RESULT_RETENTION) + with self._guarded_iteration("pruning"): + self.prune_inactive_servers(timeout_seconds=self.PRUNE_TIMEOUT) + self.requeue_stuck_processing_tasks(threshold=180) + self.prune_old_task_results(older_than_seconds=self.TASK_RESULT_RETENTION) time.sleep(self.PRUNE_CHECK_INTERVAL) def check_middleware(self, middleware_event: str,task: Optional[Task] = None, error: Optional[Exception] = None): diff --git a/modelq/app/redis_retry.py b/modelq/app/redis_retry.py index 9bfdba1..6fa427b 100644 --- a/modelq/app/redis_retry.py +++ b/modelq/app/redis_retry.py @@ -1,3 +1,4 @@ +import random import time import redis from redis.exceptions import ConnectionError, TimeoutError @@ -10,12 +11,29 @@ class _RedisWithRetry: Any callable attribute (e.g. get, set, blpop, xadd …) is executed with a retry loop that catches *ConnectionError* and *TimeoutError* from redis‑py - and re‑issues the call after a fixed delay. Retries indefinitely until - the connection succeeds. + and re‑issues the call after a short, jittered delay. Retries indefinitely + until the connection succeeds. """ RETRYABLE = (ConnectionError, TimeoutError) - RETRY_DELAY = 30 # seconds between retry attempts + # Recovery from a dead connection costs socket_timeout (detect) + this + # (wait), so this value is the tail of every stall. Kept short. + RETRY_DELAY = 15 # seconds between retry attempts + # Connection loss is a SHARED event, not an independent one: on 2026-08-15 + # every worker on a node logged the same failure in the same second, + # because one upstream path change broke all their connections at once. A + # fixed delay would march that whole herd back into Redis in lockstep. + # Jitter spreads the reconnect over a window instead. Set to 0 for + # deterministic delays in tests. + RETRY_JITTER = 0.5 # +/- this fraction of RETRY_DELAY + + @classmethod + def _next_delay(cls) -> float: + """RETRY_DELAY spread over [1-jitter, 1+jitter] of itself.""" + if not cls.RETRY_JITTER: + return cls.RETRY_DELAY + low, high = 1.0 - cls.RETRY_JITTER, 1.0 + cls.RETRY_JITTER + return cls.RETRY_DELAY * random.uniform(low, high) def __init__(self, client: redis.Redis): self._client = client @@ -34,8 +52,9 @@ def _wrapped(*args, **kwargs): return attr(*args, **kwargs) except self.RETRYABLE as exc: attempt += 1 + delay = self._next_delay() logger.warning( f"Redis '{name}' failed ({exc.__class__.__name__}: {exc}). " - f"Retrying in {self.RETRY_DELAY}s (attempt {attempt})") - time.sleep(self.RETRY_DELAY) + f"Retrying in {delay:.1f}s (attempt {attempt})") + time.sleep(delay) return _wrapped \ No newline at end of file diff --git a/tests/test_connection_liveness.py b/tests/test_connection_liveness.py new file mode 100644 index 0000000..fcfbdaa --- /dev/null +++ b/tests/test_connection_liveness.py @@ -0,0 +1,136 @@ +"""Regression tests for the 2026-08-15 silent worker stall. + +A dropped TCP connection to Redis left every worker parked forever in an +unbounded BLPOP: a pure read-wait transmits nothing, so the kernel never +retransmits, never gives up, and never raises. The process stayed up, HTTP +health checks kept passing, and the queue grew unattended for hours. + +These tests pin the three properties that make that impossible now: + 1. every queue read is bounded by a timeout, + 2. the connection pool can detect a half-open socket, + 3. a background loop outlives an exception thrown by its body. +""" + +import socket +import threading +import time +from unittest.mock import MagicMock, patch + +import fakeredis +import pytest + +from modelq import ModelQ +from modelq.app.base import _tcp_keepalive_options + + +@pytest.fixture +def modelq_instance(): + return ModelQ(redis_client=fakeredis.FakeStrictRedis()) + + +# --------------------------------------------------------------------------- +# 1. the queue read must be bounded +# --------------------------------------------------------------------------- + +def test_worker_blpop_passes_a_timeout(modelq_instance): + """The worker must never issue an unbounded BLPOP. + + This is the actual outage. `blpop(key)` with no timeout blocks forever on a + socket Redis has already forgotten. + """ + seen = {} + stop = threading.Event() + + def fake_blpop(key, *args, **kwargs): + seen["args"] = args + seen["kwargs"] = kwargs + stop.set() + # Behave like a real timed-out BLPOP so the worker loops rather than + # trying to decode a task. + time.sleep(0.01) + return None + + modelq_instance.redis_client = MagicMock(wraps=modelq_instance.redis_client) + modelq_instance.redis_client.blpop.side_effect = fake_blpop + + modelq_instance.start_workers(no_of_workers=1) + assert stop.wait(timeout=5), "worker never called blpop" + + timeout = seen["kwargs"].get("timeout", seen["args"][0] if seen["args"] else None) + assert timeout is not None, "BLPOP was issued without a timeout" + assert timeout > 0, f"BLPOP timeout must be positive, got {timeout!r}" + + +def test_blpop_timeout_stays_below_socket_timeout(): + """redis-py applies the socket read timeout to blocking commands too. + + If BLPOP_TIMEOUT ever crept above SOCKET_TIMEOUT, every idle poll would + raise instead of returning None — turning an idle worker into a log flood. + """ + assert ModelQ.BLPOP_TIMEOUT < ModelQ.SOCKET_TIMEOUT + + +# --------------------------------------------------------------------------- +# 2. the pool must be able to notice a dead socket +# --------------------------------------------------------------------------- + +def test_connection_pool_enables_half_open_detection(): + with patch("modelq.app.base.redis.ConnectionPool") as pool, \ + patch("modelq.app.base.redis.Redis"): + ModelQ(host="redis.invalid", port=6379, password="x", username=None) + + kwargs = pool.call_args.kwargs + assert kwargs["socket_keepalive"] is True + assert kwargs["health_check_interval"] > 0 + assert kwargs["socket_timeout"] == ModelQ.SOCKET_TIMEOUT + assert kwargs["socket_connect_timeout"] == ModelQ.SOCKET_CONNECT_TIMEOUT + assert kwargs["retry_on_timeout"] is True + + +def test_keepalive_options_are_valid_for_this_platform(): + """Every key must be a real socket constant, or the pool raises at connect.""" + options = _tcp_keepalive_options() + valid = { + getattr(socket, name) + for name in ("TCP_KEEPIDLE", "TCP_KEEPALIVE", "TCP_KEEPINTVL", "TCP_KEEPCNT") + if hasattr(socket, name) + } + assert set(options).issubset(valid) + assert all(isinstance(v, int) and v > 0 for v in options.values()) + + +# --------------------------------------------------------------------------- +# 3. a background loop must survive a transient error +# --------------------------------------------------------------------------- + +def test_guarded_iteration_swallows_and_continues(modelq_instance): + modelq_instance.BACKGROUND_LOOP_BACKOFF = 0 + with modelq_instance._guarded_iteration("unit-test"): + raise ConnectionError("transient redis blip") + # Reaching here at all is the assertion: the exception did not propagate. + + +def test_heartbeat_loop_survives_a_raising_body(modelq_instance): + """The pruning/heartbeat threads died outright on 2026-08-15. + + One raise used to unwind the thread permanently. Now the loop must still be + calling its body after the failure. + """ + modelq_instance.BACKGROUND_LOOP_BACKOFF = 0 + modelq_instance.HEARTBEAT_INTERVAL = 0.01 + calls = [] + done = threading.Event() + + def exploding_heartbeat(): + calls.append(time.time()) + if len(calls) == 1: + raise ConnectionError("transient redis blip") + if len(calls) >= 3: + done.set() + + modelq_instance.heartbeat = exploding_heartbeat + threading.Thread(target=modelq_instance._heartbeat_loop, daemon=True).start() + + assert done.wait(timeout=5), ( + f"loop stopped after {len(calls)} call(s); it died on the first raise" + ) diff --git a/tests/test_redis_retry_delay.py b/tests/test_redis_retry_delay.py new file mode 100644 index 0000000..5bfe93a --- /dev/null +++ b/tests/test_redis_retry_delay.py @@ -0,0 +1,73 @@ +"""Tests for the reconnect delay in _RedisWithRetry. + +Recovery from a dead Redis connection costs socket_timeout (to detect) plus +RETRY_DELAY (to wait), so RETRY_DELAY is the tail of every stall. It also has +to spread a synchronised herd: on 2026-08-15 every worker on a node logged the +same connection failure in the same second, because one upstream path change +broke all of them at once. +""" + +from unittest.mock import patch + +import pytest +from redis.exceptions import ConnectionError as RedisConnectionError + +from modelq.app.redis_retry import _RedisWithRetry + + +class _Flaky: + """Fails `fail_times` then succeeds, recording each call.""" + + def __init__(self, fail_times): + self.fail_times = fail_times + self.calls = 0 + + def ping(self): + self.calls += 1 + if self.calls <= self.fail_times: + raise RedisConnectionError("connection reset") + return True + + +def test_retry_delay_is_short_enough_to_bound_a_stall(): + assert _RedisWithRetry.RETRY_DELAY <= 15, ( + "RETRY_DELAY is the tail of every reconnect; keep it small" + ) + + +def test_jittered_delay_stays_within_bounds(): + """Never zero (busy-loop) and never far above the nominal delay.""" + lo = _RedisWithRetry.RETRY_DELAY * (1 - _RedisWithRetry.RETRY_JITTER) + hi = _RedisWithRetry.RETRY_DELAY * (1 + _RedisWithRetry.RETRY_JITTER) + seen = {round(_RedisWithRetry._next_delay(), 4) for _ in range(200)} + for d in seen: + assert lo <= d <= hi, f"{d} outside [{lo}, {hi}]" + assert min(seen) > 0, "a zero delay would busy-loop against a down Redis" + + +def test_jitter_actually_varies_the_delay(): + """Without this, a synchronised fleet retries in lockstep.""" + seen = {round(_RedisWithRetry._next_delay(), 6) for _ in range(50)} + assert len(seen) > 1, "delay is constant — jitter is not being applied" + + +def test_jitter_can_be_disabled_for_determinism(): + with patch.object(_RedisWithRetry, "RETRY_JITTER", 0): + assert _RedisWithRetry._next_delay() == _RedisWithRetry.RETRY_DELAY + + +def test_retry_loop_recovers_and_sleeps_the_jittered_delay(): + """The point of the whole wrapper: a transient error must not be fatal.""" + flaky = _Flaky(fail_times=2) + slept = [] + + with patch("modelq.app.redis_retry.time.sleep", slept.append): + wrapped = _RedisWithRetry(flaky) + assert wrapped.ping() is True + + assert flaky.calls == 3, "should have retried twice then succeeded" + assert len(slept) == 2, "should sleep once per failed attempt" + lo = _RedisWithRetry.RETRY_DELAY * (1 - _RedisWithRetry.RETRY_JITTER) + hi = _RedisWithRetry.RETRY_DELAY * (1 + _RedisWithRetry.RETRY_JITTER) + for s in slept: + assert lo <= s <= hi