Merge pull request #893 from cbcoutinho/fix/309-embed-resilience
fix(vector): retry transient embed errors so a pod rollover drops 0 docs
This commit is contained in:
Chris Coutinho
@@ -272,6 +272,18 @@ document_parse_failed_total = Counter(
|
||||
["reason"], # reason: timeout | oom | error
|
||||
)
|
||||
|
||||
# Documents dropped after exhausting in-process indexing retries (the scanner
|
||||
# re-picks them on a later full scan, so this is "dropped for this cycle", not
|
||||
# "lost forever"). Labelled by classified cause so the embed-drop rate from a
|
||||
# transient backend-pod rollover (connection/timeout) is alertable distinctly
|
||||
# from a persistent fault (card 309).
|
||||
vector_ingest_dropped_total = Counter(
|
||||
"astrolabe_vector_ingest_dropped_total",
|
||||
"Documents dropped after exhausting indexing retries, by cause",
|
||||
# reason: connection | timeout | rate_limit | server | qdrant | other
|
||||
["reason"],
|
||||
)
|
||||
|
||||
# --- Tier-0 classifier (shadow mode) -----------------------------------------
|
||||
#
|
||||
# The classifier runs a cheap pre-pass per PDF and recommends a starting tier.
|
||||
@@ -692,6 +704,16 @@ def record_document_parse_failed(reason: str) -> None:
|
||||
document_parse_failed_total.labels(reason=reason).inc()
|
||||
|
||||
|
||||
def record_ingest_dropped(reason: str) -> None:
|
||||
"""Record a document dropped after exhausting in-process indexing retries.
|
||||
|
||||
Args:
|
||||
reason: ``connection`` | ``timeout`` | ``rate_limit`` | ``server`` |
|
||||
``qdrant`` | ``other`` (classified from the terminal exception).
|
||||
"""
|
||||
vector_ingest_dropped_total.labels(reason=reason).inc()
|
||||
|
||||
|
||||
def record_document_classification(
|
||||
recommended_tier: str,
|
||||
flags: set[str],
|
||||
|
||||
@@ -1,8 +1,10 @@
|
||||
"""Shared rate-limit retry helper for provider modules.
|
||||
"""Shared transient-error retry helper for provider modules.
|
||||
|
||||
OpenAI and Mistral both retry on 429 with the same exponential-backoff curve;
|
||||
extracting the loop here keeps the two provider modules thin and lets future
|
||||
providers (Bedrock throttling, etc.) reuse the same primitive.
|
||||
OpenAI and Mistral retry transient failures (429 rate limits, plus connection
|
||||
drops / timeouts / 5xx for the embedding path) on the same exponential-backoff
|
||||
curve; extracting the loop here keeps the two provider modules thin and lets
|
||||
future providers (Bedrock throttling, etc.) reuse the same primitive. The
|
||||
``should_retry`` predicate decides which caught exceptions are transient.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -23,23 +25,26 @@ MAX_RETRY_DELAY = 60.0
|
||||
T = TypeVar("T")
|
||||
|
||||
|
||||
def retry_on_rate_limit(
|
||||
exception_type: type[BaseException],
|
||||
is_rate_limit: Callable[[BaseException], bool] = lambda _exc: True,
|
||||
def retry_on_transient(
|
||||
exception_type: type[BaseException] | tuple[type[BaseException], ...],
|
||||
should_retry: Callable[[BaseException], bool] = lambda _exc: True,
|
||||
*,
|
||||
provider_name: str = "provider",
|
||||
label: str = "rate limit",
|
||||
) -> Callable[[Callable[..., Awaitable[T]]], Callable[..., Awaitable[T]]]:
|
||||
"""Build a decorator that retries on rate-limit exceptions.
|
||||
"""Build a decorator that retries transient exceptions with backoff.
|
||||
|
||||
Args:
|
||||
exception_type: Catch this exception class (e.g. ``openai.RateLimitError``,
|
||||
``mistralai.client.errors.SDKError``).
|
||||
is_rate_limit: Predicate that decides whether a caught exception is
|
||||
actually a rate-limit (vs. some other error of the same class).
|
||||
Defaults to "always True" — appropriate when ``exception_type`` is
|
||||
already a rate-limit-specific class.
|
||||
exception_type: Catch this exception class (or tuple of classes), e.g.
|
||||
``openai.APIError`` or ``mistralai.client.errors.SDKError``.
|
||||
should_retry: Predicate that decides whether a caught exception is
|
||||
transient (and so retryable) vs. a permanent error of the same
|
||||
class. Defaults to "always True" — appropriate when
|
||||
``exception_type`` is already transient-specific (e.g. a 429 class).
|
||||
provider_name: Used in log messages so operators can tell which
|
||||
provider exhausted retries.
|
||||
label: Short noun for the log message ("rate limit", "transient error")
|
||||
so the line accurately names what was retried.
|
||||
"""
|
||||
|
||||
def decorator(func: Callable[..., Awaitable[T]]) -> Callable[..., Awaitable[T]]:
|
||||
@@ -51,23 +56,32 @@ def retry_on_rate_limit(
|
||||
for attempt in range(1, MAX_RETRIES + 1):
|
||||
try:
|
||||
return await func(*args, **kwargs)
|
||||
except exception_type as e:
|
||||
if not is_rate_limit(e):
|
||||
# exception_type is constrained by the signature to a
|
||||
# BaseException subclass or a tuple of them; the dynamic catch is
|
||||
# the whole point of this reusable helper.
|
||||
except exception_type as e: # NOSONAR(S5708)
|
||||
if not should_retry(e):
|
||||
raise
|
||||
last_error = e
|
||||
if attempt < MAX_RETRIES:
|
||||
logger.warning(
|
||||
"%s rate limit hit (attempt %d/%d), retrying in %.1fs...",
|
||||
"%s %s (attempt %d/%d): %r; retrying in %.1fs...",
|
||||
provider_name,
|
||||
label,
|
||||
attempt,
|
||||
MAX_RETRIES,
|
||||
e,
|
||||
retry_delay,
|
||||
)
|
||||
await anyio.sleep(retry_delay)
|
||||
retry_delay = min(retry_delay * 2, MAX_RETRY_DELAY)
|
||||
|
||||
logger.error(
|
||||
"%s rate limit exceeded after %d attempts", provider_name, MAX_RETRIES
|
||||
"%s %s not resolved after %d attempts: %r",
|
||||
provider_name,
|
||||
label,
|
||||
MAX_RETRIES,
|
||||
last_error,
|
||||
)
|
||||
if last_error is None: # pragma: no cover — loop above always sets this
|
||||
raise RuntimeError("retry loop exited without capturing an error")
|
||||
@@ -76,3 +90,7 @@ def retry_on_rate_limit(
|
||||
return wrapper
|
||||
|
||||
return decorator
|
||||
|
||||
|
||||
# Back-compat alias: the helper was originally rate-limit-specific.
|
||||
retry_on_rate_limit = retry_on_transient
|
||||
|
||||
@@ -13,7 +13,7 @@ import logging
|
||||
from mistralai.client import Mistral
|
||||
from mistralai.client.errors import SDKError
|
||||
|
||||
from ._retry import retry_on_rate_limit
|
||||
from ._retry import retry_on_transient
|
||||
from .base import Provider
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@@ -30,13 +30,25 @@ BATCH_SIZE = 64
|
||||
_NO_EMBEDDING_MODEL_MSG = "Embedding not supported - no embedding_model configured"
|
||||
|
||||
|
||||
def _is_rate_limit(exc: BaseException) -> bool:
|
||||
"""True only for HTTP 429 SDKErrors."""
|
||||
return getattr(exc, "status_code", None) == 429
|
||||
def _is_transient(exc: BaseException) -> bool:
|
||||
"""Retry HTTP 429 (rate limit) and 5xx (server/transient) SDKErrors.
|
||||
|
||||
Scope is deliberately SDK-level: only ``SDKError`` (an HTTP-status error) is
|
||||
caught by the decorator, so a pure connection drop that the Mistral SDK
|
||||
surfaces as a bare ``httpx``/``ConnectionError`` is NOT retried here. The
|
||||
primary pod-rollover resilience target (card 309) is the gateway path via
|
||||
the OpenAI-compatible client, which does cover connection errors; direct
|
||||
Mistral is a self-hoster fallback where 429/5xx is the common transient.
|
||||
"""
|
||||
status = getattr(exc, "status_code", None)
|
||||
return status == 429 or (isinstance(status, int) and status >= 500)
|
||||
|
||||
|
||||
_retry_429 = retry_on_rate_limit(
|
||||
SDKError, is_rate_limit=_is_rate_limit, provider_name="Mistral"
|
||||
_retry_transient = retry_on_transient(
|
||||
SDKError,
|
||||
should_retry=_is_transient,
|
||||
provider_name="Mistral",
|
||||
label="transient error",
|
||||
)
|
||||
|
||||
|
||||
@@ -90,7 +102,7 @@ class MistralProvider(Provider):
|
||||
def supports_generation(self) -> bool:
|
||||
return False
|
||||
|
||||
@_retry_429
|
||||
@_retry_transient
|
||||
async def embed(self, text: str) -> list[float]:
|
||||
"""Generate an embedding for a single text."""
|
||||
if not self.supports_embeddings:
|
||||
@@ -169,7 +181,7 @@ class MistralProvider(Provider):
|
||||
|
||||
return all_embeddings, total_tokens
|
||||
|
||||
@_retry_429
|
||||
@_retry_transient
|
||||
async def _embed_batch_request(
|
||||
self, batch: list[str]
|
||||
) -> tuple[list[list[float]], int]:
|
||||
|
||||
@@ -8,16 +8,37 @@ Supports:
|
||||
|
||||
import logging
|
||||
|
||||
from openai import AsyncOpenAI, RateLimitError
|
||||
from openai import APIConnectionError, APIError, APIStatusError, AsyncOpenAI
|
||||
|
||||
from ._retry import retry_on_rate_limit
|
||||
from ._retry import retry_on_transient
|
||||
from .base import Provider
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# OpenAI's RateLimitError is itself a 429-specific class, so the default
|
||||
# is_rate_limit predicate ("always True") matches the previous behavior.
|
||||
_retry_429 = retry_on_rate_limit(RateLimitError, provider_name="OpenAI")
|
||||
|
||||
def _is_transient(exc: BaseException) -> bool:
|
||||
"""Whether an OpenAI APIError is transient and worth retrying.
|
||||
|
||||
Covers the failures seen dropping documents during a backend-pod rollover
|
||||
(card 309): ``APIConnectionError`` / ``APITimeoutError`` (brief gateway
|
||||
unreachability) and 429 / 5xx status errors. Permanent 4xx (auth, bad
|
||||
request) are NOT retried — they would fail identically every attempt.
|
||||
"""
|
||||
if isinstance(exc, APIConnectionError): # incl. APITimeoutError
|
||||
return True
|
||||
if isinstance(exc, APIStatusError):
|
||||
return exc.status_code == 429 or exc.status_code >= 500
|
||||
return False
|
||||
|
||||
|
||||
# Catch the APIError base (parent of both APIConnectionError and APIStatusError)
|
||||
# and let the predicate decide; non-transient errors re-raise immediately.
|
||||
_retry_transient = retry_on_transient(
|
||||
APIError,
|
||||
should_retry=_is_transient,
|
||||
provider_name="OpenAI",
|
||||
label="transient error",
|
||||
)
|
||||
|
||||
|
||||
# Well-known embedding dimensions for OpenAI models
|
||||
@@ -95,7 +116,7 @@ class OpenAIProvider(Provider):
|
||||
"""Whether this provider supports text generation."""
|
||||
return self.generation_model is not None
|
||||
|
||||
@_retry_429
|
||||
@_retry_transient
|
||||
async def embed(self, text: str) -> list[float]:
|
||||
"""
|
||||
Generate embedding vector for text.
|
||||
@@ -208,7 +229,7 @@ class OpenAIProvider(Provider):
|
||||
|
||||
return all_embeddings, total_tokens
|
||||
|
||||
@_retry_429
|
||||
@_retry_transient
|
||||
async def _embed_batch_request(
|
||||
self, batch: list[str]
|
||||
) -> tuple[list[list[float]], int]:
|
||||
@@ -262,7 +283,11 @@ class OpenAIProvider(Provider):
|
||||
)
|
||||
return self._dimension
|
||||
|
||||
@_retry_429
|
||||
# Transient retry intentionally covers generation too (RAG sampling path):
|
||||
# a pod rollover breaks generation as readily as embedding. Worst case adds
|
||||
# ~30s (5 attempts, 2s→60s backoff) to an interactive call hitting a
|
||||
# sustained connection issue, which is preferable to a hard failure.
|
||||
@_retry_transient
|
||||
async def generate(self, prompt: str, max_tokens: int = 500) -> str:
|
||||
"""
|
||||
Generate text from a prompt.
|
||||
|
||||
@@ -9,6 +9,7 @@ import uuid
|
||||
from typing import Any, cast
|
||||
|
||||
import anyio
|
||||
import httpx
|
||||
from anyio.abc import TaskStatus
|
||||
from anyio.streams.memory import MemoryObjectReceiveStream
|
||||
from qdrant_client.models import PointStruct
|
||||
@@ -23,6 +24,7 @@ from nextcloud_mcp_server.observability.metrics import (
|
||||
record_document_parse_failed,
|
||||
record_embedding,
|
||||
record_embedding_tokens,
|
||||
record_ingest_dropped,
|
||||
record_qdrant_operation,
|
||||
record_vector_sync_processing,
|
||||
update_vector_sync_queue_size,
|
||||
@@ -57,6 +59,53 @@ logger = logging.getLogger(__name__)
|
||||
_ATTR_CHUNK_COUNT = "vector_sync.chunk_count"
|
||||
|
||||
|
||||
def _drop_reason(exc: BaseException) -> str:
|
||||
"""Classify a terminal indexing failure into a metric label.
|
||||
|
||||
Distinguishes the transient backend-pod-rollover causes (connection /
|
||||
timeout — the ones provider-level retry should now ride through, card 309)
|
||||
from persistent faults, so ``astrolabe_vector_ingest_dropped_total`` is
|
||||
alertable per cause. Descends through nested ExceptionGroups to the first
|
||||
leaf so a doubly-wrapped cause isn't mislabelled ``other``. Best-effort:
|
||||
unknown causes fall back to ``other``.
|
||||
"""
|
||||
# An anyio task group can wrap the real cause (and nest groups when sub-tasks
|
||||
# use their own groups); descend to the first concrete leaf. Best-effort: a
|
||||
# group bundling several distinct failures is labelled by whichever leaf
|
||||
# sorts first, not by a "mixed" bucket.
|
||||
while isinstance(exc, BaseExceptionGroup) and exc.exceptions:
|
||||
exc = exc.exceptions[0]
|
||||
|
||||
# Raw httpx transport errors from direct Nextcloud API calls (the nc_client
|
||||
# uses httpx directly); the openai checks below catch the SDK-wrapped
|
||||
# variants of the same failure modes.
|
||||
if isinstance(exc, httpx.TimeoutException):
|
||||
return "timeout"
|
||||
if isinstance(exc, httpx.ConnectError):
|
||||
return "connection"
|
||||
|
||||
# openai.* is always installed (provider dep) but import lazily to keep this
|
||||
# helper cheap and decoupled from a specific SDK version's surface.
|
||||
try:
|
||||
import openai # noqa: PLC0415
|
||||
|
||||
if isinstance(exc, openai.APITimeoutError):
|
||||
return "timeout"
|
||||
if isinstance(exc, openai.APIConnectionError):
|
||||
return "connection"
|
||||
if isinstance(exc, openai.RateLimitError):
|
||||
return "rate_limit"
|
||||
if isinstance(exc, openai.APIStatusError):
|
||||
return "server" if exc.status_code >= 500 else "other"
|
||||
except ImportError: # pragma: no cover — openai is a hard dependency
|
||||
pass
|
||||
|
||||
# Qdrant client errors surface from its own module namespace.
|
||||
if type(exc).__module__.startswith("qdrant_client"):
|
||||
return "qdrant"
|
||||
return "other"
|
||||
|
||||
|
||||
def assign_page_numbers(chunks, page_boundaries):
|
||||
"""Assign page numbers to chunks based on page boundaries.
|
||||
|
||||
@@ -358,6 +407,13 @@ async def process_document(
|
||||
(3) suits the in-process SQLite pool, which has no durable retry. The
|
||||
procrastinate worker passes ``1`` so durable retry is owned by the
|
||||
queue (and survives worker crashes), avoiding compounding 3×N retries.
|
||||
|
||||
Retry layering: the embedding provider adds its own transient retry (5
|
||||
attempts, 2s→60s backoff — card 309) *inside* each of these attempts. On the
|
||||
in-process path (max_retries=3) a sustained outage therefore costs up to
|
||||
5×3=15 provider calls (~90s wall-clock) before the document is dropped and
|
||||
re-picked on the next scan; the procrastinate path (max_retries=1) caps it at
|
||||
one outer attempt (~30s) and defers. Don't stack a third retry layer here.
|
||||
"""
|
||||
start_time = time.time()
|
||||
|
||||
@@ -469,11 +525,13 @@ async def process_document(
|
||||
await anyio.sleep(retry_delay)
|
||||
retry_delay *= 2 # Exponential backoff
|
||||
else:
|
||||
reason = _drop_reason(e)
|
||||
logger.error(
|
||||
"Failed to index %s_%s after %s retries: %s",
|
||||
"Failed to index %s_%s after %s retries (%s): %s",
|
||||
doc_task.doc_type,
|
||||
doc_task.doc_id,
|
||||
max_retries,
|
||||
reason,
|
||||
format_exception_group(e),
|
||||
extra={
|
||||
"doc_id": doc_task.doc_id,
|
||||
@@ -481,12 +539,21 @@ async def process_document(
|
||||
"attempt": max_retries,
|
||||
"max_retries": max_retries,
|
||||
"status": "error",
|
||||
"drop_reason": reason,
|
||||
},
|
||||
)
|
||||
# Record the failed Qdrant upsert. The processing-error
|
||||
# metric is recorded once by the outer handler below, so
|
||||
# exhausted-retry failures aren't double-counted.
|
||||
record_qdrant_operation("upsert", "error")
|
||||
# Count a failed Qdrant upsert ONLY when Qdrant was the
|
||||
# failing component; an embed/connection failure exhausts
|
||||
# retries before Qdrant is ever called, so attributing it
|
||||
# to mcp_qdrant_operations_total{error} would inflate that
|
||||
# signal. The cause is captured by record_ingest_dropped
|
||||
# instead, and the processing-error metric is recorded
|
||||
# once by the outer handler below (no double-count). The
|
||||
# document is NOT marked failed, so the next scan re-picks
|
||||
# it (re-queue via the scan loop, card 309).
|
||||
if reason == "qdrant":
|
||||
record_qdrant_operation("upsert", "error")
|
||||
record_ingest_dropped(reason)
|
||||
raise
|
||||
|
||||
except Exception:
|
||||
|
||||
@@ -9,7 +9,7 @@ from nextcloud_mcp_server.providers.mistral import (
|
||||
BATCH_SIZE,
|
||||
MISTRAL_EMBEDDING_DIMENSIONS,
|
||||
MistralProvider,
|
||||
_is_rate_limit,
|
||||
_is_transient,
|
||||
)
|
||||
|
||||
|
||||
@@ -318,14 +318,58 @@ async def test_mistral_embed_with_usage_single(mock_mistral_client):
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_mistral_is_rate_limit_predicate():
|
||||
"""_is_rate_limit returns True only for SDKErrors with status_code == 429."""
|
||||
def test_mistral_is_transient_predicate():
|
||||
"""_is_transient retries 429 (rate limit) and 5xx (server/transient) SDKErrors."""
|
||||
err_429 = MagicMock(spec=SDKError)
|
||||
err_429.status_code = 429
|
||||
err_500 = MagicMock(spec=SDKError)
|
||||
err_500.status_code = 500
|
||||
err_400 = MagicMock(spec=SDKError)
|
||||
err_400.status_code = 400
|
||||
|
||||
assert _is_rate_limit(err_429) is True
|
||||
assert _is_rate_limit(err_500) is False
|
||||
assert _is_transient(err_429) is True
|
||||
assert _is_transient(err_500) is True # broadened to 5xx (card 309)
|
||||
assert _is_transient(err_400) is False # permanent client error
|
||||
# ValueError has no status_code attr → getattr returns None → False.
|
||||
assert _is_rate_limit(ValueError()) is False
|
||||
assert _is_transient(ValueError()) is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_mistral_embed_retries_on_5xx(mock_mistral_client, monkeypatch):
|
||||
"""A 5xx SDKError is retried end-to-end (not just classified by the predicate)."""
|
||||
from nextcloud_mcp_server.providers import _retry
|
||||
|
||||
monkeypatch.setattr(_retry.anyio, "sleep", AsyncMock(return_value=None))
|
||||
|
||||
# Real SDKError instance (so `except SDKError` catches it) with a 5xx status.
|
||||
err = SDKError.__new__(SDKError)
|
||||
err.status_code = 500
|
||||
|
||||
mock_mistral_client.embeddings.create_async = AsyncMock(
|
||||
side_effect=[err, _make_response([[0.1, 0.2, 0.3]])]
|
||||
)
|
||||
provider = MistralProvider(api_key="test-key", embedding_model="mistral-embed")
|
||||
|
||||
embedding = await provider.embed("hello")
|
||||
assert embedding == [0.1, 0.2, 0.3]
|
||||
assert mock_mistral_client.embeddings.create_async.await_count == 2
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_mistral_embed_batch_retries_on_5xx(mock_mistral_client, monkeypatch):
|
||||
"""The batch path (_embed_batch_request) shares the transient retry too."""
|
||||
from nextcloud_mcp_server.providers import _retry
|
||||
|
||||
monkeypatch.setattr(_retry.anyio, "sleep", AsyncMock(return_value=None))
|
||||
|
||||
err = SDKError.__new__(SDKError)
|
||||
err.status_code = 503
|
||||
|
||||
mock_mistral_client.embeddings.create_async = AsyncMock(
|
||||
side_effect=[err, _make_response([[0.1, 0.2], [0.3, 0.4]])]
|
||||
)
|
||||
provider = MistralProvider(api_key="test-key", embedding_model="mistral-embed")
|
||||
|
||||
embeddings = await provider.embed_batch(["a", "b"])
|
||||
assert embeddings == [[0.1, 0.2], [0.3, 0.4]]
|
||||
assert mock_mistral_client.embeddings.create_async.await_count == 2
|
||||
|
||||
@@ -330,3 +330,152 @@ async def test_openai_close(mock_openai_client):
|
||||
|
||||
await provider.close()
|
||||
mock_openai_client.close.assert_called_once()
|
||||
|
||||
|
||||
# --- transient-error retry (card 309) ----------------------------------------
|
||||
|
||||
|
||||
def _req():
|
||||
import httpx
|
||||
|
||||
return httpx.Request("POST", "https://gw/v1/embeddings")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_is_transient_classifies_retryable_errors():
|
||||
"""Connection / timeout / 429 / 5xx are transient; 4xx and others are not."""
|
||||
import httpx
|
||||
from openai import (
|
||||
APIConnectionError,
|
||||
APITimeoutError,
|
||||
BadRequestError,
|
||||
InternalServerError,
|
||||
RateLimitError,
|
||||
)
|
||||
|
||||
from nextcloud_mcp_server.providers.openai import _is_transient
|
||||
|
||||
req = _req()
|
||||
assert _is_transient(APIConnectionError(request=req)) is True
|
||||
assert _is_transient(APITimeoutError(request=req)) is True
|
||||
assert (
|
||||
_is_transient(
|
||||
RateLimitError("rl", response=httpx.Response(429, request=req), body=None)
|
||||
)
|
||||
is True
|
||||
)
|
||||
assert (
|
||||
_is_transient(
|
||||
InternalServerError(
|
||||
"boom", response=httpx.Response(500, request=req), body=None
|
||||
)
|
||||
)
|
||||
is True
|
||||
)
|
||||
# Permanent client errors must NOT be retried.
|
||||
assert (
|
||||
_is_transient(
|
||||
BadRequestError("bad", response=httpx.Response(400, request=req), body=None)
|
||||
)
|
||||
is False
|
||||
)
|
||||
assert _is_transient(ValueError("unrelated")) is False
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_embed_retries_on_connection_error(mock_openai_client, monkeypatch):
|
||||
"""A transient APIConnectionError (pod rollover) is retried, not dropped."""
|
||||
from openai import APIConnectionError
|
||||
|
||||
from nextcloud_mcp_server.providers import _retry
|
||||
|
||||
monkeypatch.setattr(_retry.anyio, "sleep", AsyncMock(return_value=None))
|
||||
|
||||
mock_embedding_data = MagicMock()
|
||||
mock_embedding_data.embedding = [0.1, 0.2, 0.3]
|
||||
mock_response = MagicMock()
|
||||
mock_response.data = [mock_embedding_data]
|
||||
|
||||
# First call raises a transient connection error, second succeeds.
|
||||
create = AsyncMock(side_effect=[APIConnectionError(request=_req()), mock_response])
|
||||
mock_openai_client.embeddings.create = create
|
||||
provider = OpenAIProvider(
|
||||
api_key="test-key", embedding_model="text-embedding-3-small"
|
||||
)
|
||||
|
||||
result = await provider.embed("hello")
|
||||
assert result == [0.1, 0.2, 0.3]
|
||||
assert create.await_count == 2 # one failure, one success
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_embed_batch_retries_on_connection_error(mock_openai_client, monkeypatch):
|
||||
"""The batch path (`_embed_batch_request`) shares the transient retry too."""
|
||||
from openai import APIConnectionError
|
||||
|
||||
from nextcloud_mcp_server.providers import _retry
|
||||
|
||||
monkeypatch.setattr(_retry.anyio, "sleep", AsyncMock(return_value=None))
|
||||
|
||||
data = MagicMock()
|
||||
data.embedding = [0.4, 0.5, 0.6]
|
||||
data.index = 0
|
||||
mock_response = MagicMock()
|
||||
mock_response.data = [data]
|
||||
mock_response.usage.total_tokens = 7
|
||||
|
||||
create = AsyncMock(side_effect=[APIConnectionError(request=_req()), mock_response])
|
||||
mock_openai_client.embeddings.create = create
|
||||
provider = OpenAIProvider(
|
||||
api_key="test-key", embedding_model="text-embedding-3-small"
|
||||
)
|
||||
|
||||
embeddings, tokens = await provider.embed_batch_with_usage(["text"])
|
||||
assert embeddings == [[0.4, 0.5, 0.6]]
|
||||
assert tokens == 7
|
||||
assert create.await_count == 2
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_generate_retries_on_connection_error(mock_openai_client, monkeypatch):
|
||||
"""generate() shares the transient retry (RAG sampling survives a rollover)."""
|
||||
from openai import APIConnectionError
|
||||
|
||||
from nextcloud_mcp_server.providers import _retry
|
||||
|
||||
monkeypatch.setattr(_retry.anyio, "sleep", AsyncMock(return_value=None))
|
||||
|
||||
choice = MagicMock()
|
||||
choice.message.content = "Generated response"
|
||||
mock_response = MagicMock()
|
||||
mock_response.choices = [choice]
|
||||
|
||||
create = AsyncMock(side_effect=[APIConnectionError(request=_req()), mock_response])
|
||||
mock_openai_client.chat.completions.create = create
|
||||
provider = OpenAIProvider(api_key="test-key", generation_model="gpt-4o-mini")
|
||||
|
||||
text = await provider.generate("prompt")
|
||||
assert text == "Generated response"
|
||||
assert create.await_count == 2
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_generate_does_not_retry_on_bad_request(mock_openai_client, monkeypatch):
|
||||
"""generate() fast-fails (no retry) on a permanent 4xx."""
|
||||
import httpx
|
||||
from openai import BadRequestError
|
||||
|
||||
from nextcloud_mcp_server.providers import _retry
|
||||
|
||||
monkeypatch.setattr(_retry.anyio, "sleep", AsyncMock(return_value=None))
|
||||
|
||||
err = BadRequestError(
|
||||
"bad", response=httpx.Response(400, request=_req()), body=None
|
||||
)
|
||||
create = AsyncMock(side_effect=err)
|
||||
mock_openai_client.chat.completions.create = create
|
||||
provider = OpenAIProvider(api_key="test-key", generation_model="gpt-4o-mini")
|
||||
|
||||
with pytest.raises(BadRequestError):
|
||||
await provider.generate("prompt")
|
||||
assert create.await_count == 1 # no retry on a permanent 4xx
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
"""Unit tests for the shared rate-limit retry decorator."""
|
||||
"""Unit tests for the shared transient-error retry decorator."""
|
||||
|
||||
from unittest.mock import AsyncMock
|
||||
|
||||
@@ -26,9 +26,7 @@ async def test_retry_succeeds_after_429():
|
||||
"""A 429 followed by success returns the success value."""
|
||||
calls = {"n": 0}
|
||||
|
||||
@_retry.retry_on_rate_limit(
|
||||
_FakeError, is_rate_limit=lambda e: e.status_code == 429
|
||||
)
|
||||
@_retry.retry_on_transient(_FakeError, should_retry=lambda e: e.status_code == 429)
|
||||
async def flaky():
|
||||
calls["n"] += 1
|
||||
if calls["n"] < 3:
|
||||
@@ -41,13 +39,11 @@ async def test_retry_succeeds_after_429():
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_retry_reraises_non_rate_limit_immediately():
|
||||
"""A non-rate-limit error of the same class is re-raised on first hit."""
|
||||
async def test_retry_reraises_when_predicate_returns_false():
|
||||
"""An error the predicate rejects is re-raised on first hit (no retry)."""
|
||||
calls = {"n": 0}
|
||||
|
||||
@_retry.retry_on_rate_limit(
|
||||
_FakeError, is_rate_limit=lambda e: e.status_code == 429
|
||||
)
|
||||
@_retry.retry_on_transient(_FakeError, should_retry=lambda e: e.status_code == 429)
|
||||
async def boom():
|
||||
calls["n"] += 1
|
||||
raise _FakeError(500)
|
||||
@@ -62,9 +58,7 @@ async def test_retry_gives_up_after_max_retries():
|
||||
"""After MAX_RETRIES failed attempts the last error is re-raised."""
|
||||
calls = {"n": 0}
|
||||
|
||||
@_retry.retry_on_rate_limit(
|
||||
_FakeError, is_rate_limit=lambda e: e.status_code == 429
|
||||
)
|
||||
@_retry.retry_on_transient(_FakeError, should_retry=lambda e: e.status_code == 429)
|
||||
async def always_429():
|
||||
calls["n"] += 1
|
||||
raise _FakeError(429)
|
||||
@@ -79,7 +73,7 @@ async def test_retry_default_predicate_treats_all_as_rate_limit():
|
||||
"""Default predicate (`lambda _: True`) retries every caught exception."""
|
||||
calls = {"n": 0}
|
||||
|
||||
@_retry.retry_on_rate_limit(_FakeError)
|
||||
@_retry.retry_on_transient(_FakeError)
|
||||
async def fail_once():
|
||||
calls["n"] += 1
|
||||
if calls["n"] < 2:
|
||||
@@ -95,9 +89,37 @@ async def test_retry_default_predicate_treats_all_as_rate_limit():
|
||||
async def test_retry_does_not_catch_unrelated_exceptions():
|
||||
"""Exceptions of a different class bypass the decorator entirely."""
|
||||
|
||||
@_retry.retry_on_rate_limit(_FakeError)
|
||||
@_retry.retry_on_transient(_FakeError)
|
||||
async def value_error():
|
||||
raise ValueError("nope")
|
||||
|
||||
with pytest.raises(ValueError, match="nope"):
|
||||
await value_error()
|
||||
|
||||
|
||||
class _ConnError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_retry_accepts_tuple_of_exception_types():
|
||||
"""A tuple of exception classes is caught (the OpenAI transient set shape)."""
|
||||
calls = {"n": 0}
|
||||
|
||||
@_retry.retry_on_transient((_FakeError, _ConnError))
|
||||
async def flaky():
|
||||
calls["n"] += 1
|
||||
if calls["n"] == 1:
|
||||
raise _ConnError("dropped")
|
||||
if calls["n"] == 2:
|
||||
raise _FakeError(503)
|
||||
return "ok"
|
||||
|
||||
assert await flaky() == "ok"
|
||||
assert calls["n"] == 3
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_retry_on_rate_limit_is_backcompat_alias():
|
||||
"""The old name still resolves to the generalized helper."""
|
||||
assert _retry.retry_on_rate_limit is _retry.retry_on_transient
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
"""Unit tests for the embed-drop classifier (card 309).
|
||||
|
||||
``processor._drop_reason`` maps a terminal indexing failure to a metric label
|
||||
so the transient backend-pod-rollover causes (connection / timeout) are
|
||||
alertable on ``astrolabe_vector_ingest_dropped_total`` distinctly from
|
||||
persistent faults.
|
||||
"""
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from nextcloud_mcp_server.vector import processor
|
||||
|
||||
|
||||
def _req() -> httpx.Request:
|
||||
return httpx.Request("POST", "https://gw/v1/embeddings")
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_httpx_connect_and_timeout_classified():
|
||||
assert processor._drop_reason(httpx.ConnectError("refused")) == "connection"
|
||||
assert processor._drop_reason(httpx.ReadTimeout("slow")) == "timeout"
|
||||
assert processor._drop_reason(httpx.ConnectTimeout("slow")) == "timeout"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_openai_errors_classified():
|
||||
from openai import (
|
||||
APIConnectionError,
|
||||
APITimeoutError,
|
||||
InternalServerError,
|
||||
RateLimitError,
|
||||
)
|
||||
|
||||
req = _req()
|
||||
assert processor._drop_reason(APIConnectionError(request=req)) == "connection"
|
||||
assert processor._drop_reason(APITimeoutError(request=req)) == "timeout"
|
||||
assert (
|
||||
processor._drop_reason(
|
||||
RateLimitError("rl", response=httpx.Response(429, request=req), body=None)
|
||||
)
|
||||
== "rate_limit"
|
||||
)
|
||||
assert (
|
||||
processor._drop_reason(
|
||||
InternalServerError(
|
||||
"boom", response=httpx.Response(503, request=req), body=None
|
||||
)
|
||||
)
|
||||
== "server"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_exception_group_unwraps_to_leaf():
|
||||
group = BaseExceptionGroup(
|
||||
"unhandled errors in a TaskGroup", [httpx.ConnectError("refused")]
|
||||
)
|
||||
assert processor._drop_reason(group) == "connection"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_nested_exception_group_descends_to_leaf():
|
||||
"""A doubly-wrapped group must still classify by its leaf, not 'other'."""
|
||||
nested = BaseExceptionGroup(
|
||||
"outer", [BaseExceptionGroup("inner", [httpx.ReadTimeout("slow")])]
|
||||
)
|
||||
assert processor._drop_reason(nested) == "timeout"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_qdrant_namespace_classified():
|
||||
from qdrant_client.http.exceptions import UnexpectedResponse
|
||||
|
||||
exc = UnexpectedResponse(500, "err", b"", headers=None)
|
||||
assert processor._drop_reason(exc) == "qdrant"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
def test_unknown_error_falls_back_to_other():
|
||||
assert processor._drop_reason(ValueError("nope")) == "other"
|
||||
|
||||
|
||||
@pytest.mark.unit
|
||||
async def test_process_document_records_drop_on_exhausted_retries(mocker):
|
||||
"""Exhausting retries in process_document increments the drop counter with
|
||||
the classified reason (and re-raises so the outer handler counts the error)."""
|
||||
from nextcloud_mcp_server.vector.scanner import DocumentTask
|
||||
|
||||
doc_task = DocumentTask(
|
||||
user_id="alice",
|
||||
doc_id="42",
|
||||
doc_type="note",
|
||||
operation="index",
|
||||
modified_at=0,
|
||||
)
|
||||
|
||||
mocker.patch.object(
|
||||
processor,
|
||||
"get_qdrant_client",
|
||||
mocker.AsyncMock(return_value=mocker.MagicMock()),
|
||||
)
|
||||
mocker.patch.object(
|
||||
processor, "_index_document", side_effect=httpx.ConnectError("refused")
|
||||
)
|
||||
rec = mocker.patch.object(processor, "record_ingest_dropped")
|
||||
|
||||
with pytest.raises(httpx.ConnectError):
|
||||
await processor.process_document(doc_task, mocker.MagicMock(), max_retries=1)
|
||||
|
||||
rec.assert_called_once_with("connection")
|
||||
Reference in New Issue
Block a user