feat: Add 10/20/30/40s backoff on 429 for Voyage and OpenAI embedding providers

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: GoodbyePlanet

server/embeddings/openai.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import logging
 
 import httpx
@@ -13,6 +14,7 @@
 # OpenAI accepts up to 2048 inputs per request; 128 is conservative and matches
 # Voyage's cap so behaviour is uniform across providers.
 _BATCH_SIZE = 128
+_BACKOFF_DELAYS = [10, 20, 30, 40]
 
 _NATIVE_DIMENSIONS: dict[str, int] = {
     "text-embedding-3-large": 3072,
@@ -63,7 +65,18 @@ async def embed_batch(self, texts: list[str]) -> list[list[float]]:
             body: dict = {"model": self._model, "input": batch}
             if self._dims_override is not None:
                 body["dimensions"] = self._dims_override
-            resp = await self._client.post(_API_URL, json=body)
+            for attempt in range(4):
+                resp = await self._client.post(_API_URL, json=body)
+                if resp.status_code != 429:
+                    break
+                retry_after = float(resp.headers.get("Retry-After", 0))
+                wait = retry_after if retry_after > 0 else _BACKOFF_DELAYS[attempt]
+                logger.warning(
+                    "OpenAI rate-limited (429) — retrying in %.0fs (attempt %d/4)",
+                    wait,
+                    attempt + 1,
+                )
+                await asyncio.sleep(wait)
             resp.raise_for_status()
             data = resp.json()
             batch_vectors = [item["embedding"] for item in data.get("data", [])]

server/embeddings/voyage.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import logging
 
 import httpx
@@ -12,6 +13,7 @@
 _API_URL = "https://api.voyageai.com/v1/embeddings"
 # Voyage caps batch size at 128 inputs per request.
 _BATCH_SIZE = 128
+_BACKOFF_DELAYS = [10, 20, 30, 40]
 
 # Native output dimensions for known models. Some models (voyage-code-3, voyage-3,
 # voyage-3-large) accept an `output_dimension` API parameter to shrink/grow this;
@@ -67,9 +69,7 @@ async def embed_query(self, text: str) -> list[float]:
         vectors = await self._embed([text], input_type="query")
         return vectors[0] if vectors else []
 
-    async def _embed(
-        self, texts: list[str], input_type: str
-    ) -> list[list[float]]:
+    async def _embed(self, texts: list[str], input_type: str) -> list[list[float]]:
         if not texts:
             return []
         all_vectors: list[list[float]] = []
@@ -82,7 +82,18 @@ async def _embed(
             }
             if self._dims_override is not None:
                 body["output_dimension"] = self._dims_override
-            resp = await self._client.post(_API_URL, json=body)
+            for attempt in range(4):
+                resp = await self._client.post(_API_URL, json=body)
+                if resp.status_code != 429:
+                    break
+                retry_after = float(resp.headers.get("Retry-After", 0))
+                wait = retry_after if retry_after > 0 else _BACKOFF_DELAYS[attempt]
+                logger.warning(
+                    "Voyage rate-limited (429) — retrying in %.0fs (attempt %d/4)",
+                    wait,
+                    attempt + 1,
+                )
+                await asyncio.sleep(wait)
             resp.raise_for_status()
             data = resp.json()
             batch_vectors = [item["embedding"] for item in data.get("data", [])]

tests/embeddings/test_jina.py

@@ -55,9 +55,7 @@ async def test_embed_batch_chunks_at_32(provider):
 @respx.mock
 async def test_embed_batch_supports_openai_style_response(provider):
     respx.post("http://tei-test:80/embed").mock(
-        return_value=httpx.Response(
-            200, json={"data": [{"embedding": [0.5] * 768}]}
-        )
+        return_value=httpx.Response(200, json={"data": [{"embedding": [0.5] * 768}]})
     )
     vectors = await provider.embed_batch(["hello"])
     assert vectors == [[0.5] * 768]

tests/embeddings/test_voyage.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import json
 
 import httpx
@@ -124,3 +125,39 @@ async def test_embed_batch_passes_output_dimension_when_overridden(monkeypatch):
 
 async def test_embed_batch_empty(provider):
     assert await provider.embed_batch([]) == []
+
+
+@respx.mock
+async def test_rate_limit_backoff_delays(provider, monkeypatch):
+    sleep_calls: list[float] = []
+
+    async def fake_sleep(seconds: float) -> None:
+        sleep_calls.append(seconds)
+
+    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
+    respx.post("https://api.voyageai.com/v1/embeddings").mock(
+        side_effect=[
+            httpx.Response(429),
+            httpx.Response(429),
+            httpx.Response(429),
+            httpx.Response(200, json=_vectors_response(["a"])),
+        ]
+    )
+    await provider.embed_batch(["a"])
+    assert sleep_calls == [10, 20, 30]
+
+
+@respx.mock
+async def test_rate_limit_all_retries_exhausted_raises(provider, monkeypatch):
+    sleep_calls: list[float] = []
+
+    async def fake_sleep(seconds: float) -> None:
+        sleep_calls.append(seconds)
+
+    monkeypatch.setattr(asyncio, "sleep", fake_sleep)
+    respx.post("https://api.voyageai.com/v1/embeddings").mock(
+        return_value=httpx.Response(429)
+    )
+    with pytest.raises(httpx.HTTPStatusError):
+        await provider.embed_batch(["a"])
+    assert sleep_calls == [10, 20, 30, 40]