Files
mcp-nextcloud/nextcloud_mcp_server/vector/placeholder.py
T
Chris CoutinhoandClaude Opus 4.7 719b3b5034 fix(vector): normalize doc_id to str + add Qdrant keyword payload indexes
Production was logging two cascading classes of Qdrant errors against the
welcomed-malamute deployment:

1. HTTP 400 — "Bad request: Index required but not found for \"doc_id\" of
   one of the following types: [keyword]". The collection was created via
   create_collection() with no payload indexes, so any FieldCondition
   filter on doc_id failed at the Qdrant layer (placeholder writes/reads,
   eviction, search context lookups).

2. Compounding the missing index, producers wrote a mix of int and str
   doc_ids: webhook_parser stringified node_id, scanner stringified note
   IDs, news IDs, and deck card IDs — but the file scanner passed the
   numeric file_id through unchanged. A keyword index would not have
   covered both kinds even if it had existed.

This change:

- Normalizes doc_id to str at every producer site (scanner.py:459,
  DocumentTask.doc_id, indexed_*_ids reads from Qdrant).
- Tightens str|int annotations to str across placeholder.py,
  eviction.py, search/verification.py, search/context.py,
  SearchResult.id, and the auth/api visualization endpoints.
- Defensive str() coercion on doc_id reads in semantic.py /
  bm25_hybrid.py / vector/visualization.py for the transition window
  before the backfill runs.
- Adds an idempotent startup migration in get_qdrant_client():
  - _ensure_keyword_payload_indexes creates KEYWORD indexes for
    doc_id, user_id, and doc_type (tolerates "already exists" 400s).
  - _backfill_doc_id_to_string scrolls the collection once and rewrites
    int doc_ids to str. Skipped after a quick sample shows no legacy
    int payloads.
- Public API preserved: SemanticSearchResult.id stays int via explicit
  int(r.id) narrowing in server/semantic.py — surfaces a TypeError with
  actionable context if a future doc_type ships non-numeric ids.
- Documents the startup migration in docs/configuration.md.

Tests: 11 new unit tests in tests/unit/vector/test_qdrant_client.py
covering happy path / already-exists / unrelated-400 for the index
helpers, and sample-skip / mixed-batch rewrite / payload=None edge cases
for the backfill. 889 unit tests pass.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-08 19:30:51 +02:00

306 lines
9.2 KiB
Python

"""Placeholder point management for Qdrant state tracking.
Placeholders are zero-vector points stored in Qdrant to track document processing
state. They prevent duplicate work by marking documents as "in-flight" during the
gap between scanner queuing and processor completion.
Architecture:
- Scanner writes placeholders when queuing documents for processing
- Processor deletes placeholders and writes real vectors after processing
- All user-facing queries filter out placeholders (is_placeholder: False)
Placeholders contain:
- Zero vectors (dimension from embedding service)
- is_placeholder: True flag (for filtering)
- status: "pending", "processing", "completed", "failed"
- modified_at, etag from source document
- queued_at timestamp
"""
import logging
import time
import uuid
from qdrant_client import models
from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.embedding import get_embedding_service
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
logger = logging.getLogger(__name__)
def _generate_placeholder_id(doc_type: str, doc_id: str) -> str:
"""Generate deterministic UUID for placeholder point.
Args:
doc_type: Document type (note, file, etc.)
doc_id: Document ID
Returns:
UUID string for point ID
"""
point_name = f"{doc_type}:{doc_id}:placeholder"
return str(uuid.uuid5(uuid.NAMESPACE_DNS, point_name))
async def write_placeholder_point(
doc_id: str,
doc_type: str,
user_id: str,
modified_at: int,
etag: str = "",
file_path: str | None = None,
) -> None:
"""Write a placeholder point to Qdrant to mark document as queued.
This should be called by the scanner BEFORE queuing a document for processing.
The placeholder prevents duplicate work if the scanner runs again before
processing completes.
Args:
doc_id: Document ID (always str — see DocumentTask)
doc_type: Document type (note, file, etc.)
user_id: User ID who owns the document
modified_at: Document modification timestamp
etag: Document ETag (if available)
file_path: File path (for files only)
Raises:
Exception: If Qdrant write fails
"""
try:
qdrant_client = await get_qdrant_client()
settings = get_settings()
embedding_service = get_embedding_service()
# Get dimension dynamically (never hardcode)
dimension = embedding_service.get_dimension()
# Create zero vectors
zero_dense = [0.0] * dimension
# Create empty sparse vector for placeholders
# Use models.SparseVector with empty indices/values
empty_sparse = models.SparseVector(indices=[], values=[])
# Generate deterministic point ID
point_id = _generate_placeholder_id(doc_type, doc_id)
# Build payload
payload = {
"user_id": user_id,
"doc_id": doc_id,
"doc_type": doc_type,
"is_placeholder": True,
"status": "pending",
"modified_at": modified_at,
"etag": etag,
"queued_at": int(time.time()),
}
# Add file_path for files
if doc_type == "file" and file_path:
payload["file_path"] = file_path
# Create placeholder point
point = PointStruct(
id=point_id,
vector={
"dense": zero_dense,
"sparse": empty_sparse, # Empty sparse vector for placeholders
},
payload=payload,
)
# Upsert to Qdrant
await qdrant_client.upsert(
collection_name=settings.get_collection_name(),
points=[point],
wait=True,
)
logger.debug(
f"Wrote placeholder for {doc_type}_{doc_id} (user={user_id}, "
f"modified_at={modified_at})"
)
except Exception as e:
logger.error(
f"Failed to write placeholder for {doc_type}_{doc_id}: {e}",
exc_info=True,
)
raise
async def query_document_metadata(
doc_id: str,
doc_type: str,
user_id: str,
) -> dict | None:
"""Query Qdrant for existing document entry (placeholder or real).
Returns the payload of the first matching point, which could be:
- A placeholder (is_placeholder: True)
- A real indexed document (is_placeholder: False or missing)
- None if document not in Qdrant
Args:
doc_id: Document ID
doc_type: Document type
user_id: User ID
Returns:
Payload dict if found, None otherwise
"""
try:
qdrant_client = await get_qdrant_client()
settings = get_settings()
# Query for any entry matching doc_id, doc_type, user_id
scroll_result = await qdrant_client.scroll(
collection_name=settings.get_collection_name(),
scroll_filter=Filter(
must=[
FieldCondition(key="user_id", match=MatchValue(value=user_id)),
FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
]
),
limit=1,
with_payload=True,
with_vectors=False,
)
if scroll_result[0]:
point = scroll_result[0][0]
return dict(point.payload)
return None
except Exception as e:
logger.warning(f"Error querying document metadata for {doc_type}_{doc_id}: {e}")
return None
async def delete_placeholder_point(
doc_id: str,
doc_type: str,
user_id: str,
) -> None:
"""Delete a placeholder point from Qdrant.
This should be called by the processor BEFORE writing real vectors.
We delete the placeholder to avoid duplicates, then write the real chunks.
Args:
doc_id: Document ID
doc_type: Document type
user_id: User ID
Raises:
Exception: If Qdrant delete fails
"""
try:
qdrant_client = await get_qdrant_client()
settings = get_settings()
# Delete by filter (in case there are multiple chunks from old indexing)
await qdrant_client.delete(
collection_name=settings.get_collection_name(),
points_selector=Filter(
must=[
FieldCondition(key="user_id", match=MatchValue(value=user_id)),
FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
FieldCondition(key="is_placeholder", match=MatchValue(value=True)),
]
),
)
logger.debug(f"Deleted placeholder for {doc_type}_{doc_id} (user={user_id})")
except Exception as e:
logger.error(
f"Failed to delete placeholder for {doc_type}_{doc_id}: {e}",
exc_info=True,
)
raise
async def update_placeholder_status(
doc_id: str,
doc_type: str,
user_id: str,
status: str,
) -> None:
"""Update the status field of a placeholder point.
Status values:
- "pending": Queued for processing
- "processing": Currently being processed
- "completed": Processing completed successfully
- "failed": Processing failed
Args:
doc_id: Document ID
doc_type: Document type
user_id: User ID
status: New status value
Raises:
Exception: If Qdrant update fails
"""
try:
qdrant_client = await get_qdrant_client()
settings = get_settings()
# Update payload using set_payload
await qdrant_client.set_payload(
collection_name=settings.get_collection_name(),
payload={"status": status},
points=Filter(
must=[
FieldCondition(key="user_id", match=MatchValue(value=user_id)),
FieldCondition(key="doc_id", match=MatchValue(value=doc_id)),
FieldCondition(key="doc_type", match=MatchValue(value=doc_type)),
FieldCondition(key="is_placeholder", match=MatchValue(value=True)),
]
),
)
logger.debug(
f"Updated placeholder status for {doc_type}_{doc_id} to '{status}' "
f"(user={user_id})"
)
except Exception as e:
logger.warning(
f"Failed to update placeholder status for {doc_type}_{doc_id}: {e}"
)
# Don't raise - status updates are non-critical
def get_placeholder_filter() -> FieldCondition:
"""Get a filter condition to exclude placeholders from queries.
Add this to all user-facing search/visualization queries to ensure
placeholders are never returned to users.
Returns:
FieldCondition that filters out is_placeholder: True
Example:
Filter(
must=[
get_placeholder_filter(), # Exclude placeholders
FieldCondition(key="user_id", match=MatchValue(value=user_id)),
]
)
"""
return FieldCondition(
key="is_placeholder",
match=MatchValue(value=False),
)