mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-09 03:18:44 +00:00
test(e2e): retry provider-transient statuses at the transport with bounded backoff (#35824)
* test(e2e): retry provider-transient statuses at the transport with bounded backoff The Anthropic passthrough cost test failed a full-suite run on a real 529 overloaded_error. Passthrough routes forward provider responses verbatim and bypass the router's num_retries, so provider blips reach the harness only on those paths. Following standard practice, the retry is scoped to the dependency boundary instead of rerunning tests: only the enumerated transient statuses (500/502/503/504/529, the set production SDKs retry by default) are retried, with bounded exponential backoff and a printed line per retry so flakiness stays visible in run logs. 429 is deliberately excluded: the quota suites assert the proxy's own rate-limit and budget 429s, and a transport that absorbed them would break those tests. Network errors and timeouts are not retried either, so a hang surfaces as a hang. request_with_retry takes injected callables, and the new harness tests pin the contract with protocol fakes, no monkeypatching * test(e2e): narrow the transport retry to 529, the one status the proxy cannot emit Greptile's review is right that status-only classification could absorb an intermittently failing proxy: at the transport a 500/502/503/504 from the proxy is indistinguishable from one it relayed, and the proxy is the system under test. 529 is the only status litellm provably never originates (Anthropic's overload signal, forwarded verbatim on passthrough) and the only transient observed across the full-suite runs, so the set shrinks to exactly that. The canary tests now also pin 500/502/503/504 as never retried
This commit is contained in:
yuneng-jiang
GitHub
parent
05204795cf
commit
e64536c425
2 changed files with 189 additions and 46 deletions
|
|
@ -10,7 +10,9 @@ requests itself imports.
|
|||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Generic, Iterator, Literal, NewType, TypeVar, cast
|
||||
import time
|
||||
from collections.abc import Callable
|
||||
from typing import Generic, Iterator, Literal, NewType, Protocol, TypeVar, cast
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
|
@ -301,6 +303,49 @@ def _params(params: BaseModel | None) -> dict[str, str]:
|
|||
return {key: str(value) for key, value in dumped.items()}
|
||||
|
||||
|
||||
TRANSIENT_STATUSES: frozenset[int] = frozenset({529})
|
||||
RETRY_ATTEMPTS: int = 3
|
||||
RETRY_BACKOFF_SECONDS: float = 0.5
|
||||
|
||||
|
||||
class RetryableResponse(Protocol):
|
||||
status_code: int
|
||||
|
||||
def close(self) -> None: ...
|
||||
|
||||
|
||||
def request_with_retry[T: RetryableResponse](
|
||||
issue: Callable[[], T], *, sleep: Callable[[float], None] = time.sleep
|
||||
) -> T:
|
||||
"""Bounded retry on statuses attributable to the PROVIDER, never the proxy.
|
||||
|
||||
The system under test is the proxy, so the transport may only absorb
|
||||
statuses the proxy itself cannot emit; today that is exactly 529, the
|
||||
Anthropic overloaded_error passed through verbatim (their own SDK retries
|
||||
it too). 500/502/503/504 stay first-class failures: at this layer a 5xx
|
||||
from the proxy is indistinguishable from one it relayed, and retrying them
|
||||
could mask an intermittently failing proxy. Widen the set only for a
|
||||
status litellm provably never originates, with an observed flake in hand.
|
||||
|
||||
Also deliberately NOT retried: 429, because this suite asserts the proxy's
|
||||
own rate-limit and budget 429s; network errors and timeouts, because a
|
||||
hang should surface as a hang instead of doubling the wall clock. Every
|
||||
retry prints, so flakiness stays visible in the run log instead of
|
||||
vanishing into green."""
|
||||
for attempt in range(1, RETRY_ATTEMPTS):
|
||||
resp = issue()
|
||||
if resp.status_code not in TRANSIENT_STATUSES:
|
||||
return resp
|
||||
delay = RETRY_BACKOFF_SECONDS * (1 << (attempt - 1))
|
||||
print(
|
||||
f"e2e-http: transient {resp.status_code}; retry {attempt}/{RETRY_ATTEMPTS - 1} in {delay}s",
|
||||
flush=True,
|
||||
)
|
||||
resp.close()
|
||||
sleep(delay)
|
||||
return issue()
|
||||
|
||||
|
||||
def _classify[R: BaseModel](
|
||||
resp: requests.Response, response_type: type[R]
|
||||
) -> Result[R]:
|
||||
|
|
@ -325,11 +370,13 @@ def post[R: BaseModel](
|
|||
timeout: float = 30.0,
|
||||
) -> Result[R]:
|
||||
try:
|
||||
resp = requests.post(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
resp = request_with_retry(
|
||||
lambda: requests.post(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
)
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
return NetworkError(message=str(exc))
|
||||
|
|
@ -345,11 +392,13 @@ def get[R: BaseModel](
|
|||
timeout: float = 30.0,
|
||||
) -> Result[R]:
|
||||
try:
|
||||
resp = requests.get(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
params=params.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
resp = request_with_retry(
|
||||
lambda: requests.get(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
params=params.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
)
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
return NetworkError(message=str(exc))
|
||||
|
|
@ -386,12 +435,14 @@ def delete[R: BaseModel](
|
|||
timeout: float = 30.0,
|
||||
) -> Result[R]:
|
||||
try:
|
||||
resp = requests.delete(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
params=_params(params),
|
||||
timeout=timeout,
|
||||
resp = request_with_retry(
|
||||
lambda: requests.delete(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
params=_params(params),
|
||||
timeout=timeout,
|
||||
)
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
return NetworkError(message=str(exc))
|
||||
|
|
@ -407,11 +458,13 @@ def patch[R: BaseModel](
|
|||
timeout: float = 30.0,
|
||||
) -> Result[R]:
|
||||
try:
|
||||
resp = requests.patch(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
resp = request_with_retry(
|
||||
lambda: requests.patch(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
)
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
return NetworkError(message=str(exc))
|
||||
|
|
@ -427,11 +480,13 @@ def put[R: BaseModel](
|
|||
timeout: float = 30.0,
|
||||
) -> Result[R]:
|
||||
try:
|
||||
resp = requests.put(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
resp = request_with_retry(
|
||||
lambda: requests.put(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
)
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
return NetworkError(message=str(exc))
|
||||
|
|
@ -442,11 +497,13 @@ def probe(
|
|||
url: URL, *, headers: BaseModel, params: BaseModel, timeout: float = 30.0
|
||||
) -> ProbeResult:
|
||||
try:
|
||||
resp = requests.get(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
params=params.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
resp = request_with_retry(
|
||||
lambda: requests.get(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
params=params.model_dump(by_alias=True, exclude_none=True),
|
||||
timeout=timeout,
|
||||
)
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
return ProbeResult(status_code=-1, body=str(exc))
|
||||
|
|
@ -544,13 +601,15 @@ def send(
|
|||
status rather than a typed JSON model (e.g. a budget block is a non-2xx). With
|
||||
``stream=True`` the SSE body is consumed and its events counted instead."""
|
||||
try:
|
||||
resp = requests.post(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
params=_params(params),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
stream=stream,
|
||||
timeout=timeout,
|
||||
resp = request_with_retry(
|
||||
lambda: requests.post(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
params=_params(params),
|
||||
json=json.model_dump(by_alias=True, exclude_none=True),
|
||||
stream=stream,
|
||||
timeout=timeout,
|
||||
)
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
return StreamingResponse(status_code=-1, body=str(exc))
|
||||
|
|
@ -585,13 +644,15 @@ def upload[R: BaseModel](
|
|||
dumped: dict[str, object] = form.model_dump(by_alias=True, exclude_none=True)
|
||||
data = {key: str(value) for key, value in dumped.items()}
|
||||
try:
|
||||
resp = requests.post(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
params=_params(params),
|
||||
data=data,
|
||||
files={file_field: (filename, content, file_content_type)},
|
||||
timeout=timeout,
|
||||
resp = request_with_retry(
|
||||
lambda: requests.post(
|
||||
str(url),
|
||||
headers=_headers(headers),
|
||||
params=_params(params),
|
||||
data=data,
|
||||
files={file_field: (filename, content, file_content_type)},
|
||||
timeout=timeout,
|
||||
)
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
return NetworkError(message=str(exc))
|
||||
|
|
|
|||
82
tests/e2e/test_e2e_http.py
Normal file
82
tests/e2e/test_e2e_http.py
Normal file
|
|
@ -0,0 +1,82 @@
|
|||
"""Harness coverage for the transport's transient-retry policy.
|
||||
|
||||
No proxy needed and no ``e2e`` marker: this pins the retry CONTRACT, which is
|
||||
load-bearing for the whole suite. Only statuses the proxy itself cannot emit
|
||||
may ever be retried (today exactly 529, Anthropic's overload signal): 429 must
|
||||
stay unretried because the quota suites assert the proxy's own rate-limit and
|
||||
budget 429s, and proxy-capable 5xx must stay unretried or an intermittently
|
||||
failing proxy would slip through green. The fakes satisfy the
|
||||
RetryableResponse protocol directly, so nothing here imports requests or
|
||||
monkeypatches anything.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Callable, Sequence
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
import pytest
|
||||
|
||||
from e2e_http import RETRY_ATTEMPTS, TRANSIENT_STATUSES, request_with_retry
|
||||
|
||||
|
||||
@dataclass
|
||||
class FakeResponse:
|
||||
status_code: int
|
||||
close_calls: int = 0
|
||||
|
||||
def close(self) -> None:
|
||||
self.close_calls += 1
|
||||
|
||||
|
||||
@dataclass
|
||||
class SleepRecorder:
|
||||
delays: list[float] = field(default_factory=list)
|
||||
|
||||
def __call__(self, seconds: float) -> None:
|
||||
self.delays.append(seconds)
|
||||
|
||||
|
||||
def _issue_from(responses: Sequence[FakeResponse]) -> Callable[[], FakeResponse]:
|
||||
it = iter(responses)
|
||||
return lambda: next(it)
|
||||
|
||||
|
||||
class TestTransientRetryPolicy:
|
||||
def test_transient_set_is_only_statuses_the_proxy_cannot_emit(self) -> None:
|
||||
assert TRANSIENT_STATUSES == frozenset({529})
|
||||
assert 429 not in TRANSIENT_STATUSES
|
||||
|
||||
@pytest.mark.parametrize("status", [200, 201, 400, 401, 404, 422, 500, 502, 503, 504])
|
||||
def test_non_transient_status_returns_immediately(self, status: int) -> None:
|
||||
responses = (FakeResponse(status), FakeResponse(200))
|
||||
sleep = SleepRecorder()
|
||||
result = request_with_retry(_issue_from(responses), sleep=sleep)
|
||||
assert result is responses[0]
|
||||
assert sleep.delays == []
|
||||
assert responses[0].close_calls == 0
|
||||
|
||||
def test_429_is_never_retried(self) -> None:
|
||||
responses = (FakeResponse(429), FakeResponse(200))
|
||||
sleep = SleepRecorder()
|
||||
result = request_with_retry(_issue_from(responses), sleep=sleep)
|
||||
assert result is responses[0]
|
||||
assert sleep.delays == []
|
||||
assert responses[0].close_calls == 0
|
||||
|
||||
def test_overloaded_529_retries_with_backoff_then_returns_the_success(self) -> None:
|
||||
responses = (FakeResponse(529), FakeResponse(200))
|
||||
sleep = SleepRecorder()
|
||||
result = request_with_retry(_issue_from(responses), sleep=sleep)
|
||||
assert result is responses[1]
|
||||
assert sleep.delays == [0.5]
|
||||
assert responses[0].close_calls == 1
|
||||
assert responses[1].close_calls == 0
|
||||
|
||||
def test_persistent_transient_is_bounded_and_returns_the_last_response(self) -> None:
|
||||
responses = tuple(FakeResponse(529) for _ in range(RETRY_ATTEMPTS + 1))
|
||||
sleep = SleepRecorder()
|
||||
result = request_with_retry(_issue_from(responses), sleep=sleep)
|
||||
assert result is responses[RETRY_ATTEMPTS - 1]
|
||||
assert sleep.delays == [0.5, 1.0]
|
||||
assert [r.close_calls for r in responses] == [1, 1, 0, 0]
|
||||
Loading…
Add table
Reference in a new issue