Merge pull request #801 from cbcoutinho/fix/bm25-thread-init

fix(embedding): instantiate BM25 singleton off the event loop

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

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(

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():

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"
+    )