admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(vector): retry transient embed errors so a pod rollover drops 0 docs

Browse Source 
From card 309 (OHR-Bench smoke-test triage): during a backend-pod rollover the
embedding endpoint was briefly unreachable, and openai.APIConnectionError /
ConnectError propagated unretried (the provider only retried 429). Documents
exhausted the 3 in-process retries and were dropped for that scan cycle.

Broaden the provider-level retry to the transient set -- APIConnectionError,
APITimeoutError, 429, and 5xx -- on the existing exponential backoff (2s->60s,
5 attempts), so a few seconds of retry rides through the rollover. Permanent
4xx (auth, bad request) still re-raise immediately. Generalize the shared
_retry helper (retry_on_rate_limit -> retry_on_transient, predicate renamed to
should_retry, accurate log label) with a back-compat alias; Mistral gets 429+5xx
for parity. The production gateway path inherits this via GatewayProvider, which
delegates to the decorated OpenAIProvider methods.

Add astrolabe_vector_ingest_dropped_total{reason}, incremented when a document
exhausts retries, classified (connection|timeout|rate_limit|server|qdrant|other)
by _drop_reason so the embed-drop rate is alertable per cause. Dropped docs are
NOT marked failed, so the next full scan re-picks them (re-queue via scan loop).

Refs: Deck board 12 card 309 (AC #1 no permanently-dropped docs; embed-drop
metric for AC #5).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-11 05:22:31 +02:00
co-authored by Claude Opus 4.8
parent 457c115ef4
commit 258ee96f4c
9 changed files with 344 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/observability/metrics.py
+22
View File
@@ -161,6 +161,18 @@ vector_sync_processing_duration_seconds = Histogram(
buckets=(0.1, 0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0),
)
# 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). astrolabe_ prefix: pipeline metric.
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"],
)
vector_sync_queue_size = Gauge(
"mcp_vector_sync_queue_size",
"Current number of documents in processing queue",
@@ -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],
nextcloud_mcp_server/providers/_retry.py
+31 -17
View File
@@ -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]]:
@@ -52,22 +57,27 @@ def retry_on_rate_limit(
try:
return await func(*args, **kwargs)
except exception_type as e:
if not is_rate_limit(e):
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",
provider_name,
label,
MAX_RETRIES,
)
if last_error is None: # pragma: no cover — loop above always sets this
raise RuntimeError("retry loop exited without capturing an error")
@@ -76,3 +86,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
nextcloud_mcp_server/providers/mistral.py
+12 -8
View File
@@ -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,17 @@ 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."""
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 +94,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 +173,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]:
nextcloud_mcp_server/providers/openai.py
+29 -8
View File
@@ -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,7 @@ class OpenAIProvider(Provider):
)
return self._dimension
@_retry_429
@_retry_transient
async def generate(self, prompt: str, max_tokens: int = 500) -> str:
"""
Generate text from a prompt.
nextcloud_mcp_server/vector/processor.py
+53 -2
View File
@@ -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,
@@ -56,6 +58,47 @@ 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. Unwraps a single ExceptionGroup leaf. Best-effort:
unknown causes fall back to ``other``.
"""
# An anyio task group can wrap the real cause; classify the first leaf.
if isinstance(exc, BaseExceptionGroup) and exc.exceptions:
exc = exc.exceptions[0]
# httpx transport errors (raised by the OpenAI/gateway client underneath).
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.
@@ -455,11 +498,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,
e,
extra={
"doc_id": doc_task.doc_id,
@@ -467,12 +512,18 @@ 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.
# exhausted-retry failures aren't double-counted. The
# drop counter is labelled by cause so a transient
# rollover (connection/timeout) is alertable distinctly.
# The document is NOT marked failed, so the next scan
# re-picks it (re-queue via the scan loop, card 309).
record_qdrant_operation("upsert", "error")
record_ingest_dropped(reason)
raise
except Exception: