From aba131d3cf43a2aff3ae466be18cb7ffc1efa33f Mon Sep 17 00:00:00 2001 From: ishaan-berri <155045088+ishaan-berri@users.noreply.github.com> Date: Wed, 6 May 2026 15:32:55 -0700 Subject: [PATCH] fix: Vertex Anthropic streaming status error hangs (#27310) * Fix streaming HTTP status error hangs Co-authored-by: ishaan-berri * Fix sync streaming HTTP status error hangs Co-authored-by: ishaan-berri * Cap sync streaming error read workers Co-authored-by: ishaan-berri --------- Co-authored-by: oss-agent-shin <279349115+oss-agent-shin@users.noreply.github.com> Co-authored-by: ishaan-berri --- litellm/llms/custom_httpx/http_handler.py | 53 +++++++-- .../llms/custom_httpx/test_http_handler.py | 102 ++++++++++++++++++ 2 files changed, 149 insertions(+), 6 deletions(-) diff --git a/litellm/llms/custom_httpx/http_handler.py b/litellm/llms/custom_httpx/http_handler.py index dd955c23d2..af18c66667 100644 --- a/litellm/llms/custom_httpx/http_handler.py +++ b/litellm/llms/custom_httpx/http_handler.py @@ -1,4 +1,5 @@ import asyncio +import concurrent.futures import inspect import os import socket @@ -133,6 +134,11 @@ _DEFAULT_TIMEOUT = httpx.Timeout( timeout=COMPLETION_HTTP_FALLBACK_SECONDS, connect=HTTP_HANDLER_CONNECT_TIMEOUT_SECONDS, ) +_STREAMING_ERROR_BODY_READ_TIMEOUT_SECONDS = 5.0 +_STREAMING_ERROR_BODY_READ_EXECUTOR = concurrent.futures.ThreadPoolExecutor( + max_workers=50, + thread_name_prefix="litellm-streaming-error-body-read", +) def _prepare_request_data_and_content( @@ -386,17 +392,30 @@ def _safe_get_response_text(response: httpx.Response) -> str: return "" -async def _safe_aread_response(response: httpx.Response) -> bytes: +async def _safe_aread_response( + response: httpx.Response, timeout: Optional[float] = None +) -> bytes: """Safely read async response body, falling back to empty bytes on errors.""" try: + if timeout is not None: + return await asyncio.wait_for(response.aread(), timeout=timeout) return await response.aread() except Exception: return b"" -def _safe_read_response(response: httpx.Response) -> bytes: +def _safe_read_response( + response: httpx.Response, timeout: Optional[float] = None +) -> bytes: """Safely read sync response body, falling back to empty bytes on errors.""" try: + if timeout is not None: + future = _STREAMING_ERROR_BODY_READ_EXECUTOR.submit(response.read) + try: + return future.result(timeout=timeout) + except Exception: + response.close() + return b"" return response.read() except Exception: return b"" @@ -405,8 +424,19 @@ def _safe_read_response(response: httpx.Response) -> bytes: def _raise_masked_sync_error(e: httpx.HTTPStatusError, stream: bool) -> None: """Raise a MaskedHTTPStatusError for sync HTTP handlers.""" if stream: - _body = mask_sensitive_info(_safe_read_response(e.response)) - raise MaskedHTTPStatusError(e, message=_body, text=_body) from None + try: + _body = mask_sensitive_info( + _safe_read_response( + e.response, + timeout=_STREAMING_ERROR_BODY_READ_TIMEOUT_SECONDS, + ) + ) + raise MaskedHTTPStatusError(e, message=_body, text=_body) from None + finally: + try: + e.response.close() + except Exception: + pass _text = mask_sensitive_info(_safe_get_response_text(e.response)) raise MaskedHTTPStatusError(e, message=_text, text=_text) from None @@ -414,8 +444,19 @@ def _raise_masked_sync_error(e: httpx.HTTPStatusError, stream: bool) -> None: async def _raise_masked_async_error(e: httpx.HTTPStatusError, stream: bool) -> None: """Raise a MaskedHTTPStatusError for async HTTP handlers.""" if stream: - _body = mask_sensitive_info(await _safe_aread_response(e.response)) - raise MaskedHTTPStatusError(e, message=_body, text=_body) from None + try: + _body = mask_sensitive_info( + await _safe_aread_response( + e.response, + timeout=_STREAMING_ERROR_BODY_READ_TIMEOUT_SECONDS, + ) + ) + raise MaskedHTTPStatusError(e, message=_body, text=_body) from None + finally: + try: + await e.response.aclose() + except Exception: + pass _text = mask_sensitive_info(_safe_get_response_text(e.response)) raise MaskedHTTPStatusError(e, message=_text, text=_text) from None 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 dd52304a70..26f50e8e49 100644 --- a/tests/test_litellm/llms/custom_httpx/test_http_handler.py +++ b/tests/test_litellm/llms/custom_httpx/test_http_handler.py @@ -1,8 +1,10 @@ +import asyncio import io import os import pathlib import ssl import sys +import threading from unittest.mock import MagicMock, patch import certifi @@ -18,11 +20,111 @@ from litellm.llms.custom_httpx.aiohttp_transport import LiteLLMAiohttpTransport from litellm.llms.custom_httpx.http_handler import ( AsyncHTTPHandler, HTTPHandler, + MaskedHTTPStatusError, _get_httpx_client, get_ssl_configuration, ) +@pytest.mark.asyncio +async def test_async_post_streaming_status_error_should_not_wait_forever_for_body( + monkeypatch, +): + """ + Vertex Anthropic streamRawPredict can return a pre-stream 4xx where the + streamed error body never terminates. The handler must still surface the + status promptly instead of blocking the downstream client. + """ + + class HangingErrorStream(httpx.AsyncByteStream): + async def __aiter__(self): + await asyncio.Event().wait() + if False: + yield b"" + + async def mock_handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 400, + request=request, + headers={"content-type": "application/json"}, + stream=HangingErrorStream(), + ) + + monkeypatch.setattr( + "litellm.llms.custom_httpx.http_handler._STREAMING_ERROR_BODY_READ_TIMEOUT_SECONDS", + 0.01, + ) + + litellm_handler = AsyncHTTPHandler() + await litellm_handler.client.aclose() + litellm_handler.client = httpx.AsyncClient( + transport=httpx.MockTransport(mock_handler) + ) + try: + with pytest.raises(MaskedHTTPStatusError) as exc_info: + await asyncio.wait_for( + litellm_handler.post( + "https://vertex.example/streamRawPredict", + stream=True, + ), + timeout=0.2, + ) + + assert exc_info.value.status_code == 400 + assert exc_info.value.response.status_code == 400 + finally: + await litellm_handler.close() + + +def test_sync_post_streaming_status_error_should_not_wait_forever_for_body( + monkeypatch, +): + """ + Keep the sync streaming error path aligned with the async path so a + non-terminating streamed error body cannot block a worker thread forever. + """ + + class HangingSyncErrorStream(httpx.SyncByteStream): + def __init__(self): + self.closed_event = threading.Event() + + def __iter__(self): + self.closed_event.wait() + if False: + yield b"" + + def close(self): + self.closed_event.set() + + def mock_handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 400, + request=request, + headers={"content-type": "application/json"}, + stream=HangingSyncErrorStream(), + ) + + monkeypatch.setattr( + "litellm.llms.custom_httpx.http_handler._STREAMING_ERROR_BODY_READ_TIMEOUT_SECONDS", + 0.01, + ) + + litellm_handler = HTTPHandler() + litellm_handler.client.close() + litellm_handler.client = httpx.Client(transport=httpx.MockTransport(mock_handler)) + try: + with pytest.raises(MaskedHTTPStatusError) as exc_info: + litellm_handler.post( + "https://vertex.example/streamRawPredict", + stream=True, + ) + + assert exc_info.value.status_code == 400 + assert exc_info.value.response.status_code == 400 + finally: + litellm_handler.close() + + @pytest.mark.asyncio async def test_ssl_security_level(monkeypatch): # Ensure aiohttp transport is enabled for this test