fix(redis): re-establish async cluster connections after a node restart

litellm/_redis.py

@@ -23,7 +23,11 @@
     GCPIAMCredentialProvider,
     _generate_gcp_iam_access_token,
 )
-from litellm.constants import REDIS_CONNECTION_POOL_TIMEOUT, REDIS_SOCKET_TIMEOUT
+from litellm.constants import (
+    REDIS_CLUSTER_HEALTH_CHECK_INTERVAL,
+    REDIS_CONNECTION_POOL_TIMEOUT,
+    REDIS_SOCKET_TIMEOUT,
+)
 from litellm.litellm_core_utils.sensitive_data_masker import SensitiveDataMasker
 
 from ._logging import verbose_logger
@@ -102,6 +106,8 @@ def _get_redis_cluster_kwargs(client=None):
         "max_connections",
         "socket_timeout",
         "socket_connect_timeout",
+        "health_check_interval",
+        "socket_keepalive",
     }
 
     return available_args
@@ -579,6 +585,13 @@ def get_redis_async_client(
             new_startup_nodes.append(ClusterNode(**item))
         cluster_kwargs.pop("startup_nodes", None)
 
+        # Default to a periodic health check + TCP keepalive so a connection silently dropped
+        # by a cluster restart (e.g. ElastiCache Serverless maintenance) is revalidated and
+        # reconnected before reuse instead of stalling in re-initialization; an explicit value
+        # from config still wins.
+        cluster_kwargs.setdefault("health_check_interval", REDIS_CLUSTER_HEALTH_CHECK_INTERVAL)
+        cluster_kwargs.setdefault("socket_keepalive", True)
+
         # Create async RedisCluster with IAM token as password if available
         cluster_client = async_redis.RedisCluster(
             startup_nodes=new_startup_nodes,

litellm/constants.py

@@ -332,6 +332,10 @@
 REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD = int(os.getenv("REDIS_CIRCUIT_BREAKER_FAILURE_THRESHOLD", 5))
 REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT = int(os.getenv("REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT", 60))
 REDIS_CIRCUIT_BREAKER_ENABLED = os.getenv("REDIS_CIRCUIT_BREAKER_ENABLED", "true").lower() == "true"
+# Seconds of idle before a Redis cluster connection is validated with a PING and
+# reconnected if dead, so a connection silently dropped by a cluster restart
+# (e.g. ElastiCache Serverless maintenance) is not reused while broken
+REDIS_CLUSTER_HEALTH_CHECK_INTERVAL = 25
 # Default Redis major version to assume when version cannot be determined
 # Using 7 as it's the modern version that supports LPOP with count parameter
 DEFAULT_REDIS_MAJOR_VERSION = int(os.getenv("DEFAULT_REDIS_MAJOR_VERSION", 7))

tests/test_litellm/test_redis.py

@@ -12,6 +12,7 @@
     get_redis_connection_pool,
     get_redis_url_from_environment,
 )
+from litellm.constants import REDIS_CLUSTER_HEALTH_CHECK_INTERVAL
 from litellm._redis_credential_provider import (
     GCPIAMCredentialProvider,
     _token_cache,
@@ -171,6 +172,46 @@ def test_socket_timeouts_in_cluster_kwargs():
     assert "socket_connect_timeout" in kwargs
 
 
+def test_reconnect_kwargs_in_cluster_kwargs():
+    """Health check and keepalive must survive the cluster kwarg allow-list so
+    operators can tune Redis cluster reconnection behavior via config."""
+    kwargs = _get_redis_cluster_kwargs()
+    assert "health_check_interval" in kwargs
+    assert "socket_keepalive" in kwargs
+
+
+@patch("litellm._redis.async_redis.RedisCluster")
+def test_async_cluster_sets_reconnect_defaults(mock_cluster_cls):
+    """
+    The async RedisCluster client must be built with a periodic health check and
+    TCP keepalive so a connection silently dropped by a cluster restart (e.g.
+    ElastiCache Serverless maintenance) is revalidated and reconnected before
+    reuse instead of stalling in re-initialization. Regression for LIT-4083.
+    """
+    get_redis_async_client(startup_nodes=[{"host": "cluster-node", "port": 6379}])
+
+    mock_cluster_cls.assert_called_once()
+    call_kwargs = mock_cluster_cls.call_args[1]
+    assert call_kwargs["health_check_interval"] == REDIS_CLUSTER_HEALTH_CHECK_INTERVAL
+    assert call_kwargs["health_check_interval"] > 0
+    assert call_kwargs["socket_keepalive"] is True
+
+
+@patch("litellm._redis.async_redis.RedisCluster")
+def test_async_cluster_reconnect_defaults_are_overridable(mock_cluster_cls):
+    """An explicit health_check_interval / socket_keepalive from config must win
+    over the built-in reconnect defaults."""
+    get_redis_async_client(
+        startup_nodes=[{"host": "cluster-node", "port": 6379}],
+        health_check_interval=7,
+        socket_keepalive=False,
+    )
+
+    call_kwargs = mock_cluster_cls.call_args[1]
+    assert call_kwargs["health_check_interval"] == 7
+    assert call_kwargs["socket_keepalive"] is False
+
+
 def test_get_redis_async_client_with_connection_pool():
     """Test that connection_pool parameter is properly passed to Redis client"""
     # Create a mock connection pool