From fbe70ecd9cd9e52a9661d9e9f5753f9467a22e47 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 19:30:10 +0200 Subject: [PATCH 1/4] 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 From fc3e0f28f607246d98dfd3ca73b301173159ac43 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 19:39:46 +0200 Subject: [PATCH 2/4] fix: add metrics-interval validator + type/test gaps (review #850) - Add Validator("VECTOR_SYNC_METRICS_REFRESH_INTERVAL", gte=1) so a 0/negative value can't turn the publish loop into a busy-spin. - Annotate count_indexed's qdrant_client param as AsyncQdrantClient. - Add tests: exact kwarg is forwarded to qdrant count, and the placeholder filter matches False (excludes placeholders) with chunk_index pinned to 0. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/config.py | 1 + .../vector/metrics_publisher.py | 3 ++- tests/unit/vector/test_metrics_publisher.py | 25 +++++++++++++++++++ 3 files changed, 28 insertions(+), 1 deletion(-) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 0a09b378..334508cb 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -271,6 +271,7 @@ _dynaconf = Dynaconf( Validator("VECTOR_SYNC_SCAN_INTERVAL", gte=1), Validator("VECTOR_SYNC_PROCESSOR_WORKERS", gte=1), Validator("VECTOR_SYNC_QUEUE_MAX_SIZE", gte=1), + Validator("VECTOR_SYNC_METRICS_REFRESH_INTERVAL", gte=1), Validator("VECTOR_SYNC_USER_POLL_INTERVAL", gte=1), Validator("VERIFICATION_CONCURRENCY", gte=1), Validator("DOCUMENT_CHUNK_SIZE", gte=1), diff --git a/nextcloud_mcp_server/vector/metrics_publisher.py b/nextcloud_mcp_server/vector/metrics_publisher.py index bee70bc1..87998201 100644 --- a/nextcloud_mcp_server/vector/metrics_publisher.py +++ b/nextcloud_mcp_server/vector/metrics_publisher.py @@ -24,6 +24,7 @@ from typing import Any import anyio from anyio.abc import TaskStatus +from qdrant_client import AsyncQdrantClient from qdrant_client.models import FieldCondition, Filter, MatchValue from nextcloud_mcp_server.config import get_settings @@ -41,7 +42,7 @@ logger = logging.getLogger(__name__) async def count_indexed( - qdrant_client, collection: str, *, exact: bool = True + qdrant_client: AsyncQdrantClient, collection: str, *, exact: bool = True ) -> tuple[int, int]: """Return ``(documents, chunks)`` indexed in the collection. diff --git a/tests/unit/vector/test_metrics_publisher.py b/tests/unit/vector/test_metrics_publisher.py index 784f0411..77ad7ddf 100644 --- a/tests/unit/vector/test_metrics_publisher.py +++ b/tests/unit/vector/test_metrics_publisher.py @@ -53,6 +53,31 @@ class TestCountIndexed: assert _must_keys(chunks_filter) == ["is_placeholder"] assert _must_keys(docs_filter) == ["is_placeholder", "chunk_index"] + async def test_placeholder_filter_excludes_placeholders(self) -> None: + # is_placeholder must match False (exclude), not True (which would count + # the in-flight placeholders as if they were indexed content). + qc = AsyncMock() + qc.count.side_effect = [_count_obj(10), _count_obj(3)] + + await mp.count_indexed(qc, _COLLECTION) + + chunks_filter = qc.count.await_args_list[0].kwargs["count_filter"] + assert chunks_filter.must[0].match.value is False + docs_filter = qc.count.await_args_list[1].kwargs["count_filter"] + # And the distinct-document filter pins chunk_index to 0. + assert docs_filter.must[0].match.value is False + assert docs_filter.must[1].match.value == 0 + + async def test_exact_kwarg_forwarded(self) -> None: + # The gauge path passes exact=False; dropping it would silently make the + # every-N-seconds refresh do exact counts on large tenants. + qc = AsyncMock() + qc.count.side_effect = [_count_obj(10), _count_obj(3)] + + await mp.count_indexed(qc, _COLLECTION, exact=False) + + assert all(call.kwargs["exact"] is False for call in qc.count.await_args_list) + class TestPublishVectorSyncMetrics: @pytest.fixture(autouse=True) From 8c9f97d6c4de6139d0c56de41d156a4b5f23cc1d Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 19:47:29 +0200 Subject: [PATCH 3/4] docs: clarify postgres-mode None + test exact=True default (review #850) - Note at both metrics-task call sites that receive_stream is None in postgres mode (get_ingest_pending falls back to procrastinate counts). - Add test_default_is_exact_true covering the status-endpoint count path. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/app.py | 4 ++++ tests/unit/vector/test_metrics_publisher.py | 9 +++++++++ third_party/astrolabe | 2 +- 3 files changed, 14 insertions(+), 1 deletion(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 6f8f0ecc..76b5a07d 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -1748,6 +1748,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # 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). + # receive_stream is None in postgres mode — get_ingest_pending + # falls back to the procrastinate job counts there. await tg.start( vector_sync_metrics_task, task_producer, @@ -1989,6 +1991,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # 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). + # receive_stream is None in postgres mode — get_ingest_pending + # falls back to the procrastinate job counts there. await tg.start( vector_sync_metrics_task, task_producer, diff --git a/tests/unit/vector/test_metrics_publisher.py b/tests/unit/vector/test_metrics_publisher.py index 77ad7ddf..c533fcda 100644 --- a/tests/unit/vector/test_metrics_publisher.py +++ b/tests/unit/vector/test_metrics_publisher.py @@ -78,6 +78,15 @@ class TestCountIndexed: assert all(call.kwargs["exact"] is False for call in qc.count.await_args_list) + async def test_default_is_exact_true(self) -> None: + # The on-demand status endpoint relies on the exact=True default. + qc = AsyncMock() + qc.count.side_effect = [_count_obj(10), _count_obj(3)] + + await mp.count_indexed(qc, _COLLECTION) + + assert all(call.kwargs["exact"] is True for call in qc.count.await_args_list) + class TestPublishVectorSyncMetrics: @pytest.fixture(autouse=True) diff --git a/third_party/astrolabe b/third_party/astrolabe index 8a26b8d9..54adb28f 160000 --- a/third_party/astrolabe +++ b/third_party/astrolabe @@ -1 +1 @@ -Subproject commit 8a26b8d9bf42649488aab29f220cde8d3764cf2c +Subproject commit 54adb28f1d10ac5c143ca28e853d4289fdd0a69b From e4d81d47d943dc48b62b38f165e61e99f885ae3f Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 20:28:44 +0200 Subject: [PATCH 4/4] feat: harmonize MCP tool + userinfo page to documents/chunks model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extend the documents-vs-chunks split to the remaining status surfaces so all three report consistently (Deck #195): - nc_get_vector_sync_status MCP tool + VectorSyncStatusResponse: add indexed_documents (distinct) and indexed_chunks; keep indexed_count as a deprecated alias of indexed_chunks. Reuses count_indexed. - userinfo HTML page (/app/vector-sync/status): show Indexed Documents AND Indexed Chunks rows; switch its count to count_indexed (which also excludes placeholder points — the old raw count included them). - /api/v1/vector-sync/status: restore indexed_count as a deprecated alias of indexed_chunks so existing consumers (integration tests, pre-#115 UI) keep working; the change is now purely additive for indexed_count. Tests: VectorSyncStatusResponse documents/chunks/alias + zeroed defaults. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/api/management.py | 4 ++- nextcloud_mcp_server/auth/userinfo_routes.py | 38 ++++++++++++-------- nextcloud_mcp_server/models/semantic.py | 16 +++++++-- nextcloud_mcp_server/server/semantic.py | 28 +++++++-------- tests/unit/test_vector_sync_status_model.py | 36 +++++++++++++++++++ third_party/astrolabe | 2 +- 6 files changed, 91 insertions(+), 33 deletions(-) create mode 100644 tests/unit/test_vector_sync_status_model.py diff --git a/nextcloud_mcp_server/api/management.py b/nextcloud_mcp_server/api/management.py index c9fd698f..0853ad2d 100644 --- a/nextcloud_mcp_server/api/management.py +++ b/nextcloud_mcp_server/api/management.py @@ -327,9 +327,11 @@ async def get_vector_sync_status(request: Request) -> JSONResponse: "status": status, # 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_chunks exposes the raw point count separately; indexed_count + # is kept as a deprecated alias of indexed_chunks for back-compat. "indexed_documents": indexed_documents, "indexed_chunks": indexed_chunks, + "indexed_count": indexed_chunks, "pending_documents": pending.pending, "ingest_queue": settings.ingest_queue, } diff --git a/nextcloud_mcp_server/auth/userinfo_routes.py b/nextcloud_mcp_server/auth/userinfo_routes.py index 85bc9771..0fcc5a66 100644 --- a/nextcloud_mcp_server/auth/userinfo_routes.py +++ b/nextcloud_mcp_server/auth/userinfo_routes.py @@ -104,7 +104,8 @@ async def _get_processing_status(request: Request) -> dict[str, Any] | None: Dictionary with processing status, or None if vector sync is disabled or components are unavailable: { - "indexed_count": int, # Number of documents in Qdrant + "indexed_documents": int, # Distinct documents in Qdrant + "indexed_chunks": int, # Total chunks/points in Qdrant "pending_count": int, # Number of documents in queue "status": str, # "syncing" or "idle" } @@ -130,30 +131,33 @@ async def _get_processing_status(request: Request) -> dict[str, Any] | None: ingest_queue=settings.ingest_queue, ) - # Get Qdrant client and query indexed count - indexed_count = 0 + # Corpus size: distinct documents AND total chunks (placeholders + # excluded — the prior count included them). + indexed_documents = 0 + indexed_chunks = 0 try: + from nextcloud_mcp_server.vector.metrics_publisher import ( # noqa: PLC0415 + count_indexed, + ) from nextcloud_mcp_server.vector.qdrant_client import ( # noqa: PLC0415 get_qdrant_client, ) qdrant_client = await get_qdrant_client() - - # Count documents in collection - count_result = await qdrant_client.count( - collection_name=settings.get_collection_name() + 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" return { - "indexed_count": indexed_count, + "indexed_documents": indexed_documents, + "indexed_chunks": indexed_chunks, "pending_count": pending.pending, "status": status, } @@ -190,12 +194,14 @@ async def vector_sync_status_fragment(request: Request) -> HTMLResponse: """ ) - indexed_count = processing_status["indexed_count"] + indexed_documents = processing_status["indexed_documents"] + indexed_chunks = processing_status["indexed_chunks"] pending_count = processing_status["pending_count"] status = processing_status["status"] # Format numbers with commas for readability - indexed_count_str = f"{indexed_count:,}" + indexed_documents_str = f"{indexed_documents:,}" + indexed_chunks_str = f"{indexed_chunks:,}" pending_count_str = f"{pending_count:,}" # Status badge color and text @@ -212,7 +218,11 @@ async def vector_sync_status_fragment(request: Request) -> HTMLResponse: - + + + + + diff --git a/nextcloud_mcp_server/models/semantic.py b/nextcloud_mcp_server/models/semantic.py index ae586ee8..35f67ddf 100644 --- a/nextcloud_mcp_server/models/semantic.py +++ b/nextcloud_mcp_server/models/semantic.py @@ -153,14 +153,26 @@ class VectorSyncStatusResponse(BaseResponse): including how many documents are indexed and how many are pending. Attributes: - indexed_count: Number of documents in Qdrant vector database + indexed_documents: Distinct documents indexed in the vector database + indexed_chunks: Total indexed chunks (vector points); ~N per document + indexed_count: DEPRECATED alias of indexed_chunks pending_count: Number of documents in processing queue status: Current sync status ("idle" or "syncing") enabled: Whether vector sync is enabled """ + indexed_documents: int = Field( + default=0, description="Distinct documents indexed in the vector database" + ) + indexed_chunks: int = Field( + default=0, description="Total indexed chunks (vector points); ~N per document" + ) indexed_count: int = Field( - default=0, description="Number of documents indexed in vector database" + default=0, + description=( + "DEPRECATED alias of indexed_chunks (the chunk/point count). Use " + "indexed_documents for the distinct-document count." + ), ) pending_count: int = Field( default=0, description="Number of documents pending processing" diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index 8febb0ae..b2b8ec0c 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -18,7 +18,6 @@ from mcp.types import ( ToolAnnotations, ) from pydantic import Field -from qdrant_client.models import Filter from nextcloud_mcp_server.auth import require_scopes from nextcloud_mcp_server.config import get_settings @@ -41,7 +40,7 @@ from nextcloud_mcp_server.search.bm25_hybrid import BM25HybridSearchAlgorithm from nextcloud_mcp_server.search.context import get_chunk_with_context from nextcloud_mcp_server.search.verification import verify_search_results from nextcloud_mcp_server.utils.validation import parse_modified_timestamp -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__) @@ -982,28 +981,27 @@ def configure_semantic_tools(mcp: FastMCP): ingest_queue=settings.ingest_queue, ) - # Get Qdrant client and query indexed count - indexed_count = 0 + # Corpus size: distinct documents AND total chunks (placeholders + # excluded). A single "indexed" figure is ambiguous because each + # document fans out to ~N chunks. + indexed_documents = 0 + indexed_chunks = 0 try: qdrant_client = await get_qdrant_client() - - # Count documents in collection, excluding placeholders - # Placeholders are zero-vector points used to track processing state - 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" return VectorSyncStatusResponse( - indexed_count=indexed_count, + indexed_documents=indexed_documents, + indexed_chunks=indexed_chunks, + indexed_count=indexed_chunks, # deprecated alias pending_count=pending.pending, status=status, enabled=True, diff --git a/tests/unit/test_vector_sync_status_model.py b/tests/unit/test_vector_sync_status_model.py new file mode 100644 index 00000000..c82caff2 --- /dev/null +++ b/tests/unit/test_vector_sync_status_model.py @@ -0,0 +1,36 @@ +"""Unit tests for VectorSyncStatusResponse documents-vs-chunks fields.""" + +import pytest + +from nextcloud_mcp_server.models.semantic import VectorSyncStatusResponse + +pytestmark = pytest.mark.unit + + +def test_vector_sync_status_documents_and_chunks() -> None: + """Exposes documents AND chunks; indexed_count is a deprecated chunks alias.""" + response = VectorSyncStatusResponse( + indexed_documents=486, + indexed_chunks=16039, + indexed_count=16039, # deprecated alias + pending_count=2214, + status="syncing", + enabled=True, + ingest_queue="memory", + ) + + data = response.model_dump() + assert data["indexed_documents"] == 486 + assert data["indexed_chunks"] == 16039 + # Alias mirrors chunks (not documents) for back-compat. + assert data["indexed_count"] == data["indexed_chunks"] + assert data["pending_count"] == 2214 + + +def test_vector_sync_status_defaults_zeroed() -> None: + """New corpus fields default to 0 (disabled / pre-sync path).""" + response = VectorSyncStatusResponse(status="disabled", enabled=False) + data = response.model_dump() + assert data["indexed_documents"] == 0 + assert data["indexed_chunks"] == 0 + assert data["indexed_count"] == 0 diff --git a/third_party/astrolabe b/third_party/astrolabe index 54adb28f..3c0e2a58 160000 --- a/third_party/astrolabe +++ b/third_party/astrolabe @@ -1 +1 @@ -Subproject commit 54adb28f1d10ac5c143ca28e853d4289fdd0a69b +Subproject commit 3c0e2a58c24694259c61dcba166d4b0d326e081e
Indexed Documents{indexed_count_str}{indexed_documents_str}
Indexed Chunks{indexed_chunks_str}
Pending Documents