fix(vector): address PR review round 16 — type-aware index check, comments
Detect pre-existing payload indexes with the wrong schema type in `_ensure_payload_indexes`. The previous "field already in existing_schema → skip" branch silently survived a collection migrated from the int-doc_id era where `doc_id` is indexed as INTEGER, letting `MatchValue(value="123")` searches keep failing with HTTP 400 on Qdrant Cloud strict mode — exactly the production failure this PR was meant to fix. New behaviour: compare `existing_schema[field].data_type` against the declared type; on mismatch log a WARNING and append to `failed_fields` so the consolidated end-of-function summary picks it up. No auto-repair (operator intervention only — see docs/configuration.md recovery procedure). New test exercises the doc_id-INTEGER scenario end-to-end and asserts both the per-field WARNING and the summary line. Clarify the `_verify_news_items` malformed-doc_id rationale: the news API has no per-item endpoint, so a malformed doc_id genuinely cannot be verified against the source of truth. We err toward false-positive (keep) over false-negative (drop) — same conservative posture as `_verify_notes` and `_verify_deck_cards`. The producer-side validation is the real security boundary; the verifier is defence-in-depth. Both the inline comment and the WARNING message now spell this out. Add a TODO in `get_last_indexed_timestamp` flagging the O(N) cost on every incremental sync tick. The previous single-page `limit=10_000` silently bounded the scroll; paginating fixed correctness but made the unbounded cost visible. The follow-up tracker (canonical TODO at `api/visualization.py`) covers migrating the max-`indexed_at` to a sentinel point or collection metadata for O(1) lookup. Consolidate the duplicate non-numeric-doc_type TODOs at `api/visualization.py:508` and `auth/viz_routes.py:570` into a single canonical comment in `visualization.py`; `viz_routes.py` is reduced to a back-reference. Removes the rot risk of "fixed in one place, forgotten in the other." The canonical comment also references the O(1) timestamp follow-up in `scanner.py`. Document the `batch_size = 256` (qdrant_client.py) vs `_DELETION_TRACKING_PAGE_SIZE = 1024` (scanner.py) split with cross-referencing comments at each site: the smaller batch is for the read-write backfill upsert path (Qdrant accepts ~256-point chunks comfortably); the larger page is for read-only deletion-tracking scrolls where no per-page write round-trip applies. Replace `assert qdrant_client is not None` in `scan_user_documents` with `cast(AsyncQdrantClient, qdrant_client)` plus an explanatory comment. `assert` is silently elided under `-O`; `cast` is the conventional zero-cost narrower for branches the type checker can't infer from the surrounding `if not initial_sync` ternary. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.7
parent
68506f96c5
commit
8f4f5c0079
@@ -505,9 +505,15 @@ async def get_chunk_context(request: Request) -> JSONResponse:
|
|||||||
# ints from MySQL auto_increment; doc_id stays a str downstream
|
# ints from MySQL auto_increment; doc_id stays a str downstream
|
||||||
# (Qdrant payload index is keyword-typed). is_valid_nextcloud_doc_id
|
# (Qdrant payload index is keyword-typed). is_valid_nextcloud_doc_id
|
||||||
# rejects "0", leading zeros, and Unicode digits that pass isdigit().
|
# rejects "0", leading zeros, and Unicode digits that pass isdigit().
|
||||||
# TODO: when chunk-context support extends to non-numeric doc_types
|
#
|
||||||
# (calendar VEVENT UIDs, CardDAV hrefs, …), relax this gate or make
|
# Canonical TODO (referenced by ``auth/viz_routes.py`` and
|
||||||
# it doc_type-aware. Today every indexed doc_type is numeric.
|
# ``vector/scanner.py:get_last_indexed_timestamp``): when chunk-
|
||||||
|
# context support extends to non-numeric doc_types (calendar VEVENT
|
||||||
|
# UIDs, CardDAV hrefs, …), relax this gate or make it doc_type-
|
||||||
|
# aware. Today every indexed doc_type is numeric. The follow-up
|
||||||
|
# tracker also covers the O(N) → O(1) migration of
|
||||||
|
# ``get_last_indexed_timestamp`` (currently re-scans every
|
||||||
|
# ``indexed_at`` on each tick).
|
||||||
if not is_valid_nextcloud_doc_id(doc_id):
|
if not is_valid_nextcloud_doc_id(doc_id):
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -561,15 +561,11 @@ async def chunk_context_endpoint(request: Request) -> JSONResponse:
|
|||||||
assert start_str is not None
|
assert start_str is not None
|
||||||
assert end_str is not None
|
assert end_str is not None
|
||||||
|
|
||||||
# Validate doc_id at the handler boundary: a malformed doc_id would
|
# Same numeric-doc_id gate as ``api/visualization.py`` — see the
|
||||||
# otherwise pass through to get_chunk_with_context and bottom out as a
|
# canonical TODO and rationale there. Kept in sync so both
|
||||||
# 404 from deep inside, not a clear 400. Nextcloud IDs are unsigned
|
# OAuth-protected and direct-access handlers reject malformed
|
||||||
# ints from MySQL auto_increment; doc_id stays a str downstream
|
# IDs at the boundary instead of bottoming out as a 404 from
|
||||||
# (Qdrant payload index is keyword-typed). is_valid_nextcloud_doc_id
|
# deep inside ``get_chunk_with_context``.
|
||||||
# rejects "0", leading zeros, and Unicode digits that pass isdigit().
|
|
||||||
# TODO: when chunk-context support extends to non-numeric doc_types
|
|
||||||
# (calendar VEVENT UIDs, CardDAV hrefs, …), relax this gate or make
|
|
||||||
# it doc_type-aware. Today every indexed doc_type is numeric.
|
|
||||||
if not is_valid_nextcloud_doc_id(doc_id):
|
if not is_valid_nextcloud_doc_id(doc_id):
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
{
|
{
|
||||||
|
|||||||
@@ -392,8 +392,18 @@ async def _verify_news_items(
|
|||||||
# SearchResult.id is always str (Qdrant payload doc_id is keyword-
|
# SearchResult.id is always str (Qdrant payload doc_id is keyword-
|
||||||
# indexed; producers stringify on write). Pass through verbatim.
|
# indexed; producers stringify on write). Pass through verbatim.
|
||||||
if not is_valid_nextcloud_doc_id(d):
|
if not is_valid_nextcloud_doc_id(d):
|
||||||
|
# The news API has no per-item endpoint, so a malformed doc_id
|
||||||
|
# cannot be verified against the source of truth. Err toward
|
||||||
|
# false-positive (keep in results) over false-negative (drop a
|
||||||
|
# potentially legitimate result) — matches the same conservative
|
||||||
|
# posture _verify_notes and _verify_deck_cards take for
|
||||||
|
# non-numeric IDs. The producer-side validation is the real
|
||||||
|
# security boundary; the verifier is defence-in-depth.
|
||||||
logger.warning(
|
logger.warning(
|
||||||
"Malformed news_item doc_id %r in verifier; keeping (cannot verify)",
|
"Malformed news_item doc_id %r in verifier; keeping to "
|
||||||
|
"avoid dropping a potentially legitimate result (news API "
|
||||||
|
"has no per-item endpoint, so cannot verify against source "
|
||||||
|
"of truth — false-positive preferred over false-negative)",
|
||||||
d,
|
d,
|
||||||
)
|
)
|
||||||
accessible.add(d)
|
accessible.add(d)
|
||||||
|
|||||||
@@ -182,9 +182,31 @@ async def _ensure_payload_indexes(
|
|||||||
failed_fields: list[str] = []
|
failed_fields: list[str] = []
|
||||||
for field, schema_type in _PAYLOAD_INDEX_FIELDS.items():
|
for field, schema_type in _PAYLOAD_INDEX_FIELDS.items():
|
||||||
if field in existing_schema:
|
if field in existing_schema:
|
||||||
# Index already present — silent skip. Logging here on every
|
# Index already present. Confirm the existing schema type matches
|
||||||
# restart would be noise that hides the genuinely interesting
|
# what we'd create — a pre-existing collection with `doc_id`
|
||||||
# "first-time creation" line in _create_one_payload_index.
|
# indexed as INTEGER (the bug this PR fixes) would otherwise
|
||||||
|
# silently survive here, and searches using
|
||||||
|
# MatchValue(value="123") would keep failing with HTTP 400 on
|
||||||
|
# Qdrant Cloud strict mode. Compare via PayloadSchemaType
|
||||||
|
# equality; PayloadIndexInfo.data_type is the same enum
|
||||||
|
# we wrote with.
|
||||||
|
existing_info = existing_schema[field]
|
||||||
|
existing_type = getattr(existing_info, "data_type", None)
|
||||||
|
if existing_type is not None and existing_type != schema_type:
|
||||||
|
logger.warning(
|
||||||
|
"Payload index on '%s' has wrong schema type "
|
||||||
|
"(got %s, expected %s); searches filtering on this "
|
||||||
|
"field will fail with HTTP 400 until the index is "
|
||||||
|
"dropped and recreated. See docs/configuration.md "
|
||||||
|
"for the recovery procedure.",
|
||||||
|
field,
|
||||||
|
getattr(existing_type, "name", existing_type),
|
||||||
|
schema_type.name,
|
||||||
|
)
|
||||||
|
failed_fields.append(field)
|
||||||
|
# Either way, skip the create call: a matching index needs no
|
||||||
|
# work, and a mismatch must not be auto-repaired (operator
|
||||||
|
# intervention only — see docs/configuration.md).
|
||||||
continue
|
continue
|
||||||
if not await _create_one_payload_index(
|
if not await _create_one_payload_index(
|
||||||
client, collection_name, field, schema_type
|
client, collection_name, field, schema_type
|
||||||
@@ -338,6 +360,12 @@ async def _backfill_doc_id_to_string(
|
|||||||
# Qdrant scroll returns next_offset as PointId | None — keep it untyped here
|
# 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.
|
# so the qdrant client's full union (UUID/int/str/PointId) flows through.
|
||||||
next_offset = None
|
next_offset = None
|
||||||
|
# Smaller than ``_DELETION_TRACKING_PAGE_SIZE = 1024`` in
|
||||||
|
# ``vector/scanner.py`` because this is a read-write path: every batch
|
||||||
|
# is followed by a ``set_payload`` upsert, and 256-point upserts are
|
||||||
|
# the working size where Qdrant comfortably accepts writes without
|
||||||
|
# timing out under load. The scanner-side scroll has no per-page write
|
||||||
|
# round-trip, so it can use a larger page.
|
||||||
batch_size = 256
|
batch_size = 256
|
||||||
# Log progress every N batches so a long-running migration on a large
|
# 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
|
# collection (≥ 50k points) doesn't look like a startup hang. At batch
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import random
|
|||||||
import time
|
import time
|
||||||
from dataclasses import dataclass
|
from dataclasses import dataclass
|
||||||
from email.utils import parsedate_to_datetime
|
from email.utils import parsedate_to_datetime
|
||||||
|
from typing import cast
|
||||||
|
|
||||||
import anyio
|
import anyio
|
||||||
from anyio.abc import TaskStatus
|
from anyio.abc import TaskStatus
|
||||||
@@ -49,6 +50,13 @@ INDEXED_DOC_TYPES: frozenset[str] = frozenset(
|
|||||||
# < 100 k point per (user_id, doc_type) case. The previous single-page
|
# < 100 k point per (user_id, doc_type) case. The previous single-page
|
||||||
# ``limit=10_000`` silently truncated deletion sets for any user past the
|
# ``limit=10_000`` silently truncated deletion sets for any user past the
|
||||||
# cap, so anything indexed beyond the first 10 k was never reconciled.
|
# cap, so anything indexed beyond the first 10 k was never reconciled.
|
||||||
|
#
|
||||||
|
# Intentionally larger than the ``batch_size = 256`` used by
|
||||||
|
# ``_backfill_doc_id_to_string`` in ``vector/qdrant_client.py``: this is a
|
||||||
|
# read-only scroll that just collects payloads (no write round-trip per
|
||||||
|
# point), so the per-page memory budget is the only relevant constraint.
|
||||||
|
# The 256 there is sized for read-write upsert batches where Qdrant
|
||||||
|
# accepts ~256-point chunks comfortably without timing out under load.
|
||||||
_DELETION_TRACKING_PAGE_SIZE: int = 1024
|
_DELETION_TRACKING_PAGE_SIZE: int = 1024
|
||||||
|
|
||||||
|
|
||||||
@@ -120,6 +128,14 @@ async def get_last_indexed_timestamp(user_id: str) -> int | None:
|
|||||||
Returns:
|
Returns:
|
||||||
Unix timestamp of most recently indexed note, or None if no notes indexed yet
|
Unix timestamp of most recently indexed note, or None if no notes indexed yet
|
||||||
"""
|
"""
|
||||||
|
# TODO: This is O(N) over a user's indexed notes on every incremental
|
||||||
|
# sync tick. Was accidentally bounded at 10 k before this PR (single-
|
||||||
|
# page scroll silently truncated); paginating fixed correctness but
|
||||||
|
# made the unbounded cost visible. Track the max ``indexed_at`` as
|
||||||
|
# collection metadata or a dedicated sentinel point so this becomes
|
||||||
|
# O(1). Out of scope for the current PR — see the chunk-context /
|
||||||
|
# vector-sync follow-up tracker (referenced by the canonical TODO at
|
||||||
|
# ``api/visualization.py``).
|
||||||
try:
|
try:
|
||||||
qdrant_client = await get_qdrant_client()
|
qdrant_client = await get_qdrant_client()
|
||||||
|
|
||||||
@@ -263,7 +279,12 @@ async def scan_user_documents(
|
|||||||
qdrant_client = await get_qdrant_client() if not initial_sync else None
|
qdrant_client = await get_qdrant_client() if not initial_sync else None
|
||||||
indexed_doc_ids = set()
|
indexed_doc_ids = set()
|
||||||
if not initial_sync:
|
if not initial_sync:
|
||||||
assert qdrant_client is not None # narrow for the type checker
|
# ``assert ... is not None`` would also narrow but raises an
|
||||||
|
# opaque AssertionError under ``-O`` and at runtime — ``cast``
|
||||||
|
# is the conventional zero-cost narrower for branches the type
|
||||||
|
# checker can't infer from the surrounding ``if not
|
||||||
|
# initial_sync`` (the ternary above ties the two together).
|
||||||
|
qdrant_client = cast(AsyncQdrantClient, qdrant_client)
|
||||||
points = await _scroll_all_points(
|
points = await _scroll_all_points(
|
||||||
qdrant_client,
|
qdrant_client,
|
||||||
collection_name=get_settings().get_collection_name(),
|
collection_name=get_settings().get_collection_name(),
|
||||||
|
|||||||
@@ -158,6 +158,49 @@ async def test_ensure_payload_indexes_skips_fields_already_indexed(mocker, caplo
|
|||||||
assert not any("doc_id" in m for m in info_messages), info_messages
|
assert not any("doc_id" in m for m in info_messages), info_messages
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.unit
|
||||||
|
async def test_ensure_payload_indexes_warns_on_wrong_schema_type(mocker, caplog):
|
||||||
|
"""Pre-existing index with wrong schema type surfaces as a WARNING.
|
||||||
|
|
||||||
|
The bug this PR fixes: a collection migrated from the int-doc_id era
|
||||||
|
can have ``doc_id`` indexed as INTEGER, which silently survives the
|
||||||
|
"field already in schema → skip" branch and lets ``MatchValue(value="123")``
|
||||||
|
keep failing with HTTP 400 on Qdrant Cloud strict mode. Confirm the
|
||||||
|
type-aware check fires a WARNING, marks the field as failed (so the
|
||||||
|
consolidated end-of-function summary picks it up), and does NOT
|
||||||
|
attempt to recreate the index — operator intervention is the only
|
||||||
|
safe path.
|
||||||
|
"""
|
||||||
|
client = mocker.AsyncMock()
|
||||||
|
# PayloadIndexInfo-like stand-in: only ``data_type`` is read.
|
||||||
|
wrong = SimpleNamespace(data_type=PayloadSchemaType.INTEGER)
|
||||||
|
client.get_collection.return_value = SimpleNamespace(
|
||||||
|
payload_schema={"doc_id": wrong}
|
||||||
|
)
|
||||||
|
|
||||||
|
with caplog.at_level("WARNING", logger="nextcloud_mcp_server.vector.qdrant_client"):
|
||||||
|
await _ensure_payload_indexes(client, "test-collection")
|
||||||
|
|
||||||
|
# No create attempt for the mismatched field.
|
||||||
|
created_fields = {
|
||||||
|
c.kwargs["field_name"] for c in client.create_payload_index.await_args_list
|
||||||
|
}
|
||||||
|
assert "doc_id" not in created_fields
|
||||||
|
|
||||||
|
warning_messages = [
|
||||||
|
r.getMessage() for r in caplog.records if r.levelname == "WARNING"
|
||||||
|
]
|
||||||
|
# Per-field warning describes both observed and expected types.
|
||||||
|
assert any(
|
||||||
|
"doc_id" in m and "INTEGER" in m and "KEYWORD" in m for m in warning_messages
|
||||||
|
), warning_messages
|
||||||
|
# Consolidated summary at end of function includes the field too.
|
||||||
|
assert any(
|
||||||
|
"Payload index creation incomplete" in m and "doc_id" in m
|
||||||
|
for m in warning_messages
|
||||||
|
), warning_messages
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.unit
|
@pytest.mark.unit
|
||||||
async def test_ensure_payload_indexes_logs_400_as_warning(mocker, caplog):
|
async def test_ensure_payload_indexes_logs_400_as_warning(mocker, caplog):
|
||||||
"""Any 400 from create_payload_index is logged at WARNING and skipped.
|
"""Any 400 from create_payload_index is logged at WARNING and skipped.
|
||||||
|
|||||||
Reference in New Issue
Block a user