fix: bound the queue read so a dead Redis socket cannot park a worker forever

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,22 +877,45 @@ 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):
         """
         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):

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

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"
+    )