fix: Vertex Anthropic streaming status error hangs (#27310)

* Fix streaming HTTP status error hangs Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com> * Fix sync streaming HTTP status error hangs Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com> * Cap sync streaming error read workers Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com> --------- Co-authored-by: oss-agent-shin <279349115+oss-agent-shin@users.noreply.github.com> Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>
2026-07-16 22:17:21 +00:00 · 2026-05-06 15:32:55 -07:00
parent c15718f9d1
commit aba131d3cf
2 changed files with 149 additions and 6 deletions
@@ -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

@@ -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