From 7e6fb09da3987c8b75062307695f9083da55feed Mon Sep 17 00:00:00 2001 From: Mateo Wang <277851410+mateo-berri@users.noreply.github.com> Date: Sun, 18 Oct 2026 13:58:34 +0000 Subject: [PATCH] fix(streaming)!: map stream read timeouts to litellm.Timeout on chat, messages, and responses (#47198) * fix(streaming): map stream read timeouts to litellm.Timeout * revert(streaming): move the timeout mapping to a later commit * fix(streaming): map stream read timeouts to litellm.Timeout on chat, messages, and responses * fix(streaming): keep stream fallback and cooldown unchanged for transport timeouts A chat stream served through the Responses bridge now keeps mid-stream fallback on a stall, because the fallback exemption also covers a litellm.Timeout whose cause is an httpx transport timeout. The Responses stream hands the raw transport error to failure callbacks again, so router cooldown sees a stall as it did before. The 408 deployment info now lists stream_timeout. * test(integration): cover a stalled hosted_vllm stream on chat, Responses, and Messages with its spend log row * fix(router): keep a fallback deployment whose stream stalled out of cooldown * test(integration): cover bridged chat, fallback and cooldown for stalled hosted_vllm streams --- .../exception_mapping_utils.py | 5 +- .../litellm_core_utils/streaming_handler.py | 20 +- litellm/responses/streaming_iterator.py | 20 +- litellm/router.py | 11 +- litellm/router_utils/cooldown_handlers.py | 5 + .../router_utils/fallback_event_handlers.py | 8 + .../test_hosted_vllm_stream_stall_wire.py | 364 ++++++++++++++++++ .../test_exception_mapping_utils.py | 20 + .../test_streaming_handler.py | 50 ++- .../unit/responses/test_streaming_iterator.py | 37 ++ .../test_fallback_event_handlers.py | 65 ++++ tests/unit/test_router_exception_redaction.py | 25 ++ 12 files changed, 617 insertions(+), 13 deletions(-) create mode 100644 tests/integration/providers/test_hosted_vllm_stream_stall_wire.py diff --git a/litellm/litellm_core_utils/exception_mapping_utils.py b/litellm/litellm_core_utils/exception_mapping_utils.py index c52bc534058..1f49cc2c4bb 100644 --- a/litellm/litellm_core_utils/exception_mapping_utils.py +++ b/litellm/litellm_core_utils/exception_mapping_utils.py @@ -2551,7 +2551,8 @@ def exception_type( ################################################################################ if ( - "Request Timeout Error" in error_str + isinstance(original_exception, httpx.TimeoutException) + or "Request Timeout Error" in error_str or "Request timed out" in error_str or "Timed out generating response" in error_str or "The read operation timed out" in error_str @@ -2559,7 +2560,7 @@ def exception_type( exception_mapping_worked = True raise Timeout( - message=f"APITimeoutError - Request timed out. Error_str: {error_str}", + message=f"APITimeoutError - Request timed out. Error_str: {error_str or type(mappable_exception).__name__}", model=model, llm_provider=custom_llm_provider, litellm_debug_info=extra_information, diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 78cfb58bedf..e0155f88de1 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -45,7 +45,7 @@ from litellm.types.utils import ( ) from litellm.types.utils import GenericStreamingChunk as GChunk -from ..exceptions import OpenAIError +from ..exceptions import OpenAIError, Timeout from .core_helpers import map_finish_reason, process_response_headers from .exception_mapping_utils import exception_type, with_deployment_api_base from .llm_response_utils.get_api_base import get_api_base @@ -88,6 +88,16 @@ def _apply_provider_fields(response: ModelResponseStream, fields: Mapping[str, o setattr(response, key, value) +def _is_transport_timeout(error: BaseException) -> bool: + return isinstance(error, httpx.TimeoutException) or ( + isinstance(error, Timeout) and isinstance(error.__cause__, httpx.TimeoutException) + ) + + +def _skips_stream_fallback(status_code: int | None) -> bool: + return status_code is not None and 400 <= status_code < 500 and status_code != 429 + + def is_async_iterable(obj: object) -> bool: """ Check if an object is an async iterable (can be used with 'async for'). @@ -2384,6 +2394,8 @@ class CustomStreamWrapper: 429 (rate-limit) is explicitly exempted from the 4xx filter because it is transient and the Router should switch to another model group. + So is a transport read timeout, which maps to a 408 Timeout, including + one an inner Responses stream already mapped for the chat bridge. An error an inner stream already wrapped (the chat-to-Responses bridge consumes a Responses stream) is rebuilt around the provider exception @@ -2448,9 +2460,9 @@ class CustomStreamWrapper: # Raise non-retriable client errors directly (skip fallback). # Exception: 429 (rate-limit) IS retriable/transient — allow it # through so the Router can switch to a different model group. - if mapped_status_code is not None and 400 <= mapped_status_code < 500 and mapped_status_code != 429: - raise mapped_exception - if original_status_code is not None and 400 <= original_status_code < 500 and original_status_code != 429: + if not _is_transport_timeout(e) and ( + _skips_stream_fallback(mapped_status_code) or _skips_stream_fallback(original_status_code) + ): raise mapped_exception raise MidStreamFallbackError( diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 145189616af..3d07f22c81a 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -293,6 +293,15 @@ def _map_stream_error_to_exception( body=error_body, headers=headers, ) + return _mapped_exception(provider_exception, model, custom_llm_provider, logging_obj) + + +def _mapped_exception( + original_exception: Exception, + model: str, + custom_llm_provider: str, + logging_obj: LiteLLMLoggingObj | None, +) -> Exception: litellm_params: Final = ( None if logging_obj is None @@ -302,7 +311,7 @@ def _map_stream_error_to_exception( return litellm.exception_type( model=model, custom_llm_provider=custom_llm_provider, - original_exception=provider_exception, + original_exception=original_exception, completion_kwargs={}, extra_kwargs=with_deployment_api_base(litellm_params), ) @@ -1044,6 +1053,11 @@ class BaseResponsesAPIStreamingIterator: raise error raise self._fallback_error(error) from error + def _raise_for_transport_timeout(self, error: httpx.TimeoutException) -> NoReturn: + self.finished = True + self._handle_failure(error) + raise _mapped_exception(error, self.model or "", self.custom_llm_provider or "", self.logging_obj) from error + def _raise_for_transport_error(self, error: httpx.ReadError | httpx.RemoteProtocolError) -> NoReturn: self._handle_failure(error) if self._output_started: @@ -1140,6 +1154,8 @@ class ResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): if self.completed_response is not None: raise StopAsyncIteration from e self._raise_for_transport_error(e) + except httpx.TimeoutException as e: + self._raise_for_transport_timeout(e) except httpx.HTTPError as e: # Handle HTTP errors self.finished = True @@ -1223,6 +1239,8 @@ class SyncResponsesAPIStreamingIterator(BaseResponsesAPIStreamingIterator): if self.completed_response is not None: raise StopIteration from e self._raise_for_transport_error(e) + except httpx.TimeoutException as e: + self._raise_for_transport_timeout(e) except httpx.HTTPError as e: # Handle HTTP errors self.finished = True diff --git a/litellm/router.py b/litellm/router.py index c30108b4e38..65ba67d9ab8 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -4377,12 +4377,15 @@ class Router: return response except litellm.Timeout as e: - deployment_request_timeout_param: Final = _timeout_debug_deployment_dict.get("litellm_params", {}).get( - "request_timeout", None + deployment_litellm_params: Final[Mapping[str, object]] = _timeout_debug_deployment_dict.get( + "litellm_params", {} ) - deployment_timeout_param = _timeout_debug_deployment_dict.get("litellm_params", {}).get("timeout", None) if litellm.expose_router_debug_in_errors: - e.message += f"\n\nDeployment Info: request_timeout: {deployment_request_timeout_param}\ntimeout: {deployment_timeout_param}" + e.message += ( + f"\n\nDeployment Info: request_timeout: {deployment_litellm_params.get('request_timeout')}" + f"\ntimeout: {deployment_litellm_params.get('timeout')}" + f"\nstream_timeout: {deployment_litellm_params.get('stream_timeout')}" + ) # Set per-deployment num_retries on exception for retry logic if deployment is not None: self._set_deployment_num_retries_on_exception(e, deployment) diff --git a/litellm/router_utils/cooldown_handlers.py b/litellm/router_utils/cooldown_handlers.py index bb64d912c4c..bbd895320bd 100644 --- a/litellm/router_utils/cooldown_handlers.py +++ b/litellm/router_utils/cooldown_handlers.py @@ -26,6 +26,7 @@ from litellm.constants import ( INTERNAL_CALL_ORIGIN_METADATA_KEY, SINGLE_DEPLOYMENT_TRAFFIC_FAILURE_THRESHOLD, ) +from litellm.exceptions import MidStreamFallbackError from litellm.litellm_core_utils.credential_accessor import CredentialAccessor from litellm.router_utils.cooldown_cache import ROUTER_COOLDOWNS_TARGET, CooldownCacheValue from litellm.router_utils.cooldown_callbacks import router_cooldown_event_callback @@ -757,6 +758,10 @@ def is_caller_timeout_408( return (finished - started).total_seconds() >= timeout +def is_stream_stall(exception: BaseException) -> bool: + return isinstance(exception, litellm.Timeout) and isinstance(exception.__cause__, MidStreamFallbackError) + + def is_caller_scoped_auth_failure(deployment: Deployment | None, exception_status: str | int) -> bool: if cast_exception_status_to_int(exception_status) not in (401, 403) or deployment is None: return False diff --git a/litellm/router_utils/fallback_event_handlers.py b/litellm/router_utils/fallback_event_handlers.py index 06bd3329cc1..bc870572f38 100644 --- a/litellm/router_utils/fallback_event_handlers.py +++ b/litellm/router_utils/fallback_event_handlers.py @@ -26,6 +26,7 @@ from litellm.router_utils.cooldown_handlers import ( is_advisor_orchestration_failure, is_caller_scoped_auth_failure, is_caller_timeout_408, + is_stream_stall, set_cooldown_deployments, ) from litellm.router_utils.router_callbacks.track_deployment_metrics import ( @@ -73,6 +74,13 @@ def _trigger_cooldown_for_failed_deployment( ) return + if is_stream_stall(exception): + verbose_router_logger.debug( + "Not triggering cooldown for fallback deployment: its stream stalled on a read timeout, " + "which stays out of cooldown as it does for the primary deployment." + ) + return + exception_status: Final[str | int] = getattr(exception, "status_code", "") # Generic API calls (files, batches, threads, rerank, ...) take a caller-supplied diff --git a/tests/integration/providers/test_hosted_vllm_stream_stall_wire.py b/tests/integration/providers/test_hosted_vllm_stream_stall_wire.py new file mode 100644 index 00000000000..fb58885e011 --- /dev/null +++ b/tests/integration/providers/test_hosted_vllm_stream_stall_wire.py @@ -0,0 +1,364 @@ +import threading +import uuid +from collections.abc import Generator, Iterator, Mapping +from contextlib import ExitStack, contextmanager +from dataclasses import dataclass +from typing import Final, Literal + +import httpx +import pytest +from integration._support.anthropic_sse import SseEvent, event_types, parse_sse +from integration._support.client import ( + Gateway, + Scenario, + eventually, + gateway_from_environment, + object_value, + string_value, +) +from integration._support.database import read_rows +from integration._support.openai_wire import ( + answering_model_discovery, + chat_stream, + openai_error, + posted_targets, + responses_stream, +) +from integration._support.process import graceful_stop_seconds, owned_proxy_process +from integration._support.redis_process import OwnedRedis, owned_redis +from integration._support.wire import Reply, Request, Wire, wire_server +from integration.routing.test_payment_required_cooldown import cooldown_config, cooled, router_settings +from integration.routing.test_payment_required_mapping import deployment +from pydantic import JsonValue + +_BACKEND: Final = "glm-5-fp8" +_HOSTED: Final = f"hosted_vllm/{_BACKEND}" +_BRIDGED: Final = "openai/gpt-5.3-codex" +_PROVIDER_KEY: Final = "synthetic-hosted-vllm-key" +_ANSWER: Final = "scripted answer before the upstream stalls" +_FALLBACK_ANSWER: Final = "scripted answer from the fallback deployment" +_PROMPT: Final = "Keep streaming while the upstream thinks." +_CHAT_TARGET: Final = "/v1/chat/completions" +_RESPONSES_TARGET: Final = "/v1/responses" +_STREAM_TIMEOUT_SECONDS: Final = 1 +_DEPLOYMENT_TIMEOUT_SECONDS: Final = 30 +_PAUSE_SECONDS: Final = 3.0 +_HOLD_SECONDS: Final = 30.0 +_ROW_SECONDS: Final = 60 +_EXTRA: Final[dict[str, JsonValue]] = {"num_retries": 0, "cache": {"no-cache": True}} +_ROWS: Final = 'SELECT status, metadata FROM "LiteLLM_SpendLogs" WHERE model_group=%s' +_EITHER_GROUP_ROWS: Final = 'SELECT status, metadata FROM "LiteLLM_SpendLogs" WHERE model_group IN (%s, %s)' +_OWNED_PROXY: Final = pytest.mark.timeout(2 * graceful_stop_seconds() + 120) + +Endpoint = Literal["chat", "responses", "messages"] +Shape = Literal["prompt", "pause_after_content", "pause_after_role", "stall_before_first", "hold_headers", "fail"] + + +def _stream_parts(target: str, answer: str) -> tuple[bytes, bytes, bytes]: + identity: Final = f"stall-{uuid.uuid4().hex}" + if target == _RESPONSES_TARGET: + return responses_stream(identity, _BACKEND, answer) + return chat_stream(identity, _BACKEND, answer) + + +def _sse(chunks: tuple[bytes, ...], *, pause: float = 0, gate: threading.Event | None = None) -> Reply: + return Reply(content_type="text/event-stream", chunks=chunks, pause_between_chunks=pause, gate_before_first=gate) + + +@dataclass(frozen=True, slots=True) +class _Peer: + shape: Shape + answer: str + release: threading.Event + + def __call__(self, request: Request) -> Reply: + assert request.method == "POST", request + assert request.target in (_CHAT_TARGET, _RESPONSES_TARGET), request + opened, delta, closed = _stream_parts(request.target, self.answer) + match self.shape: + case "fail": + return openai_error(500) + case "hold_headers": + assert self.release.wait(_HOLD_SECONDS), request + return _sse((opened, delta, closed)) + case "pause_after_content": + return _sse((opened + delta, closed), pause=_PAUSE_SECONDS) + case "pause_after_role": + return _sse((opened, delta + closed), pause=_PAUSE_SECONDS) + case "stall_before_first": + return _sse((opened, delta, closed), gate=self.release) + case "prompt": + return _sse((opened, delta, closed)) + + +@contextmanager +def _peer(shape: Shape, answer: str = _ANSWER) -> Generator[Wire]: + release: Final = threading.Event() + with wire_server(answering_model_discovery(_Peer(shape, answer, release))) as wire: + try: + yield wire + finally: + release.set() + + +def _params(wire: Wire, backend: str) -> dict[str, JsonValue]: + return { + "model": backend, + "api_base": f"{wire.url}/v1", + "api_key": _PROVIDER_KEY, + "timeout": _DEPLOYMENT_TIMEOUT_SECONDS, + "stream_timeout": _STREAM_TIMEOUT_SECONDS, + } + + +def _deployment(scenario: Scenario, wire: Wire, backend: str = _HOSTED) -> tuple[str, str]: + group: Final = f"integration-stall-{uuid.uuid4().hex[:12]}" + return group, deployment(scenario, group, _params(wire, backend)) + + +def _path(endpoint: Endpoint) -> str: + match endpoint: + case "chat": + return "/v1/chat/completions" + case "responses": + return "/v1/responses" + case "messages": + return "/v1/messages" + + +def _body(endpoint: Endpoint, model: str, extra: Mapping[str, JsonValue]) -> dict[str, JsonValue]: + match endpoint: + case "chat": + return {"model": model, "messages": [{"role": "user", "content": _PROMPT}], "stream": True, **extra} + case "responses": + return {"model": model, "input": _PROMPT, "stream": True, **extra} + case "messages": + return { + "model": model, + "messages": [{"role": "user", "content": _PROMPT}], + "max_tokens": 16, + "stream": True, + **extra, + } + + +def _streamed(gateway: Gateway, endpoint: Endpoint, model: str, **extra: JsonValue) -> httpx.Response: + return gateway.request("POST", _path(endpoint), _body(endpoint, model, {**_EXTRA, **extra})) + + +def _upstream_targets(wire: Wire) -> tuple[str, ...]: + eventually(wire.received.qsize, lambda size: size >= 1, 10) + return posted_targets(wire) + + +def _chat_errors(events: tuple[SseEvent, ...]) -> tuple[dict[str, JsonValue], ...]: + return tuple(object_value(event.data["error"]) for event in events if "error" in event.data) + + +def _error_class(message: JsonValue) -> str: + return string_value(message).split(":", 1)[0] + + +def _failure_row(query: str, groups: tuple[str, ...]) -> dict[str, JsonValue]: + rows: Final = eventually(lambda: read_rows(query, groups), lambda found: len(found) >= 1, seconds=_ROW_SECONDS) + assert [row["status"] for row in rows] == ["failure"], rows + return object_value(object_value(rows[0]["metadata"])["error_information"]) + + +def _logged_failure(model: str) -> tuple[JsonValue, JsonValue]: + error: Final = _failure_row(_ROWS, (model,)) + return error["error_code"], error["error_class"] + + +@pytest.mark.parametrize( + ("backend", "target"), + ( + pytest.param(_HOSTED, _CHAT_TARGET, id="hosted-vllm"), + pytest.param(_BRIDGED, _RESPONSES_TARGET, id="bridged-to-responses"), + ), +) +def test_a_chat_stream_that_stalls_past_its_stream_timeout_ends_with_a_408_timeout_and_logs_it( + gateway: Gateway, backend: str, target: str +) -> None: + with _peer("pause_after_content") as wire, gateway.scenario() as scenario: + model, _ = _deployment(scenario, wire, backend) + response: Final = _streamed(gateway, "chat", model) + assert response.status_code == 200, response.text + assert _ANSWER in response.text, response.text + events: Final = parse_sse(response.text) + assert _chat_errors(events[:-1]) == () and "error" in events[-1].data, response.text + error: Final = object_value(events[-1].data["error"]) + observed: Final = (error["code"], _error_class(error["message"]), *_logged_failure(model)) + assert observed == ("408", "litellm.Timeout", "408", "Timeout"), (error, observed) + assert _upstream_targets(wire) == (target,) + + +def test_a_native_responses_stream_that_stalls_past_its_stream_timeout_fails_with_request_timeout_and_logs_it( + gateway: Gateway, +) -> None: + with _peer("pause_after_content") as wire, gateway.scenario() as scenario: + model, _ = _deployment(scenario, wire) + response: Final = _streamed(gateway, "responses", model) + assert response.status_code == 200, response.text + events: Final = parse_sse(response.text) + types: Final = event_types(events) + assert "response.output_text.delta" in types, response.text + assert types[-1] == "response.failed" and "response.completed" not in types, response.text + failed: Final = object_value(object_value(events[-1].data["response"])["error"]) + observed: Final = (failed["code"], *_logged_failure(model)) + assert observed == ("request_timeout", "408", "Timeout"), (failed, observed) + assert _upstream_targets(wire) == (_RESPONSES_TARGET,) + + +def test_a_messages_stream_that_stalls_past_its_stream_timeout_ends_with_a_timeout_error_event_and_logs_it( + gateway: Gateway, +) -> None: + with _peer("pause_after_content") as wire, gateway.scenario() as scenario: + model, _ = _deployment(scenario, wire) + response: Final = _streamed(gateway, "messages", model) + assert response.status_code == 200, response.text + events: Final = parse_sse(response.text) + types: Final = event_types(events) + assert "content_block_delta" in types and "message_stop" not in types, response.text + assert types[-1] == "error", response.text + error: Final = object_value(events[-1].data["error"]) + observed: Final = (_error_class(error["message"]), *_logged_failure(model)) + assert observed == ("litellm.Timeout", "408", "Timeout"), (error, observed) + assert _upstream_targets(wire) == (_CHAT_TARGET,) + + +def test_a_chat_stream_that_never_pauses_is_answered_in_full_under_the_same_stream_timeout(gateway: Gateway) -> None: + with _peer("prompt") as wire, gateway.scenario() as scenario: + model, _ = _deployment(scenario, wire) + response: Final = _streamed(gateway, "chat", model) + assert response.status_code == 200, response.text + assert _ANSWER in response.text, response.text + assert response.text.rstrip().endswith("data: [DONE]"), response.text + events: Final = parse_sse(response.text) + assert _chat_errors(events) == (), response.text + assert _finish_reasons(events) == ("stop",), response.text + assert _upstream_targets(wire) == (_CHAT_TARGET,) + rows: Final = eventually(lambda: read_rows(_ROWS, (model,)), lambda found: len(found) >= 1, _ROW_SECONDS) + assert [row["status"] for row in rows] == ["success"], rows + + +@pytest.mark.parametrize( + ("endpoint", "shape"), + ( + pytest.param("chat", "pause_after_role", id="chat-after-the-role-chunk"), + pytest.param("messages", "stall_before_first", id="messages-before-the-first-chunk"), + ), +) +def test_a_stream_that_stalls_before_any_content_falls_back_to_the_next_deployment( + gateway: Gateway, endpoint: Endpoint, shape: Shape +) -> None: + with _peer(shape) as stalled, _peer("prompt", _FALLBACK_ANSWER) as healthy, gateway.scenario() as scenario: + primary, _ = _deployment(scenario, stalled) + fallback, _ = _deployment(scenario, healthy) + response: Final = _streamed(gateway, endpoint, primary, fallbacks=[fallback]) + assert response.status_code == 200, response.text + assert _FALLBACK_ANSWER in response.text and _ANSWER not in response.text, response.text + events: Final = parse_sse(response.text) + assert _chat_errors(events) == () and _completed(endpoint, events), response.text + assert (_upstream_targets(stalled), _upstream_targets(healthy)) == ((_CHAT_TARGET,), (_CHAT_TARGET,)) + + +@dataclass(frozen=True, slots=True) +class _CooldownRig: + gateway: Gateway + cache: OwnedRedis + + +@pytest.fixture(scope="module") +def cooldown_rig(tmp_path_factory: pytest.TempPathFactory) -> Iterator[_CooldownRig]: + directory: Final = tmp_path_factory.mktemp("stream-stall-cooldown") + with ExitStack() as stack: + upstream: Final = stack.enter_context(gateway_from_environment()) + cache: Final = stack.enter_context(owned_redis(directory)) + owned: Final = stack.enter_context( + owned_proxy_process( + upstream, + directory, + {"PROXY_CONFIG_RELOAD_INTERVAL_SECONDS": "2"}, + config=cooldown_config(directory, router_settings(cache, allowed_fails=0), prometheus=False), + ) + ) + yield _CooldownRig(owned.gateway, cache) + + +@_OWNED_PROXY +@pytest.mark.parametrize( + ("endpoint", "backend", "shape"), + ( + pytest.param("chat", _HOSTED, "pause_after_content", id="chat"), + pytest.param("chat", _HOSTED, "stall_before_first", id="chat-before-the-first-chunk"), + pytest.param("responses", _HOSTED, "pause_after_content", id="responses"), + pytest.param("chat", _BRIDGED, "pause_after_content", id="chat-bridged-to-responses"), + ), +) +def test_a_deployment_whose_stream_stalls_stays_out_of_cooldown( + cooldown_rig: _CooldownRig, endpoint: Endpoint, backend: str, shape: Shape +) -> None: + with _peer(shape) as wire, cooldown_rig.gateway.scenario() as scenario: + group, identity = _deployment(scenario, wire, backend) + response: Final = _streamed(cooldown_rig.gateway, endpoint, group) + error: Final = _failure_row(_ROWS, (group,)) + assert not cooled(cooldown_rig.cache, identity), (error, response.text) + + +@_OWNED_PROXY +def test_a_deployment_whose_stream_times_out_before_it_opens_is_cooled_down(cooldown_rig: _CooldownRig) -> None: + with _peer("hold_headers") as wire, cooldown_rig.gateway.scenario() as scenario: + group, identity = _deployment(scenario, wire) + response: Final = _streamed(cooldown_rig.gateway, "chat", group) + assert response.status_code == 408, response.text + assert eventually(lambda: cooled(cooldown_rig.cache, identity), bool, _ROW_SECONDS) + + +def _fallback_cooled(cooldown_rig: _CooldownRig, fallback_shape: Shape) -> tuple[bool, dict[str, JsonValue], str]: + with ( + _peer("fail") as failing, + _peer(fallback_shape) as fallback_wire, + cooldown_rig.gateway.scenario() as scenario, + ): + primary, _ = _deployment(scenario, failing) + fallback, identity = _deployment(scenario, fallback_wire) + response: Final = _streamed(cooldown_rig.gateway, "chat", primary, fallbacks=[fallback]) + error: Final = _failure_row(_EITHER_GROUP_ROWS, (primary, fallback)) + assert _upstream_targets(fallback_wire) == (_CHAT_TARGET,), response.text + return cooled(cooldown_rig.cache, identity), error, response.text + + +@_OWNED_PROXY +def test_a_fallback_deployment_whose_stream_stalls_stays_out_of_cooldown(cooldown_rig: _CooldownRig) -> None: + is_cooled, error, text = _fallback_cooled(cooldown_rig, "stall_before_first") + assert not is_cooled, (error, text) + + +@_OWNED_PROXY +def test_a_fallback_deployment_whose_stream_times_out_before_it_opens_is_cooled_down( + cooldown_rig: _CooldownRig, +) -> None: + is_cooled, error, text = _fallback_cooled(cooldown_rig, "hold_headers") + assert is_cooled, (error, text) + + +def _completed(endpoint: Endpoint, events: tuple[SseEvent, ...]) -> bool: + match endpoint: + case "chat": + return _finish_reasons(events) == ("stop",) + case "messages": + return event_types(events)[-1:] == ("message_stop",) + case "responses": + return event_types(events)[-1:] == ("response.completed",) + + +def _finish_reasons(events: tuple[SseEvent, ...]) -> tuple[JsonValue, ...]: + choices: Final = tuple(_first_choice(event.data) for event in events if event.data.get("choices")) + return tuple(choice["finish_reason"] for choice in choices if choice.get("finish_reason")) + + +def _first_choice(data: Mapping[str, JsonValue]) -> dict[str, JsonValue]: + choices: Final = data["choices"] + assert isinstance(choices, list), data + return object_value(choices[0]) diff --git a/tests/unit/litellm_core_utils/test_exception_mapping_utils.py b/tests/unit/litellm_core_utils/test_exception_mapping_utils.py index 731e842090a..5ee14ff530d 100644 --- a/tests/unit/litellm_core_utils/test_exception_mapping_utils.py +++ b/tests/unit/litellm_core_utils/test_exception_mapping_utils.py @@ -2781,3 +2781,23 @@ def test_missing_aws_dependency_is_not_mapped_to_provider_failure(provider, depe def test_non_aws_import_failure_keeps_provider_mapping(failure): with pytest.raises(litellm.APIConnectionError): exception_type(model="test-model", original_exception=failure, custom_llm_provider="openai") + + +@pytest.mark.parametrize( + "transport_timeout,reason", + [ + (httpx.ReadTimeout(""), "ReadTimeout"), + (httpx.ReadTimeout("Timeout on reading data from socket"), "Timeout on reading data from socket"), + (httpx.TimeoutException(""), "TimeoutException"), + ], + ids=["httpx_read_timeout", "aiohttp_read_timeout", "timeout_base_class"], +) +@pytest.mark.parametrize("provider", ["hosted_vllm", "openai", "anthropic"]) +def test_transport_timeout_maps_to_timeout( + transport_timeout: httpx.TimeoutException, reason: str, provider: str +) -> None: + with pytest.raises(litellm.Timeout) as exc_info: + exception_type(model="glm", original_exception=transport_timeout, custom_llm_provider=provider) + + assert exc_info.value.status_code == 408 + assert f"Request timed out. Error_str: {reason}" in str(exc_info.value) diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index c07fe2d1c31..50bcc14b681 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -2,7 +2,7 @@ import asyncio import json import threading import time -from typing import AsyncIterator, Final, Iterator, Literal, NoReturn, Optional +from typing import AsyncIterator, Callable, Final, Iterator, Literal, NoReturn, Optional from unittest.mock import AsyncMock, MagicMock, Mock, patch import httpx @@ -1382,7 +1382,53 @@ async def test_async_streaming_read_timeout_triggers_midstream_fallback( await response.__anext__() assert excinfo.value.is_pre_first_chunk is True - assert isinstance(excinfo.value.original_exception, Exception) + assert isinstance(excinfo.value.original_exception, litellm.Timeout) + + +def _raw_read_timeout() -> NoReturn: + raise httpx.ReadTimeout("") + + +def _bridged_read_timeout() -> NoReturn: + raise litellm.Timeout( + message="APITimeoutError - Request timed out. Error_str: ReadTimeout", model="glm", llm_provider="hosted_vllm" + ) from httpx.ReadTimeout("") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("stall", [_raw_read_timeout, _bridged_read_timeout], ids=["raw", "responses_bridge"]) +async def test_async_streaming_read_timeout_after_first_chunk_surfaces_timeout( + logging_obj: Logging, stall: Callable[[], NoReturn] +): + from litellm.exceptions import MidStreamFallbackError + + class _OneChunkThenStall: + def __init__(self) -> None: + self.sent = False + + def __aiter__(self) -> "_OneChunkThenStall": + return self + + async def __anext__(self) -> ModelResponseStream: + if self.sent: + stall() + self.sent = True + return ModelResponseStream(choices=[{"index": 0, "delta": {"content": "hello"}}]) + + response: Final = CustomStreamWrapper( + completion_stream=_OneChunkThenStall(), + model="glm", + logging_obj=logging_obj, + custom_llm_provider="hosted_vllm", + ) + + with pytest.raises(MidStreamFallbackError) as excinfo: + async for _ in response: + pass + + assert excinfo.value.is_pre_first_chunk is False + assert isinstance(excinfo.value.original_exception, litellm.Timeout) + assert excinfo.value.original_exception.status_code == 408 _AZURE_DEPLOYMENT_API_BASE: Final = "https://mid-stream-debug.openai.azure.com" diff --git a/tests/unit/responses/test_streaming_iterator.py b/tests/unit/responses/test_streaming_iterator.py index 7cdac6c7581..7eb3115117d 100644 --- a/tests/unit/responses/test_streaming_iterator.py +++ b/tests/unit/responses/test_streaming_iterator.py @@ -323,6 +323,43 @@ async def test_transport_error_after_output_started_is_not_fallback_eligible(): _assert_failure_logged_once(logging_obj, trailing_error) +_READ_TIMEOUT_PREFIXES: Final = [ + pytest.param([], id="before-output"), + pytest.param(_PARTIAL_OUTPUT_EVENTS, id="after-output"), +] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("prefix", _READ_TIMEOUT_PREFIXES) +async def test_transport_read_timeout_raises_litellm_timeout(prefix): + logging_obj: Final = _failure_tracking_logging_obj() + stall: Final = httpx.ReadTimeout("") + iterator: Final = _make_iterator(sse_events=prefix, logging_obj=logging_obj, trailing_error=stall) + + with pytest.raises(litellm.Timeout) as exc_info: + async for _ in iterator: + pass + + assert exc_info.value.status_code == 408 + assert "Request timed out. Error_str: ReadTimeout" in str(exc_info.value) + assert exc_info.value.__cause__ is stall + _assert_failure_logged_once(logging_obj, stall) + + +@pytest.mark.parametrize("prefix", _READ_TIMEOUT_PREFIXES) +def test_sync_transport_read_timeout_raises_litellm_timeout(prefix): + iterator: Final = _make_sync_iterator( + sse_events=prefix, logging_obj=_logging_obj_stub(), trailing_error=httpx.ReadTimeout("") + ) + + with pytest.raises(litellm.Timeout) as exc_info: + for _ in iterator: + pass + + assert exc_info.value.status_code == 408 + assert "Request timed out. Error_str: ReadTimeout" in str(exc_info.value) + + @pytest.mark.asyncio @pytest.mark.parametrize("trailer", [[], [_DONE_MARKER]], ids=["eof", "done-marker"]) async def test_stream_ending_after_partial_output_without_terminal_event_raises(trailer): diff --git a/tests/unit/router_utils/test_fallback_event_handlers.py b/tests/unit/router_utils/test_fallback_event_handlers.py index 637fdfa6873..383a1e25e98 100644 --- a/tests/unit/router_utils/test_fallback_event_handlers.py +++ b/tests/unit/router_utils/test_fallback_event_handlers.py @@ -1,6 +1,7 @@ import asyncio import importlib import json +import time from collections.abc import AsyncIterator from datetime import datetime, timedelta from types import MappingProxyType @@ -16,6 +17,8 @@ from litellm import Router from litellm.integrations.custom_logger import CustomLogger from litellm.llms.anthropic.pass_through.messages.fake_stream_iterator import FakeAnthropicMessagesStreamIterator from litellm.litellm_core_utils import get_llm_provider_logic +from litellm.litellm_core_utils.litellm_logging import Logging +from litellm.litellm_core_utils.streaming_handler import CustomStreamWrapper from litellm.router_utils.cooldown_handlers import mark_advisor_orchestration_failure from litellm.router_utils.fallback_event_handlers import ( MID_STREAM_FALLBACK_CONTROLS_KEY, @@ -1083,6 +1086,68 @@ class TestTriggerCooldownForFailedDeployment: mock_set_cooldown.assert_called_once() +async def _stalled_stream_error() -> Exception: + from litellm.router import _aread_first_chunk + + async def _stall(**kwargs: object) -> NoReturn: + raise httpx.ReadTimeout("Timeout on reading data from socket") + + stream: Final = CustomStreamWrapper( + completion_stream=None, + model="gpt-5.6", + logging_obj=Logging( + model="gpt-5.6", + messages=[{"role": "user", "content": "Hey"}], + stream=True, + call_type="acompletion", + start_time=time.time(), + litellm_call_id="stalled-fallback", + function_id="stalled-fallback", + ), + custom_llm_provider="openai", + make_call=_stall, + ) + try: + await _aread_first_chunk(stream) + except litellm.Timeout as stalled: + return stalled + raise AssertionError("a stream that stalls before its first chunk must raise litellm.Timeout") + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("stream_stalled", "cooled"), + [(True, ()), (False, ("fallback-deployment",))], + ids=["stream-stall", "non-stream-timeout"], +) +async def test_a_fallback_deployment_whose_stream_stalled_stays_out_of_cooldown( + stream_stalled: bool, cooled: tuple[str, ...] +) -> None: + router: Final = litellm.Router( + model_list=[ + { + "model_name": "fallback-model", + "litellm_params": {"model": "openai/gpt-5.6", "api_key": "sk-fake"}, + "model_info": {"id": "fallback-deployment"}, + } + ], + allowed_fails=0, + cooldown_time=60, + num_retries=0, + ) + error: Final = ( + await _stalled_stream_error() + if stream_stalled + else litellm.Timeout(message="timeout", model="gpt-5.6", llm_provider="openai") + ) + error.failed_deployment_id = "fallback-deployment" + + _trigger_cooldown_for_failed_deployment(litellm_router=router, kwargs={}, exception=error) + + active: Final = router.cooldown_cache.get_active_cooldowns(model_ids=["fallback-deployment"], parent_otel_span=None) + assert tuple(entry[0] for entry in active) == cooled + + class TestRunAsyncFallbackTriggersCooldown: class RouterWithLoggingKwarg: fallback_access_check = None diff --git a/tests/unit/test_router_exception_redaction.py b/tests/unit/test_router_exception_redaction.py index 5b52496fc1c..5f9a5ba06a8 100644 --- a/tests/unit/test_router_exception_redaction.py +++ b/tests/unit/test_router_exception_redaction.py @@ -37,6 +37,8 @@ non-content-policy error, regardless of whether `fallbacks` is set. from __future__ import annotations +from typing import Final + import pytest import litellm @@ -313,6 +315,29 @@ async def test_flag_on_shows_deployment_timeout_debug(monkeypatch: pytest.Monkey assert "Deployment Info: request_timeout:" in msg, msg +@pytest.mark.asyncio +async def test_flag_on_deployment_timeout_debug_names_stream_timeout(monkeypatch: pytest.MonkeyPatch): + monkeypatch.setattr(litellm, "expose_router_debug_in_errors", True) + router: Final = Router( + model_list=[ + { + "model_name": _INTERNAL_MODEL_GROUP_NAME, + "litellm_params": {"model": "gpt-4o", "api_key": "key", "stream_timeout": 7}, + }, + ], + num_retries=0, + ) + with pytest.raises(litellm.Timeout) as excinfo: + await router.acompletion( + model=_INTERNAL_MODEL_GROUP_NAME, + messages=[{"role": "user", "content": "hi"}], + stream=True, + mock_timeout=True, + timeout=0.001, + ) + assert "\nstream_timeout: 7" in excinfo.value.message, excinfo.value.message + + # --- Site 3: ContentPolicyViolationError fallback hint (no fallback set) ----