From 06e8013e6c8b084a241f60cf3c5e4de9645c1ff8 Mon Sep 17 00:00:00 2001 From: Yassin Kortam Date: Thu, 16 Jul 2026 01:28:37 +0300 Subject: [PATCH] fix(logging): stop pinning large request payloads past request end (#33455) Three process-lifetime retention points kept full request payloads (messages included) alive after the request finished. Under bursts of large-token traffic (~73K tokens/request mean) this presented as stepwise RSS growth that never returned to baseline, ending in OOM: 1. Logging.pre_call/post_call stored their entire locals() (messages, the Logging object, complete_input_dict) in the module-level litellm.error_logs dict, pinning the most recent request's payload per worker forever. Nothing reads that dict; the writes are removed. 2. LLMCachingHandler.request_kwargs kept litellm_logging_obj inside the stored kwargs while the handler itself hangs off logging_obj._llm_caching_handler, closing a reference cycle (Logging -> LLMCachingHandler -> kwargs -> Logging). Cyclic payloads are only reclaimed by generational GC, so megabytes of dead request data lingered until a rare gen-2 pass, and the transient copies fragment the allocator into a permanent RSS high-water mark. The handler now drops litellm_logging_obj from its stored kwargs; the caching layer never reads it. 3. The router stored every request's kwargs in the ITPM/OTPM contextvar even when no deployment configures itpm/otpm. Pooled resources created mid-request (e.g. redis connections) capture the asyncio context, extending that pin far past the request. The slot is now populated only for deployments with io token limits and overwritten with None otherwise. Live-proxy verification (bursts of 30 x ~300KB requests, PII guardrail + prometheus + redis cache): unfixed grows 16-29MB per burst without release; fixed grows under 1MB per burst after warmup and flattens. Resolves LIT-4434 --- litellm/caching/caching_handler.py | 22 ++++++- litellm/litellm_core_utils/litellm_logging.py | 2 - litellm/router.py | 2 +- .../io_token_rate_limit_check.py | 11 +++- .../caching/test_caching_handler.py | 28 +++++++++ .../test_litellm_logging.py | 16 +++++ .../test_router/test_io_token_rate_limits.py | 62 +++++++++++++++++++ 7 files changed, 135 insertions(+), 8 deletions(-) diff --git a/litellm/caching/caching_handler.py b/litellm/caching/caching_handler.py index c860f8e540d..b17e055c7ea 100644 --- a/litellm/caching/caching_handler.py +++ b/litellm/caching/caching_handler.py @@ -85,6 +85,22 @@ class CachingHandlerResponse(BaseModel): in_memory_cache_obj = InMemoryCache() +def _drop_logging_obj_from_kwargs(request_kwargs: dict[str, object]) -> dict[str, object]: + """ + The caching handler is stored on the Logging object + (``logging_obj._llm_caching_handler``), so keeping ``litellm_logging_obj`` + inside ``request_kwargs`` closes a reference cycle + (Logging -> LLMCachingHandler -> kwargs -> Logging) that keeps the full + request payload (messages included) alive until a generational GC pass + instead of being freed by refcount when the request ends. Nothing in the + caching layer reads the logging object from these kwargs; cache-key + generation ignores litellm-internal params. + """ + if "litellm_logging_obj" not in request_kwargs: + return request_kwargs + return {k: v for k, v in request_kwargs.items() if k != "litellm_logging_obj"} + + def _is_chat_completion_cached_dict(cached_result: dict) -> bool: cached_id = cached_result.get("id") if isinstance(cached_id, str) and cached_id.startswith("chatcmpl"): @@ -118,7 +134,7 @@ class LLMCachingHandler: self.async_streaming_chunks: List[ModelResponse] = [] self.sync_streaming_chunks: List[ModelResponse] = [] - self.request_kwargs = request_kwargs + self.request_kwargs = _drop_logging_obj_from_kwargs(request_kwargs) self.preset_cache_key: Optional[str] = None self.original_function = original_function self.start_time = start_time @@ -297,7 +313,7 @@ class LLMCachingHandler: new_kwargs.pop("metadata", None) if new_kwargs.get("stream") is True and "cache_key" not in new_kwargs: new_kwargs["cache_key"] = litellm.cache.get_cache_key(**new_kwargs) - self.request_kwargs = new_kwargs + self.request_kwargs = _drop_logging_obj_from_kwargs(new_kwargs) print_verbose("Checking Sync Cache") cached_result = litellm.cache.get_cache(**new_kwargs) if cached_result is not None: @@ -693,7 +709,7 @@ class LLMCachingHandler: new_kwargs.pop("metadata", None) if new_kwargs.get("stream") is True and "cache_key" not in new_kwargs: new_kwargs["cache_key"] = litellm.cache.get_cache_key(**new_kwargs) - self.request_kwargs = new_kwargs + self.request_kwargs = _drop_logging_obj_from_kwargs(new_kwargs) cached_result: Optional[Any] = None if call_type == CallTypes.aembedding.value: if isinstance(new_kwargs["input"], str): diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index 5494ea11b36..268dcc3df78 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -925,7 +925,6 @@ class Logging(LiteLLMLoggingBaseClass): def pre_call(self, input, api_key, model=None, additional_args={}): # Log the exact input to the LLM API - litellm.error_logs["PRE_CALL"] = locals() try: self._pre_call( input=input, @@ -1135,7 +1134,6 @@ class Logging(LiteLLMLoggingBaseClass): def post_call(self, original_response, input=None, api_key=None, additional_args={}): # Log the exact result from the LLM API, for streaming - log the type of response received - litellm.error_logs["POST_CALL"] = locals() if isinstance(original_response, dict): original_response = json.dumps(original_response, default=str) try: diff --git a/litellm/router.py b/litellm/router.py index 46794ac7a24..f408d030b8b 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -2986,7 +2986,7 @@ class Router: # here before it's wiped below, instead of relying on that attempt's # (possibly still-pending) failure event to do it. refund_stale_reservation_before_retry(self.cache, kwargs) - set_io_token_rate_limit_request_kwargs(kwargs) + set_io_token_rate_limit_request_kwargs(kwargs, store_in_context=deployment_has_io_token_limits(deployment)) ## DEPLOYMENT-LEVEL TAGS deployment_tags = deployment.get("litellm_params", {}).get("tags") diff --git a/litellm/router_utils/pre_call_checks/io_token_rate_limit_check.py b/litellm/router_utils/pre_call_checks/io_token_rate_limit_check.py index 62c99a1c9f6..803fdc4b353 100644 --- a/litellm/router_utils/pre_call_checks/io_token_rate_limit_check.py +++ b/litellm/router_utils/pre_call_checks/io_token_rate_limit_check.py @@ -43,14 +43,21 @@ ITPM_CACHE_KEY = "_litellm_itpm_cache_key" OTPM_CACHE_KEY = "_litellm_otpm_cache_key" -def set_io_token_rate_limit_request_kwargs(kwargs: Optional[dict[str, Any]]) -> None: +def set_io_token_rate_limit_request_kwargs(kwargs: Optional[dict[str, Any]], store_in_context: bool = True) -> None: # The reservation sentinels are server-only, but `metadata` is caller # controlled on proxy requests. Strip any client-supplied copies here (this # runs before the router stashes its own reservation) so a forged # reservation can't drive the post-call reconcile/refund against an # arbitrary counter and bypass the configured limits. _clear_reservation_from_kwargs(kwargs) - _io_token_rate_limit_request_kwargs.set(kwargs) + # The context slot pins the entire request kwargs (messages included) for + # the lifetime of the surrounding context, which outlives the request when + # the context is captured by pooled resources (e.g. a redis connection + # created mid-request). Only ITPM/OTPM-limited deployments read it, so for + # every other deployment overwrite the slot with None instead of the + # kwargs; overwriting (rather than skipping) also releases a previous + # request's kwargs when a context is reused. + _io_token_rate_limit_request_kwargs.set(kwargs if store_in_context else None) def get_io_token_rate_limit_request_kwargs() -> Optional[dict[str, Any]]: diff --git a/tests/test_litellm/caching/test_caching_handler.py b/tests/test_litellm/caching/test_caching_handler.py index 01327529410..1136a0b7e7b 100644 --- a/tests/test_litellm/caching/test_caching_handler.py +++ b/tests/test_litellm/caching/test_caching_handler.py @@ -556,3 +556,31 @@ async def test_embedding_cache_falls_back_to_token_counter_for_legacy_entries(): assert cache_hit # token_counter over "hello world" yields a nonzero count — fallback path still runs assert response.usage.prompt_tokens > 0 + + +def test_request_kwargs_does_not_retain_logging_obj(): + """ + The caching handler lives on logging_obj._llm_caching_handler, so keeping + litellm_logging_obj inside request_kwargs closes a reference cycle + (Logging -> LLMCachingHandler -> kwargs -> Logging). That cycle keeps the + full request payload alive until a generational GC pass instead of being + freed by refcount when the request finishes; under bursts of large-token + requests this presents as stepwise RSS growth that never returns to + baseline. Other kwargs (messages included) must be preserved. + """ + logging_obj = MagicMock() + kwargs = { + "model": "gpt-4o", + "messages": [{"role": "user", "content": "hello"}], + "litellm_logging_obj": logging_obj, + } + + handler = LLMCachingHandler( + original_function=MagicMock(), + request_kwargs=kwargs, + start_time=datetime.now(), + ) + + assert "litellm_logging_obj" not in handler.request_kwargs + assert handler.request_kwargs["messages"] == kwargs["messages"] + assert handler.request_kwargs["model"] == "gpt-4o" diff --git a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py index 34a87c95611..f99cfb953ba 100644 --- a/tests/test_litellm/litellm_core_utils/test_litellm_logging.py +++ b/tests/test_litellm/litellm_core_utils/test_litellm_logging.py @@ -3780,3 +3780,19 @@ def test_zero_token_video_usage_preserves_duration_seconds(logging_obj): assert payload["metadata"]["usage_object"]["duration_seconds"] == 4.0 assert payload["total_tokens"] == 0 assert payload["completion_tokens"] == 0 + + +def test_pre_call_does_not_pin_request_in_module_state(logging_obj): + """ + pre_call/post_call must not stash their locals (full messages, the Logging + object, complete_input_dict) into module-level state. That pinned the most + recent request's entire payload in memory for the life of the worker, + which with multi-hundred-KB requests is a permanent per-worker leak. + """ + litellm.error_logs.clear() + big_input = [{"role": "user", "content": "x" * 10_000}] + + logging_obj.pre_call(input=big_input, api_key="sk-test") + logging_obj.post_call(original_response='{"ok": true}', input=big_input, api_key="sk-test") + + assert litellm.error_logs == {} diff --git a/tests/test_litellm/test_router/test_io_token_rate_limits.py b/tests/test_litellm/test_router/test_io_token_rate_limits.py index 939e3189596..a5a68271111 100644 --- a/tests/test_litellm/test_router/test_io_token_rate_limits.py +++ b/tests/test_litellm/test_router/test_io_token_rate_limits.py @@ -977,3 +977,65 @@ class TestRouterIOTokenIntegration: assert info is not None assert info.itpm == 100 assert info.otpm == 20 + + +class TestContextSlotRetention: + def test_setter_stores_kwargs_only_for_io_limited_deployments(self): + """ + The context slot pins the entire request kwargs (messages included) + for the lifetime of the surrounding asyncio context, and pooled + resources created mid-request (e.g. redis connections) capture that + context, extending the pin far past the request. Only ITPM/OTPM + pre-call checks read the slot, so the setter must store None for + deployments without io token limits and still clear reservation + sentinels from kwargs either way. + """ + kwargs = { + "messages": [{"role": "user", "content": "x" * 1000}], + "metadata": {ITPM_RESERVED_KEY: 999, ITPM_CACHE_KEY: "forged"}, + } + set_io_token_rate_limit_request_kwargs(kwargs, store_in_context=False) + assert get_io_token_rate_limit_request_kwargs() is None + assert ITPM_RESERVED_KEY not in kwargs["metadata"] + assert ITPM_CACHE_KEY not in kwargs["metadata"] + + set_io_token_rate_limit_request_kwargs(kwargs, store_in_context=True) + assert get_io_token_rate_limit_request_kwargs() is kwargs + + set_io_token_rate_limit_request_kwargs(kwargs, store_in_context=False) + assert get_io_token_rate_limit_request_kwargs() is None + + @pytest.mark.asyncio + async def test_router_does_not_pin_kwargs_without_io_limits(self): + router = Router( + model_list=[ + { + "model_name": "plain", + "litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-test"}, + } + ] + ) + set_io_token_rate_limit_request_kwargs(None) + kwargs = {"messages": [{"role": "user", "content": "hello"}], "metadata": {}} + deployment = router.get_deployment_by_model_group_name("plain") + assert deployment is not None + router._update_kwargs_with_deployment(deployment=deployment.model_dump(), kwargs=kwargs) + assert get_io_token_rate_limit_request_kwargs() is None + + @pytest.mark.asyncio + async def test_router_pins_kwargs_for_io_limited_deployment(self): + router = Router( + model_list=[ + { + "model_name": "limited", + "litellm_params": {"model": "openai/gpt-4o-mini", "api_key": "sk-test", "itpm": 100}, + } + ], + optional_pre_call_checks=["enforce_model_rate_limits"], + ) + set_io_token_rate_limit_request_kwargs(None) + kwargs = {"messages": [{"role": "user", "content": "hello"}], "metadata": {}} + deployment = router.get_deployment_by_model_group_name("limited") + assert deployment is not None + router._update_kwargs_with_deployment(deployment=deployment.model_dump(), kwargs=kwargs) + assert get_io_token_rate_limit_request_kwargs() is kwargs