diff --git a/litellm/caching/redis_cache.py b/litellm/caching/redis_cache.py index f1c80eaacbe..dc11f520e9a 100644 --- a/litellm/caching/redis_cache.py +++ b/litellm/caching/redis_cache.py @@ -305,6 +305,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) @@ -344,7 +348,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: 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: if "no running event loop" in str(e): verbose_logger.debug("Ignoring async redis ping. No running event loop.") @@ -370,7 +376,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=end_time - start_time, @@ -378,6 +384,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 @@ -387,7 +395,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=end_time - start_time, @@ -395,6 +403,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 @@ -566,7 +576,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, @@ -575,13 +585,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -591,6 +603,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]]: @@ -699,7 +713,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -710,6 +724,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), @@ -735,7 +751,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -746,11 +762,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -762,6 +780,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), @@ -822,7 +842,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, @@ -832,12 +852,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -848,6 +870,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", @@ -883,7 +907,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -894,6 +918,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", @@ -909,7 +935,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_success_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -919,10 +945,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -933,6 +961,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", @@ -980,7 +1010,7 @@ class RedisCache(BaseCache): end_time = time.time() _duration = end_time - start_time - 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, @@ -990,12 +1020,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1006,6 +1038,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), @@ -1176,7 +1210,7 @@ class RedisCache(BaseCache): end_time = time.time() _duration = end_time - start_time - 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, @@ -1187,11 +1221,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1203,6 +1239,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) @@ -1237,7 +1275,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, @@ -1247,6 +1285,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'. @@ -1264,7 +1304,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, @@ -1275,6 +1315,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 @@ -1323,20 +1365,22 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, 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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1344,6 +1388,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 @@ -1477,7 +1523,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, @@ -1487,12 +1533,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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1503,6 +1551,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), @@ -1562,20 +1612,22 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, 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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1583,6 +1635,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 @@ -1631,19 +1685,21 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, 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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1651,6 +1707,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), @@ -1696,13 +1754,15 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, 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): @@ -1721,7 +1781,7 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, @@ -1729,6 +1789,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 @@ -1820,19 +1882,21 @@ class RedisCache(BaseCache): ## LOGGING ## end_time = time.time() _duration = end_time - start_time - 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, 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( # rebind-ok: one log task per exit path self.service_logger_obj.async_service_failure_hook( service=ServiceTypes.REDIS, duration=_duration, @@ -1840,6 +1904,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), 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..b940e05e240 --- /dev/null +++ b/tests/test_litellm/test_redis_cache_task_refs.py @@ -0,0 +1,255 @@ +"""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", "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 + + +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"),)), + ("async_increment", ("k", 1.0)), + ("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", CALLS, ids=CALL_IDS) +async def test_failure_path_holds_the_service_log_task(cache: RedisCache, method_name: str, args: tuple) -> None: + """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 getattr(cache, method_name)(*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 +@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.""" + with driven(cache, method_name, failing=False): + await getattr(cache, method_name)(*args) + + 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 + + +@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