mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-24 05:45:11 +00:00
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
This commit is contained in:
Mateo Wang
GitHub
parent
5012083b72
commit
7e6fb09da3
12 changed files with 617 additions and 13 deletions
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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])
|
||||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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) ----
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue