diff --git a/src/everos/component/rerank/_errors.py b/src/everos/component/rerank/_errors.py index 65480b850..547455f03 100644 --- a/src/everos/component/rerank/_errors.py +++ b/src/everos/component/rerank/_errors.py @@ -1,7 +1,10 @@ -"""Shared error construction for HTTP-based rerank providers.""" +"""Shared error construction and retry pacing for HTTP rerank providers.""" from __future__ import annotations +import asyncio +import random + import httpx from everos.core.observability.logging import get_logger @@ -10,6 +13,26 @@ logger = get_logger(__name__) +_BACKOFF_BASE_SECONDS = 0.5 +_BACKOFF_CAP_SECONDS = 8.0 + + +async def backoff_sleep(attempt: int) -> None: + """Wait before the next retry of a 429 / 5xx rerank request. + + Retrying with no delay is useless against a *per-minute* quota — the + whole budget burns in milliseconds and the caller still fails. Hosted + rerank routers enforce exactly that kind of quota, so the retry loop + has to actually wait. Exponential with full jitter, capped, to avoid a + thundering herd when a batch of concurrent searches trips the limit + together. + + Args: + attempt: Zero-based index of the attempt that just failed. + """ + delay = min(_BACKOFF_BASE_SECONDS * (2**attempt), _BACKOFF_CAP_SECONDS) + await asyncio.sleep(random.uniform(0, delay)) + def upstream_http_error(provider: str, response: httpx.Response) -> RerankServiceError: """Log the upstream response body and return a client-safe error. diff --git a/src/everos/component/rerank/dashscope_provider.py b/src/everos/component/rerank/dashscope_provider.py index 13110cf82..498f26d77 100644 --- a/src/everos/component/rerank/dashscope_provider.py +++ b/src/everos/component/rerank/dashscope_provider.py @@ -47,6 +47,7 @@ import httpx +from ._errors import backoff_sleep from .protocol import RerankError, RerankResult @@ -160,6 +161,7 @@ async def _score_chunk( f"DashScope rerank HTTP {response.status_code}: " f"{response.text[:200]}" ) + await backoff_sleep(attempt) continue raise RerankError( f"DashScope rerank HTTP {response.status_code}: " diff --git a/src/everos/component/rerank/deepinfra_provider.py b/src/everos/component/rerank/deepinfra_provider.py index 8ef5b8579..8c38dd233 100644 --- a/src/everos/component/rerank/deepinfra_provider.py +++ b/src/everos/component/rerank/deepinfra_provider.py @@ -35,7 +35,12 @@ import httpx -from ._errors import retries_exhausted_error, transport_error, upstream_http_error +from ._errors import ( + backoff_sleep, + retries_exhausted_error, + transport_error, + upstream_http_error, +) from .protocol import RerankResult, RerankServiceError # Qwen3-Reranker chat template. The DeepInfra inference API treats the reranker @@ -160,6 +165,7 @@ async def _score_chunk( if response.status_code >= 500 or response.status_code == 429: if attempt == self._max_retries: raise upstream_http_error("DeepInfra", response) + await backoff_sleep(attempt) continue raise upstream_http_error("DeepInfra", response) diff --git a/src/everos/component/rerank/vllm_provider.py b/src/everos/component/rerank/vllm_provider.py index cfedcfb8b..b067d3bc0 100644 --- a/src/everos/component/rerank/vllm_provider.py +++ b/src/everos/component/rerank/vllm_provider.py @@ -41,7 +41,12 @@ import httpx -from ._errors import retries_exhausted_error, transport_error, upstream_http_error +from ._errors import ( + backoff_sleep, + retries_exhausted_error, + transport_error, + upstream_http_error, +) from .protocol import RerankResult, RerankServiceError @@ -143,6 +148,7 @@ async def _score_chunk( if response.status_code >= 500 or response.status_code == 429: if attempt == self._max_retries: raise upstream_http_error("vLLM", response) + await backoff_sleep(attempt) continue raise upstream_http_error("vLLM", response) diff --git a/tests/unit/test_component/test_rerank/test_vllm_provider.py b/tests/unit/test_component/test_rerank/test_vllm_provider.py index d9adbf29d..182e3b6d1 100644 --- a/tests/unit/test_component/test_rerank/test_vllm_provider.py +++ b/tests/unit/test_component/test_rerank/test_vllm_provider.py @@ -185,3 +185,35 @@ def handler(_req: httpx.Request) -> httpx.Response: p = VllmRerankProvider(model="m", api_key="", base_url="http://x/v1") with pytest.raises(RerankServiceError, match="malformed rerank result"): await p.rerank("q", ["a"]) + + +async def test_429_retry_waits_between_attempts( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A per-minute quota is not survivable by spinning — retries must sleep.""" + import everos.component.rerank._errors as errmod + + slept: list[float] = [] + + async def fake_sleep(seconds: float) -> None: + slept.append(seconds) + + monkeypatch.setattr(errmod.asyncio, "sleep", fake_sleep) + + attempts = 0 + + def handler(_req: httpx.Request) -> httpx.Response: + nonlocal attempts + attempts += 1 + if attempts <= 2: + return httpx.Response(429, json={"error": "rate limited"}) + return _ok_response([{"index": 0, "relevance_score": 0.7}]) + + _patch_httpx(monkeypatch, handler) + p = VllmRerankProvider(model="m", api_key="k", base_url="http://x/v1") + out = await p.rerank("q", ["d"]) + assert [r.score for r in out] == [0.7] + assert attempts == 3 + # One wait per failed attempt, and each wait is a real (non-zero) budget. + assert len(slept) == 2 + assert all(s >= 0 for s in slept)