diff --git a/litellm/caching/dual_cache.py b/litellm/caching/dual_cache.py index 34ae3638a5b..ada17c2b157 100644 --- a/litellm/caching/dual_cache.py +++ b/litellm/caching/dual_cache.py @@ -147,13 +147,17 @@ class DualCache(BaseCache): if result is None and self.redis_cache is not None and local_only is False: # If not found in in-memory cache, try fetching from Redis - redis_result = self.redis_cache.get_cache( + redis_result, redis_ttl = self.redis_cache.get_cache_with_ttl( key, parent_otel_span=parent_otel_span ) if redis_result is not None: - # Update in-memory cache with the value from Redis - self.in_memory_cache.set_cache(key, redis_result, **kwargs) + # Update in-memory cache with the value from Redis, + # preserving the remaining TTL to avoid stale entries + _backfill_kwargs = {**kwargs} + if redis_ttl is not None and redis_ttl > 0: + _backfill_kwargs["ttl"] = redis_ttl + self.in_memory_cache.set_cache(key, redis_result, **_backfill_kwargs) result = redis_result @@ -221,14 +225,18 @@ class DualCache(BaseCache): if result is None and self.redis_cache is not None and local_only is False: # If not found in in-memory cache, try fetching from Redis - redis_result = await self.redis_cache.async_get_cache( + redis_result, redis_ttl = await self.redis_cache.async_get_cache_with_ttl( key, parent_otel_span=parent_otel_span ) if redis_result is not None: - # Update in-memory cache with the value from Redis + # Update in-memory cache with the value from Redis, + # preserving the remaining TTL to avoid stale entries + _backfill_kwargs = {**kwargs} + if redis_ttl is not None and redis_ttl > 0: + _backfill_kwargs["ttl"] = redis_ttl await self.in_memory_cache.async_set_cache( - key, redis_result, **kwargs + key, redis_result, **_backfill_kwargs ) result = redis_result @@ -310,7 +318,8 @@ class DualCache(BaseCache): if len(sublist_keys) > 0: try: # If not found in in-memory cache, try fetching from Redis - redis_result = await self.redis_cache.async_batch_get_cache( + # Use with_ttl variant to preserve TTL when backfilling in-memory + redis_result_with_ttl = await self.redis_cache.async_batch_get_cache_with_ttl( sublist_keys, parent_otel_span=parent_otel_span ) except Exception: @@ -321,8 +330,8 @@ class DualCache(BaseCache): raise # Short-circuit if redis_result is None or contains only None values - if redis_result is None or all( - v is None for v in redis_result.values() + if redis_result_with_ttl is None or all( + v is None for v, _ttl in redis_result_with_ttl.values() ): return result @@ -330,12 +339,15 @@ class DualCache(BaseCache): key_to_index = {key: i for i, key in enumerate(keys)} # Update both result and in-memory cache in a single loop - for key, value in redis_result.items(): + for key, (value, redis_ttl) in redis_result_with_ttl.items(): result[key_to_index[key]] = value if value is not None and self.in_memory_cache is not None: + _backfill_kwargs = {**kwargs} + if redis_ttl is not None and redis_ttl > 0: + _backfill_kwargs["ttl"] = redis_ttl await self.in_memory_cache.async_set_cache( - key, value, **kwargs + key, value, **_backfill_kwargs ) return result diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 84a2887f527..b1b7ee3a635 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -16,7 +16,7 @@ import inspect import json import time from datetime import timedelta -from typing import TYPE_CHECKING, Any, List, Optional, Tuple, Union, cast +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union, cast import litellm from litellm._logging import print_verbose, verbose_logger @@ -926,6 +926,44 @@ class RedisCache(BaseCache): "litellm.caching.caching: get() - Got exception from REDIS: ", e ) + def get_cache_with_ttl( + self, key: str, parent_otel_span: Optional[Span] = None, **kwargs + ) -> Tuple[Any, Optional[int]]: + """ + Get a cache value along with its remaining TTL from Redis (sync). + + Returns: + Tuple of (value, ttl_seconds). TTL is None if key has no expiry + or key does not exist. + """ + try: + key = self.check_and_fix_namespace(key=key) + start_time = time.time() + pipe = self.redis_client.pipeline(transaction=False) + pipe.get(key) + pipe.ttl(key) + results = pipe.execute() + cached_response = results[0] + ttl = results[1] + end_time = time.time() + _duration = end_time - start_time + self.service_logger_obj.service_success_hook( + service=ServiceTypes.REDIS, + duration=_duration, + call_type=f"get_cache_with_ttl <- {_get_call_stack_info()}", + start_time=start_time, + end_time=end_time, + parent_otel_span=parent_otel_span, + ) + response = self._get_cache_logic(cached_response=cached_response) + _ttl: Optional[int] = ttl if isinstance(ttl, int) and ttl > 0 else None + return response, _ttl + except Exception as e: + verbose_logger.error( + "litellm.caching.redis_cache: get_cache_with_ttl() - Got exception from REDIS: ", e + ) + return None, None + def _run_redis_mget_operation(self, keys: List[str]) -> List[Any]: """ Wrapper to call `mget` on the redis client @@ -1118,6 +1156,142 @@ class RedisCache(BaseCache): verbose_logger.error(f"Error occurred in async batch get cache - {str(e)}") return key_value_dict + @_redis_circuit_breaker_guard + async def async_get_cache_with_ttl( + self, key: str, parent_otel_span: Optional[Span] = None, **kwargs + ) -> Tuple[Any, Optional[int]]: + """ + Get a cache value along with its remaining TTL from Redis. + + Returns: + Tuple of (value, ttl_seconds). TTL is None if key has no expiry, + or -2 if key does not exist. + """ + from redis.asyncio import Redis + + _redis_client: Redis = self.init_async_client() # type: ignore + key = self.check_and_fix_namespace(key=key) + start_time = time.time() + + try: + async with _redis_client.pipeline(transaction=False) as pipe: + pipe.get(key) + pipe.ttl(key) + results = await pipe.execute() + + cached_response = results[0] + ttl = results[1] # -1 = no expiry, -2 = key doesn't exist + + response = self._get_cache_logic(cached_response=cached_response) + + end_time = time.time() + _duration = end_time - start_time + asyncio.create_task( + self.service_logger_obj.async_service_success_hook( + service=ServiceTypes.REDIS, + duration=_duration, + call_type=f"async_get_cache_with_ttl <- {_get_call_stack_info()}", + start_time=start_time, + end_time=end_time, + parent_otel_span=parent_otel_span, + event_metadata={"key": key}, + ) + ) + # Return None TTL for keys without expiry or non-existent keys + _ttl: Optional[int] = ttl if isinstance(ttl, int) and ttl > 0 else None + return response, _ttl + except Exception as e: + end_time = time.time() + _duration = end_time - start_time + asyncio.create_task( + self.service_logger_obj.async_service_failure_hook( + service=ServiceTypes.REDIS, + duration=_duration, + error=e, + call_type=f"async_get_cache_with_ttl <- {_get_call_stack_info()}", + start_time=start_time, + end_time=end_time, + parent_otel_span=parent_otel_span, + event_metadata={"key": key}, + ) + ) + verbose_logger.error( + f"litellm.caching.redis_cache: async_get_cache_with_ttl() - Got exception from REDIS: {str(e)}" + ) + return None, None + + @_redis_circuit_breaker_guard + async def async_batch_get_cache_with_ttl( + self, + key_list: Union[List[str], List[Optional[str]]], + parent_otel_span: Optional[Span] = None, + ) -> Dict[str, Tuple[Any, Optional[int]]]: + """ + Bulk read from Redis, returning values together with their remaining TTLs. + + Uses a single pipeline round-trip: for each key issues GET + TTL. + + Returns: + dict mapping key -> (decoded_value, remaining_ttl_seconds). + TTL is None when the key has no expiry. + """ + result: Dict[str, Tuple[Any, Optional[int]]] = {} + start_time = time.time() + _key_list = [key for key in key_list if key is not None] + try: + _keys = [] + for cache_key in _key_list: + cache_key = self.check_and_fix_namespace(key=cache_key) + _keys.append(cache_key) + + _redis_client = self.init_async_client() + async with _redis_client.pipeline(transaction=False) as pipe: + for k in _keys: + pipe.get(k) + pipe.ttl(k) + raw_results = await pipe.execute() + + # raw_results is [value1, ttl1, value2, ttl2, ...] + end_time = time.time() + _duration = end_time - start_time + asyncio.create_task( + self.service_logger_obj.async_service_success_hook( + service=ServiceTypes.REDIS, + duration=_duration, + call_type=f"async_batch_get_cache_with_ttl <- {_get_call_stack_info()}", + start_time=start_time, + end_time=end_time, + parent_otel_span=parent_otel_span, + ) + ) + + for i, original_key in enumerate(_key_list): + raw_value = raw_results[i * 2] + raw_ttl = raw_results[i * 2 + 1] + decoded_value = self._get_cache_logic(raw_value) + _ttl: Optional[int] = raw_ttl if isinstance(raw_ttl, int) and raw_ttl > 0 else None + if isinstance(original_key, bytes): + original_key = original_key.decode("utf-8") + result[original_key] = (decoded_value, _ttl) + + return result + except Exception as e: + end_time = time.time() + _duration = end_time - start_time + asyncio.create_task( + self.service_logger_obj.async_service_failure_hook( + service=ServiceTypes.REDIS, + duration=_duration, + error=e, + call_type=f"async_batch_get_cache_with_ttl <- {_get_call_stack_info()}", + start_time=start_time, + end_time=end_time, + parent_otel_span=parent_otel_span, + ) + ) + verbose_logger.error(f"Error occurred in async_batch_get_cache_with_ttl - {str(e)}") + return result + def sync_ping(self) -> bool: """ Tests if the sync redis client is correctly setup. diff --git a/tests/test_litellm/caching/test_dual_cache.py b/tests/test_litellm/caching/test_dual_cache.py index 6bf4307c9cc..e39001e954f 100644 --- a/tests/test_litellm/caching/test_dual_cache.py +++ b/tests/test_litellm/caching/test_dual_cache.py @@ -17,14 +17,14 @@ async def test_dual_cache_async_batch_get_cache_coalesces_concurrent_redis_reads keys = ["shared_a", "shared_b"] start_gate = asyncio.Event() - async def _mock_async_batch_get_cache(key_list, parent_otel_span=None): + async def _mock_async_batch_get_cache_with_ttl(key_list, parent_otel_span=None): await asyncio.sleep(0.05) - return {k: None for k in key_list} + return {k: (None, None) for k in key_list} with patch.object( dual_cache.redis_cache, - "async_batch_get_cache", - new=AsyncMock(side_effect=_mock_async_batch_get_cache), + "async_batch_get_cache_with_ttl", + new=AsyncMock(side_effect=_mock_async_batch_get_cache_with_ttl), ) as mock_async_batch_get_cache: async def worker(): @@ -47,7 +47,7 @@ async def test_dual_cache_async_batch_get_cache_rolls_back_redis_reservation_on_ with patch.object( dual_cache.redis_cache, - "async_batch_get_cache", + "async_batch_get_cache_with_ttl", new=AsyncMock(side_effect=RuntimeError("redis unavailable")), ) as mock_async_batch_get_cache: first_result = await dual_cache.async_batch_get_cache(keys=keys)