Add TTL-aware cache methods to Redis and dual cache

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

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.

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)