mirror of
https://github.com/BerriAI/litellm.git
synced 2026-11-14 09:03:16 +00:00
fix(router): keep async_increment_count inside the type discipline gate
This commit is contained in:
parent
b0aac7f513
commit
e7c153ee72
4 changed files with 28 additions and 16 deletions
|
|
@ -590,14 +590,15 @@ class DualCache(BaseCache):
|
|||
) -> int | None:
|
||||
"""``async_increment_cache`` for whole-number counters: Redis gets INCRBY, so the key holds an
|
||||
integer. Returns the new count, the in-memory one when Redis fails, or None with no backend."""
|
||||
result: int | None = None
|
||||
in_memory: Final = (
|
||||
None
|
||||
if self.in_memory_cache is None
|
||||
else int(await self.in_memory_cache.async_increment(key, value, ttl=ttl))
|
||||
)
|
||||
if self.redis_cache is None or local_only:
|
||||
return in_memory
|
||||
try:
|
||||
if self.in_memory_cache is not None:
|
||||
result = int(await self.in_memory_cache.async_increment(key, value, ttl=ttl))
|
||||
if self.redis_cache is not None and local_only is False:
|
||||
result = await self.redis_cache.async_increment_count(
|
||||
key, value, ttl=ttl, parent_otel_span=parent_otel_span
|
||||
)
|
||||
return await self.redis_cache.async_increment_count(key, value, ttl=ttl, parent_otel_span=parent_otel_span)
|
||||
except Exception as e: # noqa: BLE001 # same tolerance as async_increment_cache: a Redis failure keeps the in-memory count
|
||||
log_redis_failure(
|
||||
verbose_logger,
|
||||
|
|
@ -605,7 +606,7 @@ class DualCache(BaseCache):
|
|||
"Redis async_increment_count failed, falling back to in-memory result",
|
||||
e,
|
||||
)
|
||||
return result
|
||||
return in_memory
|
||||
|
||||
async def async_increment_cache_post_call(
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -1542,15 +1542,15 @@ class RedisCache(BaseCache):
|
|||
_redis_client: Final = self.init_async_client()
|
||||
start_time: Final = time.time()
|
||||
_used_ttl: Final = self.get_ttl(ttl=ttl)
|
||||
key = self.check_and_fix_namespace(key=key)
|
||||
namespaced_key: Final = self.check_and_fix_namespace(key=key)
|
||||
try:
|
||||
async with _redis_client.pipeline(transaction=False) as pipe:
|
||||
pipe.incrby(name=key, amount=value)
|
||||
pipe.ttl(key)
|
||||
pipe.incrby(name=namespaced_key, amount=value)
|
||||
pipe.ttl(namespaced_key)
|
||||
result, current_ttl = await pipe.execute()
|
||||
if _used_ttl is not None and current_ttl == -1:
|
||||
await _redis_client.expire(key, _used_ttl)
|
||||
end_time = time.time()
|
||||
await _redis_client.expire(namespaced_key, _used_ttl)
|
||||
end_time: Final = time.time()
|
||||
asyncio.create_task(
|
||||
self.service_logger_obj.async_service_success_hook(
|
||||
service=ServiceTypes.REDIS,
|
||||
|
|
@ -1564,16 +1564,16 @@ class RedisCache(BaseCache):
|
|||
)
|
||||
return int(result)
|
||||
except Exception as e:
|
||||
end_time = time.time()
|
||||
failed_at: Final = time.time()
|
||||
asyncio.create_task(
|
||||
self.service_logger_obj.async_service_failure_hook(
|
||||
service=ServiceTypes.REDIS,
|
||||
duration=end_time - start_time,
|
||||
duration=failed_at - start_time,
|
||||
error=e,
|
||||
call_type="async_increment_count",
|
||||
caller=_get_call_stack_info(),
|
||||
start_time=start_time,
|
||||
end_time=end_time,
|
||||
end_time=failed_at,
|
||||
parent_otel_span=parent_otel_span,
|
||||
)
|
||||
)
|
||||
|
|
|
|||
|
|
@ -1052,3 +1052,4 @@ async def test_async_increment_count_falls_back_to_the_in_memory_count_when_redi
|
|||
assert await dual_cache.async_increment_count("deployment:d1:allowed_fails", 1, ttl=60.0) == 1
|
||||
assert await dual_cache.async_increment_count("deployment:d1:allowed_fails", 1, local_only=True) == 2
|
||||
redis_cache.async_increment_count.assert_awaited_once()
|
||||
|
||||
|
|
|
|||
|
|
@ -1758,3 +1758,13 @@ async def test_async_increment_count_uses_incrby_and_sets_the_ttl_only_on_a_new_
|
|||
assert all(type(count) is int for count in (first, second))
|
||||
assert [command for command, _, _ in client.commands] == ["incrby", "ttl", "expire", "incrby", "ttl"]
|
||||
assert client.ttls == {"deployment:d1:allowed_fails": 60}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_async_increment_count_reraises_a_redis_failure(monkeypatch, redis_no_ping):
|
||||
monkeypatch.setenv("REDIS_HOST", "https://my-test-host")
|
||||
redis_cache: Final = RedisCache()
|
||||
client: Final = MagicMock()
|
||||
client.pipeline.side_effect = ConnectionError("redis down")
|
||||
with patch.object(redis_cache, "init_async_client", return_value=client), pytest.raises(ConnectionError):
|
||||
await redis_cache.async_increment_count("deployment:d1:allowed_fails", 1, ttl=60.0)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue