diff --git a/tests/e2e/e2e_http.py b/tests/e2e/e2e_http.py index 471587b93c4..dfb342e34ba 100644 --- a/tests/e2e/e2e_http.py +++ b/tests/e2e/e2e_http.py @@ -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)) diff --git a/tests/e2e/test_e2e_http.py b/tests/e2e/test_e2e_http.py new file mode 100644 index 00000000000..007801a797a --- /dev/null +++ b/tests/e2e/test_e2e_http.py @@ -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]