From e0b7afb4b132ced97c124ec373f9e00d56024ad9 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 17 May 2026 19:33:24 +0200 Subject: [PATCH] fix(embedding): instantiate BM25 singleton off the event loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The ``BM25SparseEmbeddingProvider.__init__`` calls ``fastembed.SparseTextEmbedding(model_name="Qdrant/bm25")`` which downloads ~50 MB of model weights from HuggingFace and loads them into memory — observed >5 s wall-clock in production. The inference methods (``encode_async``, ``encode_batch_async``) already wrap work in ``anyio.to_thread.run_sync``, so the design intent is clearly to keep FastEmbed off the event loop. That protection just didn't cover the constructor. Symptom in the Astrolabe Cloud per-tenant deploy (deck #102 smoke): ~30–90 s after a user enables semantic search, the pod tips into a SIGKILL-restart cycle. Loki shows a single log line Initializing BM25 sparse embedding provider: Qdrant/bm25 followed by nothing else from the event loop until exitCode 137. Kubernetes ``/health/live`` httpGet probe timeout=5s fires 6 times in a row, kubelet kills the container, restart, repeat. Fix: switch ``get_bm25_service()`` to an async accessor that wraps the first-time construction in ``anyio.to_thread.run_sync``. Two existing call sites (``vector/processor.py:603``, ``search/bm25_hybrid.py:123``) update to ``await``. Both are already inside async functions so the await is free. New unit test pins the invariant by monkey-patching ``BM25SparseEmbeddingProvider.__init__`` with ``time.sleep(1)`` and asserting a concurrent ``anyio.sleep(0.05)`` finishes promptly — the test fails if the constructor ever runs back on the event loop. Same pattern exists in ``OllamaEmbeddingProvider.__init__`` (sync ``httpx.get`` health-check). Ollama isn't enabled in any current deploy; filed as a follow-up. Refs: - Astrolabe Cloud deck card #102 (smoke discovery) - Sibling fix #799 (NullPool for cross-loop-asyncpg, same class of "anyio bites you in production" bug) Verified: - ``uv run pytest tests/unit/`` — 1027 passed - ``uv run ruff check`` clean on touched files - ``uv run ty check`` clean on touched files - New tests pass Co-Authored-By: Claude Opus 4.7 (1M context) --- nextcloud_mcp_server/embedding/service.py | 31 +++++- nextcloud_mcp_server/search/bm25_hybrid.py | 2 +- nextcloud_mcp_server/vector/processor.py | 2 +- tests/unit/test_embedding_service.py | 108 +++++++++++++++++++++ 4 files changed, 139 insertions(+), 4 deletions(-) create mode 100644 tests/unit/test_embedding_service.py diff --git a/nextcloud_mcp_server/embedding/service.py b/nextcloud_mcp_server/embedding/service.py index 83fed5d6..32e53c9a 100644 --- a/nextcloud_mcp_server/embedding/service.py +++ b/nextcloud_mcp_server/embedding/service.py @@ -6,6 +6,8 @@ New code should use nextcloud_mcp_server.providers.get_provider() directly. import logging +import anyio + from nextcloud_mcp_server.providers import get_provider from .bm25_provider import BM25SparseEmbeddingProvider @@ -89,14 +91,39 @@ def get_embedding_service() -> EmbeddingService: _bm25_service: BM25SparseEmbeddingProvider | None = None -def get_bm25_service() -> BM25SparseEmbeddingProvider: +async def get_bm25_service() -> BM25SparseEmbeddingProvider: """ Get singleton BM25 sparse embedding service instance. + Lazily instantiates the singleton off the event loop. The + ``BM25SparseEmbeddingProvider`` constructor calls + ``fastembed.SparseTextEmbedding(model_name="Qdrant/bm25")`` which + downloads ~50 MB of model weights from HuggingFace and loads them + into memory — observed >5 s wall-clock in production, enough to + stall the calling thread. The encode methods on the provider + already offload via ``anyio.to_thread.run_sync``; this routes the + first-time init through the same path so the event loop stays + responsive (kubernetes ``/health/live`` httpGet probe in + particular). + + The singleton is process-wide so the per-pod cost is paid once. + Subsequent calls hit the warm path and return after a single + non-blocking await. + + Concurrent first callers race: two coroutines can both observe + ``_bm25_service is None`` and both enter ``run_sync``. We accept + the duplicate model load over an ``asyncio.Lock``, which has its + own cross-loop hazards under anyio TaskGroups (see PR #799 for + that class of bug). The duplicate is bounded — FastEmbed caches + the downloaded weights on disk after the first call, so loser(s) + pick up cheaply. + Returns: Global BM25SparseEmbeddingProvider instance """ global _bm25_service if _bm25_service is None: - _bm25_service = BM25SparseEmbeddingProvider() + _bm25_service = await anyio.to_thread.run_sync( # type: ignore[attr-defined] # ty: ignore[unresolved-attribute] + BM25SparseEmbeddingProvider + ) return _bm25_service diff --git a/nextcloud_mcp_server/search/bm25_hybrid.py b/nextcloud_mcp_server/search/bm25_hybrid.py index ddf3b7ba..8f60c580 100644 --- a/nextcloud_mcp_server/search/bm25_hybrid.py +++ b/nextcloud_mcp_server/search/bm25_hybrid.py @@ -120,7 +120,7 @@ class BM25HybridSearchAlgorithm(SearchAlgorithm): # Generate sparse embedding for BM25 keyword search with trace_operation("search.get_bm25_service"): - bm25_service = get_bm25_service() + bm25_service = await get_bm25_service() with trace_operation("search.sparse_embedding_bm25"): sparse_embedding = await bm25_service.encode_async(query) logger.debug( diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 6f88793e..d77f1868 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -600,7 +600,7 @@ async def _index_document( "vector_sync.chunk_count": len(chunk_texts), }, ): - bm25_service = get_bm25_service() + bm25_service = await get_bm25_service() sparse_embeddings = await bm25_service.encode_batch(chunk_texts) async def generate_highlights(): diff --git a/tests/unit/test_embedding_service.py b/tests/unit/test_embedding_service.py new file mode 100644 index 00000000..e6601247 --- /dev/null +++ b/tests/unit/test_embedding_service.py @@ -0,0 +1,108 @@ +"""Unit tests for the embedding singleton accessors. + +Pins the invariant that ``get_bm25_service()`` instantiates its +singleton OFF the event loop. The BM25 provider's constructor calls +``fastembed.SparseTextEmbedding(model_name=...)`` which downloads +~50 MB of model weights from HuggingFace and loads them into memory +— observed >5 s wall-clock in production. If a future refactor +inlines the construction back into the coroutine, kubernetes +``/health/live`` httpGet probes will start timing out and pods will +crashloop (regression seen in the Astrolabe Cloud deck #102 smoke +that prompted this fix). +""" + +from __future__ import annotations + +import time + +import anyio +import pytest + +from nextcloud_mcp_server.embedding import service as svc + +pytestmark = pytest.mark.unit + + +@pytest.fixture(autouse=True) +def _reset_singleton(monkeypatch: pytest.MonkeyPatch) -> None: + """Drop the singleton so each test sees a cold path.""" + monkeypatch.setattr(svc, "_bm25_service", None) + + +async def test_get_bm25_service_runs_init_off_event_loop( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A slow synchronous BM25 init must NOT block other coroutines. + + Simulate the FastEmbed model download with a 1 s blocking sleep + inside the provider's ``__init__``. While that init is in flight, + a concurrent ``anyio.sleep(0.05)`` must finish promptly — proving + the init was offloaded to a worker thread rather than blocking the + event loop. + + Without the ``anyio.to_thread.run_sync`` wrapper this test would + fail: the concurrent sleep would not be scheduled until after the + blocking constructor returned. + """ + init_block_seconds = 1.0 + concurrent_sleep_seconds = 0.05 + + class _BlockingFake: + def __init__(self) -> None: + # Stand-in for SparseTextEmbedding's slow constructor. + time.sleep(init_block_seconds) + + monkeypatch.setattr(svc, "BM25SparseEmbeddingProvider", _BlockingFake) + + concurrent_elapsed: float | None = None + + async def _other_work() -> None: + nonlocal concurrent_elapsed + started = time.monotonic() + await anyio.sleep(concurrent_sleep_seconds) + concurrent_elapsed = time.monotonic() - started + + started = time.monotonic() + async with anyio.create_task_group() as tg: + tg.start_soon(svc.get_bm25_service) + tg.start_soon(_other_work) + total_elapsed = time.monotonic() - started + + assert concurrent_elapsed is not None + # The concurrent task should finish in roughly the time of its own + # anyio.sleep — not delayed by the 1 s constructor. Generous bound + # for flakiness; the meaningful contrast is "well under 1 s". + assert concurrent_elapsed < 0.5, ( + f"concurrent anyio.sleep took {concurrent_elapsed:.3f}s — " + "event loop was blocked by BM25SparseEmbeddingProvider init " + "(regression: missing anyio.to_thread.run_sync wrapper)" + ) + # Total wall-clock should be roughly the init time (the longer of + # the two), confirming the tasks ran in parallel. + assert total_elapsed >= init_block_seconds * 0.9 + assert total_elapsed < init_block_seconds + 0.5 + + +async def test_get_bm25_service_caches_singleton( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Second call must return the same instance without re-running + the constructor — the singleton contract is the whole reason + ``get_bm25_service`` exists.""" + call_count = 0 + + class _CountingFake: + def __init__(self) -> None: + nonlocal call_count + call_count += 1 + + monkeypatch.setattr(svc, "BM25SparseEmbeddingProvider", _CountingFake) + + first = await svc.get_bm25_service() + second = await svc.get_bm25_service() + + assert first is second + assert call_count == 1, ( + f"BM25SparseEmbeddingProvider was constructed {call_count} times — " + "the singleton accessor should construct exactly once" + )