From 6dcc7664a332ced4cbfc35bd3b85df518b1a0163 Mon Sep 17 00:00:00 2001 From: BeiMu-new <2690322907@qq.com> Date: Fri, 16 Oct 2026 00:43:07 +0800 Subject: [PATCH] fix(embedding): leave provider 5xx retries to the layers that own them A store-level retry for a provider 5xx stacked on top of the retries the embedding wrapper and the provider SDK already perform, so a batch the SDK answered after 12 attempts became 36. Retry ownership stays with those layers; the store answers a provider-wide 5xx once and gives up. Transport failures keep their retry because they describe a connection that no lower layer has reported, and the opt-in quota policy is untouched. Tests updated to pin the new boundary: a 503 is answered once for the OpenAI-shaped, Gemini-shaped and DashScope-shaped errors, while a 429 keeps its retry from the rate-limit branch. --- .../embedding_store/local_embedding_store.py | 26 ++++++++---------- tests/unit/test_local_embedding_store.py | 27 +++++++++++++------ 2 files changed, 30 insertions(+), 23 deletions(-) diff --git a/reme/components/embedding_store/local_embedding_store.py b/reme/components/embedding_store/local_embedding_store.py index 142d9365..d5a9d03d 100644 --- a/reme/components/embedding_store/local_embedding_store.py +++ b/reme/components/embedding_store/local_embedding_store.py @@ -330,13 +330,17 @@ class LocalEmbeddingStore(BaseEmbeddingStore): return None if self._is_provider_wide_failure(error): # Credentials, permissions and server outages describe the endpoint, not any - # single item: isolation cannot fix them, so retry the transient ones with the - # usual backoff and give up on the rest instead of multiplying the requests. - if self._is_transient_provider_failure(error) and attempt < self.max_retries - 1: - delay = 2**attempt - self.logger.warning(f"Embedding provider error; retrying in {delay:.1f}s") - await asyncio.sleep(delay) - continue + # single item: isolation cannot fix them, so give up on the whole batch + # instead of multiplying the rejected requests. + # + # A 5xx is deliberately *not* retried here. The embedding wrapper and the + # provider SDK already retry transient server errors, and every extra layer + # that retries multiplies the attempts and the waiting time of the layers + # below it -- a batch the SDK answered after 12 attempts became 36 once this + # store retried on top of it. Retry ownership therefore stays with those + # layers: a caller that wants a longer budget configures it there instead of + # stacking another silent loop here. Transport failures keep their retry + # above because they describe a connection that no lower layer has reported. self.logger.exception("Embedding request failed: provider-wide error, not splitting") stop() break @@ -491,14 +495,6 @@ class LocalEmbeddingStore(BaseEmbeddingStore): parts.append(str(body)) return " ".join(parts).lower() - @staticmethod - def _is_transient_provider_failure(error: Exception) -> bool: - """Return whether a provider-wide failure is worth retrying with the usual backoff.""" - status = LocalEmbeddingStore._http_status(error) - if status is not None: - return 500 <= status < 600 - return type(error).__name__ == "InternalServerError" - @staticmethod def _error_code(error: Exception) -> str | None: """Return the provider's business error code, across the supported error shapes. diff --git a/tests/unit/test_local_embedding_store.py b/tests/unit/test_local_embedding_store.py index e93f81e0..87d71a37 100644 --- a/tests/unit/test_local_embedding_store.py +++ b/tests/unit/test_local_embedding_store.py @@ -1075,8 +1075,8 @@ def test_provider_wide_credentials_errors_are_not_split_into_per_item_requests() run(go()) -def test_provider_wide_server_errors_are_retried_without_splitting(monkeypatch): - """A 500 is transient, so it is retried with the usual backoff and never split per item.""" +def test_provider_wide_server_errors_are_not_retried_without_splitting(monkeypatch): + """A 500 is left to the layers that own retries, so this store asks exactly once.""" async def go(): sleeps = [] @@ -1090,8 +1090,10 @@ def test_provider_wide_server_errors_are_retried_without_splitting(monkeypatch): monkeypatch.setattr(asyncio, "sleep", fake_sleep) assert await store._call_with_retry([f"item{i}" for i in range(10)]) is None - assert provider.batch_sizes == [10, 10, 10] - assert sleeps == [1.0, 2.0] + # One request only: the wrapper and the SDK already retry a 5xx, and retrying here + # as well multiplied their attempts (12 -> 36 in the reported reproduction). + assert provider.batch_sizes == [10] + assert sleeps == [] assert store.is_healthy is False run(go()) @@ -1301,7 +1303,12 @@ def test_http_status_ignores_a_string_code(): def test_google_genai_service_failures_are_not_split_into_per_item_requests(monkeypatch): - """A Gemini 401 / 429 / 503 describes the service, so the batch is never split per item.""" + """A Gemini 401 / 429 / 503 describes the service, so the batch is never split per item. + + A 503 is answered once: the wrapper and the SDK own the retry for a 5xx, and retrying + here as well multiplied their attempts. A 429 keeps its retry because the rate-limit + branch above this classifier is the one place that reacts to it. + """ async def go(): sleeps = [] @@ -1313,7 +1320,7 @@ def test_google_genai_service_failures_are_not_split_into_per_item_requests(monk for error, expected_batches in ( (GenAiClientError(401), [10]), - (GenAiServerError(503), [10, 10, 10]), + (GenAiServerError(503), [10]), (GenAiClientError(429), [10, 10, 10]), ): sleeps.clear() @@ -1332,7 +1339,11 @@ def test_google_genai_service_failures_are_not_split_into_per_item_requests(monk def test_dashscope_runtime_errors_are_not_split_into_per_item_requests(monkeypatch): - """A DashScope failure arrives as a plain RuntimeError; the status lives in its message.""" + """A DashScope failure arrives as a plain RuntimeError; the status lives in its message. + + As with the Gemini shapes, a 503 is answered once (the lower layers own that retry) and + a 429 keeps the retry from the rate-limit branch. + """ async def go(): sleeps = [] @@ -1342,7 +1353,7 @@ def test_dashscope_runtime_errors_are_not_split_into_per_item_requests(monkeypat monkeypatch.setattr(asyncio, "sleep", fake_sleep) - for status, expected_batches in ((401, [10]), (503, [10, 10, 10]), (429, [10, 10, 10])): + for status, expected_batches in ((401, [10]), (503, [10]), (429, [10, 10, 10])): sleeps.clear() provider = AlwaysProviderWideFailureAsEmbedding(DashScopeRuntimeError.for_status(status)) store = LocalEmbeddingStore(