From 760fab109cca8a666b6d0403fbdb3604af86f57c Mon Sep 17 00:00:00 2001 From: 9qeklajc Date: Wed, 23 Sep 2026 21:27:36 +0200 Subject: [PATCH] fix(upstream): retry only upstream-answered transient 5xx in place --- routstr/core/exceptions.py | 7 +++ routstr/core/settings.py | 5 +- routstr/proxy.py | 36 ++++-------- routstr/upstream/base.py | 2 + routstr/upstream/ehbp.py | 1 + routstr/upstream/gemini_messages.py | 1 + routstr/upstream/messages_dispatch.py | 1 + tests/integration/test_failover_billing.py | 67 ++++++++++++++++++---- tests/unit/test_model_path_routing.py | 10 +--- 9 files changed, 82 insertions(+), 48 deletions(-) diff --git a/routstr/core/exceptions.py b/routstr/core/exceptions.py index b8dcdc38..88e0d370 100644 --- a/routstr/core/exceptions.py +++ b/routstr/core/exceptions.py @@ -18,6 +18,11 @@ class UpstreamError(Exception): string-matching the message. ``details`` holds optional structured, redaction-safe context. Both default to ``None`` for backwards compatibility. + + ``from_upstream_response`` is True only when ``status_code`` is the status + the upstream itself answered with, as opposed to a status this proxy chose + for a transport failure, timeout or internal fault. Callers use it to + decide whether a status is safe to retry. """ def __init__( @@ -26,11 +31,13 @@ class UpstreamError(Exception): status_code: int = 502, code: str | None = None, details: dict[str, object] | None = None, + from_upstream_response: bool = False, ): self.message = message self.status_code = status_code self.code = code self.details = details + self.from_upstream_response = from_upstream_response super().__init__(message) diff --git a/routstr/core/settings.py b/routstr/core/settings.py index 791464d0..77d623e7 100644 --- a/routstr/core/settings.py +++ b/routstr/core/settings.py @@ -36,10 +36,7 @@ class Settings(BaseSettings): # Core upstream_base_url: str = Field(default="", env="UPSTREAM_BASE_URL") upstream_api_key: str = Field(default="", env="UPSTREAM_API_KEY") - # Extra attempts against the SAME upstream when it answers a request with a - # transient 5xx (502/503/504). An edge/gateway blip is usually gone by the - # next attempt, so retrying in place avoids both a failed request and an - # unnecessary failover to a pricier provider. 0 disables the retry. + # Extra attempts against the same upstream on a transient 5xx. 0 disables. upstream_5xx_retry_attempts: int = Field( default=1, ge=0, env="UPSTREAM_5XX_RETRY_ATTEMPTS" ) diff --git a/routstr/proxy.py b/routstr/proxy.py index 99458f8c..887c81f2 100644 --- a/routstr/proxy.py +++ b/routstr/proxy.py @@ -395,27 +395,12 @@ def _forwarding_allowed(path: str, method: str) -> bool: return method in _allowed_methods_for(_canonical_api_path(path)) -# Upstream statuses worth re-trying against the SAME provider before failing -# over. All three are gateway/edge conditions that a retry usually clears; a -# 500 is excluded because it is as likely to be a deterministic rejection that -# would fail identically on the next attempt. +# Gateway conditions a retry usually clears. 500 is excluded: as likely to be a +# deterministic rejection that fails identically on the next attempt. _RETRYABLE_UPSTREAM_5XX = frozenset({502, 503, 504}) - -# Backoff before a same-upstream retry. Short: the client is still waiting, and -# a gateway blip clears in well under a second. Scaled by attempt number. _UPSTREAM_5XX_RETRY_BACKOFF_SECONDS = 0.5 -def _is_retryable_upstream_5xx(status_code: int | None) -> bool: - """True for transient gateway conditions safe to retry on the same upstream.""" - return status_code in _RETRYABLE_UPSTREAM_5XX - - -def _upstream_5xx_retry_attempts() -> int: - """Configured extra same-upstream attempts (0 disables the retry).""" - return max(0, int(getattr(settings, "upstream_5xx_retry_attempts", 0) or 0)) - - @proxy_router.api_route("/{path:path}", methods=["GET", "POST"], response_model=None) async def proxy( request: Request, path: str, session: AsyncSession = Depends(get_session) @@ -803,7 +788,7 @@ async def _proxy( await _finish_read_transaction(session) max_cost_for_model = candidate_max - retries_left = _upstream_5xx_retry_attempts() + retries_left = settings.upstream_5xx_retry_attempts retry_index = 0 headers = upstream.prepare_headers(dict(request.headers)) @@ -858,12 +843,15 @@ async def _proxy( reservation_snapshot, ) except UpstreamError as e: - # A transient gateway 5xx (502/503/504) is worth one more try - # against the SAME upstream before failing over: the request - # body is already buffered, the reservation is untouched, and - # nothing has been streamed to the client yet, so the retry - # cannot double-bill or duplicate content. - if _is_retryable_upstream_5xx(e.status_code) and retries_left > 0: + # Only a gateway status the upstream itself answered with: + # re-sending the buffered body cannot double-bill. A 502 this + # proxy invented for a transport error or timeout is not + # retried — that request may already be running upstream. + if ( + e.from_upstream_response + and e.status_code in _RETRYABLE_UPSTREAM_5XX + and retries_left > 0 + ): retries_left -= 1 retry_index += 1 logger.warning( diff --git a/routstr/upstream/base.py b/routstr/upstream/base.py index 488d07b1..f270442d 100644 --- a/routstr/upstream/base.py +++ b/routstr/upstream/base.py @@ -3151,6 +3151,7 @@ class BaseUpstreamProvider: status_code=response.status_code, code=rate_limit.code if rate_limit else None, details=rate_limit.as_details() if rate_limit else None, + from_upstream_response=True, ) try: @@ -3526,6 +3527,7 @@ class BaseUpstreamProvider: status_code=response.status_code, code=rate_limit.code if rate_limit else None, details=rate_limit.as_details() if rate_limit else None, + from_upstream_response=True, ) try: diff --git a/routstr/upstream/ehbp.py b/routstr/upstream/ehbp.py index 90b92719..0b077709 100644 --- a/routstr/upstream/ehbp.py +++ b/routstr/upstream/ehbp.py @@ -879,6 +879,7 @@ async def forward_ehbp_request( f"EHBP upstream {provider_type} returned {resp.status_code} " f"for model {model_obj.id}: {body_preview[:200] or ''}", status_code=resp.status_code, + from_upstream_response=True, ) # Check for usage metrics in response headers (non-streaming) or diff --git a/routstr/upstream/gemini_messages.py b/routstr/upstream/gemini_messages.py index 11440b91..d822f70f 100644 --- a/routstr/upstream/gemini_messages.py +++ b/routstr/upstream/gemini_messages.py @@ -348,6 +348,7 @@ async def _post_and_stream( raise UpstreamError( f"Upstream error via gemini compat: {body_text}", status_code=response.status_code, + from_upstream_response=True, ) return client, response diff --git a/routstr/upstream/messages_dispatch.py b/routstr/upstream/messages_dispatch.py index c7e698e3..2856256c 100644 --- a/routstr/upstream/messages_dispatch.py +++ b/routstr/upstream/messages_dispatch.py @@ -567,6 +567,7 @@ async def dispatch_anthropic_messages( status_code=status_for_classify, code=rate_limit.code if rate_limit else None, details=rate_limit.as_details() if rate_limit else None, + from_upstream_response=True, ) from exc if not client_stream and hasattr(result, "__aiter__"): diff --git a/tests/integration/test_failover_billing.py b/tests/integration/test_failover_billing.py index 22ed947c..f82b3248 100644 --- a/tests/integration/test_failover_billing.py +++ b/tests/integration/test_failover_billing.py @@ -197,8 +197,7 @@ async def test_failover_serve_billed_at_serving_providers_rate( assert response.status_code == 200 payload = response.json() - # The winner is tried twice: its 502 is retried on the SAME upstream before - # the fallback is consulted at all. + # The winner is tried twice: its 502 is retried in place before failover. assert [r.url.host for r in sent_requests] == [ "cheap.example.com", "cheap.example.com", @@ -327,7 +326,7 @@ async def test_same_id_failover_settles_at_serving_price( ) assert response.status_code == 200 - # The winner's 502 is retried on the SAME upstream before the fallback. + # The winner's 502 is retried in place before failover. assert [r.url.host for r in sent_requests] == [ "cheap.example.com", "cheap.example.com", @@ -469,7 +468,7 @@ async def test_usd_cost_serve_carries_serving_providers_fee( ) assert response.status_code == 200 - # The winner's 502 is retried on the SAME upstream before the fallback. + # The winner's 502 is retried in place before failover. assert [r.url.host for r in sent_requests] == [ "cheap.example.com", "cheap.example.com", @@ -609,7 +608,7 @@ async def test_failover_reserves_serving_candidates_envelope( ) assert response.status_code == 200 - # The winner's 502 is retried on the SAME upstream before the fallback. + # The winner's 502 is retried in place before failover. assert [r.url.host for r in sent_requests] == [ "cheap.example.com", "cheap.example.com", @@ -636,13 +635,11 @@ async def test_transient_502_retries_same_upstream_before_failing_over( authenticated_client: AsyncClient, dual_provider_maps: tuple[_StaticProvider, _StaticProvider], ) -> None: - """A transient 502 is retried on the SAME upstream, not failed over at once. + """A transient 502 is retried on the same upstream, not failed over at once. - The winning provider answers the first attempt with a gateway 502 and the - second with a completion, so the retry must land on the same host and the - pricier fallback must never be contacted. The request is therefore billed at - the winner's rate (0.001/0.002 sats per token -> 2_000 msats) rather than at - the fallback's 10_000. + The winner answers the first attempt with a 502 and the second with a + completion, so the pricier fallback is never contacted and the request is + billed at the winner's rate (2_000 msats, not the fallback's 10_000). """ sent_requests: list[httpx.Request] = [] cheap_attempts = 0 @@ -681,7 +678,7 @@ async def test_transient_502_retries_same_upstream_before_failing_over( ) assert response.status_code == 200 - # Retried in place: the same host twice, the fallback never consulted. + # Retried in place: same host twice, fallback never consulted. assert [r.url.host for r in sent_requests] == [ "cheap.example.com", "cheap.example.com", @@ -690,3 +687,49 @@ async def test_transient_502_retries_same_upstream_before_failing_over( assert payload["model"] == "prova/dual-model" # Billed at the winner's rate, not the fallback's 10_000. assert payload["cost"]["total_msats"] == 2_000 + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_transport_failure_is_not_retried_in_place( + authenticated_client: AsyncClient, + dual_provider_maps: tuple[_StaticProvider, _StaticProvider], +) -> None: + """A transport failure fails over at once instead of retrying in place. + + The proxy maps a connect/timeout error to a 502 of its own, so the upstream + may already have accepted and billed the request: re-sending it is not safe. + """ + sent_requests: list[httpx.Request] = [] + + async def fake_transport( + request: httpx.Request, *args: Any, **kwargs: Any + ) -> httpx.Response: + sent_requests.append(request) + if request.url.host == "cheap.example.com": + raise httpx.ConnectError("connection refused", request=request) + return _successful_upstream_response() + + with ( + patch( + "httpx.AsyncHTTPTransport.handle_async_request", + side_effect=fake_transport, + ), + patch( + "routstr.payment.cost_calculation.sats_usd_price", + return_value=0.0005, + ), + ): + response = await authenticated_client.post( + "/v1/chat/completions", + json={ + "model": "dual-model", + "messages": [{"role": "user", "content": "hello"}], + }, + ) + + assert response.status_code == 200 + assert [r.url.host for r in sent_requests] == [ + "cheap.example.com", + "expensive.example.com", + ] diff --git a/tests/unit/test_model_path_routing.py b/tests/unit/test_model_path_routing.py index 30bd322a..4dfa2ef8 100644 --- a/tests/unit/test_model_path_routing.py +++ b/tests/unit/test_model_path_routing.py @@ -487,9 +487,7 @@ async def test_pinned_exception_does_not_fall_back() -> None: request, [(MagicMock(), first), (MagicMock(), fallback)] ) assert response.status_code == 503 - # The pin still binds the route to `first`: the 503 is retried on the SAME - # upstream, and the route is never relaxed to the fallback. - assert first.forward_request.await_count == 2 + first.forward_request.assert_awaited_once() fallback.forward_request.assert_not_awaited() @@ -548,11 +546,7 @@ async def test_ehbp_pin_does_not_fall_back(cashu: bool) -> None: request, [(MagicMock(), selected), (MagicMock(), fallback)] ) assert response.status_code == 503 - # The pin still binds the route to `selected`. The bearer/EHBP dispatch runs - # inside the candidate loop, so its 503 is retried on the same upstream; the - # x-cashu dispatch happens earlier and is not retried. Neither relaxes the - # pin to the fallback. - assert forward.await_count == (1 if cashu else 2) + forward.assert_awaited_once() assert forward.await_args is not None assert forward.await_args.kwargs["upstream"] is selected