fix(redis): free each request's Redis batch scope once the request ends (#50751)

* fix(spend): respect use_redis_transaction_buffer when selecting the SpendLogs ledger

* revert: keep the SpendLogs ledger independent of use_redis_transaction_buffer

Measured on a live proxy, ledger on and ledger off grow RSS at the same rate per request,
so gating the ledger on the Redis buffer moved deployments off it without fixing the growth.

* fix(redis): free each request's Redis batch scope once the request ends

The native request batch keeps a copy of the context it is built in. It was built inside the
request scope, so that context pointed back at the scope, and the cycle collector cannot see
through the native object. Every request's batches, spend counter batches and native Redis
state stayed alive, growing worker RSS about 40 KiB per request and slowing each request as
the heap grew. Build the native batch in the context the scope was opened from instead.

litellm/caching/redis_batch.py

@@ -16,7 +16,7 @@ import logging
 import time
 import weakref
 from collections.abc import Awaitable, Callable, Generator, Mapping, Sequence
-from contextvars import ContextVar, Token
+from contextvars import ContextVar, Token, copy_context
 from dataclasses import dataclass, field
 from datetime import timedelta
 from enum import Enum, auto
@@ -528,6 +528,7 @@ class RequestRedisBatches:
         "_deadline",
         "_deadline_flush",
         "_native",
+        "_native_context",
         "_native_factory",
         "_native_flush_hooks",
         "_native_misses",
@@ -551,6 +552,7 @@ class RequestRedisBatches:
         self._native_factory: Final = native_factory
         self._native_send_factory: Final = native_send_factory
         self._native: NativeRequestRedis | None = None
+        self._native_context: Final = copy_context()
         self._native_flush_hooks: Final[list[Callable[[], None]]] = []  # mutable-ok: append-only registry
         self._native_misses: frozenset[str] = frozenset()
         self._deadline: asyncio.TimerHandle | None = None
@@ -580,9 +582,12 @@ class RequestRedisBatches:
         return batch
 
     def native(self) -> NativeRequestRedis:
-        """The request's native batches, built on first use so requests that touch no native store pay nothing."""
+        """The request's native batches, built on first use so requests that touch no native store pay nothing.
+
+        They keep a copy of the context they are built in, and the cycle collector cannot see through native
+        objects, so they are built in the context this scope was opened from, which cannot reach the scope."""
         if self._native is None:
-            self._native = self._native_factory(self.post_call_deadline)
+            self._native = self._native_context.run(self._native_factory, self.post_call_deadline)
         return self._native
 
     def native_post_call(self) -> NativeRequestRedis:

tests/unit/caching/test_redis_batch.py

@@ -3,9 +3,11 @@
 from __future__ import annotations
 
 import asyncio
+import gc
 import hashlib
 import json
 import uuid
+import weakref
 from collections.abc import Awaitable, Callable, Sequence
 from dataclasses import dataclass, field
 from datetime import timedelta
@@ -19,6 +21,7 @@ from litellm._service_logger import ServiceLogging
 from litellm.caching.redis_batch import (
     MIXED_PIPELINE_TARGET,
     RedisBatch,
+    RequestRedisBatches,
     active_request_redis_batch,
     request_redis_batch_scope,
 )
@@ -418,6 +421,17 @@ def test_request_scope_hands_out_one_batch_per_backend_and_nests() -> None:
     assert active_request_redis_batch(cache_a) is None
 
 
+def test_a_request_scope_that_built_its_native_batches_is_freed_once_it_closes() -> None:
+    def serve_request() -> weakref.ref[RequestRedisBatches]:
+        with request_redis_batch_scope() as batches:
+            batches.native()
+            return weakref.ref(batches)
+
+    closed: Final = serve_request()
+    gc.collect()
+    assert closed() is None
+
+
 @pytest.mark.asyncio
 async def test_a_key_an_mget_read_as_absent_stays_known_missing_until_something_sets_it() -> None:
     cache, client = make()