From 250816eaebc650a2c16a95de429ccba610549051 Mon Sep 17 00:00:00 2001 From: Maximilian Roos Date: Mon, 27 Jul 2026 13:30:09 -0700 Subject: [PATCH 1/7] fix(http_handler): keep a handler alive while its responses are still streaming Closing an httpx client tears down the connection pool every response from that client is streaming through. litellm caches HTTP handlers for one hour, and nothing in a streaming response's reference graph points back at the handler -- httpx binds the response to the transport stream and the pool, never to the client's wrapper. So when the cache entry expires, the handler's last reference goes with it while its responses are still being read, GC runs `__del__`, and every in-flight body on that pool is cut. The clock is per process and anchored at first client construction, which is why the failures arrive in hourly bursts. Rather than guard the finalizer, make it unreachable while the client is in use. `create_client` installs one httpx `response` event hook that stamps the owning handler onto `response.extensions`, so a response keeps its handler alive; the handler is collected once its last response is gone and the existing `__del__` then closes the client exactly as it does today. Both `__del__` bodies are unchanged -- only the timing moves, so the disposal path cannot regress. The hook closes over a weak reference to the handler, never a strong one: a strong one would make the handler immortal and its client would never be closed at all. There is a test for that specific invariant. This reaches the paths a fix in `post()` cannot: the proxy passthrough routes take `.client` off the handler and stream from the raw client, so the handler is a function local that dies on return. An event hook on the client still sees those responses. Everything used is public httpx surface -- the `event_hooks` setter and `Response.extensions`. That matters because litellm's default transport is aiohttp, whose connector exposes no public way to ask whether a connection is in use, so a fix built on httpcore's pool would pass under DISABLE_AIOHTTP_TRANSPORT=true and silently do nothing by default. The regression test is parameterized over both transports. The sync twin in `HTTPHandler` has the same defect and gets the same treatment. Fixes #24929 --- litellm/llms/custom_httpx/http_handler.py | 41 +++++- .../llms/custom_httpx/test_http_handler.py | 135 ++++++++++++++++++ 2 files changed, 175 insertions(+), 1 deletion(-) diff --git a/litellm/llms/custom_httpx/http_handler.py b/litellm/llms/custom_httpx/http_handler.py index 5cec763bb5d..42f1d7b74da 100644 --- a/litellm/llms/custom_httpx/http_handler.py +++ b/litellm/llms/custom_httpx/http_handler.py @@ -6,6 +6,7 @@ import socket import ssl import sys import time +import weakref from typing import ( TYPE_CHECKING, Any, @@ -506,6 +507,41 @@ class MaskedHTTPStatusError(httpx.HTTPStatusError): self.status_code = original_error.response.status_code +_CLIENT_OWNER_EXTENSION = "litellm_client_owner" + + +def _pin_owner_on_async_client(owner: "AsyncHTTPHandler", client: httpx.AsyncClient) -> None: + """Record the handler that owns `client` on every response that client returns.""" + owner_ref = weakref.ref(owner) + existing_hooks = client.event_hooks + + async def _pin_owner(response: httpx.Response) -> None: + handler = owner_ref() + if handler is not None: + response.extensions[_CLIENT_OWNER_EXTENSION] = handler + + client.event_hooks = { + "request": list(existing_hooks.get("request", [])), + "response": [_pin_owner, *existing_hooks.get("response", [])], + } + + +def _pin_owner_on_sync_client(owner: "HTTPHandler", client: httpx.Client) -> None: + """Record the handler that owns `client` on every response that client returns.""" + owner_ref = weakref.ref(owner) + existing_hooks = client.event_hooks + + def _pin_owner(response: httpx.Response) -> None: + handler = owner_ref() + if handler is not None: + response.extensions[_CLIENT_OWNER_EXTENSION] = handler + + client.event_hooks = { + "request": list(existing_hooks.get("request", [])), + "response": [_pin_owner, *existing_hooks.get("response", [])], + } + + class AsyncHTTPHandler: def __init__( self, @@ -553,7 +589,7 @@ class AsyncHTTPHandler: # Get default headers (User-Agent, overridable via LITELLM_USER_AGENT) default_headers = get_default_headers() - return httpx.AsyncClient( + client = httpx.AsyncClient( transport=transport, event_hooks=event_hooks, timeout=timeout, @@ -562,6 +598,8 @@ class AsyncHTTPHandler: headers=default_headers, follow_redirects=True, ) + _pin_owner_on_async_client(self, client) + return client async def close(self): # Close the client when you're done with it @@ -1099,6 +1137,7 @@ class HTTPHandler: headers=default_headers, follow_redirects=True, ) + _pin_owner_on_sync_client(self, self.client) else: self.client = client diff --git a/tests/test_litellm/llms/custom_httpx/test_http_handler.py b/tests/test_litellm/llms/custom_httpx/test_http_handler.py index 87d67e0e8b7..9f9513689fa 100644 --- a/tests/test_litellm/llms/custom_httpx/test_http_handler.py +++ b/tests/test_litellm/llms/custom_httpx/test_http_handler.py @@ -1,10 +1,14 @@ import asyncio +import gc import io import os import pathlib import ssl import sys import threading +import time +import weakref +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from unittest.mock import MagicMock, patch import certifi @@ -793,3 +797,134 @@ class TestDefaultCachedClientTimeoutHonorsRequestTimeout: litellm.in_memory_llm_clients_cache = LLMClientCache() client = get_async_httpx_client(llm_provider=LlmProviders.BEDROCK) assert client.timeout.read == 300.0 + + +class _ChunkedSSEServer: + """In-process HTTP server that answers every request with chunked SSE frames.""" + + def __init__(self, frame_count: int = 6, frame_delay: float = 0.05) -> None: + self.frame_count = frame_count + self.frame_delay = frame_delay + parent = self + + class _Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def _stream(self): + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + try: + for index in range(parent.frame_count): + frame = f"data: frame-{index}\n\n".encode() + self.wfile.write(b"%x\r\n" % len(frame) + frame + b"\r\n") + self.wfile.flush() + time.sleep(parent.frame_delay) + self.wfile.write(b"0\r\n\r\n") + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError): + pass + + do_GET = _stream + do_POST = _stream + + def log_message(self, *args): + pass + + self._server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + self.url = f"http://127.0.0.1:{self._server.server_address[1]}/stream" + + def __enter__(self): + threading.Thread(target=self._server.serve_forever, daemon=True).start() + return self + + def __exit__(self, *exc_info): + self._server.shutdown() + self._server.server_close() + + +class TestHandlerCollectionDoesNotAbortInFlightStreams: + """Regression guard for https://github.com/BerriAI/litellm/issues/24929""" + + @pytest.mark.asyncio + @pytest.mark.parametrize("disable_aiohttp_transport", [False, True]) + async def test_async_stream_survives_handler_collection(self, monkeypatch, disable_aiohttp_transport): + monkeypatch.delenv("DISABLE_AIOHTTP_TRANSPORT", raising=False) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", disable_aiohttp_transport) + monkeypatch.setattr(litellm, "force_ipv4", False) + + with _ChunkedSSEServer() as server: + handler = AsyncHTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) + client = handler.client + response = await client.send(client.build_request("GET", server.url), stream=True) + del handler + + async def read_frames(body: httpx.Response) -> int: + total = 0 + async for chunk in body.aiter_bytes(): + total += chunk.count(b"data: frame-") + gc.collect() + return total + + frames = await asyncio.wait_for(read_frames(response), timeout=20) + + assert frames == 6 + assert client.is_closed is False + + del response + gc.collect() + for _ in range(200): + if client.is_closed: + break + await asyncio.sleep(0.01) + assert client.is_closed is True + + def test_sync_stream_survives_handler_collection(self, monkeypatch): + monkeypatch.setattr(litellm, "force_ipv4", False) + + with _ChunkedSSEServer() as server: + handler = HTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) + client = handler.client + response = client.send(client.build_request("GET", server.url), stream=True) + del handler + + frames = 0 + for chunk in response.iter_bytes(): + frames += chunk.count(b"data: frame-") + gc.collect() + + assert frames == 6 + assert client.is_closed is False + + del response + gc.collect() + assert client.is_closed is True + + def test_client_does_not_keep_its_handler_alive(self, monkeypatch): + monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) + handler = AsyncHTTPHandler() + client = handler.client + handler_ref = weakref.ref(handler) + + del handler + gc.collect() + + assert handler_ref() is None + assert isinstance(client, httpx.AsyncClient) + + @pytest.mark.asyncio + async def test_caller_supplied_event_hooks_still_fire(self): + seen = [] + + async def user_response_hook(response: httpx.Response) -> None: + seen.append(response.status_code) + + with _ChunkedSSEServer(frame_count=1, frame_delay=0.0) as server: + handler = AsyncHTTPHandler(event_hooks={"request": [], "response": [user_response_hook]}) + try: + response = await handler.client.get(server.url) + assert response.status_code == 200 + assert seen == [200] + finally: + await handler.close() From 99ae409fd308b0de1a42b74e705d88ea9c23bfce Mon Sep 17 00:00:00 2001 From: Maximilian Roos Date: Tue, 28 Jul 2026 09:18:38 -0700 Subject: [PATCH 2/7] Revert "fix(http_handler): keep a handler alive while its responses are still streaming" This reverts commit 250816eaebc650a2c16a95de429ccba610549051. That approach kept both `__del__` finalizers and made them unreachable while a response was live, by stamping the owning handler onto `response.extensions` from a client event hook. It only covers the shapes where a response is in flight. A caller that keeps just `handler.client`, and a throwaway handler wrapped around a client someone else owns, have no response to pin the handler to, so the client is still closed underneath them. The following commit deletes the finalizers instead, which covers those shapes and removes code rather than adding it. --- litellm/llms/custom_httpx/http_handler.py | 41 +----- .../llms/custom_httpx/test_http_handler.py | 135 ------------------ 2 files changed, 1 insertion(+), 175 deletions(-) diff --git a/litellm/llms/custom_httpx/http_handler.py b/litellm/llms/custom_httpx/http_handler.py index 42f1d7b74da..5cec763bb5d 100644 --- a/litellm/llms/custom_httpx/http_handler.py +++ b/litellm/llms/custom_httpx/http_handler.py @@ -6,7 +6,6 @@ import socket import ssl import sys import time -import weakref from typing import ( TYPE_CHECKING, Any, @@ -507,41 +506,6 @@ class MaskedHTTPStatusError(httpx.HTTPStatusError): self.status_code = original_error.response.status_code -_CLIENT_OWNER_EXTENSION = "litellm_client_owner" - - -def _pin_owner_on_async_client(owner: "AsyncHTTPHandler", client: httpx.AsyncClient) -> None: - """Record the handler that owns `client` on every response that client returns.""" - owner_ref = weakref.ref(owner) - existing_hooks = client.event_hooks - - async def _pin_owner(response: httpx.Response) -> None: - handler = owner_ref() - if handler is not None: - response.extensions[_CLIENT_OWNER_EXTENSION] = handler - - client.event_hooks = { - "request": list(existing_hooks.get("request", [])), - "response": [_pin_owner, *existing_hooks.get("response", [])], - } - - -def _pin_owner_on_sync_client(owner: "HTTPHandler", client: httpx.Client) -> None: - """Record the handler that owns `client` on every response that client returns.""" - owner_ref = weakref.ref(owner) - existing_hooks = client.event_hooks - - def _pin_owner(response: httpx.Response) -> None: - handler = owner_ref() - if handler is not None: - response.extensions[_CLIENT_OWNER_EXTENSION] = handler - - client.event_hooks = { - "request": list(existing_hooks.get("request", [])), - "response": [_pin_owner, *existing_hooks.get("response", [])], - } - - class AsyncHTTPHandler: def __init__( self, @@ -589,7 +553,7 @@ class AsyncHTTPHandler: # Get default headers (User-Agent, overridable via LITELLM_USER_AGENT) default_headers = get_default_headers() - client = httpx.AsyncClient( + return httpx.AsyncClient( transport=transport, event_hooks=event_hooks, timeout=timeout, @@ -598,8 +562,6 @@ class AsyncHTTPHandler: headers=default_headers, follow_redirects=True, ) - _pin_owner_on_async_client(self, client) - return client async def close(self): # Close the client when you're done with it @@ -1137,7 +1099,6 @@ class HTTPHandler: headers=default_headers, follow_redirects=True, ) - _pin_owner_on_sync_client(self, self.client) else: self.client = client diff --git a/tests/test_litellm/llms/custom_httpx/test_http_handler.py b/tests/test_litellm/llms/custom_httpx/test_http_handler.py index 9f9513689fa..87d67e0e8b7 100644 --- a/tests/test_litellm/llms/custom_httpx/test_http_handler.py +++ b/tests/test_litellm/llms/custom_httpx/test_http_handler.py @@ -1,14 +1,10 @@ import asyncio -import gc import io import os import pathlib import ssl import sys import threading -import time -import weakref -from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from unittest.mock import MagicMock, patch import certifi @@ -797,134 +793,3 @@ class TestDefaultCachedClientTimeoutHonorsRequestTimeout: litellm.in_memory_llm_clients_cache = LLMClientCache() client = get_async_httpx_client(llm_provider=LlmProviders.BEDROCK) assert client.timeout.read == 300.0 - - -class _ChunkedSSEServer: - """In-process HTTP server that answers every request with chunked SSE frames.""" - - def __init__(self, frame_count: int = 6, frame_delay: float = 0.05) -> None: - self.frame_count = frame_count - self.frame_delay = frame_delay - parent = self - - class _Handler(BaseHTTPRequestHandler): - protocol_version = "HTTP/1.1" - - def _stream(self): - self.send_response(200) - self.send_header("Content-Type", "text/event-stream") - self.send_header("Transfer-Encoding", "chunked") - self.end_headers() - try: - for index in range(parent.frame_count): - frame = f"data: frame-{index}\n\n".encode() - self.wfile.write(b"%x\r\n" % len(frame) + frame + b"\r\n") - self.wfile.flush() - time.sleep(parent.frame_delay) - self.wfile.write(b"0\r\n\r\n") - self.wfile.flush() - except (BrokenPipeError, ConnectionResetError): - pass - - do_GET = _stream - do_POST = _stream - - def log_message(self, *args): - pass - - self._server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) - self.url = f"http://127.0.0.1:{self._server.server_address[1]}/stream" - - def __enter__(self): - threading.Thread(target=self._server.serve_forever, daemon=True).start() - return self - - def __exit__(self, *exc_info): - self._server.shutdown() - self._server.server_close() - - -class TestHandlerCollectionDoesNotAbortInFlightStreams: - """Regression guard for https://github.com/BerriAI/litellm/issues/24929""" - - @pytest.mark.asyncio - @pytest.mark.parametrize("disable_aiohttp_transport", [False, True]) - async def test_async_stream_survives_handler_collection(self, monkeypatch, disable_aiohttp_transport): - monkeypatch.delenv("DISABLE_AIOHTTP_TRANSPORT", raising=False) - monkeypatch.setattr(litellm, "disable_aiohttp_transport", disable_aiohttp_transport) - monkeypatch.setattr(litellm, "force_ipv4", False) - - with _ChunkedSSEServer() as server: - handler = AsyncHTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) - client = handler.client - response = await client.send(client.build_request("GET", server.url), stream=True) - del handler - - async def read_frames(body: httpx.Response) -> int: - total = 0 - async for chunk in body.aiter_bytes(): - total += chunk.count(b"data: frame-") - gc.collect() - return total - - frames = await asyncio.wait_for(read_frames(response), timeout=20) - - assert frames == 6 - assert client.is_closed is False - - del response - gc.collect() - for _ in range(200): - if client.is_closed: - break - await asyncio.sleep(0.01) - assert client.is_closed is True - - def test_sync_stream_survives_handler_collection(self, monkeypatch): - monkeypatch.setattr(litellm, "force_ipv4", False) - - with _ChunkedSSEServer() as server: - handler = HTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) - client = handler.client - response = client.send(client.build_request("GET", server.url), stream=True) - del handler - - frames = 0 - for chunk in response.iter_bytes(): - frames += chunk.count(b"data: frame-") - gc.collect() - - assert frames == 6 - assert client.is_closed is False - - del response - gc.collect() - assert client.is_closed is True - - def test_client_does_not_keep_its_handler_alive(self, monkeypatch): - monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) - handler = AsyncHTTPHandler() - client = handler.client - handler_ref = weakref.ref(handler) - - del handler - gc.collect() - - assert handler_ref() is None - assert isinstance(client, httpx.AsyncClient) - - @pytest.mark.asyncio - async def test_caller_supplied_event_hooks_still_fire(self): - seen = [] - - async def user_response_hook(response: httpx.Response) -> None: - seen.append(response.status_code) - - with _ChunkedSSEServer(frame_count=1, frame_delay=0.0) as server: - handler = AsyncHTTPHandler(event_hooks={"request": [], "response": [user_response_hook]}) - try: - response = await handler.client.get(server.url) - assert response.status_code == 200 - assert seen == [200] - finally: - await handler.close() From 3f00e6fa46eefd22e2cebb1fa4c8f63c8863d412 Mon Sep 17 00:00:00 2001 From: Maximilian Roos Date: Tue, 28 Jul 2026 09:18:51 -0700 Subject: [PATCH 3/7] fix(http_handler): don't close the httpx client when a handler is collected `AsyncHTTPHandler` and `HTTPHandler` each closed their httpx client from `__del__`. Closing an httpx client tears down the connection pool every in-flight response is streaming through, and it permanently invalidates the client for later requests. Neither is safe on garbage collection, because a handler is routinely collected while its client is still in use: - Streaming. Nothing in a response's reference graph points back at the handler: httpx binds the response to the transport stream and the pool, never to the wrapper that built them. A cached handler therefore becomes collectable mid-stream when its one-hour TTL expires, and every body still being read off that pool is cut. That is #24929; the window recurs hourly per process, because the TTL clock starts at first client construction. - Borrowed clients. `litellm/a2a_protocol/main.py` and `litellm/proxy/pass_through_endpoints/pass_through_endpoints.py` keep only `handler.client`. Both take the handler from `get_async_httpx_client`, so the cache pins it for the same one-hour TTL and then lets it go on eviction, while the borrowed client is still serving a longer-lived consumer: `create_a2a_client` hands its client to the a2a SDK and documents it as "create client once, reuse for multiple requests". The next request on that client raises `RuntimeError: Cannot send a request, as the client has been closed.` - Someone else's client. `litellm/llms/azure/azure.py` wraps `litellm.client_session` in a throwaway `HTTPHandler` for a single image generation, and `HTTPHandler.close()` closes whatever client it was handed. One such call left the user's shared session closed for the rest of the process. `LLMClientCache` already documents the invariant these finalizers broke: evicted clients "may still be in use by in-flight requests", so the cache deliberately does not close them and leaves them to normal garbage collection. Closing on collection is precisely what that rules out. Deleting both finalizers removes the cause instead of narrowing the window. The async one was never dependable regardless: it needs a running event loop, so collection outside one silently did nothing. Explicit `close()`/`aclose()` is untouched, and `close_litellm_async_clients()` still closes cached async clients at exit. The cost is that an evicted client is no longer closed eagerly. On the default aiohttp transport that shows up as five lines of asyncio ERROR output per evicted handler ("Unclosed client session", "Unclosed connector", and their context keys) with no change in descriptors or RSS, since aiohttp's own connector finalizer closes the connections as it warns. On the httpcore transport there is no log output, but idle keep-alive sockets from evicted pools are held until a generation-2 collection reclaims them: over 5000 evictions that plateaus at 56 descriptors and ~1 MB of RSS, and with 20 concurrent requests per pool it peaks around 650 descriptors before each gen-2 pass returns it to baseline. Bounded by the GC cycle, not by the eviction count. Fixes #24929 --- litellm/llms/custom_httpx/http_handler.py | 12 - .../test_handler_gc_does_not_close_client.py | 284 ++++++++++++++++++ 2 files changed, 284 insertions(+), 12 deletions(-) create mode 100644 tests/test_litellm/llms/custom_httpx/test_handler_gc_does_not_close_client.py diff --git a/litellm/llms/custom_httpx/http_handler.py b/litellm/llms/custom_httpx/http_handler.py index 5cec763bb5d..d510432b3b4 100644 --- a/litellm/llms/custom_httpx/http_handler.py +++ b/litellm/llms/custom_httpx/http_handler.py @@ -879,12 +879,6 @@ class AsyncHTTPHandler: response.raise_for_status() return response - def __del__(self) -> None: - try: - asyncio.get_running_loop().create_task(self.close()) - except Exception: - pass - @staticmethod def _create_async_transport( ssl_context: Optional[ssl.SSLContext] = None, @@ -1345,12 +1339,6 @@ class HTTPHandler: except Exception as e: raise e - def __del__(self) -> None: - try: - self.close() - except Exception: - pass - def _create_sync_transport(self) -> Optional[HTTPTransport]: """ Create an HTTP transport with IPv4 only if litellm.force_ipv4 is True. diff --git a/tests/test_litellm/llms/custom_httpx/test_handler_gc_does_not_close_client.py b/tests/test_litellm/llms/custom_httpx/test_handler_gc_does_not_close_client.py new file mode 100644 index 00000000000..8c95691a9c9 --- /dev/null +++ b/tests/test_litellm/llms/custom_httpx/test_handler_gc_does_not_close_client.py @@ -0,0 +1,284 @@ +""" +Garbage-collecting an HTTP handler must not close the httpx client it holds. + +``HTTPHandler`` and ``AsyncHTTPHandler`` used to close their client from +``__del__``. Closing an httpx client tears down the connection pool that every +in-flight response is streaming through, and it permanently invalidates the +client for future requests -- including for callers who only ever borrowed +``handler.client``. Since nothing in a response's reference graph points back +at the handler, and since litellm caches handlers behind a one-hour TTL, the +handler routinely became collectable while its client was still in use. + +``LLMClientCache`` documents the invariant this broke: evicted clients "may +still be in use by in-flight requests", so they are left to normal garbage +collection rather than closed eagerly. A finalizer that closes on collection +defeats exactly that. + +Each test below is one shape the finalizers broke; all of them fail if either +``__del__`` comes back. Async cases run on both transports, because litellm +defaults to aiohttp and only uses httpcore when aiohttp is disabled. + +Unlike the rest of ``tests/test_litellm/``, these tests need a real connection +pool rather than a mock: a mocked transport goes on yielding chunks after its +client is closed, so the very teardown under test is what a mock cannot +reproduce. The server here is the hermetic, credential-free +``ThreadingHTTPServer`` on an ephemeral loopback port already used by +``tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py``. + +Related: https://github.com/BerriAI/litellm/issues/24929 +""" + +import asyncio +import gc +import os +import sys +import threading +import time +import weakref +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +import httpx +import pytest + +sys.path.insert(0, os.path.abspath("../../../..")) + +import litellm +from litellm.caching.llm_caching_handler import LLMClientCache +from litellm.llms.custom_httpx.http_handler import ( + AsyncHTTPHandler, + HTTPHandler, + get_async_httpx_client, +) +from litellm.types.utils import LlmProviders + +FRAME_COUNT = 6 +# Generous: the server emits all frames in ~0.3s. A client whose pool was torn +# down mid-stream can stall silently instead of raising, so reads are bounded. +READ_TIMEOUT_SECONDS = 15.0 + +BOTH_TRANSPORTS = pytest.mark.parametrize("disable_aiohttp_transport", [False, True], ids=["aiohttp", "httpcore"]) + +# Every test here rests on the handler actually being collected at the ``del``. +# If something ever pins it, the test would pass while guarding nothing, so each +# one checks the premise. The check comes last: a reintroduced finalizer also +# fails it, by resurrecting the handler into the task it creates for ``close()``, +# and the transport error is the more useful thing to see first. +HANDLER_NOT_COLLECTED = "handler was not collected; this test no longer exercises the finalizer path" + + +class _ChunkedSSEServer: + """In-process HTTP/1.1 server that answers every request with chunked SSE frames.""" + + def __init__(self, frame_count: int = FRAME_COUNT, frame_delay: float = 0.05) -> None: + self.frame_count = frame_count + self.frame_delay = frame_delay + parent = self + + class _Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def _stream(self): + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Transfer-Encoding", "chunked") + self.end_headers() + try: + for index in range(parent.frame_count): + frame = f"data: frame-{index}\n\n".encode() + self.wfile.write(b"%x\r\n" % len(frame) + frame + b"\r\n") + self.wfile.flush() + time.sleep(parent.frame_delay) + self.wfile.write(b"0\r\n\r\n") + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError): + pass + + do_GET = _stream + do_POST = _stream + + def log_message(self, *args): + pass + + self._server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + self.url = f"http://127.0.0.1:{self._server.server_address[1]}/stream" + + def __enter__(self): + threading.Thread(target=self._server.serve_forever, daemon=True).start() + return self + + def __exit__(self, *exc_info): + self._server.shutdown() + self._server.server_close() + + +def _select_transport(monkeypatch, disable_aiohttp_transport: bool) -> None: + monkeypatch.delenv("DISABLE_AIOHTTP_TRANSPORT", raising=False) + monkeypatch.setattr(litellm, "disable_aiohttp_transport", disable_aiohttp_transport) + monkeypatch.setattr(litellm, "force_ipv4", False) + + +async def _read_frames(response: httpx.Response) -> int: + """Count SSE frames, collecting garbage between chunks so a finalizer has every chance to fire. + + The body is joined before counting: a chunk boundary can fall inside the + marker, which a per-chunk count would miss. + """ + chunks = [] + async for chunk in response.aiter_bytes(): + chunks.append(chunk) + gc.collect() + return b"".join(chunks).count(b"data: frame-") + + +@pytest.mark.asyncio +@BOTH_TRANSPORTS +async def test_async_stream_survives_handler_collection(monkeypatch, disable_aiohttp_transport): + """A response being streamed keeps working after its handler is collected.""" + _select_transport(monkeypatch, disable_aiohttp_transport) + + with _ChunkedSSEServer() as server: + handler = AsyncHTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) + client = handler.client + try: + response = await client.send(client.build_request("GET", server.url), stream=True) + + # The handler loses its last reference while the body is still streaming. + ref = weakref.ref(handler) + del handler + gc.collect() + + frames = await asyncio.wait_for(_read_frames(response), timeout=READ_TIMEOUT_SECONDS) + assert frames == FRAME_COUNT + assert client.is_closed is False + assert ref() is None, HANDLER_NOT_COLLECTED + finally: + await client.aclose() + + +@pytest.mark.asyncio +@BOTH_TRANSPORTS +async def test_borrowed_async_client_outlives_its_handler(monkeypatch, disable_aiohttp_transport): + """A caller that keeps only ``handler.client`` can still send requests once the handler is gone. + + This is the shape at litellm/a2a_protocol/main.py (``httpx_client = + _async_handler.client``, handed to the a2a SDK) and at + litellm/proxy/pass_through_endpoints/pass_through_endpoints.py (``async_client + = async_client_obj.client``). Both take the handler from + ``get_async_httpx_client``, so the cache pins it for + ``_DEFAULT_TTL_FOR_HTTPX_CLIENTS`` (one hour) and then lets it go on eviction, + at which point it is collected while the borrowed client is still serving a + longer-lived consumer: ``create_a2a_client`` hands its client to the a2a SDK + and documents it as "create client once, reuse for multiple requests". + """ + _select_transport(monkeypatch, disable_aiohttp_transport) + + with _ChunkedSSEServer(frame_count=1, frame_delay=0.0) as server: + handler = AsyncHTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) + client = handler.client + try: + ref = weakref.ref(handler) + del handler + gc.collect() + # A finalizer would close the client from a task, so let the loop turn. + await asyncio.sleep(0.05) + + assert client.is_closed is False + response = await client.get(server.url) + assert response.status_code == 200 + assert ref() is None, HANDLER_NOT_COLLECTED + finally: + await client.aclose() + + +def test_sync_handler_collection_does_not_close_a_caller_owned_client(monkeypatch): + """A throwaway handler wrapped around someone else's client must not close it. + + litellm/llms/azure/azure.py builds ``HTTPHandler(client=litellm.client_session)`` + for a single image generation and drops it. With a finalizer, that one call + left the user's shared session closed for the rest of the process. + """ + monkeypatch.setattr(litellm, "force_ipv4", False) + + with _ChunkedSSEServer(frame_count=1, frame_delay=0.0) as server: + caller_client = httpx.Client(timeout=httpx.Timeout(10.0, connect=5.0)) + monkeypatch.setattr(litellm, "client_session", caller_client) + try: + handler = HTTPHandler(client=litellm.client_session) + ref = weakref.ref(handler) + del handler + gc.collect() + + assert caller_client.is_closed is False + assert caller_client.get(server.url).status_code == 200 + assert ref() is None, HANDLER_NOT_COLLECTED + finally: + caller_client.close() + + +def test_sync_stream_survives_handler_collection(monkeypatch): + """A sync response being streamed keeps working after its handler is collected. + + litellm/main.py builds a sync handler only for non-streaming calls, commented + "Keep this here, otherwise, the httpx.client closes and streaming is + impossible" -- a workaround for this finalizer rather than a fix for it. + """ + monkeypatch.setattr(litellm, "force_ipv4", False) + + with _ChunkedSSEServer() as server: + handler = HTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) + client = handler.client + try: + response = client.send(client.build_request("GET", server.url), stream=True) + + # The handler loses its last reference while the body is still streaming. + ref = weakref.ref(handler) + del handler + gc.collect() + + # Joined before counting, as in ``_read_frames``. + chunks = [] + for chunk in response.iter_bytes(): + chunks.append(chunk) + gc.collect() + + assert b"".join(chunks).count(b"data: frame-") == FRAME_COUNT + assert client.is_closed is False + assert ref() is None, HANDLER_NOT_COLLECTED + finally: + client.close() + + +@pytest.mark.asyncio +@BOTH_TRANSPORTS +async def test_cached_handler_eviction_does_not_abort_an_in_flight_stream(monkeypatch, disable_aiohttp_transport): + """Evicting a cached handler mid-stream leaves the stream alone. + + ``get_async_httpx_client`` caches handlers for an hour. When that TTL + expires the cache drops the only reference to a handler whose client is + still streaming -- the production shape of #24929, and the case + ``LLMClientCache`` documents as "may still be in use by in-flight requests". + """ + _select_transport(monkeypatch, disable_aiohttp_transport) + monkeypatch.setattr(litellm, "in_memory_llm_clients_cache", LLMClientCache()) + + with _ChunkedSSEServer() as server: + handler = get_async_httpx_client(llm_provider=LlmProviders.OPENAI) + client = handler.client + try: + response = await client.send(client.build_request("GET", server.url), stream=True) + + # An hour passes: the TTL expires and the cache lets the handler go. + ref = weakref.ref(handler) + litellm.in_memory_llm_clients_cache.flush_cache() + del handler + gc.collect() + + frames = await asyncio.wait_for(_read_frames(response), timeout=READ_TIMEOUT_SECONDS) + assert frames == FRAME_COUNT + assert client.is_closed is False + + # And the evicted client is still usable for the next request. + assert (await client.get(server.url)).status_code == 200 + assert ref() is None, HANDLER_NOT_COLLECTED + finally: + await client.aclose() From c44f62abb50218477c73b72cc2493dc110eb712c Mon Sep 17 00:00:00 2001 From: Maximilian Roos Date: Tue, 28 Jul 2026 19:08:06 -0700 Subject: [PATCH 4/7] test(http_handler): move the GC lifecycle tests to tests/local_testing tests/test_litellm/readme.md states that directory can only contain mocked tests. These need a real connection pool, because a mocked transport goes on yielding chunks after its client is closed, which is the teardown under test. tests/local_testing is the tree for tests that open a real socket, and CircleCI's local_testing_part1 job picks the file up by its name. No change to the tests themselves beyond the sys.path depth and a docstring line naming the new location. --- .../test_handler_gc_does_not_close_client.py | 11 +++++------ 1 file changed, 5 insertions(+), 6 deletions(-) rename tests/{test_litellm/llms/custom_httpx => local_testing}/test_handler_gc_does_not_close_client.py (96%) diff --git a/tests/test_litellm/llms/custom_httpx/test_handler_gc_does_not_close_client.py b/tests/local_testing/test_handler_gc_does_not_close_client.py similarity index 96% rename from tests/test_litellm/llms/custom_httpx/test_handler_gc_does_not_close_client.py rename to tests/local_testing/test_handler_gc_does_not_close_client.py index 8c95691a9c9..3874405a261 100644 --- a/tests/test_litellm/llms/custom_httpx/test_handler_gc_does_not_close_client.py +++ b/tests/local_testing/test_handler_gc_does_not_close_client.py @@ -18,12 +18,11 @@ Each test below is one shape the finalizers broke; all of them fail if either ``__del__`` comes back. Async cases run on both transports, because litellm defaults to aiohttp and only uses httpcore when aiohttp is disabled. -Unlike the rest of ``tests/test_litellm/``, these tests need a real connection -pool rather than a mock: a mocked transport goes on yielding chunks after its +These live here rather than under ``tests/test_litellm/`` because they need a +real connection pool: a mocked transport goes on yielding chunks after its client is closed, so the very teardown under test is what a mock cannot -reproduce. The server here is the hermetic, credential-free -``ThreadingHTTPServer`` on an ephemeral loopback port already used by -``tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_handler.py``. +reproduce. The server is a hermetic, credential-free ``ThreadingHTTPServer`` +on an ephemeral loopback port, and needs no network access beyond it. Related: https://github.com/BerriAI/litellm/issues/24929 """ @@ -40,7 +39,7 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import httpx import pytest -sys.path.insert(0, os.path.abspath("../../../..")) +sys.path.insert(0, os.path.abspath("../..")) import litellm from litellm.caching.llm_caching_handler import LLMClientCache From 5972227b6d2bbe01d43e373ec5931c7cc6821658 Mon Sep 17 00:00:00 2001 From: Maximilian Roos Date: Mon, 7 Sep 2026 22:07:12 -0700 Subject: [PATCH 5/7] fix(http_handler): keep a handler alive while a response it issued is still reading _handler_may_close_client withholds the finalizer's close from a client that someone else references. A streaming response is not one of those referrers: it holds the connection it reads from, never the client, so the refcount says "sole referrer" for exactly the client that is busiest, and __del__ tears the pool down mid-body. Both handlers, both transports, and it is the shape #24929 reports: a provider's streaming call returns the response and drops the handler, and get_async_httpx_client lets a cached handler go after an hour. A streaming send now anchors the handler to the response it returns, so the handler is finalized once the caller is done with the body and the ordinary close applies with nothing in flight. The anchor is a weakref.finalize, which holds the handler in its own registry rather than on the response. That keeps the handler out of the response's reference cycle, so it is finalized by refcount and can still schedule an async close, rather than inside a cyclic collection that reaps its aiohttp session in the same pass; and a handler serving several streams is released only once all of them are done, because each anchor holds it separately. Only a streaming send anchors. A non-streaming response has been read in full by the time the method returns, so pinning the handler to it would delay every client close behind whatever the caller does with the response. The alternative was to ask the connection pool whether a request was in flight, reusing EvictedClientCloser's check. It reads client._transport, so it answers "idle" for any client with a proxy configured, where httpx routes through client._mounts; and having found the client busy it can only poll, which never terminates for a response the caller abandons unread, since httpx leaves that connection checked out. The response's own lifetime is the condition both were approximating. --- litellm/llms/custom_httpx/http_handler.py | 40 +++ tests/local_testing/conftest.py | 3 + .../test_handler_gc_does_not_close_client.py | 302 ++++++++++-------- 3 files changed, 210 insertions(+), 135 deletions(-) diff --git a/litellm/llms/custom_httpx/http_handler.py b/litellm/llms/custom_httpx/http_handler.py index e1f0fc9e7d3..612d5997ec5 100644 --- a/litellm/llms/custom_httpx/http_handler.py +++ b/litellm/llms/custom_httpx/http_handler.py @@ -7,6 +7,7 @@ import ssl import sys import threading import time +import weakref from collections.abc import AsyncIterable, Callable, Iterable, Mapping from http.cookiejar import CookieJar, DefaultCookiePolicy from types import MappingProxyType @@ -178,6 +179,33 @@ def _handler_may_close_client(client_refcount: int, owns_client: bool) -> bool: return owns_client and client_refcount <= _CLIENT_REFCOUNT_WHEN_HANDLER_IS_SOLE_REFERRER +def _drop_streaming_anchor(_handler: object) -> None: + """Release a handler anchored to a streaming response. See ``_anchor_handler_to``. + + The work is the reference held until this point, so there is nothing to do here. + """ + + +def _anchor_handler_to(response: httpx.Response, handler: object) -> None: + """Keep the handler alive for as long as a streaming response can still read. + + A body still arriving reads through the handler's connection pool, and closing + the client tears that pool down. The refcount ``_handler_may_close_client`` + reads cannot see that body: the reference graph runs response -> stream -> + connection and stops there, so a client carrying one looks exactly like an + unreferenced client, and the finalizer closes it mid-body. + + ``weakref.finalize`` holds the handler in its own registry rather than on the + response, which matters twice. The handler stays out of the response's + reference cycle, so it is finalized by refcount once the anchor drops and can + still schedule an async close, instead of being finalized inside a cyclic + collection that reaps its aiohttp session in the same pass. And a handler + serving several streams collects only once every one of them is done, because + each anchor holds it separately. + """ + weakref.finalize(response, _drop_streaming_anchor, handler) + + def blocked_cookie_jar() -> CookieJar: """A jar that stores no response cookie and sends none, for httpx clients. @@ -704,6 +732,8 @@ class AsyncHTTPHandler: content=request_content, ) response: Final = await self.client.send(req, stream=stream) + if stream: + _anchor_handler_to(response, self) response.raise_for_status() return response except (httpx.RemoteProtocolError, httpx.ConnectError): @@ -898,6 +928,8 @@ class AsyncHTTPHandler: content=request_content, ) response: Final = await self.client.send(req, stream=stream) + if stream: + _anchor_handler_to(response, self) response.raise_for_status() return response except (httpx.RemoteProtocolError, httpx.ConnectError): @@ -1362,6 +1394,8 @@ class HTTPHandler: content=request_content, ) response: Final = self.client.send(req, stream=stream) + if stream: + _anchor_handler_to(response, self) response.raise_for_status() return response except httpx.TimeoutException: @@ -1412,6 +1446,8 @@ class HTTPHandler: content=request_content, ) response: Final = self.client.send(req, stream=stream) + if stream: + _anchor_handler_to(response, self) response.raise_for_status() return response except httpx.TimeoutException: @@ -1462,6 +1498,8 @@ class HTTPHandler: content=request_content, ) response: Final = self.client.send(req, stream=stream) + if stream: + _anchor_handler_to(response, self) return response except httpx.TimeoutException: raise litellm.Timeout( @@ -1511,6 +1549,8 @@ class HTTPHandler: content=request_content, ) response: Final = self.client.send(req, stream=stream) + if stream: + _anchor_handler_to(response, self) response.raise_for_status() return response except httpx.TimeoutException: diff --git a/tests/local_testing/conftest.py b/tests/local_testing/conftest.py index 5535a62bb81..228457f4d55 100644 --- a/tests/local_testing/conftest.py +++ b/tests/local_testing/conftest.py @@ -75,6 +75,9 @@ _VCR_INCOMPATIBLE_FILES = frozenset( "test_router_caching.py", # Hits the local fake OpenAI endpoint on 127.0.0.1; nothing to record. "test_fake_openai_endpoint.py", + # Needs the real connection pool a collected handler tears down; vcrpy + # patches the transport that pool lives in. + "test_handler_gc_does_not_close_client.py", } ) diff --git a/tests/local_testing/test_handler_gc_does_not_close_client.py b/tests/local_testing/test_handler_gc_does_not_close_client.py index 3874405a261..1a6ab1b1827 100644 --- a/tests/local_testing/test_handler_gc_does_not_close_client.py +++ b/tests/local_testing/test_handler_gc_does_not_close_client.py @@ -1,36 +1,39 @@ """ -Garbage-collecting an HTTP handler must not close the httpx client it holds. +Collecting an HTTP handler must not abort a response that is still on the wire. -``HTTPHandler`` and ``AsyncHTTPHandler`` used to close their client from -``__del__``. Closing an httpx client tears down the connection pool that every -in-flight response is streaming through, and it permanently invalidates the -client for future requests -- including for callers who only ever borrowed -``handler.client``. Since nothing in a response's reference graph points back -at the handler, and since litellm caches handlers behind a one-hour TTL, the -handler routinely became collectable while its client was still in use. +``HTTPHandler`` and ``AsyncHTTPHandler`` close their client from ``__del__``. +Closing a client tears down the connection pool, which aborts every response +still streaming through it. ``_handler_may_close_client`` already withholds the +close from a client someone else holds, but a streaming response holds the +connection it is reading from and never the client, so the refcount it reads +says "sole referrer" for exactly the client that is busiest. The handler is +routinely collectable at that moment: a provider's streaming call returns the +response and drops the handler, and ``get_async_httpx_client`` caches handlers +behind a one-hour TTL and then lets them go. -``LLMClientCache`` documents the invariant this broke: evicted clients "may -still be in use by in-flight requests", so they are left to normal garbage -collection rather than closed eagerly. A finalizer that closes on collection -defeats exactly that. +The fix anchors the handler to the streaming response, so these tests turn on +*when* the handler is collected rather than on whether it is: pinned while the +body can still arrive, released once the caller is done with the response. -Each test below is one shape the finalizers broke; all of them fail if either -``__del__`` comes back. Async cases run on both transports, because litellm -defaults to aiohttp and only uses httpcore when aiohttp is disabled. +Nothing here re-tests the shapes ``_handler_may_close_client`` covers -- a +borrowed ``handler.client``, a caller-supplied client, an evicted-but-held +client. Those are pinned in ``tests/test_litellm/llms/custom_httpx/ +test_http_handler.py``. What is uncovered there is the in-flight response, so no +test here may keep the client in a local: that inflates the very refcount under +test, and the test then passes on a broken handler. They hold weak references +instead, which the refcount does not count. These live here rather than under ``tests/test_litellm/`` because they need a real connection pool: a mocked transport goes on yielding chunks after its client is closed, so the very teardown under test is what a mock cannot -reproduce. The server is a hermetic, credential-free ``ThreadingHTTPServer`` -on an ephemeral loopback port, and needs no network access beyond it. +reproduce. The server is a hermetic, credential-free ``ThreadingHTTPServer`` on +an ephemeral loopback port, and needs no network access beyond it. Related: https://github.com/BerriAI/litellm/issues/24929 """ import asyncio import gc -import os -import sys import threading import time import weakref @@ -39,8 +42,6 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import httpx import pytest -sys.path.insert(0, os.path.abspath("../..")) - import litellm from litellm.caching.llm_caching_handler import LLMClientCache from litellm.llms.custom_httpx.http_handler import ( @@ -54,15 +55,12 @@ FRAME_COUNT = 6 # Generous: the server emits all frames in ~0.3s. A client whose pool was torn # down mid-stream can stall silently instead of raising, so reads are bounded. READ_TIMEOUT_SECONDS = 15.0 +RELEASE_TIMEOUT_SECONDS = 3.0 BOTH_TRANSPORTS = pytest.mark.parametrize("disable_aiohttp_transport", [False, True], ids=["aiohttp", "httpcore"]) -# Every test here rests on the handler actually being collected at the ``del``. -# If something ever pins it, the test would pass while guarding nothing, so each -# one checks the premise. The check comes last: a reintroduced finalizer also -# fails it, by resurrecting the handler into the task it creates for ``close()``, -# and the transport error is the more useful thing to see first. -HANDLER_NOT_COLLECTED = "handler was not collected; this test no longer exercises the finalizer path" +STILL_PINNED = "the handler was released while its response could still read" +NOT_RELEASED = "the handler outlived the response that was holding it" class _ChunkedSSEServer: @@ -129,93 +127,44 @@ async def _read_frames(response: httpx.Response) -> int: return b"".join(chunks).count(b"data: frame-") +async def _wait_until(is_done, failure: str) -> None: + deadline = time.monotonic() + RELEASE_TIMEOUT_SECONDS + while time.monotonic() < deadline: + if is_done(): + return + await asyncio.sleep(0.05) + pytest.fail(failure) + + @pytest.mark.asyncio @BOTH_TRANSPORTS async def test_async_stream_survives_handler_collection(monkeypatch, disable_aiohttp_transport): - """A response being streamed keeps working after its handler is collected.""" + """A response still streaming keeps working after its handler goes out of scope. + + The caller holds the response and nothing else, which is what a provider's + streaming path is left with once ``post(..., stream=True)`` has returned. + """ _select_transport(monkeypatch, disable_aiohttp_transport) with _ChunkedSSEServer() as server: handler = AsyncHTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) - client = handler.client - try: - response = await client.send(client.build_request("GET", server.url), stream=True) + response = await handler.post(server.url, stream=True) - # The handler loses its last reference while the body is still streaming. - ref = weakref.ref(handler) - del handler - gc.collect() + ref = weakref.ref(handler) + del handler + gc.collect() + await asyncio.sleep(0) # let any close the finalizer scheduled run - frames = await asyncio.wait_for(_read_frames(response), timeout=READ_TIMEOUT_SECONDS) - assert frames == FRAME_COUNT - assert client.is_closed is False - assert ref() is None, HANDLER_NOT_COLLECTED - finally: - await client.aclose() + assert ref() is not None, STILL_PINNED + assert await asyncio.wait_for(_read_frames(response), timeout=READ_TIMEOUT_SECONDS) == FRAME_COUNT - -@pytest.mark.asyncio -@BOTH_TRANSPORTS -async def test_borrowed_async_client_outlives_its_handler(monkeypatch, disable_aiohttp_transport): - """A caller that keeps only ``handler.client`` can still send requests once the handler is gone. - - This is the shape at litellm/a2a_protocol/main.py (``httpx_client = - _async_handler.client``, handed to the a2a SDK) and at - litellm/proxy/pass_through_endpoints/pass_through_endpoints.py (``async_client - = async_client_obj.client``). Both take the handler from - ``get_async_httpx_client``, so the cache pins it for - ``_DEFAULT_TTL_FOR_HTTPX_CLIENTS`` (one hour) and then lets it go on eviction, - at which point it is collected while the borrowed client is still serving a - longer-lived consumer: ``create_a2a_client`` hands its client to the a2a SDK - and documents it as "create client once, reuse for multiple requests". - """ - _select_transport(monkeypatch, disable_aiohttp_transport) - - with _ChunkedSSEServer(frame_count=1, frame_delay=0.0) as server: - handler = AsyncHTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) - client = handler.client - try: - ref = weakref.ref(handler) - del handler - gc.collect() - # A finalizer would close the client from a task, so let the loop turn. - await asyncio.sleep(0.05) - - assert client.is_closed is False - response = await client.get(server.url) - assert response.status_code == 200 - assert ref() is None, HANDLER_NOT_COLLECTED - finally: - await client.aclose() - - -def test_sync_handler_collection_does_not_close_a_caller_owned_client(monkeypatch): - """A throwaway handler wrapped around someone else's client must not close it. - - litellm/llms/azure/azure.py builds ``HTTPHandler(client=litellm.client_session)`` - for a single image generation and drops it. With a finalizer, that one call - left the user's shared session closed for the rest of the process. - """ - monkeypatch.setattr(litellm, "force_ipv4", False) - - with _ChunkedSSEServer(frame_count=1, frame_delay=0.0) as server: - caller_client = httpx.Client(timeout=httpx.Timeout(10.0, connect=5.0)) - monkeypatch.setattr(litellm, "client_session", caller_client) - try: - handler = HTTPHandler(client=litellm.client_session) - ref = weakref.ref(handler) - del handler - gc.collect() - - assert caller_client.is_closed is False - assert caller_client.get(server.url).status_code == 200 - assert ref() is None, HANDLER_NOT_COLLECTED - finally: - caller_client.close() + del response + gc.collect() + assert ref() is None, NOT_RELEASED def test_sync_stream_survives_handler_collection(monkeypatch): - """A sync response being streamed keeps working after its handler is collected. + """The sync handler closes inline from its finalizer, so a stream must hold it off. litellm/main.py builds a sync handler only for non-streaming calls, commented "Keep this here, otherwise, the httpx.client closes and streaming is @@ -225,26 +174,115 @@ def test_sync_stream_survives_handler_collection(monkeypatch): with _ChunkedSSEServer() as server: handler = HTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) - client = handler.client - try: - response = client.send(client.build_request("GET", server.url), stream=True) + response = handler.post(server.url, stream=True) - # The handler loses its last reference while the body is still streaming. - ref = weakref.ref(handler) - del handler + ref = weakref.ref(handler) + del handler + gc.collect() + assert ref() is not None, STILL_PINNED + + # Joined before counting, as in ``_read_frames``. + chunks = [] + for chunk in response.iter_bytes(): + chunks.append(chunk) gc.collect() + assert b"".join(chunks).count(b"data: frame-") == FRAME_COUNT - # Joined before counting, as in ``_read_frames``. - chunks = [] - for chunk in response.iter_bytes(): - chunks.append(chunk) - gc.collect() + del response + gc.collect() + assert ref() is None, NOT_RELEASED - assert b"".join(chunks).count(b"data: frame-") == FRAME_COUNT - assert client.is_closed is False - assert ref() is None, HANDLER_NOT_COLLECTED - finally: - client.close() + +@pytest.mark.asyncio +@BOTH_TRANSPORTS +async def test_an_abandoned_stream_still_releases_its_handler(monkeypatch, disable_aiohttp_transport): + """A caller that drops a stream unread must not pin the handler for good. + + Tying the handler to the response's own lifetime is what bounds this. No + deadline, and no poll of the connection's state, can tell an abandoned body + from one the upstream is merely slow to finish: httpx leaves the connection + checked out until the response is read or closed, and a legitimate stream is + bounded only by how long the upstream keeps sending. + """ + _select_transport(monkeypatch, disable_aiohttp_transport) + + with _ChunkedSSEServer() as server: + handler = AsyncHTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) + client_ref = weakref.ref(handler.client) + response = await handler.post(server.url, stream=True) + + ref = weakref.ref(handler) + del handler, response + gc.collect() + + assert ref() is None, NOT_RELEASED + await _wait_until( + lambda: client_ref() is None or client_ref().is_closed, + "the client outlived the abandoned stream without being closed", + ) + + +@pytest.mark.asyncio +@BOTH_TRANSPORTS +async def test_the_pool_is_released_once_the_stream_it_carried_ends(monkeypatch, disable_aiohttp_transport): + """Holding the finalizer off must defer the close, not drop it. + + Otherwise a collected handler leaks its pool for every streaming request it + was carrying, and on aiohttp warns "Unclosed client session" when the + collector eventually takes it. The pool and the session are children of the + client, so keeping one here does not inflate the refcount the finalizer + reads, the way keeping the client would. + """ + _select_transport(monkeypatch, disable_aiohttp_transport) + + with _ChunkedSSEServer() as server: + handler = AsyncHTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) + transport = handler.client._transport + if disable_aiohttp_transport: + pool = transport._pool + + def is_released() -> bool: + return pool.connections == [] + else: + session = transport._get_valid_client_session() + + def is_released() -> bool: + return session.closed + + response = await handler.post(server.url, stream=True) + + del handler, transport + gc.collect() + assert not is_released(), "the pool was torn down while it was still carrying a body" + + assert await asyncio.wait_for(_read_frames(response), timeout=READ_TIMEOUT_SECONDS) == FRAME_COUNT + del response + gc.collect() + + await _wait_until(is_released, "the pool outlived the stream it carried, unclosed") + + +@pytest.mark.asyncio +@BOTH_TRANSPORTS +async def test_a_non_streaming_response_does_not_pin_its_handler(monkeypatch, disable_aiohttp_transport): + """Only a body that can still arrive holds the handler. + + A non-streaming response has been read in full by the time ``post`` returns, + so pinning the handler to it would delay every client close behind whatever + the caller goes on to do with the response. + """ + _select_transport(monkeypatch, disable_aiohttp_transport) + + with _ChunkedSSEServer(frame_count=1, frame_delay=0.0) as server: + handler = AsyncHTTPHandler(timeout=httpx.Timeout(10.0, connect=5.0)) + response = await handler.post(server.url) + assert response.status_code == 200 + + ref = weakref.ref(handler) + del handler + gc.collect() + + assert ref() is None, "a fully-read response pinned its handler" @pytest.mark.asyncio @@ -254,30 +292,24 @@ async def test_cached_handler_eviction_does_not_abort_an_in_flight_stream(monkey ``get_async_httpx_client`` caches handlers for an hour. When that TTL expires the cache drops the only reference to a handler whose client is - still streaming -- the production shape of #24929, and the case - ``LLMClientCache`` documents as "may still be in use by in-flight requests". + still streaming -- the production shape of #24929. """ _select_transport(monkeypatch, disable_aiohttp_transport) monkeypatch.setattr(litellm, "in_memory_llm_clients_cache", LLMClientCache()) with _ChunkedSSEServer() as server: handler = get_async_httpx_client(llm_provider=LlmProviders.OPENAI) - client = handler.client - try: - response = await client.send(client.build_request("GET", server.url), stream=True) + response = await handler.post(server.url, stream=True) - # An hour passes: the TTL expires and the cache lets the handler go. - ref = weakref.ref(handler) - litellm.in_memory_llm_clients_cache.flush_cache() - del handler - gc.collect() + # An hour passes: the TTL expires and the cache lets the handler go. + ref = weakref.ref(handler) + litellm.in_memory_llm_clients_cache.flush_cache() + del handler + gc.collect() - frames = await asyncio.wait_for(_read_frames(response), timeout=READ_TIMEOUT_SECONDS) - assert frames == FRAME_COUNT - assert client.is_closed is False + assert ref() is not None, STILL_PINNED + assert await asyncio.wait_for(_read_frames(response), timeout=READ_TIMEOUT_SECONDS) == FRAME_COUNT - # And the evicted client is still usable for the next request. - assert (await client.get(server.url)).status_code == 200 - assert ref() is None, HANDLER_NOT_COLLECTED - finally: - await client.aclose() + del response + gc.collect() + assert ref() is None, NOT_RELEASED From 6e6ab5c0ebafd01700ef23bf02368a70638323ef Mon Sep 17 00:00:00 2001 From: Maximilian Roos Date: Mon, 7 Sep 2026 22:38:32 -0700 Subject: [PATCH 6/7] test(http_handler): cover the anchor in the tree Codecov measures tests/local_testing runs on CircleCI, which does not feed the patch-coverage check, so the anchor read as half-uncovered. These three drive it over a mock transport, which is enough for the lifetime contract even though it cannot reproduce the teardown the loopback tests exist for. --- .../llms/custom_httpx/test_http_handler.py | 68 +++++++++++++++++++ 1 file changed, 68 insertions(+) diff --git a/tests/test_litellm/llms/custom_httpx/test_http_handler.py b/tests/test_litellm/llms/custom_httpx/test_http_handler.py index 9e64bfafa54..4590a361026 100644 --- a/tests/test_litellm/llms/custom_httpx/test_http_handler.py +++ b/tests/test_litellm/llms/custom_httpx/test_http_handler.py @@ -1025,6 +1025,74 @@ def test_handed_out_sync_client_pool_survives_handler_collection(keepalive_serve consumer_client.close() +def _mock_transport() -> httpx.MockTransport: + """Answers anything with a short body, left unread when the caller asked to stream.""" + + def respond(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, request=request, content=b"ab") + + return httpx.MockTransport(respond) + + +@pytest.mark.asyncio +async def test_a_streaming_response_holds_its_handler_until_it_is_released(): + """The finalizer must not run while a body this handler issued can still arrive. + + ``_handler_may_close_client`` cannot see that body: it holds the connection it + reads from and never the client. Anchoring the handler to the response is what + withholds the close, and releasing it is what still delivers one. + """ + handler = AsyncHTTPHandler() + handler.client._transport = _mock_transport() + ref = weakref.ref(handler) + response = await handler.post("https://example.invalid/stream", stream=True) + + del handler + gc.collect() + assert ref() is not None, "the handler was released while its response could still read" + + assert await response.aread() == b"ab" + del response + gc.collect() + assert ref() is None, "the handler outlived the response that was holding it" + + +@pytest.mark.asyncio +async def test_a_fully_read_response_does_not_hold_its_handler(): + """A non-streaming response is complete when ``post`` returns, so it anchors nothing. + + Otherwise every client close would wait on whatever the caller does next with + a response it has already read. + """ + handler = AsyncHTTPHandler() + handler.client._transport = _mock_transport() + ref = weakref.ref(handler) + response = await handler.post("https://example.invalid/whole") + assert response.content == b"ab" + + del handler + gc.collect() + + assert ref() is None, "a fully-read response pinned its handler" + + +def test_a_sync_streaming_response_holds_its_handler_until_it_is_released(): + """The sync finalizer closes inline, so the same anchor has to hold it off.""" + handler = HTTPHandler() + handler.client._transport = _mock_transport() + ref = weakref.ref(handler) + response = handler.post("https://example.invalid/stream", stream=True) + + del handler + gc.collect() + assert ref() is not None, "the handler was released while its response could still read" + + assert response.read() == b"ab" + del response + gc.collect() + assert ref() is None, "the handler outlived the response that was holding it" + + def test_sync_close_leaves_caller_supplied_client_open(): supplied = httpx.Client() handler = HTTPHandler(client=supplied) From ffa08e16d6fb575589309e008c9bbf2ffe96ee63 Mon Sep 17 00:00:00 2001 From: Maximilian Roos Date: Mon, 7 Sep 2026 23:02:02 -0700 Subject: [PATCH 7/7] test(http_handler): parametrize the anchor tests over every streaming send post was the only method the mock-transport tests reached, leaving the anchor in async delete and in sync patch/put/delete uncovered in the tree Codecov measures. Parametrizing also means a method added later is covered here rather than being the one that forgets to anchor. --- .../llms/custom_httpx/test_http_handler.py | 56 +++++++++++-------- 1 file changed, 34 insertions(+), 22 deletions(-) diff --git a/tests/test_litellm/llms/custom_httpx/test_http_handler.py b/tests/test_litellm/llms/custom_httpx/test_http_handler.py index 4590a361026..40420bebee4 100644 --- a/tests/test_litellm/llms/custom_httpx/test_http_handler.py +++ b/tests/test_litellm/llms/custom_httpx/test_http_handler.py @@ -1034,27 +1034,56 @@ def _mock_transport() -> httpx.MockTransport: return httpx.MockTransport(respond) +RELEASED_TOO_EARLY = "the handler was released while its response could still read" +NEVER_RELEASED = "the handler outlived the response that was holding it" + +# Every method that can hand back a body the caller has not read yet, which is +# every one that passes stream= down to send(). Parametrized so a method added +# later is covered here rather than being the one that forgets to anchor. +ASYNC_STREAMING_SENDS = ["post", "delete"] +SYNC_STREAMING_SENDS = ["post", "patch", "put", "delete"] + + @pytest.mark.asyncio -async def test_a_streaming_response_holds_its_handler_until_it_is_released(): +@pytest.mark.parametrize("method", ASYNC_STREAMING_SENDS) +async def test_a_streaming_response_holds_its_handler_until_it_is_released(method): """The finalizer must not run while a body this handler issued can still arrive. ``_handler_may_close_client`` cannot see that body: it holds the connection it reads from and never the client. Anchoring the handler to the response is what - withholds the close, and releasing it is what still delivers one. + withholds the close, and releasing the anchor is what still delivers one. """ handler = AsyncHTTPHandler() handler.client._transport = _mock_transport() ref = weakref.ref(handler) - response = await handler.post("https://example.invalid/stream", stream=True) + response = await getattr(handler, method)("https://example.invalid/stream", stream=True) del handler gc.collect() - assert ref() is not None, "the handler was released while its response could still read" + assert ref() is not None, RELEASED_TOO_EARLY assert await response.aread() == b"ab" del response gc.collect() - assert ref() is None, "the handler outlived the response that was holding it" + assert ref() is None, NEVER_RELEASED + + +@pytest.mark.parametrize("method", SYNC_STREAMING_SENDS) +def test_a_sync_streaming_response_holds_its_handler_until_it_is_released(method): + """The sync finalizer closes inline, so the same anchor has to hold it off.""" + handler = HTTPHandler() + handler.client._transport = _mock_transport() + ref = weakref.ref(handler) + response = getattr(handler, method)("https://example.invalid/stream", stream=True) + + del handler + gc.collect() + assert ref() is not None, RELEASED_TOO_EARLY + + assert response.read() == b"ab" + del response + gc.collect() + assert ref() is None, NEVER_RELEASED @pytest.mark.asyncio @@ -1076,23 +1105,6 @@ async def test_a_fully_read_response_does_not_hold_its_handler(): assert ref() is None, "a fully-read response pinned its handler" -def test_a_sync_streaming_response_holds_its_handler_until_it_is_released(): - """The sync finalizer closes inline, so the same anchor has to hold it off.""" - handler = HTTPHandler() - handler.client._transport = _mock_transport() - ref = weakref.ref(handler) - response = handler.post("https://example.invalid/stream", stream=True) - - del handler - gc.collect() - assert ref() is not None, "the handler was released while its response could still read" - - assert response.read() == b"ab" - del response - gc.collect() - assert ref() is None, "the handler outlived the response that was holding it" - - def test_sync_close_leaves_caller_supplied_client_open(): supplied = httpx.Client() handler = HTTPHandler(client=supplied)