diff --git a/providers/anthropic_messages.py b/providers/anthropic_messages.py index ee7ca341..22f808f9 100644 --- a/providers/anthropic_messages.py +++ b/providers/anthropic_messages.py @@ -360,9 +360,9 @@ class AnthropicMessagesTransport(BaseProvider): try: async def _validated_stream_send() -> httpx.Response: - """Send request; raise inside retry loop on 429 so rate limiter can backoff.""" + """Send request; raise inside retry loop on 429/503 so limiter can backoff.""" send_response = await self._send_stream_request(body) - if send_response.status_code == 429: + if send_response.status_code in (429, 503): await send_response.aclose() send_response.raise_for_status() if send_response.status_code != 200: diff --git a/providers/error_mapping.py b/providers/error_mapping.py index 44c33a50..ebd1a76d 100644 --- a/providers/error_mapping.py +++ b/providers/error_mapping.py @@ -50,6 +50,14 @@ def map_error( return InvalidRequestError(message, raw_error=str(e)) if isinstance(e, openai.InternalServerError): raw_message = str(e) + sdk_status = getattr(e, "status_code", None) + if sdk_status == 503: + stable = APIError("_", status_code=503) + return APIError( + get_user_facing_error_message(stable), + status_code=503, + raw_error=str(e), + ) if "overloaded" in raw_message.lower() or "capacity" in raw_message.lower(): return OverloadedError(message, raw_error=raw_message) return APIError(message, status_code=500, raw_error=str(e)) diff --git a/providers/rate_limit.py b/providers/rate_limit.py index 59e42910..a3019aed 100644 --- a/providers/rate_limit.py +++ b/providers/rate_limit.py @@ -12,10 +12,32 @@ import openai from loguru import logger from core.rate_limit import StrictSlidingWindowLimiter +from core.trace import trace_event T = TypeVar("T") +def retryable_upstream_status(exc: BaseException) -> int | None: + """Return HTTP-like status codes that qualify for reactive backoff retries. + + ``429`` and ``503`` use the same exponential backoff plus scoped limiter + blocking semantics as today's rate-limit path. + """ + if isinstance(exc, openai.RateLimitError): + return 429 + if isinstance(exc, httpx.HTTPStatusError): + status = exc.response.status_code + if status in (429, 503): + return status + return None + if isinstance(exc, openai.APIError): + status = getattr(exc, "status_code", None) + if isinstance(status, int) and status == 503: + return 503 + return None + return None + + class GlobalRateLimiter: """ Global singleton rate limiter that blocks all requests @@ -26,7 +48,7 @@ class GlobalRateLimiter: may be open simultaneously, independent of the sliding window. Proactive limits - throttles requests to stay within API limits. - Reactive limits - pauses all requests when a 429 is hit. + Reactive limits - pauses all requests when a 429 or 503 retry backoff is active. Concurrency limit - caps simultaneously open streams. """ @@ -203,10 +225,11 @@ class GlobalRateLimiter: jitter: float = 1.0, **kwargs: Any, ) -> Any: - """Execute an async callable with rate limiting and retry on 429. + """Execute an async callable with rate limiting and retry on transient limits. - Waits for the proactive limiter before each attempt. On 429, applies - exponential backoff with jitter before retrying. + Waits for the proactive limiter before each attempt. On ``429`` (rate limit) + or ``503`` (service unavailable), applies exponential backoff with jitter + and sets the reactive block before retrying. Args: fn: Async callable to execute. @@ -222,43 +245,51 @@ class GlobalRateLimiter: The last exception if all retries are exhausted. """ last_exc: Exception | None = None + total_attempts = 1 + max_retries - for attempt in range(1 + max_retries): + for attempt in range(total_attempts): await self.wait_if_blocked() try: return await fn(*args, **kwargs) - except openai.RateLimitError as e: - last_exc = e - if attempt >= max_retries: - logger.warning( - f"Rate limit retry exhausted after {max_retries} retries" - ) - break - - delay = min(base_delay * (2**attempt), max_delay) - delay += random.uniform(0, jitter) - logger.warning( - f"Rate limited (429), attempt {attempt + 1}/{max_retries + 1}. " - f"Retrying in {delay:.1f}s..." - ) - self.set_blocked(delay) - await asyncio.sleep(delay) - except httpx.HTTPStatusError as e: - if e.response.status_code != 429: + except Exception as e: + status = retryable_upstream_status(e) + if status is None: raise + + label = ( + "Rate limited (429)" + if status == 429 + else "Upstream unavailable (503)" + ) last_exc = e if attempt >= max_retries: logger.warning( - f"HTTP 429 retry exhausted after {max_retries} retries" + "{} retry exhausted after {} retries (attempts={})", + label, + max_retries, + total_attempts, ) break delay = min(base_delay * (2**attempt), max_delay) delay += random.uniform(0, jitter) + attempt_no = attempt + 1 logger.warning( - f"HTTP 429 from upstream, attempt {attempt + 1}/{max_retries + 1}. " - f"Retrying in {delay:.1f}s..." + "{}, attempt {}/{}. Retrying in {:.1f}s...", + label, + attempt_no, + total_attempts, + delay, + ) + trace_event( + stage="provider", + event="provider.retry.scheduled", + source="provider", + status_code=status, + attempt=attempt_no, + max_attempts=total_attempts, + delay_s=round(delay, 3), ) self.set_blocked(delay) await asyncio.sleep(delay) diff --git a/tests/providers/test_anthropic_messages_429_retry.py b/tests/providers/test_anthropic_messages_429_retry.py index e64bbb71..58fd923c 100644 --- a/tests/providers/test_anthropic_messages_429_retry.py +++ b/tests/providers/test_anthropic_messages_429_retry.py @@ -1,4 +1,4 @@ -"""Native Anthropic transport: HTTP 429 is retried inside execute_with_retry.""" +"""Native Anthropic transport: HTTP 429 and 503 are retried inside execute_with_retry.""" from contextlib import asynccontextmanager from unittest.mock import AsyncMock, MagicMock, patch @@ -80,6 +80,108 @@ async def test_native_stream_retries_on_http_429_then_streams(provider_config): GlobalRateLimiter.reset_instance() +@pytest.mark.asyncio +async def test_native_stream_retries_on_http_503_then_streams(provider_config): + """First response 503 (closed), second 200 streams; send is called twice.""" + GlobalRateLimiter.reset_instance() + try: + provider = NativeProvider(provider_config) + req = MockRequest() + request_obj = httpx.Request("POST", "https://custom.test/v1/messages") + ok_lines = [ + "event: message_start", + 'data: {"type":"message_start"}', + "", + ] + ok_response = FakeResponse(lines=ok_lines) + unavailable = FakeResponse(status_code=503, text="Service Unavailable") + + send_calls = {"n": 0} + + async def send_side_effect(*_a, **_kw): + send_calls["n"] += 1 + if send_calls["n"] == 1: + return unavailable + return ok_response + + with ( + patch.object(provider._client, "build_request", return_value=request_obj), + patch.object( + provider._client, + "send", + new_callable=AsyncMock, + side_effect=send_side_effect, + ), + patch( + "asyncio.sleep", + new_callable=AsyncMock, + ), + ): + events = [e async for e in provider.stream_response(req)] + + assert send_calls["n"] == 2 + assert unavailable.is_closed + assert ok_response.is_closed + assert events == [ + "event: message_start\n", + 'data: {"type":"message_start"}\n', + "\n", + ] + finally: + GlobalRateLimiter.reset_instance() + + +@pytest.mark.asyncio +async def test_native_stream_503_retry_exhausted(provider_config): + """Repeated HTTP 503 exhausts execute_with_retry; emits overloaded-style message.""" + GlobalRateLimiter.reset_instance() + try: + + @asynccontextmanager + async def _slot(): + yield + + with patch("providers.anthropic_messages.GlobalRateLimiter") as mock_gl: + instance = mock_gl.get_scoped_instance.return_value + real = GlobalRateLimiter( + rate_limit=100, + rate_window=60, + max_concurrency=5, + ) + instance.wait_if_blocked = real.wait_if_blocked + instance.execute_with_retry = real.execute_with_retry + instance.set_blocked = real.set_blocked + instance.concurrency_slot.side_effect = _slot + + provider = NativeProvider(provider_config) + req = MockRequest() + + unavailable = FakeResponse(status_code=503, text="Service Unavailable") + + with ( + patch.object( + provider._client, "build_request", return_value=MagicMock() + ), + patch.object( + provider._client, + "send", + new_callable=AsyncMock, + return_value=unavailable, + ) as mock_send, + patch("asyncio.sleep", new_callable=AsyncMock), + ): + events = [e async for e in provider.stream_response(req)] + + assert mock_send.await_count == 4 + assert unavailable.is_closed + assert_canonical_stream_error_envelope( + events, + user_message_substr="Provider is currently overloaded", + ) + finally: + GlobalRateLimiter.reset_instance() + + @pytest.mark.asyncio async def test_non_429_http_error_not_retried(provider_config): """HTTP 500 from upstream is not retried; single send.""" diff --git a/tests/providers/test_error_mapping.py b/tests/providers/test_error_mapping.py index 2a180e79..f41091d3 100644 --- a/tests/providers/test_error_mapping.py +++ b/tests/providers/test_error_mapping.py @@ -88,6 +88,18 @@ class TestMapError: assert isinstance(result, APIError) assert result.status_code == 500 + def test_internal_server_error_503_maps_to_api_error_with_status(self): + """InternalServerError carrying HTTP 503 retains 503 for stable user messaging.""" + exc = _make_openai_error( + openai.InternalServerError, + message="503", + status_code=503, + ) + result = map_error(exc) + assert isinstance(result, APIError) + assert result.status_code == 503 + assert "temporarily unavailable" in result.message.lower() + def test_generic_api_error(self): """openai.APIError -> APIError with original status_code.""" exc = _make_openai_error( diff --git a/tests/providers/test_openai_compat_503_retry.py b/tests/providers/test_openai_compat_503_retry.py new file mode 100644 index 00000000..79f08ce1 --- /dev/null +++ b/tests/providers/test_openai_compat_503_retry.py @@ -0,0 +1,100 @@ +"""OpenAI-compat transports: HTTP 503 uses the same execute_with_retry path as 429.""" + +from unittest.mock import AsyncMock, MagicMock, patch + +import openai +import pytest +from httpx import Request, Response + +from config.nim import NimSettings +from providers.base import ProviderConfig +from providers.nvidia_nim import NvidiaNimProvider +from providers.rate_limit import GlobalRateLimiter +from tests.providers.test_nvidia_nim import MockRequest + + +def _internal_503() -> openai.InternalServerError: + return openai.InternalServerError( + "unavailable", + response=Response(503, request=Request("POST", "http://x")), + body={}, + ) + + +@pytest.mark.asyncio +async def test_nim_stream_retries_on_openai_503_then_streams(): + GlobalRateLimiter.reset_instance() + try: + config = ProviderConfig( + api_key="test_key", + base_url="https://test.api.nvidia.com/v1", + rate_limit=100, + rate_window=60, + http_read_timeout=600.0, + http_write_timeout=15.0, + http_connect_timeout=5.0, + ) + provider = NvidiaNimProvider(config, nim_settings=NimSettings()) + req = MockRequest() + + mock_chunk = MagicMock() + mock_chunk.choices = [ + MagicMock( + delta=MagicMock(content="Hi", reasoning_content=""), + finish_reason=None, + ) + ] + mock_chunk.usage = None + + async def mock_stream(): + yield mock_chunk + + with ( + patch.object( + provider._client.chat.completions, + "create", + new_callable=AsyncMock, + ) as mock_create, + patch("asyncio.sleep", new_callable=AsyncMock), + ): + mock_create.side_effect = [_internal_503(), mock_stream()] + events = [e async for e in provider.stream_response(req)] + + assert mock_create.await_count == 2 + assert any("Hi" in e for e in events) + finally: + GlobalRateLimiter.reset_instance() + + +@pytest.mark.asyncio +async def test_nim_stream_openai_503_exhausted_emits_user_message(): + GlobalRateLimiter.reset_instance() + try: + config = ProviderConfig( + api_key="test_key", + base_url="https://test.api.nvidia.com/v1", + rate_limit=100, + rate_window=60, + http_read_timeout=600.0, + http_write_timeout=15.0, + http_connect_timeout=5.0, + ) + provider = NvidiaNimProvider(config, nim_settings=NimSettings()) + req = MockRequest() + + with ( + patch.object( + provider._client.chat.completions, + "create", + new_callable=AsyncMock, + ) as mock_create, + patch("asyncio.sleep", new_callable=AsyncMock), + ): + mock_create.side_effect = _internal_503() + events = [e async for e in provider.stream_response(req)] + + assert mock_create.await_count == 4 + blob = "".join(events) + assert "temporarily unavailable" in blob.lower() + finally: + GlobalRateLimiter.reset_instance() diff --git a/tests/providers/test_provider_rate_limit.py b/tests/providers/test_provider_rate_limit.py index 7d592808..bae58fd7 100644 --- a/tests/providers/test_provider_rate_limit.py +++ b/tests/providers/test_provider_rate_limit.py @@ -278,6 +278,88 @@ class TestProviderRateLimiter: assert result == "ok" assert call_count == 2 + @pytest.mark.asyncio + async def test_execute_with_retry_succeeds_on_openai_internal_server_error_503( + self, + ): + """503 as openai.InternalServerError then success.""" + import openai + from httpx import Request, Response + + GlobalRateLimiter.reset_instance() + limiter = GlobalRateLimiter.get_instance(rate_limit=100, rate_window=60) + + def make_503(): + return openai.InternalServerError( + "unavailable", + response=Response(503, request=Request("POST", "http://x")), + body={}, + ) + + call_count = 0 + + async def fail_then_ok(): + nonlocal call_count + call_count += 1 + if call_count == 1: + raise make_503() + return "ok" + + result = await limiter.execute_with_retry( + fail_then_ok, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0 + ) + assert result == "ok" + assert call_count == 2 + + @pytest.mark.asyncio + async def test_execute_with_retry_succeeds_on_httpx_503(self): + """HTTP 503 as httpx.HTTPStatusError then success.""" + import httpx + from httpx import Request, Response + + limiter = GlobalRateLimiter.get_instance(rate_limit=100, rate_window=60) + + call_count = 0 + + async def fail_then_ok(): + nonlocal call_count + call_count += 1 + if call_count == 1: + r = Response(503, request=Request("POST", "http://x"), text="busy") + raise httpx.HTTPStatusError( + "Service Unavailable", request=r.request, response=r + ) + return "ok" + + result = await limiter.execute_with_retry( + fail_then_ok, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0 + ) + assert result == "ok" + assert call_count == 2 + + @pytest.mark.asyncio + async def test_execute_with_retry_exhaust_openai_503_raises(self): + """When all 503 retries exhausted (openai), last InternalServerError is raised.""" + import openai + from httpx import Request, Response + + GlobalRateLimiter.reset_instance() + limiter = GlobalRateLimiter.get_instance(rate_limit=100, rate_window=60) + + exc = openai.InternalServerError( + "unavailable", + response=Response(503, request=Request("POST", "http://x")), + body={}, + ) + + async def always_503(): + raise exc + + with pytest.raises(openai.InternalServerError): + await limiter.execute_with_retry( + always_503, max_retries=2, base_delay=0.01, max_delay=0.1, jitter=0 + ) + @pytest.mark.asyncio async def test_max_concurrency_zero_raises(self): """max_concurrency <= 0 raises ValueError."""