From 258ee96f4cc3a08feaf428f95dad9cf00d28d356 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 11 Jun 2026 05:22:31 +0200 Subject: [PATCH] fix(vector): retry transient embed errors so a pod rollover drops 0 docs 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) --- nextcloud_mcp_server/observability/metrics.py | 22 +++++ nextcloud_mcp_server/providers/_retry.py | 48 +++++++---- nextcloud_mcp_server/providers/mistral.py | 20 +++-- nextcloud_mcp_server/providers/openai.py | 37 +++++++-- nextcloud_mcp_server/vector/processor.py | 55 ++++++++++++- tests/unit/providers/test_mistral.py | 15 ++-- tests/unit/providers/test_openai.py | 82 +++++++++++++++++++ tests/unit/providers/test_retry.py | 46 ++++++++--- tests/unit/test_processor_drop_reason.py | 72 ++++++++++++++++ 9 files changed, 344 insertions(+), 53 deletions(-) create mode 100644 tests/unit/test_processor_drop_reason.py diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 8e18adb8..4b12fae3 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -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], diff --git a/nextcloud_mcp_server/providers/_retry.py b/nextcloud_mcp_server/providers/_retry.py index 8fb8011e..fbeb9fee 100644 --- a/nextcloud_mcp_server/providers/_retry.py +++ b/nextcloud_mcp_server/providers/_retry.py @@ -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 diff --git a/nextcloud_mcp_server/providers/mistral.py b/nextcloud_mcp_server/providers/mistral.py index bacdadf0..25adceb5 100644 --- a/nextcloud_mcp_server/providers/mistral.py +++ b/nextcloud_mcp_server/providers/mistral.py @@ -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]: diff --git a/nextcloud_mcp_server/providers/openai.py b/nextcloud_mcp_server/providers/openai.py index dfa7a8a1..3fb04fe3 100644 --- a/nextcloud_mcp_server/providers/openai.py +++ b/nextcloud_mcp_server/providers/openai.py @@ -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. diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 38c04b33..c721b79e 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -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: diff --git a/tests/unit/providers/test_mistral.py b/tests/unit/providers/test_mistral.py index 20bbce9a..2ca8ba09 100644 --- a/tests/unit/providers/test_mistral.py +++ b/tests/unit/providers/test_mistral.py @@ -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,17 @@ 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 diff --git a/tests/unit/providers/test_openai.py b/tests/unit/providers/test_openai.py index fa70fa17..4dfed9f2 100644 --- a/tests/unit/providers/test_openai.py +++ b/tests/unit/providers/test_openai.py @@ -330,3 +330,85 @@ 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", "http://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] + + calls = {"n": 0} + + async def _flaky(*args, **kwargs): + calls["n"] += 1 + if calls["n"] == 1: + raise APIConnectionError(request=_req()) + return mock_response + + mock_openai_client.embeddings.create = _flaky + 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 calls["n"] == 2 # one failure, one success diff --git a/tests/unit/providers/test_retry.py b/tests/unit/providers/test_retry.py index 305acdd3..79f98ff0 100644 --- a/tests/unit/providers/test_retry.py +++ b/tests/unit/providers/test_retry.py @@ -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: @@ -45,9 +43,7 @@ async def test_retry_reraises_non_rate_limit_immediately(): """A non-rate-limit error of the same class is re-raised on first hit.""" 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 diff --git a/tests/unit/test_processor_drop_reason.py b/tests/unit/test_processor_drop_reason.py new file mode 100644 index 00000000..c03ff359 --- /dev/null +++ b/tests/unit/test_processor_drop_reason.py @@ -0,0 +1,72 @@ +"""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", "http://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_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"