diff --git a/nextcloud_mcp_server/vector/qdrant_client.py b/nextcloud_mcp_server/vector/qdrant_client.py index 6209bc42..dda203cf 100644 --- a/nextcloud_mcp_server/vector/qdrant_client.py +++ b/nextcloud_mcp_server/vector/qdrant_client.py @@ -53,6 +53,7 @@ async def _ensure_keyword_payload_indexes( """ collection_info = await client.get_collection(collection_name) existing_schema = collection_info.payload_schema or {} + failed_fields: list[str] = [] for field in _KEYWORD_PAYLOAD_FIELDS: if field in existing_schema: @@ -86,6 +87,20 @@ async def _ensure_keyword_payload_indexes( e.status_code, body_text, ) + failed_fields.append(field) + + # A single per-field ERROR line is easy to miss in startup noise. Surface + # the partial-failure summary at WARNING so operators auditing the log + # for the post-startup state see a single line listing every missing + # index. See docs/configuration.md for the recovery procedure. + if failed_fields: + logger.warning( + "Payload index creation incomplete on '%s' — fields without indexes: %s. " + "Searches filtering on these fields will fail with HTTP 400 " + "(`Index required but not found`) until the next successful restart.", + collection_name, + ", ".join(failed_fields), + ) async def _backfill_doc_id_to_string( @@ -136,10 +151,15 @@ async def _backfill_doc_id_to_string( rewritten = 0 scanned = 0 + batch_num = 0 # Qdrant scroll returns next_offset as PointId | None — keep it untyped here # so the qdrant client's full union (UUID/int/str/PointId) flows through. next_offset = None batch_size = 256 + # Log progress every N batches so a long-running migration on a large + # collection (≥ 50k points) doesn't look like a startup hang. At batch + # size 256, every 20 batches ≈ 5 120 points scanned. + progress_log_every = 20 # A transient Qdrant failure mid-scroll (network blip, timeout) must not # crash startup. The singleton in get_qdrant_client is already assigned @@ -147,7 +167,10 @@ async def _backfill_doc_id_to_string( # a half-initialized state where the next call returns the cached # client and skips this migration entirely. Catch broadly, log with # exc_info, and return without writing the sentinel — the next process - # restart will retry from scratch. + # restart will retry from scratch. The sentinel write is NOT covered by + # this try/except: a failure there means the data migration succeeded + # and only the short-circuit marker is missing, which is a different + # (and milder) condition than a scroll failure. try: while True: points, next_offset = await client.scroll( @@ -160,6 +183,8 @@ async def _backfill_doc_id_to_string( if not points: break + batch_num += 1 + # Group by stringified value so points sharing a doc_id (one document # → many chunks) collapse into a single set_payload call. Point IDs # can be int/str/UUID, so widen the value type to satisfy the qdrant @@ -189,41 +214,62 @@ async def _backfill_doc_id_to_string( ) rewritten += len(point_ids) + if batch_num % progress_log_every == 0: + logger.info( + "doc_id backfill progress on '%s': scanned %d points, " + "rewrote %d so far", + collection_name, + scanned, + rewritten, + ) + if next_offset is None: break - - # Write the sentinel after a successful scroll so a future restart can - # short-circuit. Empty sparse vector mirrors the placeholder.py - # convention (vector/placeholder.py); zero dense vector is fine - # because the sentinel never participates in a search (no user_id / - # doc_id / doc_type payload to match). - sentinel_point = PointStruct( - id=_DOC_ID_BACKFILL_SENTINEL_ID, - vector={ - "dense": [0.0] * dimension, - "sparse": models.SparseVector(indices=[], values=[]), - }, - payload=dict(_DOC_ID_BACKFILL_SENTINEL_PAYLOAD), + except Exception: + logger.error( + "doc_id backfill scroll failed on '%s'; will retry on next restart", + collection_name, + exc_info=True, ) + return + + # Data backfill succeeded — write the sentinel so a future restart can + # short-circuit. Empty sparse vector mirrors the placeholder.py + # convention (vector/placeholder.py); zero dense vector is fine + # because the sentinel never participates in a search (no user_id / + # doc_id / doc_type payload to match). A failure here is non-fatal: + # the data is correct; only the short-circuit marker is missing, so + # the next restart will re-scroll an already-clean collection (idempotent + # zero-write) before retrying the upsert. + sentinel_point = PointStruct( + id=_DOC_ID_BACKFILL_SENTINEL_ID, + vector={ + "dense": [0.0] * dimension, + "sparse": models.SparseVector(indices=[], values=[]), + }, + payload=dict(_DOC_ID_BACKFILL_SENTINEL_PAYLOAD), + ) + try: await client.upsert( collection_name=collection_name, points=[sentinel_point], wait=True, ) - - logger.info( - "doc_id backfill complete: rewrote %d/%d payloads from int to str", - rewritten, - scanned, - ) except Exception: - logger.error( - "doc_id backfill failed on '%s'; will retry on next restart", + logger.warning( + "doc_id backfill data succeeded on '%s' but sentinel write failed; " + "next restart will re-scroll (idempotent zero-write on clean collection)", collection_name, exc_info=True, ) return + logger.info( + "doc_id backfill complete: rewrote %d/%d payloads from int to str", + rewritten, + scanned, + ) + async def get_qdrant_client() -> AsyncQdrantClient: """ diff --git a/tests/unit/vector/test_qdrant_client.py b/tests/unit/vector/test_qdrant_client.py index 332d0bc6..be728d48 100644 --- a/tests/unit/vector/test_qdrant_client.py +++ b/tests/unit/vector/test_qdrant_client.py @@ -401,7 +401,14 @@ async def test_backfill_logs_and_returns_when_scroll_raises(mocker, caplog): """ client = mocker.AsyncMock() client.retrieve.return_value = [] # No sentinel — backfill must run - client.scroll.side_effect = RuntimeError("boom") + + # An async-callable side_effect lets AsyncMock await the coroutine + # before the exception propagates; assigning a bare exception class + # leaks an un-awaited coroutine and trips RuntimeWarning at gc time. + async def _scroll_raises(*args, **kwargs): + raise RuntimeError("boom") + + client.scroll.side_effect = _scroll_raises with caplog.at_level("ERROR", logger="nextcloud_mcp_server.vector.qdrant_client"): await _backfill_doc_id_to_string( @@ -413,8 +420,121 @@ async def test_backfill_logs_and_returns_when_scroll_raises(mocker, caplog): client.set_payload.assert_not_awaited() errors = [r for r in caplog.records if r.levelname == "ERROR"] assert len(errors) == 1 - assert "doc_id backfill failed" in errors[0].getMessage() + assert "doc_id backfill scroll failed" in errors[0].getMessage() assert "test-collection" in errors[0].getMessage() # exc_info=True attaches the original exception to the log record. assert errors[0].exc_info is not None assert errors[0].exc_info[0] is RuntimeError + + +@pytest.mark.unit +async def test_backfill_logs_warning_when_sentinel_upsert_fails(mocker, caplog): + """Sentinel-write failure after a successful scroll logs WARNING, not ERROR. + + A failure here means the data migration succeeded but the + short-circuit marker is missing. The data is correct; only the + marker is absent, so the next restart will re-scroll an + already-clean collection (idempotent zero-write) and retry the + upsert. Differentiating this from a genuine scroll failure prevents + an "ERROR — backfill failed" log line that contradicts the + successful data state. + """ + client = mocker.AsyncMock() + client.retrieve.return_value = [] # No sentinel — backfill must run + client.scroll.return_value = ([], None) # Empty scroll — clean collection + + async def _upsert_raises(*args, **kwargs): + raise RuntimeError("sentinel write blip") + + client.upsert.side_effect = _upsert_raises + + with caplog.at_level("WARNING", logger="nextcloud_mcp_server.vector.qdrant_client"): + await _backfill_doc_id_to_string( + client, "test-collection", _backfill_dimension() + ) + + warnings = [r for r in caplog.records if r.levelname == "WARNING"] + assert len(warnings) == 1 + assert "sentinel write failed" in warnings[0].getMessage() + assert "test-collection" in warnings[0].getMessage() + assert warnings[0].exc_info is not None + assert warnings[0].exc_info[0] is RuntimeError + # No ERROR — data state is correct, not a backfill failure. + assert not [r for r in caplog.records if r.levelname == "ERROR"] + + +@pytest.mark.unit +async def test_backfill_emits_progress_log_every_20_batches(mocker, caplog): + """Long scrolls emit a progress INFO line every 20 batches. + + Operators auditing a 50k+ point collection's startup migration need + proof the server isn't hung; a single start/end pair leaves a + minutes-long silence in the log. The progress line carries the + collection name, scanned count, and rewritten count so the same + log message also acts as a heartbeat. + """ + client = mocker.AsyncMock() + client.retrieve.return_value = [] + + # Return 21 non-empty batches followed by an empty one to terminate + # the loop; every batch contains points already in str form so no + # set_payload calls happen — the test focuses on the progress log + # cadence, not the rewrite path. + str_point = SimpleNamespace(id=1, payload={"doc_id": "abc"}) + batches: list[tuple[list[SimpleNamespace], int | None]] = [ + ([str_point], 1) for _ in range(21) + ] + [([], None)] + client.scroll.side_effect = batches + + with caplog.at_level("INFO", logger="nextcloud_mcp_server.vector.qdrant_client"): + await _backfill_doc_id_to_string( + client, "test-collection", _backfill_dimension() + ) + + progress_messages = [ + r.getMessage() + for r in caplog.records + if "doc_id backfill progress on" in r.getMessage() + ] + # 21 batches → exactly one progress line at batch 20. + assert len(progress_messages) == 1 + assert "scanned 20 points" in progress_messages[0] + assert "test-collection" in progress_messages[0] + + +@pytest.mark.unit +async def test_ensure_keyword_payload_indexes_summarises_failed_fields(mocker, caplog): + """A non-400 failure surfaces both as ERROR and a WARNING summary. + + Per-field ERROR lines are easy to miss in startup noise; the + WARNING summary at the end of the loop names every field that + didn't get an index, so operators auditing the log can spot the + degraded state at a glance. + """ + client = mocker.AsyncMock() + client.get_collection.return_value = SimpleNamespace(payload_schema={}) + # Two of the three fields fail with 5xx; one succeeds. + call_count = {"n": 0} + + async def _create_index(*args, **kwargs): + call_count["n"] += 1 + if call_count["n"] != 2: + raise _make_unexpected(500, b'{"status":{"error":"boom"}}') + return None + + client.create_payload_index.side_effect = _create_index + + with caplog.at_level("WARNING", logger="nextcloud_mcp_server.vector.qdrant_client"): + await _ensure_keyword_payload_indexes(client, "test-collection") + + summary = [ + r.getMessage() + for r in caplog.records + if "Payload index creation incomplete" in r.getMessage() + ] + assert len(summary) == 1 + # Field order matches _KEYWORD_PAYLOAD_FIELDS = ("doc_id", "user_id", "doc_type") + assert "doc_id" in summary[0] + assert "doc_type" in summary[0] + assert "user_id" not in summary[0] # The one that succeeded. + assert "test-collection" in summary[0]