admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement Qdrant placeholder state management

Browse Source 
Introduces a placeholder-based state tracking system to prevent duplicate
document processing during the gap between scanner queuing and processor
completion.

**Key Changes:**

1. **Placeholder Helper Functions** (`vector/placeholder.py`):
   - `write_placeholder_point()` - Creates zero-vector placeholder when queuing
   - `query_document_metadata()` - Queries for existing entry (placeholder or real)
   - `delete_placeholder_point()` - Removes placeholder before writing real vectors
   - `get_placeholder_filter()` - Filters placeholders from user-facing queries

2. **Scanner Updates** (`vector/scanner.py`):
   - Replace `indexed_at` comparison with `modified_at` comparison
   - Write placeholder before queuing each document
   - Query per-document metadata instead of bulk-querying indexed_at
   - Fixes bug where files were resubmitted every scan cycle

3. **Processor Updates** (`vector/processor.py`):
   - Delete placeholder before upserting real vectors
   - Ensures no duplicate points in Qdrant

4. **Query Filters** (all search files):
   - Add `get_placeholder_filter()` to all user-facing queries
   - Ensures placeholders never appear in search results or visualizations
   - Applied to: bm25_hybrid.py, semantic.py, viz_routes.py, algorithms.py

**Architecture:**
- Placeholders use zero vectors with dimension from embedding service
- Payload includes `is_placeholder: True` flag for filtering
- Status field tracks: "pending", "processing", "completed", "failed"
- Deterministic UUIDs using uuid5 for consistent point IDs

**Impact:**
- Eliminates duplicate processing of same documents
- Fixes race condition where long-running documents get queued multiple times
- Prevents scanner from resubmitting files every scan cycle
- Maintains clean separation between in-flight and indexed documents

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2025-11-20 15:04:00 +01:00
co-authored by Claude
parent 944dd760ca
commit 13b2d0048c
8 changed files with 424 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/vector/scanner.py
+94 -27
View File
@@ -17,6 +17,10 @@ from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.observability.metrics import record_vector_sync_scan
from nextcloud_mcp_server.observability.tracing import trace_operation
from nextcloud_mcp_server.vector.placeholder import (
query_document_metadata,
write_placeholder_point,
)
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
logger = logging.getLogger(__name__)
@@ -184,8 +188,9 @@ async def scan_user_documents(
f"[SCAN-{scan_id}] Using pruneBefore={prune_before} to optimize data transfer"
)
# Get indexed state from Qdrant first (for incremental sync)
indexed_docs = {}
# For deletion tracking, get all doc_ids in Qdrant (for incremental sync)
# Note: We no longer bulk-query indexed_at, instead check per-document
indexed_doc_ids = set()
if not initial_sync:
qdrant_client = await get_qdrant_client()
scroll_result = await qdrant_client.scroll(
@@ -196,17 +201,14 @@ async def scan_user_documents(
FieldCondition(key="doc_type", match=MatchValue(value="note")),
]
),
with_payload=["doc_id", "indexed_at"],
with_payload=["doc_id"],
with_vectors=False,
limit=10000,
)
indexed_docs = {
point.payload["doc_id"]: point.payload["indexed_at"]
for point in scroll_result[0]
}
indexed_doc_ids = {point.payload["doc_id"] for point in scroll_result[0]}
logger.debug(f"Found {len(indexed_docs)} indexed documents in Qdrant")
logger.debug(f"Found {len(indexed_doc_ids)} indexed documents in Qdrant")
# Stream notes from Nextcloud and process immediately
note_count = 0
@@ -220,7 +222,14 @@ async def scan_user_documents(
modified_at = note.get("modified", 0)
if initial_sync:
# Send everything on first sync
# Send everything on first sync - write placeholder first
await write_placeholder_point(
doc_id=doc_id,
doc_type="note",
user_id=user_id,
modified_at=modified_at,
etag=note.get("etag", ""),
)
await send_stream.send(
DocumentTask(
user_id=user_id,
@@ -232,9 +241,7 @@ async def scan_user_documents(
)
queued += 1
else:
# Incremental sync: compare with indexed state
indexed_at = indexed_docs.get(doc_id)
# Incremental sync: check if document exists and compare modified_at
# If document reappeared, remove from potentially_deleted
doc_key = (user_id, doc_id)
if doc_key in _potentially_deleted:
@@ -243,8 +250,36 @@ async def scan_user_documents(
)
del _potentially_deleted[doc_key]
# Query Qdrant for existing entry (placeholder or real)
existing_metadata = await query_document_metadata(
doc_id=doc_id, doc_type="note", user_id=user_id
)
# Send if never indexed or modified since last index
if indexed_at is None or modified_at > indexed_at:
# Compare against stored modified_at (not indexed_at!)
needs_indexing = False
if existing_metadata is None:
# Never seen before
needs_indexing = True
elif existing_metadata.get("modified_at", 0) < modified_at:
# Document modified since last indexing
needs_indexing = True
elif existing_metadata.get("is_placeholder", False):
# Placeholder exists but processing may have failed - requeue
logger.debug(
f"Found existing placeholder for note {doc_id}, requeuing"
)
needs_indexing = True
if needs_indexing:
# Write placeholder before queuing
await write_placeholder_point(
doc_id=doc_id,
doc_type="note",
user_id=user_id,
modified_at=modified_at,
etag=note.get("etag", ""),
)
await send_stream.send(
DocumentTask(
user_id=user_id,
@@ -272,7 +307,7 @@ async def scan_user_documents(
) # Allow 1.5 scan intervals
current_time = time.time()
for doc_id in indexed_docs:
for doc_id in indexed_doc_ids:
if doc_id not in nextcloud_doc_ids:
doc_key = (user_id, doc_id)
@@ -312,8 +347,8 @@ async def scan_user_documents(
_potentially_deleted[doc_key] = current_time
# Scan tagged PDF files (after notes)
# Get indexed files from Qdrant (separate query for doc_type="file")
indexed_files = {}
# Get indexed file IDs from Qdrant (for deletion tracking)
indexed_file_ids = set()
if not initial_sync:
file_scroll_result = await qdrant_client.scroll(
collection_name=settings.get_collection_name(),
@@ -324,16 +359,15 @@ async def scan_user_documents(
]
),
limit=10000, # Reasonable limit for file count
with_payload=["doc_id", "indexed_at"],
with_payload=["doc_id"],
with_vectors=False,
)
indexed_files = {
point.payload["doc_id"]: point.payload["indexed_at"]
for point in file_scroll_result[0]
indexed_file_ids = {
point.payload["doc_id"] for point in file_scroll_result[0]
}
logger.debug(f"Found {len(indexed_files)} indexed files in Qdrant")
logger.debug(f"Found {len(indexed_file_ids)} indexed files in Qdrant")
# Scan for tagged PDF files
file_count = 0
@@ -370,7 +404,14 @@ async def scan_user_documents(
pass
if initial_sync:
# Send everything on first sync
# Send everything on first sync - write placeholder first
await write_placeholder_point(
doc_id=file_id,
doc_type="file",
user_id=user_id,
modified_at=modified_at,
file_path=file_path,
)
await send_stream.send(
DocumentTask(
user_id=user_id,
@@ -383,9 +424,7 @@ async def scan_user_documents(
)
file_queued += 1
else:
# Incremental sync: compare with indexed state
indexed_at = indexed_files.get(file_id)
# Incremental sync: check if file exists and compare modified_at
# If file reappeared, remove from potentially_deleted
file_key = (user_id, file_id)
if file_key in _potentially_deleted:
@@ -394,8 +433,36 @@ async def scan_user_documents(
)
del _potentially_deleted[file_key]
# Query Qdrant for existing entry (placeholder or real)
existing_metadata = await query_document_metadata(
doc_id=file_id, doc_type="file", user_id=user_id
)
# Send if never indexed or modified since last index
if indexed_at is None or modified_at > indexed_at:
# Compare against stored modified_at (not indexed_at!)
needs_indexing = False
if existing_metadata is None:
# Never seen before
needs_indexing = True
elif existing_metadata.get("modified_at", 0) < modified_at:
# File modified since last indexing
needs_indexing = True
elif existing_metadata.get("is_placeholder", False):
# Placeholder exists but processing may have failed - requeue
logger.debug(
f"Found existing placeholder for file {file_path} (ID: {file_id}), requeuing"
)
needs_indexing = True
if needs_indexing:
# Write placeholder before queuing
await write_placeholder_point(
doc_id=file_id,
doc_type="file",
user_id=user_id,
modified_at=modified_at,
file_path=file_path,
)
await send_stream.send(
DocumentTask(
user_id=user_id,
@@ -415,7 +482,7 @@ async def scan_user_documents(
# Check for deleted files (not initial sync)
if not initial_sync:
for file_id in indexed_files:
for file_id in indexed_file_ids:
if file_id not in nextcloud_file_ids:
file_key = (user_id, file_id)