admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: backend-agnostic vector-sync gauges (pending/documents/chunks)

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-04 19:30:10 +02:00
co-authored by Claude Opus 4.8
parent 55ea8dd358
commit fbe70ecd9c
6 changed files with 384 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/api/management.py
+15 -14
View File
@@ -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,
}
nextcloud_mcp_server/app.py
+23
View File
@@ -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
nextcloud_mcp_server/config.py
+6
View File
@@ -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",
nextcloud_mcp_server/observability/metrics.py
+39
View File
@@ -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,
nextcloud_mcp_server/vector/metrics_publisher.py
+139
View File
@@ -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()
tests/unit/vector/test_metrics_publisher.py
+162
View File
@@ -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