fix(embedding): instantiate BM25 singleton off the event loop
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) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.7
parent
63c1ff0bb4
commit
e0b7afb4b1
@@ -6,6 +6,8 @@ New code should use nextcloud_mcp_server.providers.get_provider() directly.
|
|||||||
|
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
import anyio
|
||||||
|
|
||||||
from nextcloud_mcp_server.providers import get_provider
|
from nextcloud_mcp_server.providers import get_provider
|
||||||
|
|
||||||
from .bm25_provider import BM25SparseEmbeddingProvider
|
from .bm25_provider import BM25SparseEmbeddingProvider
|
||||||
@@ -89,14 +91,39 @@ def get_embedding_service() -> EmbeddingService:
|
|||||||
_bm25_service: BM25SparseEmbeddingProvider | None = None
|
_bm25_service: BM25SparseEmbeddingProvider | None = None
|
||||||
|
|
||||||
|
|
||||||
def get_bm25_service() -> BM25SparseEmbeddingProvider:
|
async def get_bm25_service() -> BM25SparseEmbeddingProvider:
|
||||||
"""
|
"""
|
||||||
Get singleton BM25 sparse embedding service instance.
|
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:
|
Returns:
|
||||||
Global BM25SparseEmbeddingProvider instance
|
Global BM25SparseEmbeddingProvider instance
|
||||||
"""
|
"""
|
||||||
global _bm25_service
|
global _bm25_service
|
||||||
if _bm25_service is None:
|
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
|
return _bm25_service
|
||||||
|
|||||||
@@ -120,7 +120,7 @@ class BM25HybridSearchAlgorithm(SearchAlgorithm):
|
|||||||
|
|
||||||
# Generate sparse embedding for BM25 keyword search
|
# Generate sparse embedding for BM25 keyword search
|
||||||
with trace_operation("search.get_bm25_service"):
|
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"):
|
with trace_operation("search.sparse_embedding_bm25"):
|
||||||
sparse_embedding = await bm25_service.encode_async(query)
|
sparse_embedding = await bm25_service.encode_async(query)
|
||||||
logger.debug(
|
logger.debug(
|
||||||
|
|||||||
@@ -600,7 +600,7 @@ async def _index_document(
|
|||||||
"vector_sync.chunk_count": len(chunk_texts),
|
"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)
|
sparse_embeddings = await bm25_service.encode_batch(chunk_texts)
|
||||||
|
|
||||||
async def generate_highlights():
|
async def generate_highlights():
|
||||||
|
|||||||
@@ -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"
|
||||||
|
)
|
||||||
Reference in New Issue
Block a user