From 1701131017bb77dc5f34a289689a8b8122f0940f Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 19 Jun 2026 10:11:28 +0200 Subject: [PATCH 1/4] fix(vector): offload embedded Qdrant ops to a worker thread (#926) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit qdrant-client's embedded backends (:memory: and path= local mode) run every operation synchronously on the calling thread despite the AsyncQdrantClient surface — AsyncQdrantLocal contains no thread offload, and this is unchanged through the latest v1.18.x (the async surface is autogenerated from the synchronous QdrantLocal). On a CPU-constrained host a background scan of thousands of tagged files issues ~3 Qdrant queries per file, all on the event loop thread, pinning one core at 100% and stalling /health/live, /health/ready, and the outbound Nextcloud-reachability probe for minutes — the failure mode in issue #926 (health-check timeouts, "nextcloud_reachable: error"). Wrap the embedded client in a transparent proxy that offloads every coroutine-returning call to a worker thread via anyio.to_thread.run_sync, keeping the event loop responsive. A dedicated CapacityLimiter(1) serialises those offloads to preserve QdrantLocal's single-access invariant (today guaranteed implicitly by the single-threaded loop). Network mode (QDRANT_URL) is left untouched — it already does non-blocking I/O. Centralised at get_qdrant_client() so all call sites benefit. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/vector/qdrant_client.py | 98 ++++++++++++- tests/unit/vector/test_qdrant_client.py | 145 ++++++++++++++++++- 2 files changed, 240 insertions(+), 3 deletions(-) diff --git a/nextcloud_mcp_server/vector/qdrant_client.py b/nextcloud_mcp_server/vector/qdrant_client.py index 60fa994f..da807864 100644 --- a/nextcloud_mcp_server/vector/qdrant_client.py +++ b/nextcloud_mcp_server/vector/qdrant_client.py @@ -1,9 +1,14 @@ """Qdrant client wrapper.""" +import functools +import inspect import logging -from typing import Any +from collections.abc import Coroutine +from typing import Any, cast import anyio +import anyio.to_thread +from anyio import CapacityLimiter from qdrant_client import AsyncQdrantClient, models from qdrant_client.http.exceptions import UnexpectedResponse from qdrant_client.models import ( @@ -111,6 +116,84 @@ _qdrant_client: AsyncQdrantClient | None = None _qdrant_init_lock: anyio.Lock | None = None +def _drive_local_coroutine(coro: Coroutine[Any, Any, Any]) -> Any: + """Run an embedded-Qdrant coroutine to completion on the current thread. + + qdrant-client's local/in-memory backend (``AsyncQdrantLocal``) exposes an + ``async def`` surface over purely synchronous, CPU-bound work: its + coroutines never await real I/O, so they complete on the first ``send`` + (verified against qdrant-client 1.17). Driving one here — inside the worker + thread dispatched by :func:`anyio.to_thread.run_sync` — is what moves that + CPU work off the event loop. + + A genuine suspension (a non-``None`` yield) would mean the backend started + doing real async I/O, which a plain thread cannot drive correctly. Fail + loudly in that case rather than spin or silently mis-drive the coroutine. + """ + try: + while True: + yielded = coro.send(None) + if yielded is not None: + coro.close() + raise RuntimeError( + "embedded Qdrant coroutine suspended unexpectedly " + f"(yielded {yielded!r}); the thread-offload shim in " + "vector/qdrant_client.py assumes the local backend is " + "synchronous" + ) + except StopIteration as exc: + return exc.value + + +class _ThreadOffloadingQdrantClient: + """Run embedded-Qdrant operations on a worker thread to free the event loop. + + qdrant-client's embedded backends — ``:memory:`` and ``path=`` local mode — + execute every operation synchronously on the calling thread despite the + ``AsyncQdrantClient`` surface (``AsyncQdrantLocal`` contains no + ``to_thread``/executor offload). A single background scan of thousands of + tagged files therefore pins the event loop on a CPU-constrained host, so + ``/health/live`` / ``/health/ready`` probes and in-flight MCP requests stall + until the scan finishes — the failure mode reported in issue #926. + + This proxy forwards attribute access to the wrapped client and offloads + every coroutine-returning call to a worker thread, keeping the event loop + responsive. A private ``CapacityLimiter(1)`` serialises those offloads: + ``AsyncQdrantLocal`` is not safe for concurrent access, and today the + single-threaded event loop is what implicitly serialises it — the limiter + preserves that invariant once the work runs off-loop, while still letting + the loop interleave health checks and other requests between operations. + + Network mode (``QDRANT_URL``) is never wrapped: there the client performs + real non-blocking I/O and already yields to the event loop. + """ + + def __init__(self, inner: AsyncQdrantClient) -> None: + self._inner = inner + self._limiter = CapacityLimiter(1) + + def __getattr__(self, name: str) -> Any: + # Only reached for names absent from the instance __dict__; ``_inner`` + # and ``_limiter`` resolve normally and never recurse through here. + attr = getattr(self._inner, name) + if not callable(attr): + return attr + + @functools.wraps(attr) + def _maybe_offload(*args: Any, **kwargs: Any) -> Any: + result = attr(*args, **kwargs) + if inspect.iscoroutine(result): + return self._offload(result) + return result + + return _maybe_offload + + async def _offload(self, coro: Coroutine[Any, Any, Any]) -> Any: + return await anyio.to_thread.run_sync( + _drive_local_coroutine, coro, limiter=self._limiter + ) + + async def _create_one_payload_index( client: AsyncQdrantClient, collection_name: str, @@ -596,6 +679,19 @@ async def get_qdrant_client() -> AsyncQdrantClient: logger.warning("No Qdrant mode configured, defaulting to :memory:") provisional = AsyncQdrantClient(":memory:") + # Embedded Qdrant (``:memory:`` / ``path=``) runs every operation + # synchronously on the calling thread, so without this each call — + # including the O(N) startup backfill scroll below and every later + # scan/search query — would block the event loop and stall health + # checks on a CPU-constrained host (issue #926). Offload local-mode + # work to a worker thread. Network mode (mirrors the ``if + # settings.qdrant_url`` branch above) already does non-blocking I/O + # and is left untouched. + if not settings.qdrant_url: + provisional = cast( + AsyncQdrantClient, _ThreadOffloadingQdrantClient(provisional) + ) + # Get collection name (auto-generated from deployment ID + model) collection_name = settings.get_collection_name() diff --git a/tests/unit/vector/test_qdrant_client.py b/tests/unit/vector/test_qdrant_client.py index ba8f2edd..44ab90e6 100644 --- a/tests/unit/vector/test_qdrant_client.py +++ b/tests/unit/vector/test_qdrant_client.py @@ -18,6 +18,7 @@ from types import SimpleNamespace from unittest.mock import call import anyio +import anyio.lowlevel import httpx import pytest from qdrant_client.http.exceptions import UnexpectedResponse @@ -28,8 +29,10 @@ from nextcloud_mcp_server.vector.qdrant_client import ( _DOC_ID_BACKFILL_SENTINEL_ID, _PAYLOAD_INDEX_FIELDS, _backfill_doc_id_to_string, + _drive_local_coroutine, _ensure_payload_indexes, _group_int_doc_ids, + _ThreadOffloadingQdrantClient, get_qdrant_client, ) @@ -976,7 +979,11 @@ async def test_get_qdrant_client_creates_collection_on_local_mode_value_error( client = await get_qdrant_client() - assert client is provisional + # Local/in-memory mode is wrapped in the thread-offload proxy (issue #926), + # so the returned client is the proxy over ``provisional`` rather than the + # bare mock; the proxy forwards every awaited call straight through to it. + assert isinstance(client, _ThreadOffloadingQdrantClient) + assert client._inner is provisional provisional.create_collection.assert_awaited_once() # The created collection should be the auto-generated name from # settings — guards against accidental collection-name drift. @@ -1009,4 +1016,138 @@ async def test_get_qdrant_client_propagates_unrelated_value_error( with pytest.raises(ValueError, match="Bad collection_name"): await get_qdrant_client() - provisional.create_collection.assert_not_awaited() + +# --------------------------------------------------------------------------- +# Thread-offload proxy for embedded Qdrant (issue #926) +# --------------------------------------------------------------------------- + + +@pytest.mark.unit +def test_drive_local_coroutine_returns_value(): + """A non-suspending coroutine is driven to completion and its value returned.""" + + async def _coro(): + return 42 + + assert _drive_local_coroutine(_coro()) == 42 + + +@pytest.mark.unit +def test_drive_local_coroutine_propagates_exception(): + """An exception raised inside the coroutine surfaces to the driver.""" + + async def _coro(): + raise ValueError("boom") + + with pytest.raises(ValueError, match="boom"): + _drive_local_coroutine(_coro()) + + +@pytest.mark.unit +def test_drive_local_coroutine_rejects_real_suspension(): + """A coroutine that genuinely suspends (yields non-None) fails loudly. + + The embedded backend is assumed synchronous; a real ``await`` of I/O would + mean this shim can no longer drive it from a plain thread. + """ + + class _Suspends: + # Mimics a real future/coroutine that yields control to the event loop. + def __await__(self): + yield "pending-future" + + async def _coro(): + await _Suspends() + return 1 + + with pytest.raises(RuntimeError, match="suspended unexpectedly"): + _drive_local_coroutine(_coro()) + + +@pytest.mark.unit +async def test_offloading_proxy_runs_off_event_loop_thread(mocker): + """The proxy executes the wrapped coroutine on a worker thread, not the loop. + + This is the whole point of issue #926: embedded-Qdrant CPU work must not run + on the event loop thread, where it would stall health checks. + """ + import threading + + loop_thread = threading.get_ident() + worker_threads: list[int] = [] + + inner = mocker.AsyncMock() + + async def _record_thread(*args, **kwargs): + worker_threads.append(threading.get_ident()) + return "ok" + + inner.scroll.side_effect = _record_thread + + proxy = _ThreadOffloadingQdrantClient(inner) + result = await proxy.scroll("collection") + + assert result == "ok" + assert worker_threads, "wrapped coroutine never ran" + assert worker_threads[0] != loop_thread + inner.scroll.assert_awaited_once_with("collection") + + +@pytest.mark.unit +async def test_offloading_proxy_passes_through_non_callables(mocker): + """Non-callable attributes resolve straight to the wrapped client.""" + inner = mocker.AsyncMock() + inner.collection_name = "my-collection" + + proxy = _ThreadOffloadingQdrantClient(inner) + + assert proxy.collection_name == "my-collection" + + +@pytest.mark.unit +async def test_get_qdrant_client_does_not_wrap_network_mode(mocker, monkeypatch): + """Network mode (``QDRANT_URL``) returns the bare client, never the proxy. + + Remote Qdrant already does non-blocking I/O; wrapping it would needlessly + funnel every call through the serialising worker-thread limiter. + """ + from nextcloud_mcp_server.config import Settings + + settings = Settings( + qdrant_url="http://qdrant:6333", + ollama_embedding_model="nomic-embed-text", + vector_sync_enabled=False, + ) + monkeypatch.setattr( + "nextcloud_mcp_server.vector.qdrant_client.get_settings", lambda: settings + ) + embedding_service = mocker.Mock() + embedding_service.provider = mocker.Mock(spec_set=[]) + embedding_service.get_dimension = lambda: 4 + monkeypatch.setattr( + "nextcloud_mcp_server.embedding.get_embedding_service", + lambda: embedding_service, + ) + + # Drive the create path (collection "not found") so the dimension-validation + # branch — which would need a fully-shaped CollectionInfo — is skipped. + provisional = _stub_provisional( + mocker, ValueError(f"Collection {settings.get_collection_name()} not found") + ) + monkeypatch.setattr( + "nextcloud_mcp_server.vector.qdrant_client.AsyncQdrantClient", + lambda *a, **kw: provisional, + ) + + original_client = qdrant_module._qdrant_client + original_lock = qdrant_module._qdrant_init_lock + qdrant_module._qdrant_client = None + qdrant_module._qdrant_init_lock = None + try: + client = await get_qdrant_client() + finally: + qdrant_module._qdrant_client = original_client + qdrant_module._qdrant_init_lock = original_lock + + assert client is provisional + assert not isinstance(client, _ThreadOffloadingQdrantClient) From de960715c6a4fcdb96b1c78e9a330db899ad59e5 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 19 Jun 2026 10:16:57 +0200 Subject: [PATCH 2/4] test(vector): address round-1 review nits on #926 offload proxy - Use the reset_qdrant_singleton fixture in the network-mode test instead of manual save/restore boilerplate. - Add a test locking down the sync-callable forwarding branch (callable returning a non-coroutine is passed through without thread offload). - Document that _drive_local_coroutine treats a bare `yield None` as a non-suspension, and that the proxy is wrapped before the startup migrations so the O(N) backfill scroll also runs off the event loop. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/vector/qdrant_client.py | 10 ++++++ tests/unit/vector/test_qdrant_client.py | 32 ++++++++++++++------ 2 files changed, 32 insertions(+), 10 deletions(-) diff --git a/nextcloud_mcp_server/vector/qdrant_client.py b/nextcloud_mcp_server/vector/qdrant_client.py index da807864..29e760d3 100644 --- a/nextcloud_mcp_server/vector/qdrant_client.py +++ b/nextcloud_mcp_server/vector/qdrant_client.py @@ -129,6 +129,11 @@ def _drive_local_coroutine(coro: Coroutine[Any, Any, Any]) -> Any: A genuine suspension (a non-``None`` yield) would mean the backend started doing real async I/O, which a plain thread cannot drive correctly. Fail loudly in that case rather than spin or silently mis-drive the coroutine. + A bare ``yield None`` (e.g. a hand-inserted ``anyio.lowlevel.checkpoint()``) + is deliberately treated as a non-suspension and re-driven — consistent with + the local backend being purely synchronous. If a qdrant adapter ever starts + inserting real checkpoints, prefer wrapping it in network mode over relaxing + this guard. """ try: while True: @@ -687,6 +692,11 @@ async def get_qdrant_client() -> AsyncQdrantClient: # work to a worker thread. Network mode (mirrors the ``if # settings.qdrant_url`` branch above) already does non-blocking I/O # and is left untouched. + # + # Wrap here, before the ``_backfill_doc_id_to_string`` / + # ``_ensure_payload_indexes`` migrations below, so that the O(N) + # startup scroll runs off the event loop too — not just steady-state + # scan/search queries. if not settings.qdrant_url: provisional = cast( AsyncQdrantClient, _ThreadOffloadingQdrantClient(provisional) diff --git a/tests/unit/vector/test_qdrant_client.py b/tests/unit/vector/test_qdrant_client.py index 44ab90e6..49581ec2 100644 --- a/tests/unit/vector/test_qdrant_client.py +++ b/tests/unit/vector/test_qdrant_client.py @@ -1105,7 +1105,27 @@ async def test_offloading_proxy_passes_through_non_callables(mocker): @pytest.mark.unit -async def test_get_qdrant_client_does_not_wrap_network_mode(mocker, monkeypatch): +async def test_offloading_proxy_forwards_sync_callables_directly(mocker): + """A callable returning a non-coroutine is forwarded without thread offload. + + ``_maybe_offload`` only routes through the worker thread when the call + produces a coroutine; plain sync methods return their value straight + through. + """ + inner = mocker.Mock() # Mock (not AsyncMock): calls return plain values. + inner.get_fastembed_vector_params.return_value = {"size": 384} + + proxy = _ThreadOffloadingQdrantClient(inner) + result = proxy.get_fastembed_vector_params() + + assert result == {"size": 384} + inner.get_fastembed_vector_params.assert_called_once_with() + + +@pytest.mark.unit +async def test_get_qdrant_client_does_not_wrap_network_mode( + mocker, monkeypatch, reset_qdrant_singleton +): """Network mode (``QDRANT_URL``) returns the bare client, never the proxy. Remote Qdrant already does non-blocking I/O; wrapping it would needlessly @@ -1139,15 +1159,7 @@ async def test_get_qdrant_client_does_not_wrap_network_mode(mocker, monkeypatch) lambda *a, **kw: provisional, ) - original_client = qdrant_module._qdrant_client - original_lock = qdrant_module._qdrant_init_lock - qdrant_module._qdrant_client = None - qdrant_module._qdrant_init_lock = None - try: - client = await get_qdrant_client() - finally: - qdrant_module._qdrant_client = original_client - qdrant_module._qdrant_init_lock = original_lock + client = await get_qdrant_client() assert client is provisional assert not isinstance(client, _ThreadOffloadingQdrantClient) From 5b468514ec5e10e3c12311045c5a4746a0e066c2 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 19 Jun 2026 10:22:39 +0200 Subject: [PATCH 3/4] test(vector): use network-mode 404 signal + https in not-wrap test (#926) - Clear SonarCloud hotspot python:S5332 (insecure http:// URL) by using https:// for the inert placeholder Qdrant URL in the network-mode test. - Make the test faithful to production: stub the network existence-check with UnexpectedResponse(404) (what the real HTTP client raises) instead of the local-mode ValueError("not found"), per round-2 review. Co-Authored-By: Claude Opus 4.8 (1M context) --- tests/unit/vector/test_qdrant_client.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/tests/unit/vector/test_qdrant_client.py b/tests/unit/vector/test_qdrant_client.py index 49581ec2..e88ca443 100644 --- a/tests/unit/vector/test_qdrant_client.py +++ b/tests/unit/vector/test_qdrant_client.py @@ -1134,7 +1134,7 @@ async def test_get_qdrant_client_does_not_wrap_network_mode( from nextcloud_mcp_server.config import Settings settings = Settings( - qdrant_url="http://qdrant:6333", + qdrant_url="https://qdrant:6333", ollama_embedding_model="nomic-embed-text", vector_sync_enabled=False, ) @@ -1149,10 +1149,12 @@ async def test_get_qdrant_client_does_not_wrap_network_mode( lambda: embedding_service, ) - # Drive the create path (collection "not found") so the dimension-validation - # branch — which would need a fully-shaped CollectionInfo — is skipped. + # Drive the create path via the network-mode "not found" signal — an + # UnexpectedResponse(404), as the real HTTP client raises (local mode's + # ValueError("not found") is the other branch). This skips the + # dimension-validation branch that would need a fully-shaped CollectionInfo. provisional = _stub_provisional( - mocker, ValueError(f"Collection {settings.get_collection_name()} not found") + mocker, _make_unexpected(404, b'{"status":{"error":"Not found"}}') ) monkeypatch.setattr( "nextcloud_mcp_server.vector.qdrant_client.AsyncQdrantClient", From 420c8cd2d39025e4c214fe668d6cdb81e552a66a Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 19 Jun 2026 10:27:18 +0200 Subject: [PATCH 4/4] docs(vector): correct _drive_local_coroutine suspension-guard comment (#926) The prior comment claimed anyio.lowlevel.checkpoint() yields None and is re-driven, but checkpoint() yields a non-None backend object (asyncio Future / trio checkpoint) and so trips the RuntimeError guard. Clarify that only a literal bare yield/yield None is re-driven; any real awaitable is caught by the non-None guard. Comment-only; no behavior change. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/vector/qdrant_client.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/nextcloud_mcp_server/vector/qdrant_client.py b/nextcloud_mcp_server/vector/qdrant_client.py index 29e760d3..00112fe9 100644 --- a/nextcloud_mcp_server/vector/qdrant_client.py +++ b/nextcloud_mcp_server/vector/qdrant_client.py @@ -126,14 +126,15 @@ def _drive_local_coroutine(coro: Coroutine[Any, Any, Any]) -> Any: thread dispatched by :func:`anyio.to_thread.run_sync` — is what moves that CPU work off the event loop. - A genuine suspension (a non-``None`` yield) would mean the backend started - doing real async I/O, which a plain thread cannot drive correctly. Fail - loudly in that case rather than spin or silently mis-drive the coroutine. - A bare ``yield None`` (e.g. a hand-inserted ``anyio.lowlevel.checkpoint()``) - is deliberately treated as a non-suspension and re-driven — consistent with - the local backend being purely synchronous. If a qdrant adapter ever starts - inserting real checkpoints, prefer wrapping it in network mode over relaxing - this guard. + A genuine suspension would mean the backend started doing real async I/O, + which a plain thread cannot drive correctly. Any real awaitable yields a + non-``None`` object (an asyncio ``Future`` / a trio checkpoint), so the + ``yielded is not None`` guard catches it and fails loudly rather than + spinning or silently mis-driving the coroutine. Only a literal bare + ``yield`` / ``yield None`` is treated as a non-suspension and re-driven — + consistent with the local backend being purely synchronous. If a qdrant + adapter ever starts inserting real checkpoints, prefer wrapping it in + network mode over relaxing this guard. """ try: while True: