fix(redis): make Sentinel clients resilient to dropped connections

docs/source/changelog.rst

@@ -6,6 +6,12 @@ Changelog
 All notable changes to this project will be documented in this file.
 
 =======
+`6.2.1`_ -- 2026-07-08
+-------------
+Fixed
+^^^^^
+* Make Redis Sentinel clients resilient to dropped connections: set ``health_check_interval``, ``socket_keepalive`` and a ``retry`` on ``ConnectionError``/``TimeoutError``, so a transient Sentinel connection reset recovers transparently instead of raising ``MasterNotFoundError``.
+
 `6.2.0`_ -- 2026-05-18
 -------------
 Feat

remoulade/helpers/redis_client.py

@@ -3,49 +3,61 @@
 
 import redis
 import redis.asyncio as redis_async
+from redis.asyncio.retry import Retry as AsyncRetry
+from redis.backoff import ExponentialBackoff
+from redis.retry import Retry
 
+HEALTH_CHECK_INTERVAL = 30
+CONNECTION_MAX_RETRIES = 3
 
-def redis_client(url: str | None, socket_timeout: float | None = None, **parameters):
-    socket_parameters = {}
+
+def _connection_parameters(retry, socket_timeout):
+    # health_check_interval re-pings an idle connection and reconnects it before reuse; retry + retry_on_error
+    # transparently re-run a command over a fresh connection on a reset or timeout, which also recovers Sentinel
+    # master rediscovery since MasterNotFoundError is a ConnectionError.
+    parameters = {
+        "socket_keepalive": True,
+        "health_check_interval": HEALTH_CHECK_INTERVAL,
+        "retry": retry,
+        "retry_on_error": [redis.ConnectionError, redis.TimeoutError],
+    }
     if socket_timeout is not None:
-        socket_parameters = {
-            "socket_timeout": socket_timeout,
-            "socket_connect_timeout": socket_timeout,
-            "socket_keepalive": True,
-        }
+        parameters["socket_timeout"] = socket_timeout
+        parameters["socket_connect_timeout"] = socket_timeout
+    return parameters
+
+
+def redis_client(url: str | None, socket_timeout: float | None = None, **parameters):
+    connection_parameters = _connection_parameters(Retry(ExponentialBackoff(), CONNECTION_MAX_RETRIES), socket_timeout)
     if url:
         url_parsed = urlparse(url)
         if url_parsed.scheme == "sentinel":
-            sentinel_kwargs = {"password": url_parsed.password, **socket_parameters}
+            sentinel_kwargs = {"password": url_parsed.password, **connection_parameters}
             sentinel = redis.Sentinel([(url_parsed.hostname, url_parsed.port)], sentinel_kwargs=sentinel_kwargs)
             return sentinel.master_for(
                 service_name=os.path.normpath(url_parsed.path).split("/")[1],
                 password=url_parsed.password,
-                **socket_parameters,
+                **connection_parameters,
             )
         else:
-            parameters["connection_pool"] = redis.ConnectionPool.from_url(url, **socket_parameters)  # type: ignore
+            parameters["connection_pool"] = redis.ConnectionPool.from_url(url, **connection_parameters)  # type: ignore
             return redis.Redis(**parameters)
 
 
 def async_redis_client(url: str | None, socket_timeout: float | None = None, **parameters):
-    socket_parameters = {}
-    if socket_timeout is not None:
-        socket_parameters = {
-            "socket_timeout": socket_timeout,
-            "socket_connect_timeout": socket_timeout,
-            "socket_keepalive": True,
-        }
+    connection_parameters = _connection_parameters(
+        AsyncRetry(ExponentialBackoff(), CONNECTION_MAX_RETRIES), socket_timeout
+    )
     if url:
         url_parsed = urlparse(url)
         if url_parsed.scheme == "sentinel":
-            sentinel_kwargs = {"password": url_parsed.password, **socket_parameters}
+            sentinel_kwargs = {"password": url_parsed.password, **connection_parameters}
             sentinel = redis_async.Sentinel([(url_parsed.hostname, url_parsed.port)], sentinel_kwargs=sentinel_kwargs)
             return sentinel.master_for(  # type: ignore
                 service_name=os.path.normpath(url_parsed.path).split("/")[1],
                 password=url_parsed.password,
-                **socket_parameters,
+                **connection_parameters,
             )
         else:
-            parameters["connection_pool"] = redis_async.ConnectionPool.from_url(url, **socket_parameters)
+            parameters["connection_pool"] = redis_async.ConnectionPool.from_url(url, **connection_parameters)
             return redis_async.Redis(**parameters)

tests/test_helpers.py

@@ -3,14 +3,21 @@
 from queue import Queue
 
 import pytest
+from redis import ConnectionError as RedisConnectionError
+from redis.asyncio.retry import Retry as AsyncRetry
+from redis.retry import Retry
 
 import remoulade
 from remoulade import QueueJoinTimeout
 from remoulade.helpers import compute_backoff, get_actor_arguments
 from remoulade.helpers.queues import join_queue
+from remoulade.helpers.redis_client import HEALTH_CHECK_INTERVAL, async_redis_client, redis_client
 from remoulade.helpers.reduce import reduce
 from remoulade.results import Results
 
+SENTINEL_URL = "sentinel://:secret@localhost:26379/mymaster"
+PLAIN_URL = "redis://localhost:6379/0"
+
 
 def test_reduce_messages(stub_broker, stub_worker, result_backend):
     # Given a result backend
@@ -131,3 +138,32 @@ def test_join_queue_stdlib_fallback_timeout_raises_queue_join_timeout():
 
     with pytest.raises(QueueJoinTimeout):
         join_queue(queue, timeout=0.01)
+
+
+@pytest.mark.parametrize("url", [SENTINEL_URL, PLAIN_URL])
+def test_redis_client_sets_resilient_connection_parameters(url):
+    connection_kwargs = redis_client(url, socket_timeout=5.0).connection_pool.connection_kwargs
+
+    assert connection_kwargs["health_check_interval"] == HEALTH_CHECK_INTERVAL
+    assert connection_kwargs["socket_keepalive"] is True
+    assert isinstance(connection_kwargs["retry"], Retry)
+    assert RedisConnectionError in connection_kwargs["retry_on_error"]
+
+
+@pytest.mark.parametrize("url", [SENTINEL_URL, PLAIN_URL])
+def test_async_redis_client_sets_resilient_connection_parameters(url):
+    connection_kwargs = async_redis_client(url, socket_timeout=5.0).connection_pool.connection_kwargs
+
+    assert connection_kwargs["health_check_interval"] == HEALTH_CHECK_INTERVAL
+    assert connection_kwargs["socket_keepalive"] is True
+    assert isinstance(connection_kwargs["retry"], AsyncRetry)
+    assert RedisConnectionError in connection_kwargs["retry_on_error"]
+
+
+def test_redis_client_is_resilient_without_socket_timeout():
+    # The scheduler builds its client without a socket_timeout; it must still get health checks and retries.
+    connection_kwargs = redis_client(PLAIN_URL).connection_pool.connection_kwargs
+
+    assert connection_kwargs["health_check_interval"] == HEALTH_CHECK_INTERVAL
+    assert isinstance(connection_kwargs["retry"], Retry)
+    assert RedisConnectionError in connection_kwargs["retry_on_error"]