mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-11 03:38:38 +00:00
fix(cache): keep the redis-tier ttl fix inside every lint budget
CI feedback on this branch, and what it turned out to be: - basedpyright budget: the per-tier `dict(kwargs)` copies re-wrote each call site as `**per_tier_kwargs`, which is an unknown-typed argument (`reportUnknownArgumentType` went 44804 -> 44811 against a 44802 limit). Note whether the caller supplied a ttl with a `ttl_given` flag instead and keep the base's `**kwargs` call shape, so the rule adds nothing. - tests/unit/caching/test_dual_cache.py: build the regression helpers without `**kwargs` for the same reason, and rename the local ttl- recording redis mock to `_ttl_recording_redis` -- it shadowed the `_recording_redis(values)` helper already defined further down the file, which would have broken every `test_shared_batch_read_*` case. - tests/unit/caching/test_request_redis_batch_post_call.py: drop the two cases that called `async_set_cache_post_call`; DualCache has no such method (`AttributeError`, 2 failed / 566 passed in caching-local). The post-call batch path is covered by the existing #43779 test, and the new regression for #43187 moves to the pre-call file beside it. - proxy_server: revert the `dict(cache_params)` filter from the previous commit. `Cache.__init__` ends in `**kwargs`, which it forwards to the backends, so the base already tolerated `default_redis_ttl`; the filter was defensive scope creep that added the last unknown-typed argument. Refs #43187 Signed-off-by: apex-mochen <2756823972@qq.com>
This commit is contained in:
apex-mochen
parent
372e6c5334
commit
aab86f42ee
5 changed files with 98 additions and 93 deletions
|
|
@ -150,22 +150,24 @@ class DualCache(BaseCache):
|
|||
def set_cache(self, key, value, local_only: bool = False, **kwargs):
|
||||
# Update both Redis and in-memory cache
|
||||
try:
|
||||
# ``kwargs`` is this call's own dict, so injecting a ttl into it is safe. What must not
|
||||
# happen is one tier's injected ttl reaching the other, so note whether the caller
|
||||
# supplied one before either tier writes back.
|
||||
ttl_given: Final = "ttl" in kwargs
|
||||
if self.in_memory_cache is not None:
|
||||
mem_kwargs = dict(kwargs) # mutable-ok: per-tier copy; tiers must not share a ttl
|
||||
if "ttl" not in mem_kwargs and self.default_in_memory_ttl is not None:
|
||||
mem_kwargs["ttl"] = self.default_in_memory_ttl
|
||||
if not ttl_given and self.default_in_memory_ttl is not None:
|
||||
kwargs["ttl"] = self.default_in_memory_ttl
|
||||
|
||||
self.in_memory_cache.set_cache(key, value, **mem_kwargs)
|
||||
self.in_memory_cache.set_cache(key, value, **kwargs)
|
||||
|
||||
if self.redis_cache is not None and local_only is False:
|
||||
redis_kwargs = dict(kwargs) # mutable-ok: per-tier copy; tiers must not share a ttl
|
||||
if "ttl" not in redis_kwargs:
|
||||
if not ttl_given:
|
||||
redis_ttl = self.default_redis_ttl
|
||||
if redis_ttl is None:
|
||||
redis_ttl = self.default_in_memory_ttl
|
||||
if redis_ttl is not None:
|
||||
redis_kwargs["ttl"] = redis_ttl
|
||||
self.redis_cache.set_cache(key, value, **redis_kwargs)
|
||||
kwargs["ttl"] = redis_ttl
|
||||
self.redis_cache.set_cache(key, value, **kwargs)
|
||||
except Exception as e:
|
||||
print_verbose(e)
|
||||
|
||||
|
|
@ -513,21 +515,20 @@ class DualCache(BaseCache):
|
|||
async def async_set_cache(self, key, value, local_only: bool = False, **kwargs):
|
||||
print_verbose(f"async set cache: cache key: {key}; local_only: {local_only}; value: {value}")
|
||||
try:
|
||||
ttl_given: Final = "ttl" in kwargs
|
||||
if self.in_memory_cache is not None:
|
||||
mem_kwargs = dict(kwargs) # mutable-ok: per-tier copy; tiers must not share a ttl
|
||||
if "ttl" not in mem_kwargs and self.default_in_memory_ttl is not None:
|
||||
mem_kwargs["ttl"] = self.default_in_memory_ttl
|
||||
await self.in_memory_cache.async_set_cache(key, value, **mem_kwargs)
|
||||
if not ttl_given and self.default_in_memory_ttl is not None:
|
||||
kwargs["ttl"] = self.default_in_memory_ttl
|
||||
await self.in_memory_cache.async_set_cache(key, value, **kwargs)
|
||||
|
||||
if self.redis_cache is not None and local_only is False:
|
||||
redis_kwargs = dict(kwargs) # mutable-ok: per-tier copy; tiers must not share a ttl
|
||||
if "ttl" not in redis_kwargs:
|
||||
if not ttl_given:
|
||||
redis_ttl = self.default_redis_ttl
|
||||
if redis_ttl is None:
|
||||
redis_ttl = self.default_in_memory_ttl
|
||||
if redis_ttl is not None:
|
||||
redis_kwargs["ttl"] = redis_ttl
|
||||
await self.redis_cache.async_set_cache(key, value, **redis_kwargs)
|
||||
kwargs["ttl"] = redis_ttl
|
||||
await self.redis_cache.async_set_cache(key, value, **kwargs)
|
||||
except Exception as e:
|
||||
log_redis_failure(
|
||||
verbose_logger, logging.ERROR, "LiteLLM Cache: exception in async add_cache", e, with_traceback=True
|
||||
|
|
@ -567,22 +568,21 @@ class DualCache(BaseCache):
|
|||
"""
|
||||
print_verbose(f"async batch set cache: cache keys: {cache_list}; local_only: {local_only}")
|
||||
try:
|
||||
ttl_given: Final = "ttl" in kwargs
|
||||
if self.in_memory_cache is not None:
|
||||
mem_kwargs = dict(kwargs) # mutable-ok: per-tier copy; tiers must not share a ttl
|
||||
if "ttl" not in mem_kwargs and self.default_in_memory_ttl is not None:
|
||||
mem_kwargs["ttl"] = self.default_in_memory_ttl
|
||||
await self.in_memory_cache.async_set_cache_pipeline(cache_list=cache_list, **mem_kwargs)
|
||||
if not ttl_given and self.default_in_memory_ttl is not None:
|
||||
kwargs["ttl"] = self.default_in_memory_ttl
|
||||
await self.in_memory_cache.async_set_cache_pipeline(cache_list=cache_list, **kwargs)
|
||||
|
||||
if self.redis_cache is not None and local_only is False:
|
||||
redis_kwargs = dict(kwargs) # mutable-ok: per-tier copy; tiers must not share a ttl
|
||||
if "ttl" not in redis_kwargs:
|
||||
if not ttl_given:
|
||||
redis_ttl = self.default_redis_ttl
|
||||
if redis_ttl is None:
|
||||
redis_ttl = self.default_in_memory_ttl
|
||||
if redis_ttl is not None:
|
||||
redis_kwargs["ttl"] = redis_ttl
|
||||
kwargs["ttl"] = redis_ttl
|
||||
await self.redis_cache.async_set_cache_pipeline(
|
||||
cache_list=cache_list, ttl=redis_kwargs.pop("ttl", None), **redis_kwargs
|
||||
cache_list=cache_list, ttl=kwargs.pop("ttl", None), **kwargs
|
||||
)
|
||||
except Exception as e:
|
||||
log_redis_failure(
|
||||
|
|
|
|||
|
|
@ -5971,18 +5971,9 @@ class ProxyConfig:
|
|||
litellm.default_in_memory_ttl = cache_params["default_in_memory_ttl"]
|
||||
|
||||
if "default_redis_ttl" in cache_params:
|
||||
# default_redis_ttl is a DualCache/global setting, not a redis-py
|
||||
# Redis() constructor kwarg. Promote it to the global and filter it
|
||||
# out when constructing Cache (do NOT pop the caller's dict: the
|
||||
# caller snapshots cache_params to detect DB reloads and a mutation
|
||||
# here would force a cache rebuild on every poll).
|
||||
litellm.default_redis_ttl = cache_params["default_redis_ttl"]
|
||||
|
||||
# Copy first: the caller snapshots cache_params to detect DB reloads, so
|
||||
# mutating it here would force a cache rebuild on every poll.
|
||||
cache_kwargs = dict(cache_params) # mutable-ok: local filter; caller's dict is left untouched
|
||||
cache_kwargs.pop("default_redis_ttl", None)
|
||||
litellm.cache = Cache(**cache_kwargs)
|
||||
litellm.cache = Cache(**cache_params)
|
||||
|
||||
resolved_usage_cache = redis_usage_cache
|
||||
cache_backend: Final = litellm.cache.cache if litellm.cache is not None else None
|
||||
|
|
|
|||
|
|
@ -1023,43 +1023,81 @@ async def test_async_batch_reads_of_missing_keys_hit_redis_once_per_expiry_windo
|
|||
await dual_cache.async_batch_get_cache(keys)
|
||||
assert redis_cache.async_batch_get_cache.await_count == 2
|
||||
|
||||
async def _write_through(dual_cache: DualCache, write_path: str, **kwargs) -> None:
|
||||
async def _write_without_a_ttl(dual_cache: DualCache, write_path: str) -> None:
|
||||
"""One of the three write paths, called the way a caller that passes no ``ttl`` calls it."""
|
||||
if write_path == "set_cache":
|
||||
dual_cache.set_cache("ttl_key", "v", **kwargs)
|
||||
dual_cache.set_cache("ttl_key", "v")
|
||||
elif write_path == "async_set_cache":
|
||||
await dual_cache.async_set_cache("ttl_key", "v", **kwargs)
|
||||
await dual_cache.async_set_cache("ttl_key", "v")
|
||||
else:
|
||||
await dual_cache.async_set_cache_pipeline([("ttl_key", "v")], **kwargs)
|
||||
await dual_cache.async_set_cache_pipeline([("ttl_key", "v")])
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("write_path", ["set_cache", "async_set_cache", "async_set_cache_pipeline"])
|
||||
@pytest.mark.parametrize(
|
||||
("kwargs", "memory_ttl", "redis_ttl"),
|
||||
[({}, 60, 3600), ({"ttl": 99}, 99, 99)],
|
||||
ids=["tier_defaults", "explicit_ttl"],
|
||||
)
|
||||
async def test_dual_cache_writes_each_tier_with_its_own_default_ttl(write_path, kwargs, memory_ttl, redis_ttl):
|
||||
"""
|
||||
Regression for #43187: the Redis tier was written with default_in_memory_ttl,
|
||||
so a configured default_redis_ttl never took effect. An explicit ttl still
|
||||
reaches both tiers unchanged.
|
||||
"""
|
||||
in_memory_cache = InMemoryCache(default_ttl=600)
|
||||
async def _write_with_a_ttl(dual_cache: DualCache, write_path: str, ttl: float) -> None:
|
||||
"""The same three paths with an explicit ``ttl``, which must reach both tiers unchanged."""
|
||||
if write_path == "set_cache":
|
||||
dual_cache.set_cache("ttl_key", "v", ttl=ttl)
|
||||
elif write_path == "async_set_cache":
|
||||
await dual_cache.async_set_cache("ttl_key", "v", ttl=ttl)
|
||||
else:
|
||||
await dual_cache.async_set_cache_pipeline([("ttl_key", "v")], ttl=ttl)
|
||||
|
||||
|
||||
def _ttl_recording_redis() -> MagicMock:
|
||||
"""A Redis tier that records what it was written with, whichever of the two writes it uses."""
|
||||
mock_redis = MagicMock()
|
||||
mock_redis.async_set_cache = AsyncMock()
|
||||
mock_redis.async_set_cache_pipeline = AsyncMock()
|
||||
return mock_redis
|
||||
|
||||
|
||||
def _written_redis_ttl(mock_redis: MagicMock, write_path: str) -> object:
|
||||
"""The ttl the recording Redis tier was written with, on whichever of its writes that path uses."""
|
||||
return getattr(mock_redis, write_path).call_args.kwargs["ttl"]
|
||||
|
||||
|
||||
async def _tier_ttls(
|
||||
write_path: str, ttl: float | None
|
||||
) -> tuple[object, float]:
|
||||
"""Write one key through one path and report the Redis ttl it was written with, plus the
|
||||
in-memory expiry. ``ttl=None`` means the caller supplied no ``ttl`` at all."""
|
||||
in_memory_cache = InMemoryCache(default_ttl=600)
|
||||
mock_redis = _ttl_recording_redis()
|
||||
dual_cache = DualCache(
|
||||
in_memory_cache=in_memory_cache,
|
||||
redis_cache=mock_redis,
|
||||
default_in_memory_ttl=60,
|
||||
default_redis_ttl=3600,
|
||||
)
|
||||
if ttl is None:
|
||||
await _write_without_a_ttl(dual_cache, write_path)
|
||||
else:
|
||||
await _write_with_a_ttl(dual_cache, write_path, ttl)
|
||||
return _written_redis_ttl(mock_redis, write_path), in_memory_cache.ttl_dict["ttl_key"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("write_path", ["set_cache", "async_set_cache", "async_set_cache_pipeline"])
|
||||
async def test_dual_cache_writes_the_redis_tier_with_the_configured_default_redis_ttl(write_path: str):
|
||||
"""
|
||||
Regression for #43187: the Redis tier was written with default_in_memory_ttl, so a configured
|
||||
default_redis_ttl never took effect. The in-memory tier still takes its own default.
|
||||
"""
|
||||
before = time.time()
|
||||
await _write_through(dual_cache, write_path, **kwargs)
|
||||
written, expiry = await _tier_ttls(write_path, None)
|
||||
after = time.time()
|
||||
|
||||
assert getattr(mock_redis, write_path).call_args.kwargs["ttl"] == redis_ttl
|
||||
expiry = in_memory_cache.ttl_dict["ttl_key"]
|
||||
assert before + memory_ttl <= expiry <= after + memory_ttl
|
||||
assert written == 3600
|
||||
assert before + 60 <= expiry <= after + 60
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("write_path", ["set_cache", "async_set_cache", "async_set_cache_pipeline"])
|
||||
async def test_dual_cache_gives_an_explicit_ttl_to_both_tiers_unchanged(write_path: str):
|
||||
"""An explicit ``ttl`` still overrides both tier defaults rather than being replaced by them."""
|
||||
before = time.time()
|
||||
written, expiry = await _tier_ttls(write_path, 99.0)
|
||||
after = time.time()
|
||||
|
||||
assert written == 99.0
|
||||
assert before + 99 <= expiry <= after + 99
|
||||
|
|
|
|||
|
|
@ -278,43 +278,6 @@ async def test_a_slot_released_before_the_response_reaches_redis_at_once_not_on_
|
|||
assert client.pipelines == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_deferred_response_cache_set_without_a_ttl_expires_in_redis_like_the_direct_path():
|
||||
client = FakeClient(_ok_replies)
|
||||
redis_cache = PostCallFakeRedisCache(client)
|
||||
dual_cache = DualCache(redis_cache=redis_cache, in_memory_cache=InMemoryCache(), default_in_memory_ttl=300)
|
||||
|
||||
await dual_cache.async_set_cache("direct", {"id": "resp"})
|
||||
with request_redis_batch_scope():
|
||||
await dual_cache.async_set_cache_post_call("deferred", {"id": "resp"}, None)
|
||||
await flush_post_call_redis_batches()
|
||||
|
||||
(command,) = client.pipelines[0].commands
|
||||
assert (command[0], command[1], command[3]) == ("SET", "deferred", redis_cache.alone[0][2]["ttl"])
|
||||
assert command[3] == 300
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_deferred_response_cache_set_without_a_ttl_uses_default_redis_ttl_when_configured():
|
||||
client = FakeClient(_ok_replies)
|
||||
redis_cache = PostCallFakeRedisCache(client)
|
||||
dual_cache = DualCache(
|
||||
redis_cache=redis_cache,
|
||||
in_memory_cache=InMemoryCache(),
|
||||
default_in_memory_ttl=300,
|
||||
default_redis_ttl=3600,
|
||||
)
|
||||
|
||||
await dual_cache.async_set_cache("direct", {"id": "resp"})
|
||||
with request_redis_batch_scope():
|
||||
await dual_cache.async_set_cache_post_call("deferred", {"id": "resp"}, None)
|
||||
await flush_post_call_redis_batches()
|
||||
|
||||
(command,) = client.pipelines[0].commands
|
||||
assert (command[0], command[1], command[3]) == ("SET", "deferred", redis_cache.alone[0][2]["ttl"])
|
||||
assert command[3] == 3600
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_released_slot_is_free_locally_at_once_and_the_older_redis_count_does_not_overwrite_the_gauge():
|
||||
def replies(command: tuple[object, ...]) -> object:
|
||||
|
|
|
|||
|
|
@ -1024,6 +1024,19 @@ async def test_a_pipelined_management_write_without_a_ttl_expires_in_redis_like_
|
|||
assert [(c[0], c[1], c[3]) for c in client.pipelines[0].commands] == [("SET", "team_id:t1", 5)]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_pipelined_write_without_a_ttl_uses_the_configured_default_redis_ttl():
|
||||
"""Regression for #43187: the pipelined Redis SET was written with default_in_memory_ttl, so a
|
||||
configured default_redis_ttl never reached it."""
|
||||
client = FakeClient(_lua_ok_replies)
|
||||
redis_cache = FakeRedisCache(client)
|
||||
cache = DualCache(redis_cache=redis_cache, default_in_memory_ttl=5, default_redis_ttl=3600)
|
||||
with request_redis_batch_scope() as request:
|
||||
await cache.async_set_cache_pre_call("k1", {"v": 1}, None)
|
||||
await request.flush_all()
|
||||
assert [(c[0], c[1], c[3]) for c in client.pipelines[0].commands] == [("SET", "k1", 3600)]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_identity_prefetch_is_one_mget_after_which_hits_and_misses_alike_cost_no_read():
|
||||
client = FakeClient(replies)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue