diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index c365fd023..fc35722df 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -128,7 +128,10 @@ ) 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.metrics_publisher import ( + vector_density_snapshot_task, + vector_sync_metrics_task, +) from nextcloud_mcp_server.vector.oauth_sync import ( ProvisionSignal, credential_cleanup_task, @@ -2129,6 +2132,12 @@ async def spawn_worker( shutdown_event, ) + # Current-corpus chunk-density snapshot on its own slower cadence + # (heavier collection scroll). Opt-out via + # VECTOR_DENSITY_SNAPSHOT_ENABLED. + if settings.vector_density_snapshot_enabled: + await tg.start(vector_density_snapshot_task, shutdown_event) + logger.info( "Background sync tasks started: 1 scanner + %s processors (queue=%s)", ingest_transport.active_consumer_count, @@ -2360,6 +2369,12 @@ async def spawn_worker( shutdown_event, ) + # Current-corpus chunk-density snapshot on its own slower + # cadence (heavier collection scroll). Opt-out via + # VECTOR_DENSITY_SNAPSHOT_ENABLED. + if settings.vector_density_snapshot_enabled: + await tg.start(vector_density_snapshot_task, shutdown_event) + logger.info( "Background sync tasks started: 1 user manager + %s processors (queue=%s)", ingest_transport.active_consumer_count, diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 3129e0281..27ca8886a 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -114,6 +114,9 @@ "vector_sync_processor_workers": 3, "vector_sync_queue_max_size": 10000, "vector_sync_metrics_refresh_interval": 20, + "vector_density_snapshot_enabled": True, + "vector_density_snapshot_interval": 300, + "vector_density_snapshot_max_documents": 50000, "vector_ram_hnsw_overhead_factor": 1.5, "vector_sync_user_poll_interval": 60, "health_ready_refresh_interval": 15, @@ -450,6 +453,8 @@ def _resolve_settings_files() -> list[str]: 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_DENSITY_SNAPSHOT_INTERVAL", gte=1), + Validator("VECTOR_DENSITY_SNAPSHOT_MAX_DOCUMENTS", gte=1), Validator("VECTOR_RAM_HNSW_OVERHEAD_FACTOR", gte=1), Validator("VECTOR_SYNC_USER_POLL_INTERVAL", gte=1), Validator("HEALTH_READY_REFRESH_INTERVAL", gte=1), @@ -924,6 +929,14 @@ class Settings: # 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 + # Current-corpus chunk-density snapshot (vector/metrics_publisher.py: + # vector_density_snapshot_task). Scrolls the collection to recompute the + # distribution of documents CURRENTLY in Qdrant, so it runs on its own slower + # cadence than the count()-based gauges above. ``max_documents`` caps the + # scroll; hitting it sets the ``..._snapshot_truncated`` gauge (no silent cap). + vector_density_snapshot_enabled: bool = True + vector_density_snapshot_interval: int = 300 # seconds + vector_density_snapshot_max_documents: int = 50000 # HNSW-graph/segment overhead multiplier applied when estimating dense-vector # RAM (``chunks * dim * 4 bytes * factor``). ~1.5 matches the cost-to-serve # note's ~6 KB / 1024-dim observation; a deployment knob because the real @@ -1729,6 +1742,9 @@ def get_settings() -> Settings: "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_density_snapshot_enabled": "VECTOR_DENSITY_SNAPSHOT_ENABLED", + "vector_density_snapshot_interval": "VECTOR_DENSITY_SNAPSHOT_INTERVAL", + "vector_density_snapshot_max_documents": "VECTOR_DENSITY_SNAPSHOT_MAX_DOCUMENTS", "vector_ram_hnsw_overhead_factor": "VECTOR_RAM_HNSW_OVERHEAD_FACTOR", "vector_sync_user_poll_interval": "VECTOR_SYNC_USER_POLL_INTERVAL", "vector_sync_orphan_sweep_enabled": "VECTOR_SYNC_ORPHAN_SWEEP_ENABLED", diff --git a/nextcloud_mcp_server/observability/metrics.py b/nextcloud_mcp_server/observability/metrics.py index 19d80233a..1c9472246 100644 --- a/nextcloud_mcp_server/observability/metrics.py +++ b/nextcloud_mcp_server/observability/metrics.py @@ -16,14 +16,18 @@ import functools import logging +import threading import time from prometheus_client import ( + REGISTRY, Counter, Gauge, Histogram, start_http_server, ) +from prometheus_client.core import GaugeHistogramMetricFamily +from prometheus_client.registry import Collector from nextcloud_mcp_server.observability.tracing import trace_operation @@ -460,11 +464,101 @@ # tail higher and disproportionately inflate vector RAM relative to the billed # source bytes. Buckets straddle that band so the risky tail is visible. Recorded # for both index modes (density is a property of content, not of dense-vs-keyword). +# +# Shared with the current-corpus snapshot GaugeHistogram +# (``astrolabe_qdrant_chunk_density_chunks_per_mb_current``) so the ingest-flow +# panel and the current-distribution panel use identical bucket edges and are +# directly comparable. +CHUNK_DENSITY_BUCKETS = (1, 5, 10, 20, 40, 60, 91, 120, 160, 200, 300, 500) + document_chunk_density_chunks_per_mb = Histogram( "astrolabe_document_chunk_density_chunks_per_mb", "Chunks produced per MB of source content, per embedded document", ["doc_type"], - buckets=(1, 5, 10, 20, 40, 60, 91, 120, 160, 200, 300, 500), + buckets=CHUNK_DENSITY_BUCKETS, +) + +# ----------------------------------------------------------------------------- +# Current-corpus chunk-density snapshot (GaugeHistogram). +# +# Unlike the ingest-time Histogram above — which accumulates one observation per +# document as it is embedded and never decrements — this is a *snapshot* of the +# density distribution of the documents CURRENTLY resident in Qdrant, recomputed +# periodically by scrolling the collection (see +# ``vector.metrics_publisher.vector_density_snapshot_task``). A GaugeHistogram is +# the correct Prometheus type: its buckets rise and fall as the corpus changes. +# +# Fed forward-only: only documents whose Qdrant payload carries +# ``payload_keys.SOURCE_BYTES`` contribute. Documents indexed before that key +# shipped (or otherwise missing a usable source size) are counted separately in +# ``chunk_density_uncovered_documents`` so the snapshot's coverage is explicit. +# ----------------------------------------------------------------------------- +QDRANT_CHUNK_DENSITY_CURRENT_METRIC = ( + "astrolabe_qdrant_chunk_density_chunks_per_mb_current" +) + + +class _ChunkDensitySnapshotCollector(Collector): + """Custom collector exposing the current-corpus density as a GaugeHistogram. + + Holds the most recent snapshot, keyed by ``doc_type``. ``update`` replaces the + whole snapshot atomically (a fresh scroll produces a complete new picture); + ``collect`` yields one GaugeHistogram sample set per ``doc_type``. Emits + nothing until the first snapshot lands, so a scrape before the publisher's + first pass simply omits the metric rather than reporting a misleading zero. + + ``_snapshot`` maps ``doc_type -> (cumulative_buckets, gsum)`` where + ``cumulative_buckets`` is a list of ``(le_str, cumulative_count)`` including + the terminal ``"+Inf"`` bucket, matching Prometheus cumulative-bucket + semantics. ``gcount`` is the ``+Inf`` count, so it is not stored separately. + """ + + def __init__(self) -> None: + self._lock = threading.Lock() + self._snapshot: dict[str, tuple[list[tuple[str, float]], float]] = {} + + def update( + self, snapshot: dict[str, tuple[list[tuple[str, float]], float]] + ) -> None: + with self._lock: + self._snapshot = snapshot + + def collect(self): + with self._lock: + snapshot = self._snapshot + if not snapshot: + return + family = GaugeHistogramMetricFamily( + QDRANT_CHUNK_DENSITY_CURRENT_METRIC, + "Chunks per MB of source content across documents currently in Qdrant " + "(snapshot, recomputed periodically)", + labels=["doc_type"], + ) + for doc_type, (buckets, gsum) in snapshot.items(): + family.add_metric([doc_type], buckets, gsum_value=gsum) + yield family + + +chunk_density_snapshot_collector = _ChunkDensitySnapshotCollector() +REGISTRY.register(chunk_density_snapshot_collector) + +# Documents currently in Qdrant that could NOT be placed in the density snapshot +# because they carry no usable source-byte size (payload predates +# payload_keys.SOURCE_BYTES, or the value is missing/non-positive). Makes the +# forward-only coverage gap visible instead of silently shrinking the histogram. +chunk_density_uncovered_documents = Gauge( + "astrolabe_qdrant_chunk_density_uncovered_documents", + "Documents in Qdrant excluded from the chunk-density snapshot (no source_bytes)", + ["doc_type"], +) + +# 1 when the last density snapshot stopped early at the scan cap +# (vector_density_snapshot_max_documents) and so covers only a prefix of the +# collection; 0 when the whole collection was scanned. Alertable so a truncated +# snapshot is never mistaken for a complete one. +chunk_density_snapshot_truncated = Gauge( + "astrolabe_qdrant_chunk_density_snapshot_truncated", + "1 if the last chunk-density snapshot hit the document scan cap (partial)", ) documents_indexed_total = Counter( @@ -1024,6 +1118,62 @@ def record_chunk_density(doc_type: str, chunk_count: int, source_bytes: int) -> ) +def density_bucket_index(chunks_per_mb: float) -> int: + """Index into a per-bucket tally (``CHUNK_DENSITY_BUCKETS`` + overflow slot). + + Returns the position of the first bucket whose upper edge is ``>=`` the value, + or ``len(CHUNK_DENSITY_BUCKETS)`` (the trailing ``"+Inf"`` overflow slot) when + the value exceeds every finite edge. The companion tally therefore has length + ``len(CHUNK_DENSITY_BUCKETS) + 1``. Shared by the snapshot publisher so its + bucketing matches these exact edges. + """ + for idx, edge in enumerate(CHUNK_DENSITY_BUCKETS): + if chunks_per_mb <= edge: + return idx + return len(CHUNK_DENSITY_BUCKETS) + + +def update_qdrant_chunk_density_snapshot( + per_doc_type: dict[str, tuple[list[float], float]], + *, + uncovered: dict[str, int] | None = None, + truncated: bool = False, +) -> None: + """Publish one current-corpus chunk-density snapshot (GaugeHistogram + coverage). + + ``per_doc_type`` maps ``doc_type -> (bucket_counts, gsum)`` where + ``bucket_counts`` is a NON-cumulative per-bucket tally aligned to + ``CHUNK_DENSITY_BUCKETS`` with one trailing overflow (``"+Inf"``) slot + (length ``len(CHUNK_DENSITY_BUCKETS) + 1``, as produced via + ``density_bucket_index``), and ``gsum`` is the sum of observed densities. The + tally is converted to Prometheus cumulative ``(le, count)`` buckets here — the + single place cumulative-bucket semantics live — and the GaugeHistogram + snapshot is swapped atomically. + + ``uncovered`` (doc_type -> count of docs with no usable source size) and + ``truncated`` (scan hit the document cap) update the companion coverage + gauges. The uncovered gauge is fully reset each snapshot so a doc_type that + falls back to zero uncovered does not leave a stale series. + """ + edges = [str(b) for b in CHUNK_DENSITY_BUCKETS] + ["+Inf"] + snapshot: dict[str, tuple[list[tuple[str, float]], float]] = {} + for doc_type, (bucket_counts, gsum) in per_doc_type.items(): + cumulative: list[tuple[str, float]] = [] + running = 0.0 + for le, count in zip(edges, bucket_counts): + running += count + cumulative.append((le, running)) + snapshot[doc_type] = (cumulative, gsum) + chunk_density_snapshot_collector.update(snapshot) + + # Reset then repopulate so a doc_type absent this round drops to no series. + chunk_density_uncovered_documents.clear() + for doc_type, count in (uncovered or {}).items(): + chunk_density_uncovered_documents.labels(doc_type=doc_type).set(count) + + chunk_density_snapshot_truncated.set(1 if truncated else 0) + + # ============================================================================= # Decorator for Automatic Tool Instrumentation # ============================================================================= diff --git a/nextcloud_mcp_server/vector/metrics_publisher.py b/nextcloud_mcp_server/vector/metrics_publisher.py index 2e374676f..be86fc53e 100644 --- a/nextcloud_mcp_server/vector/metrics_publisher.py +++ b/nextcloud_mcp_server/vector/metrics_publisher.py @@ -30,8 +30,11 @@ from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.embedding import get_embedding_service from nextcloud_mcp_server.observability.metrics import ( + CHUNK_DENSITY_BUCKETS, + density_bucket_index, estimate_vector_bytes, update_ingest_queue_depth, + update_qdrant_chunk_density_snapshot, update_vector_sync_estimated_vector_bytes, update_vector_sync_indexed_chunks, update_vector_sync_indexed_documents, @@ -229,3 +232,145 @@ async def vector_sync_metrics_task( # Sleep until the next refresh or until shutdown, whichever comes first. with anyio.move_on_after(interval): await shutdown_event.wait() + + +async def compute_chunk_density_snapshot( + qdrant_client: AsyncQdrantClient, + collection: str, + *, + max_documents: int, + page_size: int = 1000, +) -> tuple[dict[str, tuple[list[float], float]], dict[str, int], bool]: + """Tally the current-corpus chunk-density distribution by scrolling Qdrant. + + Iterates the ``chunk_index == 0`` point of every non-placeholder document + (one point per document — the same trick ``count_indexed`` uses), reading only + ``total_chunks``, ``source_bytes`` and ``doc_type`` from the payload. For each + document carrying a usable source size it computes + ``total_chunks / (source_bytes / 1e6)`` and increments the matching + per-``doc_type`` bucket (edges = ``CHUNK_DENSITY_BUCKETS`` + a ``+Inf`` + overflow slot). Documents with no usable ``source_bytes`` (payload predates the + key, or a non-positive value) are tallied in ``uncovered`` instead of silently + shrinking the histogram. + + Returns ``(per_doc_type, uncovered, truncated)`` shaped for + ``update_qdrant_chunk_density_snapshot``: ``per_doc_type`` maps + ``doc_type -> (bucket_counts, gsum)``, ``uncovered`` maps ``doc_type -> count``, + and ``truncated`` is True when the scan stopped at ``max_documents`` (partial + snapshot). Errors propagate to the best-effort caller. + """ + n_slots = len(CHUNK_DENSITY_BUCKETS) + 1 + bucket_counts: dict[str, list[float]] = {} + gsums: dict[str, float] = {} + uncovered: dict[str, int] = {} + scanned = 0 + truncated = False + offset = None + + doc_filter = Filter( + must=[ + get_placeholder_filter(), + FieldCondition(key="chunk_index", match=MatchValue(value=0)), + ] + ) + + while True: + points, offset = await qdrant_client.scroll( + collection_name=collection, + scroll_filter=doc_filter, + with_payload=["total_chunks", payload_keys.SOURCE_BYTES, "doc_type"], + with_vectors=False, + limit=page_size, + offset=offset, + ) + for point in points: + payload = point.payload or {} + doc_type = payload.get("doc_type") or "unknown" + total_chunks = payload.get("total_chunks") + source_bytes = payload.get(payload_keys.SOURCE_BYTES) + # bool is an int subclass — exclude it so a stray True can't meter as + # 1. The isinstance checks are inline (not hoisted into a bool) so the + # type checker narrows total_chunks/source_bytes for the division. + if ( + isinstance(total_chunks, int) + and not isinstance(total_chunks, bool) + and total_chunks > 0 + and isinstance(source_bytes, (int, float)) + and not isinstance(source_bytes, bool) + and source_bytes > 0 + ): + density = total_chunks / (source_bytes / 1_000_000) + counts = bucket_counts.setdefault(doc_type, [0.0] * n_slots) + counts[density_bucket_index(density)] += 1 + gsums[doc_type] = gsums.get(doc_type, 0.0) + density + else: + uncovered[doc_type] = uncovered.get(doc_type, 0) + 1 + scanned += len(points) + # Qdrant's end-of-scroll signal (offset is None) is authoritative and is + # checked FIRST: reaching it means the whole collection was covered, so it + # is never truncated — even if the final page pushed ``scanned`` to/over + # the cap. Only when there is genuinely more to fetch (offset not None) + # AND we have already retrieved *strictly more* than the cap do we stop + # early and flag truncation. This tolerates one page of slop and avoids a + # false positive when the collection size lands exactly on the cap: Qdrant + # returns a non-None next offset even when the following scroll would come + # back empty, so ``offset is not None`` alone is not proof of more data. + if offset is None: + break + if scanned > max_documents: + truncated = True + break + + per_doc_type = {dt: (bucket_counts[dt], gsums[dt]) for dt in bucket_counts} + return per_doc_type, uncovered, truncated + + +async def publish_chunk_density_snapshot() -> None: + """Compute and publish one current-corpus chunk-density snapshot. + + Never raises: a metrics refresh must not disturb ingest. On any failure the + previously published snapshot simply remains until the next successful pass. + """ + settings = get_settings() + try: + qdrant_client = await get_qdrant_client() + per_doc_type, uncovered, truncated = await compute_chunk_density_snapshot( + qdrant_client, + settings.get_collection_name(), + max_documents=settings.vector_density_snapshot_max_documents, + ) + update_qdrant_chunk_density_snapshot( + per_doc_type, uncovered=uncovered, truncated=truncated + ) + if truncated: + logger.warning( + "Chunk-density snapshot hit the %s-document scan cap; the " + "published distribution covers only a prefix of the collection", + settings.vector_density_snapshot_max_documents, + ) + except Exception as exc: # noqa: BLE001 — metrics must not break ingest + logger.warning("Failed to publish chunk-density snapshot: %s", exc) + + +async def vector_density_snapshot_task( + shutdown_event: anyio.Event, + *, + task_status: TaskStatus = anyio.TASK_STATUS_IGNORED, +) -> None: + """Publish the current-corpus chunk-density snapshot on a slow cadence. + + Separate from ``vector_sync_metrics_task`` because the collection scroll is + materially heavier than the ``count()``-based gauges, so it runs on its own + (longer) ``vector_density_snapshot_interval``. Spawned only when both vector + sync and the snapshot are enabled (see app startup wiring). + """ + settings = get_settings() + interval = settings.vector_density_snapshot_interval + logger.info("Chunk-density snapshot publisher started (interval=%ss)", interval) + task_status.started() + + while not shutdown_event.is_set(): + await publish_chunk_density_snapshot() + # Sleep until the next snapshot or until shutdown, whichever comes first. + with anyio.move_on_after(interval): + await shutdown_event.wait() diff --git a/nextcloud_mcp_server/vector/payload_keys.py b/nextcloud_mcp_server/vector/payload_keys.py index 5ee59a891..97e8624d4 100644 --- a/nextcloud_mcp_server/vector/payload_keys.py +++ b/nextcloud_mcp_server/vector/payload_keys.py @@ -32,6 +32,16 @@ INDEX_MODE_HYBRID = "hybrid" INDEX_MODE_KEYWORD = "keyword" +# Raw source size of the document in bytes at ingestion time +# (``ingested_byte_size``: raw WebDAV binary for files, UTF-8 text size for text +# doc types). Persisted on every chunk so the current-corpus chunk-density +# snapshot (``astrolabe_qdrant_chunk_density_chunks_per_mb_current``) can compute +# chunks-per-MB from live Qdrant state — the denominator the ingest-time density +# histogram consumes but that was previously discarded after embedding. Written +# forward-only: documents indexed before this key shipped carry no value and are +# reported via the ``uncovered_documents`` gauge until re-ingested. +SOURCE_BYTES = "source_bytes" + # Fixed platform namespace for deterministic chunk point IDs (design §2.2). # Derived once from ``uuid5(NAMESPACE_DNS, "astrolabe.cloud/mcp/point-id/v1")`` # and pinned here as a literal so neither repo recomputes it. DO NOT CHANGE — diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index 19daf9958..cb1506dba 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -1794,13 +1794,19 @@ async def generate_highlights(): ), ) + # Raw source size at ingestion (raw WebDAV binary for files, UTF-8 text size + # for text doc types). Computed once here and reused for both the ingest-time + # density metric and the per-point payload (payload_keys.SOURCE_BYTES) so the + # current-corpus density snapshot can recompute chunks-per-MB from Qdrant. + source_bytes = ingested_byte_size(content_bytes, content) + # Observability-only cost signals (card #624), independent of USAGE_METERING. # Best-effort and self-contained in the helper so a metrics failure can never # disturb indexing. _record_ingest_vector_cost( doc_type=doc_task.doc_type, chunk_count=len(chunk_texts), - source_bytes=ingested_byte_size(content_bytes, content), + source_bytes=source_bytes, dense_for_doc=dense_for_doc, overhead=settings.vector_ram_hnsw_overhead_factor, ) @@ -1913,6 +1919,10 @@ async def generate_highlights(): "chunk_start_offset": chunk.start_offset, "chunk_end_offset": chunk.end_offset, "metadata_version": 2, # v2 includes position metadata + # Raw source size (bytes) at ingestion — the denominator the + # current-corpus density snapshot needs, previously discarded + # after embedding. Same on every chunk of the document. + payload_keys.SOURCE_BYTES: source_bytes, # Decomposition payload keys (design §10.2), additive. payload_keys.PROCESSOR_VERSION: "monolith-v1", payload_keys.PARSED_AT: indexed_at, diff --git a/tests/integration/test_chunk_density_snapshot.py b/tests/integration/test_chunk_density_snapshot.py new file mode 100644 index 000000000..a517d1d4f --- /dev/null +++ b/tests/integration/test_chunk_density_snapshot.py @@ -0,0 +1,115 @@ +"""Current-corpus chunk-density snapshot against a real Qdrant engine. + +Complements the mocked unit tests in +``tests/unit/vector/test_metrics_publisher.py`` by running +``compute_chunk_density_snapshot`` over an in-memory Qdrant collection: this +exercises the actual ``scroll`` pagination, the real ``get_placeholder_filter`` ++ ``chunk_index == 0`` filter, and the payload round-trip of +``payload_keys.SOURCE_BYTES`` — none of which the AsyncMock unit tests can +prove. + +It also pins the forward-only coverage contract end-to-end: a point written +with ``source_bytes`` lands in the density histogram; a legacy point without it +is reported as uncovered; a placeholder is excluded entirely. +""" + +import pytest +from qdrant_client import AsyncQdrantClient +from qdrant_client.models import Distance, PointStruct, VectorParams + +from nextcloud_mcp_server.config import get_settings +from nextcloud_mcp_server.observability.metrics import density_bucket_index +from nextcloud_mcp_server.vector import payload_keys +from nextcloud_mcp_server.vector.metrics_publisher import compute_chunk_density_snapshot + +pytestmark = pytest.mark.integration + +_DIM = 8 + + +def _point( + pid, *, doc_type, chunk_index, total_chunks, source_bytes=None, is_placeholder=False +): + payload = { + "doc_id": str(pid), + "doc_type": doc_type, + "is_placeholder": is_placeholder, + "chunk_index": chunk_index, + "total_chunks": total_chunks, + } + if source_bytes is not None: + payload[payload_keys.SOURCE_BYTES] = source_bytes + return PointStruct(id=pid, vector={"dense": [0.1] * _DIM}, payload=payload) + + +@pytest.fixture +async def seeded_collection(): + """In-memory Qdrant seeded with covered / uncovered / placeholder points.""" + client = AsyncQdrantClient(":memory:") + collection = get_settings().get_collection_name() + await client.create_collection( + collection_name=collection, + vectors_config={"dense": VectorParams(size=_DIM, distance=Distance.COSINE)}, + ) + points = [ + # Covered note: 3 chunks / 1 MB -> density 3.0 (le=5 bucket). + _point( + 1, doc_type="note", chunk_index=0, total_chunks=3, source_bytes=1_000_000 + ), + # Covered note: 100 chunks / 1 MB -> density 100.0 (le=120 bucket). + _point( + 2, doc_type="note", chunk_index=0, total_chunks=100, source_bytes=1_000_000 + ), + # A non-zero chunk of doc 2 — must be ignored (only chunk_index=0 counts). + _point( + 3, doc_type="note", chunk_index=1, total_chunks=100, source_bytes=1_000_000 + ), + # Legacy file: no source_bytes -> uncovered. + _point(4, doc_type="file", chunk_index=0, total_chunks=10), + # Placeholder -> excluded entirely (not covered, not uncovered). + _point( + 5, + doc_type="note", + chunk_index=0, + total_chunks=1, + source_bytes=1_000_000, + is_placeholder=True, + ), + ] + await client.upsert(collection_name=collection, points=points, wait=True) + yield client, collection + await client.close() + + +async def test_snapshot_covers_uncovered_and_excludes_placeholder(seeded_collection): + client, collection = seeded_collection + + per_doc_type, uncovered, truncated = await compute_chunk_density_snapshot( + client, collection, max_documents=1000 + ) + + assert truncated is False + # Placeholder excluded; legacy file counted as uncovered. + assert uncovered == {"file": 1} + + # Only the two chunk_index=0 covered notes contribute (the chunk_index=1 + # point and the placeholder are ignored). + note_counts, note_gsum = per_doc_type["note"] + assert sum(note_counts) == 2 + assert note_counts[density_bucket_index(3.0)] == 1 + assert note_counts[density_bucket_index(100.0)] == 1 + assert note_gsum == pytest.approx(103.0) + # File produced no covered docs, so it has no density series. + assert "file" not in per_doc_type + + +async def test_truncation_signalled_against_real_engine(seeded_collection): + client, collection = seeded_collection + + # Cap below the covered-document count with a tiny page size so the scroll + # has a next page when the cap is reached. + _, _, truncated = await compute_chunk_density_snapshot( + client, collection, max_documents=1, page_size=1 + ) + + assert truncated is True diff --git a/tests/unit/test_chunk_density_snapshot_metric.py b/tests/unit/test_chunk_density_snapshot_metric.py new file mode 100644 index 000000000..dfc59cd90 --- /dev/null +++ b/tests/unit/test_chunk_density_snapshot_metric.py @@ -0,0 +1,121 @@ +"""Unit tests for the current-corpus chunk-density snapshot metric. + +Covers the GaugeHistogram exposition path in ``observability/metrics.py``: + +- ``density_bucket_index`` — the shared bucketing used by the snapshot publisher. +- ``update_qdrant_chunk_density_snapshot`` → ``_ChunkDensitySnapshotCollector`` — + the non-cumulative tally is converted to cumulative ``le`` buckets and exposed + as ``astrolabe_qdrant_chunk_density_chunks_per_mb_current`` (a GaugeHistogram), + plus the ``uncovered`` and ``truncated`` coverage gauges. +""" + +from __future__ import annotations + +import pytest + +from nextcloud_mcp_server.observability.metrics import ( + CHUNK_DENSITY_BUCKETS, + density_bucket_index, + update_qdrant_chunk_density_snapshot, +) + +pytestmark = pytest.mark.unit + +_CURRENT = "astrolabe_qdrant_chunk_density_chunks_per_mb_current" +_N_SLOTS = len(CHUNK_DENSITY_BUCKETS) + 1 + + +def _tally(*densities: float) -> tuple[list[float], float]: + """Build a (bucket_counts, gsum) tally from raw density observations.""" + counts = [0.0] * _N_SLOTS + gsum = 0.0 + for d in densities: + counts[density_bucket_index(d)] += 1 + gsum += d + return counts, gsum + + +class TestDensityBucketIndex: + def test_value_maps_to_first_edge_at_or_above(self): + # edges: 1, 5, 10, 20, 40, 60, 91, 120, 160, 200, 300, 500 + assert density_bucket_index(0.5) == 0 # <= 1 + assert density_bucket_index(1) == 0 # boundary, <= 1 + assert density_bucket_index(3) == 1 # <= 5 + assert density_bucket_index(100) == 7 # <= 120 + assert density_bucket_index(91) == 6 # boundary, <= 91 + + def test_overflow_goes_to_inf_slot(self): + assert density_bucket_index(10_000) == len(CHUNK_DENSITY_BUCKETS) + + +class TestUpdateSnapshot: + def test_cumulative_buckets_gcount_gsum(self, metric_sample): + # note: densities 3 (-> le=5) and 100 (-> le=120); gsum=103, gcount=2. + update_qdrant_chunk_density_snapshot({"note": _tally(3.0, 100.0)}) + + labels = {"doc_type": "note"} + # Counts are exact integers stored as floats; approx keeps the float + # comparison well-defined (and satisfies the no-float-equality lint). + # Below the first observation nothing has accumulated yet. + assert metric_sample( + f"{_CURRENT}_bucket", {**labels, "le": "1"} + ) == pytest.approx(0) + # The le=5 observation shows from le=5 onward (cumulative). + assert metric_sample( + f"{_CURRENT}_bucket", {**labels, "le": "5"} + ) == pytest.approx(1) + assert metric_sample( + f"{_CURRENT}_bucket", {**labels, "le": "91"} + ) == pytest.approx(1) + # The le=120 observation joins the cumulative count at 120. + assert metric_sample( + f"{_CURRENT}_bucket", {**labels, "le": "120"} + ) == pytest.approx(2) + assert metric_sample( + f"{_CURRENT}_bucket", {**labels, "le": "+Inf"} + ) == pytest.approx(2) + assert metric_sample(f"{_CURRENT}_gcount", labels) == pytest.approx(2) + assert metric_sample(f"{_CURRENT}_gsum", labels) == pytest.approx(103.0) + + def test_update_replaces_previous_snapshot(self, metric_sample): + update_qdrant_chunk_density_snapshot({"note": _tally(3.0, 3.0, 3.0)}) + assert metric_sample( + f"{_CURRENT}_gcount", {"doc_type": "note"} + ) == pytest.approx(3) + # A fresh snapshot fully replaces — not accumulates. + update_qdrant_chunk_density_snapshot({"note": _tally(3.0)}) + assert metric_sample( + f"{_CURRENT}_gcount", {"doc_type": "note"} + ) == pytest.approx(1) + + def test_uncovered_and_truncated_gauges(self, metric_sample): + update_qdrant_chunk_density_snapshot( + {"note": _tally(3.0)}, + uncovered={"file": 4, "deck_card": 1}, + truncated=True, + ) + assert metric_sample( + "astrolabe_qdrant_chunk_density_uncovered_documents", + {"doc_type": "file"}, + ) == pytest.approx(4) + assert metric_sample( + "astrolabe_qdrant_chunk_density_snapshot_truncated", {} + ) == pytest.approx(1) + + def test_uncovered_gauge_reset_between_snapshots(self, metric_sample): + update_qdrant_chunk_density_snapshot( + {"note": _tally(3.0)}, uncovered={"file": 9} + ) + assert metric_sample( + "astrolabe_qdrant_chunk_density_uncovered_documents", + {"doc_type": "file"}, + ) == pytest.approx(9) + # Next snapshot has no uncovered files — the stale series must clear. + update_qdrant_chunk_density_snapshot({"note": _tally(3.0)}, uncovered={}) + assert metric_sample( + "astrolabe_qdrant_chunk_density_uncovered_documents", + {"doc_type": "file"}, + ) == pytest.approx(0) + assert metric_sample( + "astrolabe_qdrant_chunk_density_snapshot_truncated", {} + ) == pytest.approx(0) diff --git a/tests/unit/vector/test_metrics_publisher.py b/tests/unit/vector/test_metrics_publisher.py index e5e5eceb8..18c068147 100644 --- a/tests/unit/vector/test_metrics_publisher.py +++ b/tests/unit/vector/test_metrics_publisher.py @@ -337,3 +337,196 @@ async def _fake_publish(task_producer, document_receive_stream) -> None: await mp.vector_sync_metrics_task(None, None, shutdown) assert published == 1 + + +def _point(**payload) -> SimpleNamespace: + """A fake Qdrant scroll Record carrying only a payload.""" + return SimpleNamespace(payload=payload) + + +def _scroll_returning(*pages): + """Build an AsyncMock scroll returning ``(points, next_offset)`` per page. + + Each ``pages`` entry is a list of points; the offset is a sentinel for every + page except the last, which returns ``None`` to end the scroll. + """ + results = [] + for i, points in enumerate(pages): + offset = None if i == len(pages) - 1 else f"offset-{i}" + results.append((points, offset)) + return AsyncMock(side_effect=results) + + +class TestComputeChunkDensitySnapshot: + async def test_buckets_gsum_and_uncovered(self) -> None: + qc = AsyncMock() + qc.scroll = _scroll_returning( + [ + _point(doc_type="note", total_chunks=3, source_bytes=1_000_000), + _point(doc_type="note", total_chunks=100, source_bytes=1_000_000), + _point(doc_type="deck_card", total_chunks=1, source_bytes=2_000_000), + # No source_bytes -> uncovered. + _point(doc_type="file", total_chunks=10), + # Non-positive source_bytes -> uncovered. + _point(doc_type="file", total_chunks=5, source_bytes=0), + ], + ) + + per_doc_type, uncovered, truncated = await mp.compute_chunk_density_snapshot( + qc, _COLLECTION, max_documents=1000 + ) + + assert truncated is False + assert uncovered == {"file": 2} + + # note: density 3 -> idx1 (le=5), density 100 -> idx7 (le=120); gsum=103. + note_counts, note_gsum = per_doc_type["note"] + assert note_counts[1] == 1 + assert note_counts[7] == 1 + assert note_gsum == pytest.approx(103.0) + assert sum(note_counts) == 2 # only the two covered notes + + # deck_card: density 0.5 -> idx0 (le=1); gsum=0.5. + deck_counts, deck_gsum = per_doc_type["deck_card"] + assert deck_counts[0] == 1 + assert deck_gsum == pytest.approx(0.5) + + async def test_scrolls_chunk_index_zero_non_placeholder(self) -> None: + qc = AsyncMock() + qc.scroll = _scroll_returning([]) + + await mp.compute_chunk_density_snapshot(qc, _COLLECTION, max_documents=1000) + + kwargs = qc.scroll.await_args_list[0].kwargs + assert kwargs["with_vectors"] is False + assert "source_bytes" in kwargs["with_payload"] + # One point per document (chunk_index=0), placeholders excluded. + assert _must_keys(kwargs["scroll_filter"]) == ["is_placeholder", "chunk_index"] + + async def test_paginates_until_offset_none(self) -> None: + qc = AsyncMock() + qc.scroll = _scroll_returning( + [_point(doc_type="note", total_chunks=3, source_bytes=1_000_000)], + [_point(doc_type="note", total_chunks=3, source_bytes=1_000_000)], + ) + + per_doc_type, _, truncated = await mp.compute_chunk_density_snapshot( + qc, _COLLECTION, max_documents=1000 + ) + + assert qc.scroll.await_count == 2 + assert truncated is False + assert per_doc_type["note"][0][1] == 2 # both notes counted in le=5 slot + + async def test_truncates_when_corpus_strictly_exceeds_cap(self) -> None: + qc = AsyncMock() + + def note(): + return _point(doc_type="note", total_chunks=3, source_bytes=1_000_000) + + # cap=2, pages of 2: after page 2 scanned=4 (> cap) with a further page + # still to come -> genuinely truncated, stop before fetching page 3. + qc.scroll = _scroll_returning( + [note(), note()], # scanned=2 (== cap, NOT yet truncated) + [note(), note()], # scanned=4 (> cap) -> truncated + [note()], # never fetched + ) + + _, _, truncated = await mp.compute_chunk_density_snapshot( + qc, _COLLECTION, max_documents=2, page_size=2 + ) + + assert truncated is True + assert qc.scroll.await_count == 2 # third page never fetched + + async def test_exact_cap_boundary_is_not_truncated(self) -> None: + # Regression for the false-positive flagged in review: Qdrant can return a + # non-None next offset even at the exact end, so a collection sized exactly + # at the cap must NOT be reported as truncated. Model that: a full page at + # the cap with a non-None offset, then an empty page with offset=None. + qc = AsyncMock() + qc.scroll = _scroll_returning( + [ + _point(doc_type="note", total_chunks=3, source_bytes=1_000_000), + _point(doc_type="note", total_chunks=3, source_bytes=1_000_000), + ], + [], # empty trailing page, offset=None -> authoritative end + ) + + _, _, truncated = await mp.compute_chunk_density_snapshot( + qc, _COLLECTION, max_documents=2, page_size=2 + ) + + assert truncated is False + + +class TestPublishChunkDensitySnapshot: + async def test_computes_and_publishes(self, monkeypatch) -> None: + monkeypatch.setattr( + mp, + "get_settings", + lambda: SimpleNamespace( + vector_density_snapshot_max_documents=1000, + get_collection_name=lambda: _COLLECTION, + ), + ) + qc = AsyncMock() + qc.scroll = _scroll_returning( + [_point(doc_type="note", total_chunks=3, source_bytes=1_000_000)], + ) + monkeypatch.setattr(mp, "get_qdrant_client", AsyncMock(return_value=qc)) + published = MagicMock() + monkeypatch.setattr(mp, "update_qdrant_chunk_density_snapshot", published) + + await mp.publish_chunk_density_snapshot() + + published.assert_called_once() + _, kwargs = published.call_args + assert kwargs["truncated"] is False + assert kwargs["uncovered"] == {} + + async def test_qdrant_failure_is_swallowed(self, monkeypatch) -> None: + monkeypatch.setattr( + mp, + "get_settings", + lambda: SimpleNamespace( + vector_density_snapshot_max_documents=1000, + get_collection_name=lambda: _COLLECTION, + ), + ) + monkeypatch.setattr( + mp, "get_qdrant_client", AsyncMock(side_effect=RuntimeError("qdrant down")) + ) + published = MagicMock() + monkeypatch.setattr(mp, "update_qdrant_chunk_density_snapshot", published) + + # Must not raise — a metrics refresh cannot disturb ingest. + await mp.publish_chunk_density_snapshot() + + published.assert_not_called() + + +class TestVectorDensitySnapshotTask: + async def test_publishes_then_exits_on_shutdown(self, monkeypatch) -> None: + shutdown = anyio.Event() + published = 0 + + # Plain callable: AsyncMock still awaits fine, and this avoids an + # async-def-without-await that the analyzer flags. + def _fake_publish() -> None: + nonlocal published + published += 1 + shutdown.set() + + monkeypatch.setattr( + mp, "publish_chunk_density_snapshot", AsyncMock(side_effect=_fake_publish) + ) + monkeypatch.setattr( + mp, + "get_settings", + lambda: SimpleNamespace(vector_density_snapshot_interval=0), + ) + + await mp.vector_density_snapshot_task(shutdown) + + assert published == 1