From 60a9882c9239dc2e69f9099ee0853f0195e6d1d2 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sat, 9 May 2026 00:30:29 +0200 Subject: [PATCH] fix(vector): address PR review round 6 + SonarCloud findings Reviewer feedback (3 important + 3 nits): - Wrap _ensure_keyword_payload_indexes' get_collection() call in try/except. The qdrant_client singleton is already assigned by the time this function runs, so a transient timeout/DNS failure propagating out left the process holding a usable client with the migration silently skipped on every subsequent call. Now logs ERROR with exc_info and returns; next process restart retries. - Add `and "doc_id" in point.payload` guard to the four set comprehensions in scanner.py (indexed_doc_ids, indexed_file_ids, indexed_item_ids, indexed_card_ids). Previously a payload missing the doc_id key would raise KeyError and crash the entire scan. - Tighten test_ensure_keyword_payload_indexes_logs_400_as_warning to match the per-field warning prefix exactly (`startswith("Schema conflict on payload index")`), so a future change adding 400s to the partial-failure summary surfaces here as a count mismatch. - Add new-collection vs existing-collection context to the _backfill_doc_id_to_string docstring's `dimension` parameter. - Replace the misleading "rewrote 0/N from int to str" wording when no rewriting was needed with "N points scanned, none required rewriting (collection already in str form)". - Add test_ensure_keyword_payload_indexes_logs_and_returns_when_ get_collection_raises mirroring the scroll-failure test. SonarCloud (1 CRITICAL + 1 MINOR): - Refactor _backfill_doc_id_to_string to bring cognitive complexity under 15 (was 19). Extracted two pure helpers: _group_int_doc_ids (group point IDs by stringified doc_id) and _apply_backfill_writes (apply set_payload calls and return rewritten count). The main function's scroll/loop/sentinel structure is unchanged. - Add `await asyncio.sleep(0)` to the three async test side_effect helpers (_scroll_raises, _upsert_raises, _create_index) so they use an actual async feature (S7503). The async-callable shape is still required to avoid the AsyncMock unawaited-coroutine warning when side_effect raises. Co-Authored-By: Claude Opus 4.7 (1M context) --- nextcloud_mcp_server/vector/qdrant_client.py | 129 +++++++++++++------ nextcloud_mcp_server/vector/scanner.py | 8 +- tests/unit/vector/test_qdrant_client.py | 55 +++++++- 3 files changed, 150 insertions(+), 42 deletions(-) diff --git a/nextcloud_mcp_server/vector/qdrant_client.py b/nextcloud_mcp_server/vector/qdrant_client.py index dda203cf..3937b07a 100644 --- a/nextcloud_mcp_server/vector/qdrant_client.py +++ b/nextcloud_mcp_server/vector/qdrant_client.py @@ -51,7 +51,22 @@ async def _ensure_keyword_payload_indexes( operators can intervene, but keep going so the remaining fields still get indexed. """ - collection_info = await client.get_collection(collection_name) + # Mirror the broad swallow in `_backfill_doc_id_to_string`: the singleton + # in `get_qdrant_client` is already assigned by the time this function + # runs, so a transient `get_collection` failure (timeout, DNS blip) + # propagating out would leave the process holding a usable client with + # the migration silently skipped on every subsequent call. Log ERROR + # with exc_info and return; the next process restart retries from scratch. + try: + collection_info = await client.get_collection(collection_name) + except Exception: + logger.error( + "Failed to fetch collection info for '%s'; payload indexes not " + "created. Will retry on next restart.", + collection_name, + exc_info=True, + ) + return existing_schema = collection_info.payload_schema or {} failed_fields: list[str] = [] @@ -103,6 +118,56 @@ async def _ensure_keyword_payload_indexes( ) +def _group_int_doc_ids(points: list[Any]) -> tuple[dict[str, list[Any]], int]: + """Group point IDs whose payload carries an int doc_id, keyed by str(doc_id). + + Returns ``(by_value, scanned)`` where ``scanned`` is the total number of + points inspected (str / missing payloads count toward scanned but are not + grouped). Pulled out of ``_backfill_doc_id_to_string`` to keep that + function's cognitive complexity within the project's limit. + + Point IDs widen to ``Any`` to satisfy the qdrant client's + ``PointsSelector`` signature (UUID / int / str unions) without re-spelling + the full type union here. + """ + by_value: dict[str, list[Any]] = {} + scanned = 0 + for point in points: + scanned += 1 + # Qdrant client typing allows None payload even when with_payload was + # requested; defensive default so the type checker is happy. + payload = point.payload or {} + value = payload.get("doc_id") + if value is None or isinstance(value, str): + continue + by_value.setdefault(str(value), []).append(point.id) + return by_value, scanned + + +async def _apply_backfill_writes( + client: AsyncQdrantClient, + collection_name: str, + by_value: dict[str, list[Any]], +) -> int: + """Apply one ``set_payload`` per stringified doc_id; return rewritten count. + + ``wait=True`` is required because ``_ensure_keyword_payload_indexes`` runs + immediately after this function (see ``get_qdrant_client`` near the call + site) and only indexes committed data — fire-and-forget writes would + leave int payloads invisible to KEYWORD filters. + """ + rewritten = 0 + for str_val, point_ids in by_value.items(): + await client.set_payload( + collection_name=collection_name, + payload={"doc_id": str_val}, + points=point_ids, + wait=True, + ) + rewritten += len(point_ids) + return rewritten + + async def _backfill_doc_id_to_string( client: AsyncQdrantClient, collection_name: str, dimension: int ) -> None: @@ -121,11 +186,18 @@ async def _backfill_doc_id_to_string( Within each scroll batch, points sharing the same int doc_id are batched into a single ``set_payload`` call to minimize Qdrant round-trips. + Only called for **existing** collections (see the + ``if collection_name in collection_names`` branch in + ``get_qdrant_client``); brand-new collections skip the backfill since + there can be no legacy int payloads in a freshly created collection. + Args: client: Qdrant client instance. collection_name: Target collection. - dimension: Dense-vector dimension for the sentinel point's vector - (forwarded by ``get_qdrant_client`` from the embedding model). + dimension: Dense-vector dimension for the sentinel point's vector, + forwarded by ``get_qdrant_client`` from the embedding model. + Required because the sentinel is upserted into an existing + collection and must match the collection's vector schema. """ # Sentinel guard: if the migration ran successfully against this # collection on a previous start, retrieve() returns the marker point @@ -184,35 +256,9 @@ async def _backfill_doc_id_to_string( 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 - # client's PointsSelector signature without re-spelling the union. - by_value: dict[str, list[Any]] = {} - for point in points: - scanned += 1 - # Qdrant client typing allows None payload even when with_payload - # was requested; defensive default so the type checker is happy. - payload = point.payload or {} - value = payload.get("doc_id") - if value is None or isinstance(value, str): - continue - by_value.setdefault(str(value), []).append(point.id) - - for str_val, point_ids in by_value.items(): - # wait=True is required because _ensure_keyword_payload_indexes - # runs immediately after this function (see get_qdrant_client - # near the call site) and only indexes committed data — - # fire-and-forget writes would leave int payloads invisible - # to KEYWORD filters. - await client.set_payload( - collection_name=collection_name, - payload={"doc_id": str_val}, - points=point_ids, - wait=True, - ) - rewritten += len(point_ids) + by_value, batch_scanned = _group_int_doc_ids(points) + scanned += batch_scanned + rewritten += await _apply_backfill_writes(client, collection_name, by_value) if batch_num % progress_log_every == 0: logger.info( @@ -264,11 +310,20 @@ async def _backfill_doc_id_to_string( ) return - logger.info( - "doc_id backfill complete: rewrote %d/%d payloads from int to str", - rewritten, - scanned, - ) + if rewritten: + logger.info( + "doc_id backfill complete on '%s': rewrote %d/%d int payloads to str", + collection_name, + rewritten, + scanned, + ) + else: + logger.info( + "doc_id backfill complete on '%s': %d points scanned, none required " + "rewriting (collection already in str form)", + collection_name, + scanned, + ) async def get_qdrant_client() -> AsyncQdrantClient: diff --git a/nextcloud_mcp_server/vector/scanner.py b/nextcloud_mcp_server/vector/scanner.py index c483019b..6322411e 100644 --- a/nextcloud_mcp_server/vector/scanner.py +++ b/nextcloud_mcp_server/vector/scanner.py @@ -230,7 +230,7 @@ async def scan_user_documents( indexed_doc_ids = { str(point.payload["doc_id"]) for point in (scroll_result[0] or []) - if point.payload is not None + if point.payload is not None and "doc_id" in point.payload } logger.debug(f"Found {len(indexed_doc_ids)} indexed documents in Qdrant") @@ -403,7 +403,7 @@ async def scan_user_documents( indexed_file_ids = { str(point.payload["doc_id"]) for point in (file_scroll_result[0] or []) - if point.payload is not None + if point.payload is not None and "doc_id" in point.payload } logger.debug(f"Found {len(indexed_file_ids)} indexed files in Qdrant") @@ -683,7 +683,7 @@ async def scan_news_items( indexed_item_ids = { str(point.payload["doc_id"]) for point in (scroll_result[0] or []) - if point.payload is not None + if point.payload is not None and "doc_id" in point.payload } logger.debug(f"Found {len(indexed_item_ids)} indexed news items in Qdrant") @@ -862,7 +862,7 @@ async def scan_deck_cards( indexed_card_ids = { str(point.payload["doc_id"]) for point in (scroll_result[0] or []) - if point.payload is not None + if point.payload is not None and "doc_id" in point.payload } logger.debug(f"Found {len(indexed_card_ids)} indexed deck cards in Qdrant") diff --git a/tests/unit/vector/test_qdrant_client.py b/tests/unit/vector/test_qdrant_client.py index be728d48..b345d24c 100644 --- a/tests/unit/vector/test_qdrant_client.py +++ b/tests/unit/vector/test_qdrant_client.py @@ -14,6 +14,7 @@ run at startup. Producer-side normalization is exercised by the existing scanner tests. """ +import asyncio from types import SimpleNamespace from unittest.mock import call @@ -152,7 +153,12 @@ async def test_ensure_keyword_payload_indexes_logs_400_as_warning(mocker, caplog # Loop continued past the failing field; all three were attempted. assert client.create_payload_index.await_count == len(_KEYWORD_PAYLOAD_FIELDS) warnings = [r for r in caplog.records if r.levelname == "WARNING"] + # 400s do not contribute to the partial-failure summary (which fires + # only for non-400 errors), so this is the per-field warning, not the + # summary. Match the message prefix exactly so a future change adding + # 400s to the summary would surface here as a count mismatch. assert len(warnings) == 1 + assert warnings[0].getMessage().startswith("Schema conflict on payload index") assert "different schema" in warnings[0].getMessage() @@ -183,6 +189,42 @@ async def test_ensure_keyword_payload_indexes_logs_non_400_as_error(mocker, capl assert "internal server error" in msg +@pytest.mark.unit +async def test_ensure_keyword_payload_indexes_logs_and_returns_when_get_collection_raises( + mocker, caplog +): + """A get_collection failure is logged and swallowed; no indexes are attempted. + + Mirrors the broad swallow in `_backfill_doc_id_to_string`. The + qdrant_client singleton is already assigned by the time this + function runs, so re-raising would leave the process holding a + usable client with the migration silently skipped on every + subsequent call. Catching, logging, and returning preserves the + retry-on-next-restart behavior. + """ + client = mocker.AsyncMock() + + async def _get_collection_raises(*args, **kwargs): + # See _scroll_raises in the backfill section for why this is async. + await asyncio.sleep(0) + raise RuntimeError("connection refused") + + client.get_collection.side_effect = _get_collection_raises + + with caplog.at_level("ERROR", logger="nextcloud_mcp_server.vector.qdrant_client"): + await _ensure_keyword_payload_indexes(client, "test-collection") + + # No index creation was attempted — the function returned early. + client.create_payload_index.assert_not_awaited() + errors = [r for r in caplog.records if r.levelname == "ERROR"] + assert len(errors) == 1 + msg = errors[0].getMessage() + assert "Failed to fetch collection info for 'test-collection'" in msg + assert "Will retry on next restart" in msg + assert errors[0].exc_info is not None + assert errors[0].exc_info[0] is RuntimeError + + # --------------------------------------------------------------------------- # _backfill_doc_id_to_string # --------------------------------------------------------------------------- @@ -212,7 +254,10 @@ async def test_backfill_clean_collection_makes_no_writes(mocker, caplog): r.getMessage() for r in caplog.records if "backfill complete" in r.getMessage() ] assert completion_logs, "expected an INFO log line for backfill completion" - assert "0/2" in completion_logs[0] + # rewritten=0 → human-readable wording instead of the misleading + # "rewrote 0/N from int to str" formula. + assert "2 points scanned" in completion_logs[0] + assert "none required rewriting" in completion_logs[0] @pytest.mark.unit @@ -405,7 +450,11 @@ async def test_backfill_logs_and_returns_when_scroll_raises(mocker, caplog): # 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. + # The `await asyncio.sleep(0)` is a no-op event-loop yield that + # satisfies static analysis ("async function uses no async features") + # without changing observable behavior. async def _scroll_raises(*args, **kwargs): + await asyncio.sleep(0) raise RuntimeError("boom") client.scroll.side_effect = _scroll_raises @@ -444,6 +493,8 @@ async def test_backfill_logs_warning_when_sentinel_upsert_fails(mocker, caplog): client.scroll.return_value = ([], None) # Empty scroll — clean collection async def _upsert_raises(*args, **kwargs): + # See _scroll_raises above for why this is async + sleep(0). + await asyncio.sleep(0) raise RuntimeError("sentinel write blip") client.upsert.side_effect = _upsert_raises @@ -517,6 +568,8 @@ async def test_ensure_keyword_payload_indexes_summarises_failed_fields(mocker, c call_count = {"n": 0} async def _create_index(*args, **kwargs): + # See _scroll_raises above for why this is async + sleep(0). + await asyncio.sleep(0) call_count["n"] += 1 if call_count["n"] != 2: raise _make_unexpected(500, b'{"status":{"error":"boom"}}')