From c9448aeb19e753767d69c4fffa11a13a70e78086 Mon Sep 17 00:00:00 2001 From: BeiMu-new <2690322907@qq.com> Date: Fri, 9 Oct 2026 12:21:52 +0800 Subject: [PATCH] fix(embedding): keep provider-wide failures out of the per-item split Review follow-up on the P2 raised against 4d8a3c5. `_call_with_retry` sent every unrecognized exception to `_embed_individually` whenever the request had more than one item. Only rate limiting and `insufficient_quota` were excluded, so a 401, a 403 or a 5xx still split a ten-item request into ten extra single-item requests, none of which can succeed: credentials, permissions and server outages are properties of the endpoint, not of any individual input. The split path is now gated on a dedicated classifier, `_is_provider_wide_failure()`, which recognizes 401/403 and 5xx by status code (with the OpenAI SDK class names as a fallback for exceptions that do not expose one). Transient server errors keep their retry policy through the existing backoff; the remaining provider-wide failures stop the loop and report unhealthy, like the quota branch. Input-dependent rejections such as the per-item context-length limit are unchanged and still isolate one item at a time. Regression coverage: test_provider_wide_credentials_errors_are_not_split_into_per_item_requests (401/403 -> a single request of size 10) and test_provider_wide_server_errors_are_retried_without_splitting (500 -> batch sizes [10, 10, 10] with sleeps [1.0, 2.0]). The existing context-length, quota, 429 and per-item transient-retry tests are retained. Local verification, using the repo's own hook arguments: pylint -> Your code has been rated at 10.00/10, exit 0 black -> 2 files would be left unchanged flake8 -> clean Classifier discriminator check (old vs new split decision, extracted from the production file and executed): 400 context-length -> SPLIT / SPLIT 401, 403, 500, 504 -> SPLIT / NO-SPLIT 429 and insufficient_quota keep their own earlier branches. --- .../embedding_store/local_embedding_store.py | 33 +++++++++ tests/unit/test_local_embedding_store.py | 73 +++++++++++++++++++ 2 files changed, 106 insertions(+) diff --git a/reme/components/embedding_store/local_embedding_store.py b/reme/components/embedding_store/local_embedding_store.py index 1d9c93b6..d5e26e23 100644 --- a/reme/components/embedding_store/local_embedding_store.py +++ b/reme/components/embedding_store/local_embedding_store.py @@ -220,6 +220,17 @@ class LocalEmbeddingStore(BaseEmbeddingStore): self.logger.exception("Embedding request failed: quota exhausted") self.is_healthy = False 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 + self.logger.exception("Embedding request failed: provider-wide error, not splitting") + break self.logger.exception("Embedding request failed") self.is_healthy = False if len(texts) > 1: @@ -248,6 +259,28 @@ class LocalEmbeddingStore(BaseEmbeddingStore): results.append(None) return results + @staticmethod + def _is_provider_wide_failure(error: Exception) -> bool: + """Return whether an error describes the provider rather than a single input. + + Authentication, permission and server errors belong to the endpoint or to the + credentials: no individual input can be blamed, so splitting the batch cannot isolate + anything and would only multiply the rejected requests. Rate limits and quota + exhaustion have their own branches above and never reach this classifier. + """ + status = getattr(error, "status_code", None) + if isinstance(status, int) and (status in (401, 403) or 500 <= status < 600): + return True + return type(error).__name__ in ("AuthenticationError", "PermissionDeniedError", "InternalServerError") + + @staticmethod + def _is_transient_provider_failure(error: Exception) -> bool: + """Return whether a provider-wide failure is worth retrying with the usual backoff.""" + status = getattr(error, "status_code", None) + if isinstance(status, int): + return 500 <= status < 600 + return type(error).__name__ == "InternalServerError" + @staticmethod def _is_rate_limited(error: Exception) -> bool: """Recognize an OpenAI-compatible 429 response without importing a provider SDK.""" diff --git a/tests/unit/test_local_embedding_store.py b/tests/unit/test_local_embedding_store.py index da75be04..926b872c 100644 --- a/tests/unit/test_local_embedding_store.py +++ b/tests/unit/test_local_embedding_store.py @@ -200,6 +200,38 @@ class PerItemLimitThenTransientErrorAsEmbedding: return [[1.0, 0.0] for _ in texts] +class AuthenticationError(Exception): + """OpenAI-compatible 401 error used without importing the provider SDK.""" + + status_code = 401 + + +class PermissionDeniedError(Exception): + """OpenAI-compatible 403 error used without importing the provider SDK.""" + + status_code = 403 + + +class InternalServerError(Exception): + """OpenAI-compatible 500 error used without importing the provider SDK.""" + + status_code = 500 + + +class AlwaysProviderWideFailureAsEmbedding: + """Always fail with the same provider-wide error, recording every requested batch size.""" + + dimensions = 2 + + def __init__(self, error: Exception): + self.error = error + self.batch_sizes: list[int] = [] + + async def __call__(self, texts: list[str], **_kwargs): + self.batch_sizes.append(len(texts)) + raise self.error + + class BadNodeEmbeddingStore(BaseEmbeddingStore): """Embedding store that returns wrong-dimensional vectors.""" @@ -953,3 +985,44 @@ def test_per_item_fallback_keeps_a_transiently_failed_vector(monkeypatch): assert sleeps == [1.0] run(go()) + + +def test_provider_wide_credentials_errors_are_not_split_into_per_item_requests(): + """401/403 belong to the credentials: splitting ten items only multiplies the failures.""" + + async def go(): + for error in ( + AuthenticationError("invalid api key"), + PermissionDeniedError("no access to model"), + ): + provider = AlwaysProviderWideFailureAsEmbedding(error) + store = LocalEmbeddingStore(name="t_local_embedding_provider_wide", max_retries=3) + store.as_embedding = provider + + assert await store._call_with_retry([f"item{i}" for i in range(10)]) is None + assert provider.batch_sizes == [10] + assert store.is_healthy is False + + 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.""" + + async def go(): + sleeps = [] + + async def fake_sleep(delay): + sleeps.append(delay) + + provider = AlwaysProviderWideFailureAsEmbedding(InternalServerError("upstream unavailable")) + store = LocalEmbeddingStore(name="t_local_embedding_provider_server", max_retries=3) + store.as_embedding = provider + 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] + assert store.is_healthy is False + + run(go())