test(e2e): retry provider-transient statuses at the transport with bounded backoff

tests/e2e/e2e_http.py

@@ -10,7 +10,9 @@
 
 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))

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]