diff --git a/nextcloud_mcp_server/api/management.py b/nextcloud_mcp_server/api/management.py index 1ebb757d..0853ad2d 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,32 @@ 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_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/app.py b/nextcloud_mcp_server/app.py index dfb6b377..af30d55c 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, @@ -1782,6 +1783,18 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = tg, spawn_worker, settings.vector_sync_processor_workers ) + # 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, + 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 @@ -1989,6 +2002,20 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = tg, spawn_worker, settings.vector_sync_processor_workers ) + # 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). + # 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, + 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/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/config.py b/nextcloud_mcp_server/config.py index 96cce29b..334508cb 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 @@ -270,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), @@ -645,6 +647,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 +1273,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/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/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/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/nextcloud_mcp_server/vector/metrics_publisher.py b/nextcloud_mcp_server/vector/metrics_publisher.py new file mode 100644 index 00000000..87998201 --- /dev/null +++ b/nextcloud_mcp_server/vector/metrics_publisher.py @@ -0,0 +1,140 @@ +"""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 import AsyncQdrantClient +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: AsyncQdrantClient, 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/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/tests/unit/vector/test_metrics_publisher.py b/tests/unit/vector/test_metrics_publisher.py new file mode 100644 index 00000000..c533fcda --- /dev/null +++ b/tests/unit/vector/test_metrics_publisher.py @@ -0,0 +1,196 @@ +"""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"] + + 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) + + 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) + 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 diff --git a/third_party/astrolabe b/third_party/astrolabe index 8a26b8d9..3c0e2a58 160000 --- a/third_party/astrolabe +++ b/third_party/astrolabe @@ -1 +1 @@ -Subproject commit 8a26b8d9bf42649488aab29f220cde8d3764cf2c +Subproject commit 3c0e2a58c24694259c61dcba166d4b0d326e081e
Indexed Documents{indexed_count_str}{indexed_documents_str}
Indexed Chunks{indexed_chunks_str}
Pending Documents