From d86a20ad40741629816dd3592b19e49b3eb6ea7f Mon Sep 17 00:00:00 2001 From: Lin Junrong Date: Tue, 4 Aug 2026 01:20:34 +0000 Subject: [PATCH 1/4] fix: hold references to the service-logging tasks in RedisCache The event loop only keeps a weak reference to a task. Every fire-and-forget asyncio.create_task()/loop.create_task() call in this file (health pings, service success/failure hooks) can be garbage-collected mid-execution once the enclosing frame returns, silently dropping the log/ping. Hold each task in a set on the instance until it completes, mirroring the pattern already used elsewhere in the codebase for the same class of bug. Rebased onto the current litellm_internal_staging tip: the ruff-autofix modernization commit (b604e2b20c) rewrote this file's type annotations (Optional/List/Union -> |, list, etc.) and reflowed several of the exact lines this fix touches, which conflicted with the original commits. Redid the fix directly against the modernized file instead of replaying the old commits; net diff is functionally identical to the pre-rebase branch. --- litellm/caching/redis_cache.py | 128 +++++++++++++++++++++++++-------- 1 file changed, 97 insertions(+), 31 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 5fedfc5bcce..8191e4e5e0b 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -301,6 +301,10 @@ class RedisCache(BaseCache): self.service_logger_obj = kwargs.pop("service_logger_obj") else: self.service_logger_obj = ServiceLogging() + # The event loop only keeps weak references to tasks, so a service-log + # task whose only reference was the create_task() call can be collected + # before it reports. Hold it until it completes. + self._service_logging_tasks: set[asyncio.Task] = set() # mutable-ok: task registry redis_kwargs.update(kwargs) self.redis_client = get_redis_client(**redis_kwargs) @@ -340,7 +344,9 @@ class RedisCache(BaseCache): """Setup async and sync health pings for Redis.""" # ASYNC HEALTH PING try: - _ = asyncio.get_running_loop().create_task(self.ping()) + _health_ping_task = asyncio.get_running_loop().create_task(self.ping()) + self._service_logging_tasks.add(_health_ping_task) + _health_ping_task.add_done_callback(self._service_logging_tasks.discard) except Exception as e: if "no running event loop" in str(e): verbose_logger.debug("Ignoring async redis ping. No running event loop.") @@ -366,7 +372,7 @@ class RedisCache(BaseCache): loop: Final = asyncio.get_running_loop() start_time: Final = time.time() end_time: Final = start_time - loop.create_task( + _service_logging_task = loop.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=end_time - start_time, @@ -374,6 +380,8 @@ class RedisCache(BaseCache): call_type="redis_async_ping", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) except Exception: pass @@ -383,7 +391,7 @@ class RedisCache(BaseCache): loop: Final = asyncio.get_running_loop() start_time: Final = time.time() end_time: Final = start_time - loop.create_task( + _service_logging_task = loop.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=end_time - start_time, @@ -391,6 +399,8 @@ class RedisCache(BaseCache): call_type="redis_sync_ping", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) except Exception: pass @@ -563,7 +573,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -572,13 +582,15 @@ class RedisCache(BaseCache): end_time=end_time, ) ) # DO NOT SLOW DOWN CALL B/C OF THIS + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) return keys except Exception as e: # NON blocking - notify users Redis is throwing an exception ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -588,6 +600,8 @@ class RedisCache(BaseCache): end_time=end_time, ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) raise e def async_register_script(self, script: str) -> Callable[..., Awaitable[Any]]: @@ -684,7 +698,7 @@ class RedisCache(BaseCache): except Exception as e: end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -695,6 +709,8 @@ class RedisCache(BaseCache): call_type=f"async_set_cache <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error( "LiteLLM Redis Caching: async set() - Got exception from REDIS %s, key=%r, value=%r", str(e), @@ -720,7 +736,7 @@ class RedisCache(BaseCache): print_verbose(f"Successfully Set ASYNC Redis Cache: key: {key}\nValue {value}\nttl={ttl}") end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -731,11 +747,13 @@ class RedisCache(BaseCache): event_metadata={"key": key}, ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) return result except Exception as e: end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -747,6 +765,8 @@ class RedisCache(BaseCache): event_metadata={"key": key}, ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error( "LiteLLM Redis Caching: async set() - Got exception from REDIS %s, Writing value=%s", str(e), @@ -805,7 +825,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -815,12 +835,14 @@ class RedisCache(BaseCache): parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) return except Exception as e: ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -831,6 +853,8 @@ class RedisCache(BaseCache): parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error( "LiteLLM Redis Caching: async set_cache_pipeline() - Got exception from REDIS %s, Writing value=%s", @@ -866,7 +890,7 @@ class RedisCache(BaseCache): except Exception as e: end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -877,6 +901,8 @@ class RedisCache(BaseCache): call_type=f"async_set_cache_sadd <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) # NON blocking - notify users Redis is throwing an exception verbose_logger.error( "LiteLLM Redis Caching: async set() - Got exception from REDIS %s, Writing value=%s", @@ -892,7 +918,7 @@ class RedisCache(BaseCache): print_verbose(f"Successfully Set ASYNC Redis Cache SADD: key: {key}\nValue {value}\nttl={ttl}") end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -902,10 +928,12 @@ class RedisCache(BaseCache): parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) except Exception as e: end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -916,6 +944,8 @@ class RedisCache(BaseCache): parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) # NON blocking - notify users Redis is throwing an exception verbose_logger.error( "LiteLLM Redis Caching: async set_cache_sadd() - Got exception from REDIS %s, Writing value=%s", @@ -963,7 +993,7 @@ class RedisCache(BaseCache): end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -973,12 +1003,14 @@ class RedisCache(BaseCache): parent_otel_span=parent_otel_span, ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) return result except Exception as e: ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -989,6 +1021,8 @@ class RedisCache(BaseCache): parent_otel_span=parent_otel_span, ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error( "LiteLLM Redis Caching: async async_increment() - Got exception from REDIS %s, Writing value=%s", str(e), @@ -1159,7 +1193,7 @@ class RedisCache(BaseCache): end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1170,11 +1204,13 @@ class RedisCache(BaseCache): event_metadata={"key": key}, ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) return response except Exception as e: end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1186,6 +1222,8 @@ class RedisCache(BaseCache): event_metadata={"key": key}, ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) print_verbose(f"litellm.caching.caching: async get() - Got exception from REDIS: {e}") _record_swallowed_redis_failure(self._circuit_breaker, e) @@ -1220,7 +1258,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1230,6 +1268,8 @@ class RedisCache(BaseCache): parent_otel_span=parent_otel_span, ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) # Associate the results back with their keys. # 'results' is a list of values corresponding to the order of keys in 'key_list'. @@ -1247,7 +1287,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1258,6 +1298,8 @@ class RedisCache(BaseCache): parent_otel_span=parent_otel_span, ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error("Error occurred in async batch get cache - %s", e) _record_swallowed_redis_failure(self._circuit_breaker, e) return key_value_dict @@ -1306,20 +1348,22 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_ping <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) return response except Exception as e: # NON blocking - notify users Redis is throwing an exception ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1327,6 +1371,8 @@ class RedisCache(BaseCache): call_type=f"async_ping <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error("LiteLLM Redis Cache PING: - Got exception from REDIS : %s", e) raise e @@ -1460,7 +1506,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1470,12 +1516,14 @@ class RedisCache(BaseCache): parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) return results except Exception as e: ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1486,6 +1534,8 @@ class RedisCache(BaseCache): parent_otel_span=_get_parent_otel_span_from_kwargs(kwargs), ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error( "LiteLLM Redis Caching: async increment_pipeline() - Got exception from REDIS %s", str(e), @@ -1545,20 +1595,22 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_rpush <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) return response except Exception as e: # NON blocking - notify users Redis is throwing an exception ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1566,6 +1618,8 @@ class RedisCache(BaseCache): call_type=f"async_rpush <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error("LiteLLM Redis Cache RPUSH: - Got exception from REDIS : %s", e) raise e @@ -1614,19 +1668,21 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_rpush_pipeline <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) return results except Exception as e: ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1634,6 +1690,8 @@ class RedisCache(BaseCache): call_type=f"async_rpush_pipeline <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error( "LiteLLM Redis Caching: async_rpush_pipeline() - Got exception from REDIS %s", str(e), @@ -1679,13 +1737,15 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_lpop <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) # Handle result parsing if needed if isinstance(result, bytes): @@ -1704,7 +1764,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1712,6 +1772,8 @@ class RedisCache(BaseCache): call_type=f"async_lpop <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error("LiteLLM Redis Cache LPOP: - Got exception from REDIS : %s", e) raise e @@ -1803,19 +1865,21 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, call_type=f"async_lpop_pipeline <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) return results except Exception as e: ## LOGGING ## end_time = time.time() _duration = end_time - start_time - asyncio.create_task( + _service_logging_task = asyncio.create_task( self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1823,6 +1887,8 @@ class RedisCache(BaseCache): call_type=f"async_lpop_pipeline <- {_get_call_stack_info()}", ) ) + self._service_logging_tasks.add(_service_logging_task) + _service_logging_task.add_done_callback(self._service_logging_tasks.discard) verbose_logger.error( "LiteLLM Redis Caching: async_lpop_pipeline() - Got exception from REDIS %s", str(e), From f698f7d3acc6ab0c3b0e99ed5ecc121ef59e3d9a Mon Sep 17 00:00:00 2001 From: Lin Junrong Date: Mon, 3 Aug 2026 15:42:08 +0800 Subject: [PATCH 2/4] test: cover the retained service-logging tasks codecov reported 32.98% patch coverage with 63 lines missing, which is essentially all of the retention sites this PR adds. These drive each RedisCache method against a mocked client and assert the scheduled task lands in _service_logging_tasks while pending and is gone once it finishes, so holding the reference cannot become a leak. Both the failure path and the success path are covered, plus the two ping error handlers. Worth recording two things the tests had to work around: several methods call init_async_client() outside their try block, so a client that fails to construct never reaches the logging path at all and the failure has to come from the redis operation instead; and add_done_callback is delivered via call_soon, so the discard lands a tick after the task completes. All ten fail against the current upstream redis_cache.py and pass with this branch. ruff check, ruff format --check, ruff-strict and check_type_discipline.py are all clean on the new file. --- .../test_redis_cache_task_refs.py | 131 ++++++++++++++++++ 1 file changed, 131 insertions(+) create mode 100644 tests/test_litellm/test_redis_cache_task_refs.py diff --git a/tests/test_litellm/test_redis_cache_task_refs.py b/tests/test_litellm/test_redis_cache_task_refs.py new file mode 100644 index 00000000000..4732fd970d3 --- /dev/null +++ b/tests/test_litellm/test_redis_cache_task_refs.py @@ -0,0 +1,131 @@ +"""The service-logging tasks RedisCache fires must be strongly referenced. + +`asyncio.create_task` / `loop.create_task` hand the task to the event loop, +which keeps only a *weak* reference to it. A task whose only referent was the +`create_task(...)` call can be garbage collected while it is suspended, and +the service-log event it was going to emit is then silently lost. + +`RedisCache` fires these from every cache operation, on both the success and +the failure path. These tests drive each of those methods with the redis client +mocked out and assert the task lands in `_service_logging_tasks`, and that the +entry is removed once it completes so holding it cannot become a leak. +""" + +import asyncio +import contextlib +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +# litellm's package __getattr__ only resolves names it lists explicitly, so the +# submodule has to be imported before patch() can find it by path. +import litellm._redis # noqa: F401 # imported for its side effect, see above +from litellm.caching.redis_cache import RedisCache + + +@pytest.fixture +def cache() -> RedisCache: + """A RedisCache with every real redis connection stubbed out.""" + # RedisCache imports these from litellm._redis inside __init__, so they have + # to be patched at the source module rather than on redis_cache. + with ( + patch("litellm._redis.get_redis_client", return_value=MagicMock()), + patch("litellm._redis.get_redis_connection_pool", return_value=MagicMock()), + ): + redis_cache = RedisCache(host="localhost", port=6379) + + # The hooks are what get scheduled; make them awaitable no-ops so the tasks + # complete immediately instead of touching a real logging backend. + redis_cache.service_logger_obj.async_service_success_hook = AsyncMock() + redis_cache.service_logger_obj.async_service_failure_hook = AsyncMock() + return redis_cache + + +async def drain(cache: RedisCache) -> None: + """Let every scheduled task run and every done-callback be delivered.""" + # add_done_callback goes through call_soon, so the discard lands on the + # tick after the task itself finishes. + for _ in range(4): + await asyncio.sleep(0) + + +def failing_client() -> MagicMock: + """An async redis client whose every operation raises. + + The client itself has to construct successfully: several of these methods + call `init_async_client()` *outside* their try block, so making that raise + would propagate before any service-log task is ever scheduled. The failure + has to come from the redis operation instead. + """ + client = MagicMock() + for op in ("get", "set", "mget", "sadd", "incrbyfloat", "expire", "rpush", "lpop", "scan_iter", "ping"): + setattr(client, op, AsyncMock(side_effect=ConnectionError("redis is down"))) + return client + + +# Each entry drives one RedisCache method against a client that fails, which is +# the path to the service-failure hook these methods fire. +FAILING_CALLS = ( + ("async_set_cache", ("k", "v")), + ("async_get_cache", ("k",)), + ("async_batch_get_cache", (("k1", "k2"),)), + ("async_increment", ("k", 1.0)), + ("async_rpush", ("k", ("v",))), + ("async_lpop", ("k",)), + ("ping", ()), +) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method_name, args", FAILING_CALLS, ids=tuple(n for n, _ in FAILING_CALLS)) +async def test_failure_path_holds_the_service_log_task(cache: RedisCache, method_name: str, args: tuple) -> None: + method = getattr(cache, method_name) + + with patch.object(cache, "init_async_client", return_value=failing_client()): + # Several of these re-raise after logging and several swallow; either + # way the scheduled task is what this asserts on. + with contextlib.suppress(ConnectionError): + await method(*args) + + assert cache._service_logging_tasks, ( + f"{method_name} scheduled a service-log task without keeping a " + "reference to it, so the loop's weak reference is the only one" + ) + + await drain(cache) + assert not cache._service_logging_tasks, f"{method_name} left its finished task in the registry" + + +@pytest.mark.asyncio +async def test_success_path_holds_the_service_log_task(cache: RedisCache) -> None: + """The success hook is fired from a task too, and needs the same reference.""" + client = MagicMock() + client.set = AsyncMock(return_value=True) + + with patch.object(cache, "init_async_client", return_value=client): + await cache.async_set_cache("k", "v") + assert cache._service_logging_tasks + + await drain(cache) + assert not cache._service_logging_tasks + + +@pytest.mark.asyncio +async def test_ping_error_handlers_hold_their_tasks(cache: RedisCache) -> None: + """_handle_async_ping_error / _handle_sync_ping_error each fire one task.""" + error = ConnectionError("redis is down") + + cache._handle_async_ping_error(error) + assert len(cache._service_logging_tasks) == 1 + + cache._handle_sync_ping_error(error) + assert len(cache._service_logging_tasks) == 2 + + await drain(cache) + assert not cache._service_logging_tasks + + +@pytest.mark.asyncio +async def test_registry_starts_empty(cache: RedisCache) -> None: + """Nothing is scheduled just by constructing the cache.""" + assert len(cache._service_logging_tasks) == 0 From 6aad20544eb3c0ae9308a473acf0b2c23821175e Mon Sep 17 00:00:00 2001 From: Lin Junrong Date: Mon, 3 Aug 2026 20:59:00 +0800 Subject: [PATCH 3/4] test: cover every service-logging task retention site MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The first pass only drove the failure path of seven methods, which left 21 of the 31 retention sites unexercised — every success-hook site, both sites in each of the six pipeline/scan methods, the client-init failure branch, and the health ping. Drives each method twice, against a client that works and one that fails, and adds the two paths that need a different seam: the pipeline methods reach redis through a helper rather than through the client's own commands, and async_scan_iter is iterated rather than awaited so it has to fail on the call. 31/31 retention sites are now executed. Reverting redis_cache.py to the base revision fails all 31. --- .../test_redis_cache_task_refs.py | 154 ++++++++++++++++-- 1 file changed, 139 insertions(+), 15 deletions(-) diff --git a/tests/test_litellm/test_redis_cache_task_refs.py b/tests/test_litellm/test_redis_cache_task_refs.py index 4732fd970d3..b940e05e240 100644 --- a/tests/test_litellm/test_redis_cache_task_refs.py +++ b/tests/test_litellm/test_redis_cache_task_refs.py @@ -58,14 +58,75 @@ def failing_client() -> MagicMock: has to come from the redis operation instead. """ client = MagicMock() - for op in ("get", "set", "mget", "sadd", "incrbyfloat", "expire", "rpush", "lpop", "scan_iter", "ping"): + for op in ("get", "set", "mget", "sadd", "incrbyfloat", "expire", "ttl", "rpush", "lpop", "ping"): setattr(client, op, AsyncMock(side_effect=ConnectionError("redis is down"))) + # scan_iter is not awaited, it is iterated, so it has to fail on the call + # itself rather than on an await that never happens. + client.scan_iter = MagicMock(side_effect=ConnectionError("redis is down")) + client.pipeline = MagicMock(return_value=_pipeline_cm()) return client -# Each entry drives one RedisCache method against a client that fails, which is -# the path to the service-failure hook these methods fire. -FAILING_CALLS = ( +class _AsyncIter: + """`scan_iter` is consumed with `async for`, which a plain AsyncMock is not.""" + + def __init__(self, items): + self._items = list(items) + + def __aiter__(self): + return self + + async def __anext__(self): + if not self._items: + raise StopAsyncIteration + return self._items.pop(0) + + +def _pipeline_cm() -> MagicMock: + """`pipeline(transaction=False)` is entered as an async context manager.""" + pipe = MagicMock() + pipe.execute = AsyncMock(return_value=[]) + cm = MagicMock() + cm.__aenter__ = AsyncMock(return_value=pipe) + cm.__aexit__ = AsyncMock(return_value=False) + return cm + + +def working_client() -> MagicMock: + """An async redis client whose operations all succeed. + + Return values are the benign ones (a cache miss, a zero-length list): the + success hook fires either way, and that hook is what these tests are about. + """ + client = MagicMock() + client.get = AsyncMock(return_value=None) + client.set = AsyncMock(return_value=True) + client.mget = AsyncMock(return_value=[None, None]) + client.sadd = AsyncMock(return_value=1) + client.incrbyfloat = AsyncMock(return_value=1.0) + client.expire = AsyncMock(return_value=True) + client.ttl = AsyncMock(return_value=60) + client.rpush = AsyncMock(return_value=1) + client.lpop = AsyncMock(return_value=None) + client.ping = AsyncMock(return_value=True) + client.scan_iter = MagicMock(return_value=_AsyncIter([])) + client.pipeline = MagicMock(return_value=_pipeline_cm()) + return client + + +# The pipeline methods delegate the actual redis work to a helper, so that is +# the seam to drive them from rather than the individual redis commands. +PIPELINE_HELPERS = { + "async_set_cache_pipeline": "_pipeline_helper", + "async_increment_pipeline": "_pipeline_increment_helper", + "async_rpush_pipeline": "_pipeline_rpush_helper", + "async_lpop_pipeline": "_pipeline_lpop_helper", +} + + +# Every RedisCache method that fires a service-log task, with arguments that +# get it past its own early-return guards. Both hooks are exercised per method. +CALLS = ( ("async_set_cache", ("k", "v")), ("async_get_cache", ("k",)), ("async_batch_get_cache", (("k1", "k2"),)), @@ -73,19 +134,49 @@ FAILING_CALLS = ( ("async_rpush", ("k", ("v",))), ("async_lpop", ("k",)), ("ping", ()), + ("async_scan_iter", ("pattern",)), + ("async_set_cache_sadd", ("k", ["v"], None)), + ("async_set_cache_pipeline", ([("k", "v")],)), + ("async_increment_pipeline", ([{"key": "k", "increment_value": 1.0, "ttl": 60}],)), + ("async_rpush_pipeline", ([{"key": "k", "values": ["v"]}],)), + ("async_lpop_pipeline", ([{"key": "k", "count": 1}],)), ) +CALL_IDS = tuple(n for n, _ in CALLS) + + +@contextlib.contextmanager +def driven(cache: RedisCache, method_name: str, *, failing: bool): + """Point one RedisCache method at a client that either works or breaks. + + The pipeline methods reach redis through a helper rather than through the + client's own commands, so for those the helper is the seam. + """ + client = failing_client() if failing else working_client() + stack = contextlib.ExitStack() + stack.enter_context(patch.object(cache, "init_async_client", return_value=client)) + + helper = PIPELINE_HELPERS.get(method_name) + if helper is not None: + stack.enter_context( + patch.object( + cache, + helper, + AsyncMock(side_effect=ConnectionError("redis is down") if failing else None, return_value=[]), + ) + ) + with stack: + yield @pytest.mark.asyncio -@pytest.mark.parametrize("method_name, args", FAILING_CALLS, ids=tuple(n for n, _ in FAILING_CALLS)) +@pytest.mark.parametrize("method_name, args", CALLS, ids=CALL_IDS) async def test_failure_path_holds_the_service_log_task(cache: RedisCache, method_name: str, args: tuple) -> None: - method = getattr(cache, method_name) - - with patch.object(cache, "init_async_client", return_value=failing_client()): + """The service-failure hook is fired from a task that nothing else holds.""" + with driven(cache, method_name, failing=True): # Several of these re-raise after logging and several swallow; either # way the scheduled task is what this asserts on. with contextlib.suppress(ConnectionError): - await method(*args) + await getattr(cache, method_name)(*args) assert cache._service_logging_tasks, ( f"{method_name} scheduled a service-log task without keeping a " @@ -97,14 +188,47 @@ async def test_failure_path_holds_the_service_log_task(cache: RedisCache, method @pytest.mark.asyncio -async def test_success_path_holds_the_service_log_task(cache: RedisCache) -> None: +@pytest.mark.parametrize("method_name, args", CALLS, ids=CALL_IDS) +async def test_success_path_holds_the_service_log_task(cache: RedisCache, method_name: str, args: tuple) -> None: """The success hook is fired from a task too, and needs the same reference.""" - client = MagicMock() - client.set = AsyncMock(return_value=True) + with driven(cache, method_name, failing=False): + await getattr(cache, method_name)(*args) - with patch.object(cache, "init_async_client", return_value=client): - await cache.async_set_cache("k", "v") - assert cache._service_logging_tasks + assert cache._service_logging_tasks, ( + f"{method_name} scheduled its success-log task without keeping a reference to it" + ) + + await drain(cache) + assert not cache._service_logging_tasks, f"{method_name} left its finished task in the registry" + + +# These wrap init_async_client() in their own try, so a client that cannot even +# be built is a separate logged path from a redis command that fails. +CLIENT_INIT_CALLS = ( + ("async_set_cache", ("k", "v")), + ("async_set_cache_sadd", ("k", ["v"], None)), +) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("method_name, args", CLIENT_INIT_CALLS, ids=tuple(n for n, _ in CLIENT_INIT_CALLS)) +async def test_client_init_failure_holds_the_service_log_task(cache: RedisCache, method_name: str, args: tuple) -> None: + with patch.object(cache, "init_async_client", side_effect=ConnectionError("no client")): + with contextlib.suppress(ConnectionError): + await getattr(cache, method_name)(*args) + + assert cache._service_logging_tasks, f"{method_name} logged the client-init failure from a task it did not keep" + + await drain(cache) + assert not cache._service_logging_tasks + + +@pytest.mark.asyncio +async def test_health_ping_setup_holds_its_task(cache: RedisCache) -> None: + """_setup_health_pings fires the async ping from a task of its own.""" + with patch.object(cache, "ping", AsyncMock(return_value=True)): + cache._setup_health_pings() + assert cache._service_logging_tasks, "the async health ping task is not kept anywhere" await drain(cache) assert not cache._service_logging_tasks From acb6c11b17e8067788c093a9fb5a8b0d7d402cd0 Mon Sep 17 00:00:00 2001 From: Lin Junrong Date: Mon, 10 Aug 2026 01:43:08 +0800 Subject: [PATCH 4/4] fix: mark the retained task bindings as intentional rebinds The type-discipline gate (LIT010) wants every local to carry a `Final` declaration, but the success and failure paths of a method each bind `_service_logging_task` once, and `Final` forbids that second binding. Use the gate's own `# rebind-ok:` escape hatch at those sites, and `Final` for `_health_ping_task`, which is bound only once. No behaviour changes: stripping the comments reproduces the previous file byte for byte. Co-Authored-By: Claude Opus 5 --- litellm/caching/redis_cache.py | 62 +++++++++++++++++----------------- 1 file changed, 31 insertions(+), 31 deletions(-) diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index 8191e4e5e0b..f354da35c15 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -344,7 +344,7 @@ class RedisCache(BaseCache): """Setup async and sync health pings for Redis.""" # ASYNC HEALTH PING try: - _health_ping_task = asyncio.get_running_loop().create_task(self.ping()) + _health_ping_task: Final = asyncio.get_running_loop().create_task(self.ping()) self._service_logging_tasks.add(_health_ping_task) _health_ping_task.add_done_callback(self._service_logging_tasks.discard) except Exception as e: @@ -372,7 +372,7 @@ class RedisCache(BaseCache): loop: Final = asyncio.get_running_loop() start_time: Final = time.time() end_time: Final = start_time - _service_logging_task = loop.create_task( + _service_logging_task = loop.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=end_time - start_time, @@ -391,7 +391,7 @@ class RedisCache(BaseCache): loop: Final = asyncio.get_running_loop() start_time: Final = time.time() end_time: Final = start_time - _service_logging_task = loop.create_task( + _service_logging_task = loop.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=end_time - start_time, @@ -573,7 +573,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -590,7 +590,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -698,7 +698,7 @@ class RedisCache(BaseCache): except Exception as e: end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -736,7 +736,7 @@ class RedisCache(BaseCache): print_verbose(f"Successfully Set ASYNC Redis Cache: key: {key}\nValue {value}\nttl={ttl}") end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -753,7 +753,7 @@ class RedisCache(BaseCache): except Exception as e: end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -825,7 +825,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -842,7 +842,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -890,7 +890,7 @@ class RedisCache(BaseCache): except Exception as e: end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -918,7 +918,7 @@ class RedisCache(BaseCache): print_verbose(f"Successfully Set ASYNC Redis Cache SADD: key: {key}\nValue {value}\nttl={ttl}") end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -933,7 +933,7 @@ class RedisCache(BaseCache): except Exception as e: end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -993,7 +993,7 @@ class RedisCache(BaseCache): end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1010,7 +1010,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1193,7 +1193,7 @@ class RedisCache(BaseCache): end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1210,7 +1210,7 @@ class RedisCache(BaseCache): except Exception as e: end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1258,7 +1258,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1287,7 +1287,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1348,7 +1348,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1363,7 +1363,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1506,7 +1506,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1523,7 +1523,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1595,7 +1595,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1610,7 +1610,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1668,7 +1668,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1682,7 +1682,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1737,7 +1737,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1764,7 +1764,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1865,7 +1865,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1879,7 +1879,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - _service_logging_task = asyncio.create_task( + _service_logging_task = asyncio.create_task( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration,