fix(upstream): retry only upstream-answered transient 5xx in place

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)
 
 

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"
     )

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(

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:

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 '<empty>'}",
                 status_code=resp.status_code,
+                from_upstream_response=True,
             )
 
         # Check for usage metrics in response headers (non-streaming) or

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

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__"):

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",
+    ]

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