diff --git a/litellm/_redis.py b/litellm/_redis.py index a86ebd9ea9..c61582abd1 100644 --- a/litellm/_redis.py +++ b/litellm/_redis.py @@ -381,6 +381,8 @@ def get_redis_async_client( ) -> Union[async_redis.Redis, async_redis.RedisCluster]: redis_kwargs = _get_redis_client_logic(**env_overrides) if "url" in redis_kwargs and redis_kwargs["url"] is not None: + if connection_pool is not None: + return async_redis.Redis(connection_pool=connection_pool) args = _get_redis_url_kwargs(client=async_redis.Redis.from_url) url_kwargs = {} for arg in redis_kwargs: @@ -461,9 +463,16 @@ def get_redis_connection_pool(**env_overrides): redis_kwargs = _get_redis_client_logic(**env_overrides) verbose_logger.debug("get_redis_connection_pool: redis_kwargs", redis_kwargs) if "url" in redis_kwargs and redis_kwargs["url"] is not None: - return async_redis.BlockingConnectionPool.from_url( - timeout=REDIS_CONNECTION_POOL_TIMEOUT, url=redis_kwargs["url"] - ) + pool_kwargs = {"timeout": REDIS_CONNECTION_POOL_TIMEOUT, "url": redis_kwargs["url"]} + if "max_connections" in redis_kwargs: + try: + pool_kwargs["max_connections"] = int(redis_kwargs["max_connections"]) + except (TypeError, ValueError): + verbose_logger.warning( + "REDIS: invalid max_connections value %r, ignoring", + redis_kwargs["max_connections"], + ) + return async_redis.BlockingConnectionPool.from_url(**pool_kwargs) connection_class = async_redis.Connection if "ssl" in redis_kwargs: connection_class = async_redis.SSLConnection diff --git a/litellm/caching/llm_caching_handler.py b/litellm/caching/llm_caching_handler.py index 16eb824f4c..5dc16a224c 100644 --- a/litellm/caching/llm_caching_handler.py +++ b/litellm/caching/llm_caching_handler.py @@ -8,6 +8,25 @@ from .in_memory_cache import InMemoryCache class LLMClientCache(InMemoryCache): + def _remove_key(self, key: str) -> None: + """Close async clients before evicting them to prevent connection pool leaks.""" + value = self.cache_dict.get(key) + super()._remove_key(key) + if value is not None: + close_fn = getattr(value, "aclose", None) or getattr( + value, "close", None + ) + if close_fn and asyncio.iscoroutinefunction(close_fn): + try: + asyncio.get_running_loop().create_task(close_fn()) + except RuntimeError: + pass + elif close_fn and callable(close_fn): + try: + close_fn() + except Exception: + pass + def update_cache_key_with_event_loop(self, key): """ Add the event loop to the cache key, to prevent event loop closed errors. diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 03d09ecc04..dcc2df5f91 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -1105,6 +1105,10 @@ class RedisCache(BaseCache): async def disconnect(self): await self.async_redis_conn_pool.disconnect(inuse_connections=True) + try: + self.redis_client.close() + except Exception as e: + verbose_logger.debug("Error closing sync Redis client: %s", e) async def test_connection(self) -> dict: """ diff --git a/tests/test_litellm/caching/test_redis_connection_pool.py b/tests/test_litellm/caching/test_redis_connection_pool.py new file mode 100644 index 0000000000..69381653ac --- /dev/null +++ b/tests/test_litellm/caching/test_redis_connection_pool.py @@ -0,0 +1,171 @@ +""" +Regression tests for Redis connection pool leak fixes (RC1-RC5). + +Tests are pure unit tests — no Redis server required. +""" + +import asyncio +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +import redis.asyncio as async_redis + +from litellm._redis import get_redis_async_client, get_redis_connection_pool +from litellm.caching.llm_caching_handler import LLMClientCache + + +def test_url_config_uses_passed_pool(): + """When connection_pool is provided with a URL config, the client + should use the passed pool — not create a new one via from_url().""" + mock_pool = MagicMock() + + with patch("litellm._redis._get_redis_client_logic") as mock_logic: + mock_logic.return_value = {"url": "redis://localhost:6379/0"} + + client = get_redis_async_client(connection_pool=mock_pool) + + assert client.connection_pool is mock_pool + + +def test_url_config_falls_back_to_from_url_without_pool(): + """When no connection_pool is provided, URL config should still + use from_url() as before.""" + with patch("litellm._redis._get_redis_client_logic") as mock_logic: + mock_logic.return_value = {"url": "redis://localhost:6379/0"} + + client = get_redis_async_client() + + # from_url creates its own pool — just verify it's not None + assert client.connection_pool is not None + + +def test_max_connections_url_config(): + """max_connections should be respected when using URL-based config.""" + with patch("litellm._redis._get_redis_client_logic") as mock_logic: + mock_logic.return_value = { + "url": "redis://localhost:6379/0", + "max_connections": 10, + } + + pool = get_redis_connection_pool() + + assert pool.max_connections == 10 + + +def test_max_connections_url_config_string_value(): + """max_connections provided as a string (from env var) should be + cast to int.""" + with patch("litellm._redis._get_redis_client_logic") as mock_logic: + mock_logic.return_value = { + "url": "redis://localhost:6379/0", + "max_connections": "25", + } + + pool = get_redis_connection_pool() + + assert pool.max_connections == 25 + + +def test_max_connections_url_config_invalid_value(): + """Invalid max_connections should be silently ignored, falling back + to the pool default (50 for BlockingConnectionPool).""" + with patch("litellm._redis._get_redis_client_logic") as mock_logic: + mock_logic.return_value = { + "url": "redis://localhost:6379/0", + "max_connections": "not_a_number", + } + + pool = get_redis_connection_pool() + + # BlockingConnectionPool default is 50 + assert pool.max_connections == 50 + + +def test_max_connections_url_config_none_value(): + """max_connections=None should be silently ignored.""" + with patch("litellm._redis._get_redis_client_logic") as mock_logic: + mock_logic.return_value = { + "url": "redis://localhost:6379/0", + "max_connections": None, + } + + pool = get_redis_connection_pool() + + assert pool.max_connections == 50 + + +def _make_redis_cache(): + """Create a RedisCache with all external I/O mocked out.""" + mock_sync_client = MagicMock() + mock_async_pool = AsyncMock() + patches = [ + patch("litellm._redis.get_redis_client", return_value=mock_sync_client), + patch("litellm._redis.get_redis_connection_pool", return_value=mock_async_pool), + patch("litellm.caching.redis_cache.RedisCache._setup_health_pings"), + ] + for p in patches: + p.start() + + from litellm.caching.redis_cache import RedisCache + cache = RedisCache(host="localhost", port=6379) + + for p in patches: + p.stop() + + return cache, mock_sync_client, mock_async_pool + + +@pytest.mark.asyncio +async def test_disconnect_closes_sync_client(): + """disconnect() should close both the async pool and the sync client.""" + cache, mock_sync_client, mock_async_pool = _make_redis_cache() + await cache.disconnect() + + mock_async_pool.disconnect.assert_awaited_once_with(inuse_connections=True) + mock_sync_client.close.assert_called_once() + + +@pytest.mark.asyncio +async def test_disconnect_idempotent(): + """Calling disconnect() twice should not raise.""" + cache, mock_sync_client, mock_async_pool = _make_redis_cache() + mock_sync_client.close.side_effect = [None, RuntimeError("already closed")] + + await cache.disconnect() + await cache.disconnect() # should not raise + + +@pytest.mark.asyncio +async def test_eviction_calls_aclose(): + """When an async client is evicted from LLMClientCache, its aclose() + should be scheduled via create_task.""" + cache = LLMClientCache(max_size_in_memory=2, default_ttl=600) + + client = AsyncMock() + client.aclose = AsyncMock() + + cache.set_cache(key="client-0", value=client) + cache.set_cache(key="filler", value="x") + # Third insert triggers eviction of client-0 + cache.set_cache(key="trigger", value="y") + + # Let the scheduled task run + await asyncio.sleep(0.05) + + assert client.aclose.await_count > 0 + + +@pytest.mark.asyncio +async def test_eviction_non_closeable_safe(): + """Evicting plain values (strings, dicts, ints) should not crash.""" + cache = LLMClientCache(max_size_in_memory=2, default_ttl=600) + + cache.set_cache(key="str-val", value="hello") + cache.set_cache(key="dict-val", value={"foo": "bar"}) + # This evicts "str-val" — should not raise + cache.set_cache(key="int-val", value=42) + + await asyncio.sleep(0.05) + + # If we got here without exception, the test passes + assert cache.get_cache(key="int-val") == 42