From f8970a1d38a0cc9996f052dbb5d5816e359ed633 Mon Sep 17 00:00:00 2001 From: Adhik Joshi Date: Sat, 15 Aug 2026 12:46:33 +0530 Subject: [PATCH 1/2] fix: bound the queue read so a dead Redis socket cannot park a worker forever MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit On 2026-08-15 the TCP connections from the GPU fleet to the Redis box were silently dropped at ~04:05 UTC. Seven server_types stopped consuming within fifteen minutes of each other and stayed dead for hours; flux_2_dev alone lost 2h45m of throughput. Nothing restarted, nothing alerted, and the endpoints still reported healthy. The worker sat in `blpop("ml_tasks")` with no timeout. That is a pure read-wait: the client transmits nothing, so the kernel never retransmits, never hits tcp_retries2, and never raises. The thread parked forever on a socket the server had already discarded — which is why Redis reported blocked_clients:0 while every worker believed it was blocked. Threads that were *sending* commands did surface the failure, as ETIMEDOUT roughly eighteen minutes later, and reconnected normally. The reader never could. Three changes: - `blpop` now takes BLPOP_TIMEOUT and the loop treats None as "keep waiting". Completing the read is what lets a dead socket be noticed at all. The timeout is kept below SOCKET_TIMEOUT because redis-py applies the socket read deadline to blocking commands too. - The connection pool sets socket_keepalive (with per-platform TCP knobs), socket_timeout, socket_connect_timeout, health_check_interval and retry_on_timeout, so a half-open connection is reaped instead of trusted. - Background loops run their body inside `_guarded_iteration`. A bare `while True:` in a thread was one unhandled exception away from being gone permanently, and the pruning thread died exactly that way on every replica of several endpoints while the process carried on looking fine. Tests cover all three and were mutation-tested: restoring the unbounded blpop, stripping the pool options, and making the guard re-raise each turn the corresponding test red. --- modelq/app/base.py | 108 ++++++++++++++++++++---- tests/test_connection_liveness.py | 136 ++++++++++++++++++++++++++++++ 2 files changed, 229 insertions(+), 15 deletions(-) create mode 100644 tests/test_connection_liveness.py 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/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" + ) From afe49349ee48905362f4256b96353b60c9f5cd9b Mon Sep 17 00:00:00 2001 From: Adhik Joshi Date: Sat, 15 Aug 2026 15:11:57 +0530 Subject: [PATCH 2/2] fix: halve the reconnect delay to 15s and jitter it Recovery from a dead connection costs socket_timeout to detect plus RETRY_DELAY to wait, so this constant is the tail of every stall. 30s made that tail longer than the detection it followed. Jitter matters as much as the value. Connection loss here is a shared event, not an independent one: on 2026-08-15 every worker on a node logged its failure in the same second, because one upstream path change broke all their connections at once. A fixed delay marches that entire herd back into Redis in lockstep, which is the worst possible moment to arrive together. The delay is now spread over [0.5x, 1.5x] of nominal, and RETRY_JITTER = 0 restores deterministic behaviour for tests. The logged delay is now the actual one slept rather than the nominal constant, so the line stays truthful. --- modelq/app/redis_retry.py | 29 ++++++++++--- tests/test_redis_retry_delay.py | 73 +++++++++++++++++++++++++++++++++ 2 files changed, 97 insertions(+), 5 deletions(-) create mode 100644 tests/test_redis_retry_delay.py 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_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