From fbe70ecd9cd9e52a9661d9e9f5753f9467a22e47 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 19:30:10 +0200 Subject: [PATCH] feat: backend-agnostic vector-sync gauges (pending/documents/chunks) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The only queue metric, mcp_vector_sync_queue_size, was updated inline by the single-user consumer (processor_task) but never by the multi-user consumer (oauth_processor_task). On multi-user tenants (e.g. blackbox-demo, 5 users) the gauge read 0 for 24h while the live anyio buffer held ~2214 pending documents (shown by /api/v1/vector-sync/status). The "indexed" figure was also a chunk count (16039 points ≈ 480 docs) mislabelled as documents. Publish a consumer-independent snapshot from a periodic task (vector/metrics_publisher.vector_sync_metrics_task), spawned in BOTH lifespan task groups (single-user and multi-user) and every queue backend: - mcp_vector_sync_pending_documents — outstanding work via ingest_status.get_ingest_pending() (anyio buffer depth or procrastinate todo+doing); also keeps the legacy queue_size gauge meaningful on all paths. - mcp_vector_sync_indexed_documents — distinct documents, counted exactly and cheaply via the one chunk_index=0 point per document (no facet). - mcp_vector_sync_indexed_chunks — total non-placeholder points. The /api/v1/vector-sync/status endpoint now returns indexed_documents (distinct docs) AND indexed_chunks separately, so documents and chunks are no longer conflated. The publisher uses approximate Qdrant counts (every-N-seconds gauge); the on-demand endpoint counts exactly. New knob: VECTOR_SYNC_METRICS_REFRESH_INTERVAL (default 20s). Fail-safe: a metrics refresh never disturbs ingest. BREAKING CHANGE: /api/v1/vector-sync/status field `indexed_documents` now holds the distinct-document count (was the chunk count); the chunk count moved to the new `indexed_chunks` field. The Astrolabe UI + the nc_get_vector_sync_status MCP tool / userinfo page are harmonized in a follow-up (Deck #195). Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/api/management.py | 29 ++-- nextcloud_mcp_server/app.py | 23 +++ nextcloud_mcp_server/config.py | 6 + nextcloud_mcp_server/observability/metrics.py | 39 +++++ .../vector/metrics_publisher.py | 139 +++++++++++++++ tests/unit/vector/test_metrics_publisher.py | 162 ++++++++++++++++++ 6 files changed, 384 insertions(+), 14 deletions(-) create mode 100644 nextcloud_mcp_server/vector/metrics_publisher.py create mode 100644 tests/unit/vector/test_metrics_publisher.py diff --git a/nextcloud_mcp_server/api/management.py b/nextcloud_mcp_server/api/management.py index 1ebb757d..c9fd698f 100644 --- a/nextcloud_mcp_server/api/management.py +++ b/nextcloud_mcp_server/api/management.py @@ -20,13 +20,12 @@ import time from importlib.metadata import version from typing import Any -from qdrant_client.models import Filter from starlette.requests import Request from starlette.responses import JSONResponse from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.config_validators import AuthMode, detect_auth_mode -from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter +from nextcloud_mcp_server.vector.metrics_publisher import count_indexed from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client logger = logging.getLogger(__name__) @@ -307,28 +306,30 @@ async def get_vector_sync_status(request: Request) -> JSONResponse: ingest_queue=settings.ingest_queue, ) - # Get Qdrant client and query indexed count (backend-independent) - indexed_count = 0 + # Corpus size (backend-independent): distinct documents AND total + # chunks. A single "indexed" figure is ambiguous because each document + # fans out to ~N chunks, so both are reported (the UI shows both). + indexed_documents = 0 + indexed_chunks = 0 try: qdrant_client = await get_qdrant_client() - - # Count documents in collection, excluding placeholders - count_result = await qdrant_client.count( - collection_name=settings.get_collection_name(), - count_filter=Filter(must=[get_placeholder_filter()]), + indexed_documents, indexed_chunks = await count_indexed( + qdrant_client, settings.get_collection_name() ) - indexed_count = count_result.count - except Exception as e: - logger.warning("Failed to query Qdrant for indexed count: %s", e) - # Continue with indexed_count = 0 + logger.warning("Failed to query Qdrant for indexed counts: %s", e) + # Continue with zeroed counts # Determine status status = "syncing" if pending.pending > 0 else "idle" body: dict[str, object] = { "status": status, - "indexed_documents": indexed_count, + # indexed_documents is now the distinct-document count (was the chunk + # count before — the two differ by the per-document chunk fan-out). + # indexed_chunks exposes the raw point count separately. + "indexed_documents": indexed_documents, + "indexed_chunks": indexed_chunks, "pending_documents": pending.pending, "ingest_queue": settings.ingest_queue, } diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 09d6568d..6f8f0ecc 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -126,6 +126,7 @@ from nextcloud_mcp_server.server import ( ) from nextcloud_mcp_server.server.auth_tools import register_auth_tools from nextcloud_mcp_server.server.oauth_tools import register_oauth_tools +from nextcloud_mcp_server.vector.metrics_publisher import vector_sync_metrics_task from nextcloud_mcp_server.vector.oauth_sync import ( oauth_processor_task, user_manager_task, @@ -1744,6 +1745,16 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = username, ) + # Publish outstanding-work + corpus gauges on a fixed cadence, + # independent of the consumer path and queue backend (fixes the + # gauge reading 0 on the multi-user path; see metrics_publisher). + await tg.start( + vector_sync_metrics_task, + task_producer, + receive_stream, + shutdown_event, + ) + # Expose this long-lived task group to request-path code that # wants to spawn background work (e.g. ADR-019 verify-on-read # eviction). Eviction coroutines have their own try/except, so @@ -1973,6 +1984,18 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = nextcloud_host_for_sync, ) + # Publish outstanding-work + corpus gauges on a fixed + # cadence. Critical on this multi-user path: the consumer is + # oauth_processor_task, which never updated the queue gauge, + # so without this the gauge read 0 while the buffer held + # thousands of pending docs (see metrics_publisher). + await tg.start( + vector_sync_metrics_task, + task_producer, + receive_stream, + shutdown_event, + ) + # Expose this long-lived task group to request-path code # that wants to spawn background work (e.g. ADR-019 # verify-on-read eviction). Eviction coroutines have their diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 96cce29b..0a09b378 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -88,6 +88,7 @@ _DEFAULTS: dict[str, Any] = { "vector_sync_scan_interval": 300, "vector_sync_processor_workers": 3, "vector_sync_queue_max_size": 10000, + "vector_sync_metrics_refresh_interval": 20, "vector_sync_user_poll_interval": 60, # Orphan-sweep at Pod startup (card #101). When True, delete any # placeholders carrying a different / absent ``instance_id`` before @@ -645,6 +646,10 @@ class Settings: vector_sync_scan_interval: int = 300 # seconds (5 minutes) vector_sync_processor_workers: int = 3 vector_sync_queue_max_size: int = 10000 + # Cadence for the periodic gauge publisher (vector/metrics_publisher.py): + # outstanding-work + indexed documents/chunks. Decoupled from the consumer + # so the gauges are correct on every deployment mode and queue backend. + vector_sync_metrics_refresh_interval: int = 20 # seconds vector_sync_user_poll_interval: int = 60 # seconds - OAuth mode user discovery vector_sync_orphan_sweep_enabled: bool = True # card #101 # System tag marking files for vector indexing. The scanner indexes files @@ -1267,6 +1272,7 @@ def get_settings() -> Settings: "vector_sync_scan_interval": "VECTOR_SYNC_SCAN_INTERVAL", "vector_sync_processor_workers": "VECTOR_SYNC_PROCESSOR_WORKERS", "vector_sync_queue_max_size": "VECTOR_SYNC_QUEUE_MAX_SIZE", + "vector_sync_metrics_refresh_interval": "VECTOR_SYNC_METRICS_REFRESH_INTERVAL", "vector_sync_user_poll_interval": "VECTOR_SYNC_USER_POLL_INTERVAL", "vector_sync_orphan_sweep_enabled": "VECTOR_SYNC_ORPHAN_SWEEP_ENABLED", "vector_sync_pdf_tag": "VECTOR_SYNC_PDF_TAG", diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 3f936765..7b9221cd 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -166,6 +166,30 @@ vector_sync_queue_size = Gauge( "Current number of documents in processing queue", ) +# Outstanding ingest work (queued + in-flight), backend-agnostic. Published by +# the periodic vector_sync_metrics_task from ingest_status.get_ingest_pending(), +# so it is correct on every consumer path (single-user processor_task AND +# multi-user oauth_processor_task) and every queue backend (anyio buffer depth +# or procrastinate todo+doing) — unlike the per-loop update of +# ``vector_sync_queue_size``, which only ran on the single-user path. +vector_sync_pending_documents = Gauge( + "mcp_vector_sync_pending_documents", + "Outstanding ingest documents (queued or in-flight, not yet processed)", +) + +# Corpus size in the vector store. ``indexed_documents`` counts distinct +# documents (one chunk_index=0 point per document); ``indexed_chunks`` counts +# every non-placeholder point. The two differ by the chunk fan-out (~N chunks +# per document), which is why a single "indexed" figure is ambiguous. +vector_sync_indexed_documents = Gauge( + "mcp_vector_sync_indexed_documents", + "Distinct documents indexed in the vector store (non-placeholder)", +) +vector_sync_indexed_chunks = Gauge( + "mcp_vector_sync_indexed_chunks", + "Total indexed chunks (non-placeholder points) in the vector store", +) + qdrant_operations_total = Counter( "mcp_qdrant_operations_total", "Total Qdrant vector database operations", @@ -520,6 +544,21 @@ def update_vector_sync_queue_size(size: int) -> None: vector_sync_queue_size.set(size) +def update_vector_sync_pending_documents(count: int) -> None: + """Set the outstanding-ingest-work gauge (queued + in-flight documents).""" + vector_sync_pending_documents.set(count) + + +def update_vector_sync_indexed_documents(count: int) -> None: + """Set the distinct-indexed-documents gauge.""" + vector_sync_indexed_documents.set(count) + + +def update_vector_sync_indexed_chunks(count: int) -> None: + """Set the total-indexed-chunks gauge.""" + vector_sync_indexed_chunks.set(count) + + def record_document_parse( processor: str, tier: str, diff --git a/nextcloud_mcp_server/vector/metrics_publisher.py b/nextcloud_mcp_server/vector/metrics_publisher.py new file mode 100644 index 00000000..bee70bc1 --- /dev/null +++ b/nextcloud_mcp_server/vector/metrics_publisher.py @@ -0,0 +1,139 @@ +"""Periodic publisher for vector-sync outstanding-work + corpus-size gauges. + +Why a dedicated task instead of updating the gauges inline? The per-loop update +of ``mcp_vector_sync_queue_size`` only runs in the single-user consumer +(``processor_task``). The multi-user consumer (``oauth_processor_task``) drains +the same queue but never touched the gauge, so in multi-user deployments the +gauge read 0 while the live anyio buffer held thousands of pending documents +(observed on tenant-blackbox-demo: gauge 0 for 24h vs 2214 pending in the status +endpoint). This task publishes the *same* ``get_ingest_pending()`` figure the +``/api/v1/vector-sync/status`` endpoint serves, on a fixed cadence, independent +of which consumer drains the queue and of the queue backend (anyio buffer depth +or procrastinate ``todo+doing``). + +It also publishes corpus size split into documents vs chunks. ``indexed_chunks`` +is every non-placeholder point; ``indexed_documents`` is the distinct document +count, obtained exactly and cheaply by counting the ``chunk_index=0`` point each +document has (both fields are payload-indexed), avoiding a Qdrant facet pass. +""" + +from __future__ import annotations + +import logging +from typing import Any + +import anyio +from anyio.abc import TaskStatus +from qdrant_client.models import FieldCondition, Filter, MatchValue + +from nextcloud_mcp_server.config import get_settings +from nextcloud_mcp_server.observability.metrics import ( + update_vector_sync_indexed_chunks, + update_vector_sync_indexed_documents, + update_vector_sync_pending_documents, + update_vector_sync_queue_size, +) +from nextcloud_mcp_server.vector.ingest_status import get_ingest_pending +from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter +from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client + +logger = logging.getLogger(__name__) + + +async def count_indexed( + qdrant_client, collection: str, *, exact: bool = True +) -> tuple[int, int]: + """Return ``(documents, chunks)`` indexed in the collection. + + ``chunks`` is every non-placeholder point; ``documents`` is the distinct + document count via the ``chunk_index=0`` point each document carries (no + facet needed). Excludes in-flight placeholder points. + + ``exact`` is forwarded to Qdrant ``count``: the periodic gauge publisher + passes ``exact=False`` so the every-N-seconds refresh stays O(1)-ish on + large tenants, while the on-demand status endpoint keeps the default + ``exact=True`` for an accurate user-facing figure. + """ + chunks_result = await qdrant_client.count( + collection_name=collection, + count_filter=Filter(must=[get_placeholder_filter()]), + exact=exact, + ) + docs_result = await qdrant_client.count( + collection_name=collection, + count_filter=Filter( + must=[ + get_placeholder_filter(), + FieldCondition(key="chunk_index", match=MatchValue(value=0)), + ] + ), + exact=exact, + ) + return docs_result.count, chunks_result.count + + +async def publish_vector_sync_metrics( + task_producer: Any, document_receive_stream: Any +) -> None: + """Compute and publish one snapshot of the vector-sync gauges. + + Never raises: a metrics refresh must not disturb the ingest pipeline. Each + figure is published independently so a failure in one (e.g. Qdrant briefly + unreachable) does not block the others. + """ + settings = get_settings() + + # Outstanding work — the same figure the status endpoint serves, so the + # Prometheus gauge and the Astrolabe UI never disagree. + try: + pending = await get_ingest_pending( + task_producer=task_producer, + document_receive_stream=document_receive_stream, + ingest_queue=settings.ingest_queue, + ) + update_vector_sync_pending_documents(pending.pending) + # Keep the legacy gauge meaningful on every consumer path, not just the + # single-user one — existing dashboards/alerts reference it. + update_vector_sync_queue_size(pending.pending) + except Exception as exc: # noqa: BLE001 — metrics must not break ingest + logger.warning("Failed to publish pending-documents gauge: %s", exc) + + # Corpus size — documents and chunks separately (the chunk fan-out makes a + # single "indexed" number ambiguous). + try: + qdrant_client = await get_qdrant_client() + # Approximate is plenty for a gauge refreshed every few seconds and keeps + # the cost bounded on large tenants; the status endpoint counts exactly. + documents, chunks = await count_indexed( + qdrant_client, settings.get_collection_name(), exact=False + ) + update_vector_sync_indexed_documents(documents) + update_vector_sync_indexed_chunks(chunks) + except Exception as exc: # noqa: BLE001 — metrics must not break ingest + logger.warning("Failed to publish indexed-corpus gauges: %s", exc) + + +async def vector_sync_metrics_task( + task_producer: Any, + document_receive_stream: Any, + shutdown_event: anyio.Event, + *, + task_status: TaskStatus = anyio.TASK_STATUS_IGNORED, +) -> None: + """Publish the vector-sync gauges every ``vector_sync_metrics_refresh_interval``. + + Spawned in every deployment mode and queue backend so the outstanding-work + and corpus gauges are accurate regardless of which consumer drains the queue. + ``document_receive_stream`` is None in postgres mode — ``get_ingest_pending`` + falls back to the procrastinate job counts there. + """ + settings = get_settings() + interval = settings.vector_sync_metrics_refresh_interval + logger.info("Vector-sync metrics publisher started (interval=%ss)", interval) + task_status.started() + + while not shutdown_event.is_set(): + await publish_vector_sync_metrics(task_producer, document_receive_stream) + # Sleep until the next refresh or until shutdown, whichever comes first. + with anyio.move_on_after(interval): + await shutdown_event.wait() diff --git a/tests/unit/vector/test_metrics_publisher.py b/tests/unit/vector/test_metrics_publisher.py new file mode 100644 index 00000000..784f0411 --- /dev/null +++ b/tests/unit/vector/test_metrics_publisher.py @@ -0,0 +1,162 @@ +"""Unit tests for the vector-sync metrics publisher. + +Covers vector/metrics_publisher.py: the exact document/chunk counting +(documents via the chunk_index=0 point) and the fail-safe snapshot publisher +that fixes the queue gauge reading 0 on the multi-user consumer path. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock + +import anyio +import pytest + +from nextcloud_mcp_server.vector import metrics_publisher as mp +from nextcloud_mcp_server.vector.ingest_status import IngestPending + +pytestmark = pytest.mark.unit + +_COLLECTION = "test_collection" + + +def _count_obj(n: int) -> SimpleNamespace: + return SimpleNamespace(count=n) + + +def _must_keys(flt) -> list[str | None]: + return [getattr(c, "key", None) for c in (flt.must or [])] + + +class TestCountIndexed: + async def test_returns_documents_and_chunks(self) -> None: + qc = AsyncMock() + # count() is called for chunks first, then documents. + qc.count.side_effect = [_count_obj(16039), _count_obj(486)] + + documents, chunks = await mp.count_indexed(qc, _COLLECTION) + + assert (documents, chunks) == (486, 16039) + assert qc.count.await_count == 2 + + async def test_document_count_filters_on_chunk_index_zero(self) -> None: + qc = AsyncMock() + qc.count.side_effect = [_count_obj(10), _count_obj(3)] + + await mp.count_indexed(qc, _COLLECTION) + + # First call = chunks (placeholder filter only); second = documents + # (placeholder filter + chunk_index), the distinct-document trick. + chunks_filter = qc.count.await_args_list[0].kwargs["count_filter"] + docs_filter = qc.count.await_args_list[1].kwargs["count_filter"] + assert _must_keys(chunks_filter) == ["is_placeholder"] + assert _must_keys(docs_filter) == ["is_placeholder", "chunk_index"] + + +class TestPublishVectorSyncMetrics: + @pytest.fixture(autouse=True) + def _stub_settings(self, monkeypatch) -> None: + settings = SimpleNamespace( + ingest_queue="memory", + get_collection_name=lambda: _COLLECTION, + ) + monkeypatch.setattr(mp, "get_settings", lambda: settings) + + @pytest.fixture + def gauges(self, monkeypatch) -> dict[str, MagicMock]: + g = { + name: MagicMock() + for name in ( + "update_vector_sync_pending_documents", + "update_vector_sync_queue_size", + "update_vector_sync_indexed_documents", + "update_vector_sync_indexed_chunks", + ) + } + for name, mock in g.items(): + monkeypatch.setattr(mp, name, mock) + return g + + async def test_publishes_all_gauges(self, monkeypatch, gauges) -> None: + monkeypatch.setattr( + mp, + "get_ingest_pending", + AsyncMock(return_value=IngestPending(pending=2214)), + ) + qc = AsyncMock() + qc.count.side_effect = [_count_obj(16039), _count_obj(486)] + monkeypatch.setattr(mp, "get_qdrant_client", AsyncMock(return_value=qc)) + + await mp.publish_vector_sync_metrics( + task_producer=None, document_receive_stream=object() + ) + + gauges["update_vector_sync_pending_documents"].assert_called_once_with(2214) + # Legacy gauge kept meaningful on every consumer path. + gauges["update_vector_sync_queue_size"].assert_called_once_with(2214) + gauges["update_vector_sync_indexed_documents"].assert_called_once_with(486) + gauges["update_vector_sync_indexed_chunks"].assert_called_once_with(16039) + + async def test_pending_failure_does_not_block_corpus_gauges( + self, monkeypatch, gauges + ) -> None: + # get_ingest_pending raising must not stop the indexed gauges (and must + # not propagate — a metrics refresh cannot disturb ingest). + monkeypatch.setattr( + mp, "get_ingest_pending", AsyncMock(side_effect=RuntimeError("queue down")) + ) + qc = AsyncMock() + qc.count.side_effect = [_count_obj(10), _count_obj(3)] + monkeypatch.setattr(mp, "get_qdrant_client", AsyncMock(return_value=qc)) + + await mp.publish_vector_sync_metrics( + task_producer=None, document_receive_stream=object() + ) + + gauges["update_vector_sync_pending_documents"].assert_not_called() + gauges["update_vector_sync_indexed_documents"].assert_called_once_with(3) + gauges["update_vector_sync_indexed_chunks"].assert_called_once_with(10) + + async def test_qdrant_failure_does_not_block_pending_gauge( + self, monkeypatch, gauges + ) -> None: + monkeypatch.setattr( + mp, + "get_ingest_pending", + AsyncMock(return_value=IngestPending(pending=42)), + ) + monkeypatch.setattr( + mp, "get_qdrant_client", AsyncMock(side_effect=RuntimeError("qdrant down")) + ) + + await mp.publish_vector_sync_metrics( + task_producer=None, document_receive_stream=object() + ) + + gauges["update_vector_sync_pending_documents"].assert_called_once_with(42) + gauges["update_vector_sync_indexed_documents"].assert_not_called() + + +class TestVectorSyncMetricsTask: + async def test_publishes_then_exits_on_shutdown(self, monkeypatch) -> None: + shutdown = anyio.Event() + published = 0 + + async def _fake_publish(task_producer, document_receive_stream) -> None: + nonlocal published + published += 1 + shutdown.set() # one pass, then stop the loop + + monkeypatch.setattr( + mp, "publish_vector_sync_metrics", AsyncMock(side_effect=_fake_publish) + ) + monkeypatch.setattr( + mp, + "get_settings", + lambda: SimpleNamespace(vector_sync_metrics_refresh_interval=0), + ) + + await mp.vector_sync_metrics_task(None, None, shutdown) + + assert published == 1