Merge pull request #819 from cbcoutinho/fix/vector-sync-isolate-notes-scan
fix(vector-sync): isolate per-app scans so a disabled Notes app can't abort sync
This commit is contained in:
Chris Coutinho
@@ -14,6 +14,7 @@ from typing import cast
|
|||||||
import anyio
|
import anyio
|
||||||
from anyio.abc import TaskStatus
|
from anyio.abc import TaskStatus
|
||||||
from anyio.streams.memory import MemoryObjectSendStream
|
from anyio.streams.memory import MemoryObjectSendStream
|
||||||
|
from httpx import HTTPStatusError
|
||||||
from qdrant_client import AsyncQdrantClient
|
from qdrant_client import AsyncQdrantClient
|
||||||
from qdrant_client.models import FieldCondition, Filter, MatchValue, Record
|
from qdrant_client.models import FieldCondition, Filter, MatchValue, Record
|
||||||
|
|
||||||
@@ -318,163 +319,42 @@ async def scan_user_documents(
|
|||||||
|
|
||||||
logger.debug("Found %s indexed documents in Qdrant", len(indexed_doc_ids))
|
logger.debug("Found %s indexed documents in Qdrant", len(indexed_doc_ids))
|
||||||
|
|
||||||
# Stream notes from Nextcloud and process immediately
|
# Notes (isolated so an uninstalled or disabled Notes app — whose API
|
||||||
note_count = 0
|
# returns 404 — cannot abort scanning of the other apps; this mirrors the
|
||||||
|
# per-app try/except guards already wrapping files/news/deck below).
|
||||||
|
settings = get_settings()
|
||||||
|
grace_period = settings.vector_sync_scan_interval * 1.5
|
||||||
|
current_time = time.time()
|
||||||
queued = 0
|
queued = 0
|
||||||
nextcloud_doc_ids = set()
|
|
||||||
|
|
||||||
async for note in nc_client.notes.get_all_notes(prune_before=prune_before):
|
try:
|
||||||
note_count += 1
|
queued += await scan_notes(
|
||||||
doc_id = str(note["id"])
|
user_id=user_id,
|
||||||
nextcloud_doc_ids.add(doc_id)
|
send_stream=send_stream,
|
||||||
modified_at = note.get("modified", 0)
|
nc_client=nc_client,
|
||||||
|
initial_sync=initial_sync,
|
||||||
if initial_sync:
|
scan_id=scan_id,
|
||||||
# Send everything on first sync - write placeholder first
|
prune_before=prune_before,
|
||||||
await write_placeholder_point(
|
indexed_doc_ids=indexed_doc_ids,
|
||||||
doc_id=doc_id,
|
grace_period=grace_period,
|
||||||
doc_type="note",
|
current_time=current_time,
|
||||||
user_id=user_id,
|
)
|
||||||
modified_at=modified_at,
|
except HTTPStatusError as e:
|
||||||
etag=note.get("etag", ""),
|
if e.response.status_code == 404:
|
||||||
|
logger.info(
|
||||||
|
"[SCAN-%s] Notes app unavailable for %s (HTTP 404); skipping notes",
|
||||||
|
scan_id,
|
||||||
|
user_id,
|
||||||
)
|
)
|
||||||
await send_stream.send(
|
|
||||||
DocumentTask(
|
|
||||||
user_id=user_id,
|
|
||||||
doc_id=doc_id,
|
|
||||||
doc_type="note",
|
|
||||||
operation="index",
|
|
||||||
modified_at=modified_at,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
queued += 1
|
|
||||||
else:
|
else:
|
||||||
# Incremental sync: check if document exists and compare modified_at
|
logger.warning("Failed to scan notes for %s: %s", user_id, e)
|
||||||
# If document reappeared, remove from potentially_deleted
|
except Exception as e:
|
||||||
doc_key = (user_id, doc_id)
|
logger.warning("Failed to scan notes for %s: %s", user_id, e)
|
||||||
if doc_key in _potentially_deleted:
|
|
||||||
logger.debug(
|
|
||||||
"Document %s reappeared, removing from deletion grace period",
|
|
||||||
doc_id,
|
|
||||||
)
|
|
||||||
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
|
|
||||||
# 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 - check if it's stale (processing may have failed)
|
|
||||||
# Only requeue if placeholder is older than 5x scan interval
|
|
||||||
# (Large PDFs can take 3-4 minutes to process)
|
|
||||||
queued_at = existing_metadata.get("queued_at", 0)
|
|
||||||
placeholder_age = time.time() - queued_at
|
|
||||||
stale_threshold = get_settings().vector_sync_scan_interval * 5
|
|
||||||
if placeholder_age > stale_threshold:
|
|
||||||
logger.debug(
|
|
||||||
"Found stale placeholder for note %s (age=%ss), requeuing",
|
|
||||||
doc_id,
|
|
||||||
format(placeholder_age, ".1f"),
|
|
||||||
)
|
|
||||||
needs_indexing = True
|
|
||||||
else:
|
|
||||||
logger.debug(
|
|
||||||
"Skipping note %s with recent placeholder (age=%ss < %ss)",
|
|
||||||
doc_id,
|
|
||||||
format(placeholder_age, ".1f"),
|
|
||||||
format(stale_threshold, ".1f"),
|
|
||||||
)
|
|
||||||
|
|
||||||
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,
|
|
||||||
doc_id=doc_id,
|
|
||||||
doc_type="note",
|
|
||||||
operation="index",
|
|
||||||
modified_at=modified_at,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
queued += 1
|
|
||||||
|
|
||||||
# Log and record metrics after streaming
|
|
||||||
logger.info("[SCAN-%s] Found %s notes for %s", scan_id, note_count, user_id)
|
|
||||||
record_vector_sync_scan(note_count)
|
|
||||||
|
|
||||||
if initial_sync:
|
if initial_sync:
|
||||||
logger.info("Sent %s documents for initial sync: %s", queued, user_id)
|
logger.info("Sent %s documents for initial sync: %s", queued, user_id)
|
||||||
return
|
return
|
||||||
|
|
||||||
# Check for deleted documents (in Qdrant but not in Nextcloud)
|
|
||||||
# Use grace period: only delete after 2 consecutive scans confirm absence
|
|
||||||
settings = get_settings()
|
|
||||||
grace_period = (
|
|
||||||
settings.vector_sync_scan_interval * 1.5
|
|
||||||
) # Allow 1.5 scan intervals
|
|
||||||
current_time = time.time()
|
|
||||||
|
|
||||||
for doc_id in indexed_doc_ids:
|
|
||||||
if doc_id not in nextcloud_doc_ids:
|
|
||||||
doc_key = (user_id, doc_id)
|
|
||||||
|
|
||||||
if doc_key in _potentially_deleted:
|
|
||||||
# Already marked as potentially deleted, check if grace period elapsed
|
|
||||||
first_missing_time = _potentially_deleted[doc_key]
|
|
||||||
time_missing = current_time - first_missing_time
|
|
||||||
|
|
||||||
if time_missing >= grace_period:
|
|
||||||
# Grace period elapsed, send for deletion
|
|
||||||
logger.info(
|
|
||||||
"Document %s missing for %ss (>%ss grace period), sending deletion",
|
|
||||||
doc_id,
|
|
||||||
format(time_missing, ".1f"),
|
|
||||||
format(grace_period, ".1f"),
|
|
||||||
)
|
|
||||||
await send_stream.send(
|
|
||||||
DocumentTask(
|
|
||||||
user_id=user_id,
|
|
||||||
doc_id=doc_id,
|
|
||||||
doc_type="note",
|
|
||||||
operation="delete",
|
|
||||||
modified_at=0,
|
|
||||||
)
|
|
||||||
)
|
|
||||||
queued += 1
|
|
||||||
# Remove from tracking after sending deletion
|
|
||||||
del _potentially_deleted[doc_key]
|
|
||||||
else:
|
|
||||||
logger.debug(
|
|
||||||
"Document %s still missing (%ss/%ss grace period)",
|
|
||||||
doc_id,
|
|
||||||
format(time_missing, ".1f"),
|
|
||||||
format(grace_period, ".1f"),
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
# First time missing, add to grace period tracking
|
|
||||||
logger.debug(
|
|
||||||
"Document %s missing for first time, starting grace period",
|
|
||||||
doc_id,
|
|
||||||
)
|
|
||||||
_potentially_deleted[doc_key] = current_time
|
|
||||||
|
|
||||||
# Scan tagged PDF files (after notes)
|
# Scan tagged PDF files (after notes)
|
||||||
# Get indexed file IDs from Qdrant (for deletion tracking)
|
# Get indexed file IDs from Qdrant (for deletion tracking)
|
||||||
indexed_file_ids = set()
|
indexed_file_ids = set()
|
||||||
@@ -744,6 +624,181 @@ async def scan_user_documents(
|
|||||||
logger.debug("No changes detected for %s", user_id)
|
logger.debug("No changes detected for %s", user_id)
|
||||||
|
|
||||||
|
|
||||||
|
async def scan_notes(
|
||||||
|
user_id: str,
|
||||||
|
send_stream: MemoryObjectSendStream[DocumentTask],
|
||||||
|
nc_client: NextcloudClient,
|
||||||
|
initial_sync: bool,
|
||||||
|
scan_id: int,
|
||||||
|
prune_before: int | None,
|
||||||
|
indexed_doc_ids: set[str],
|
||||||
|
grace_period: float,
|
||||||
|
current_time: float,
|
||||||
|
) -> int:
|
||||||
|
"""Scan a user's Notes and queue changed notes for indexing.
|
||||||
|
|
||||||
|
Extracted into its own function (like scan_news_items / scan_deck_cards) so a
|
||||||
|
failure here -- e.g. the Notes API returning 404 because the app is not
|
||||||
|
installed -- propagates to the caller's per-app guard instead of aborting the
|
||||||
|
whole user scan. The deletion-tracking pass runs only after the Notes fetch
|
||||||
|
succeeds, so a failed fetch never mass-deletes a user's indexed notes.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Number of notes queued for processing (index + delete operations).
|
||||||
|
"""
|
||||||
|
# Stream notes from Nextcloud and process immediately
|
||||||
|
note_count = 0
|
||||||
|
queued = 0
|
||||||
|
nextcloud_doc_ids: set[str] = set()
|
||||||
|
|
||||||
|
async for note in nc_client.notes.get_all_notes(prune_before=prune_before):
|
||||||
|
note_count += 1
|
||||||
|
doc_id = str(note["id"])
|
||||||
|
nextcloud_doc_ids.add(doc_id)
|
||||||
|
modified_at = note.get("modified", 0)
|
||||||
|
|
||||||
|
if initial_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,
|
||||||
|
doc_id=doc_id,
|
||||||
|
doc_type="note",
|
||||||
|
operation="index",
|
||||||
|
modified_at=modified_at,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
queued += 1
|
||||||
|
else:
|
||||||
|
# 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:
|
||||||
|
logger.debug(
|
||||||
|
"Document %s reappeared, removing from deletion grace period",
|
||||||
|
doc_id,
|
||||||
|
)
|
||||||
|
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
|
||||||
|
# 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 - check if it's stale (processing may have failed)
|
||||||
|
# Only requeue if placeholder is older than 5x scan interval
|
||||||
|
# (Large PDFs can take 3-4 minutes to process)
|
||||||
|
queued_at = existing_metadata.get("queued_at", 0)
|
||||||
|
placeholder_age = time.time() - queued_at
|
||||||
|
stale_threshold = get_settings().vector_sync_scan_interval * 5
|
||||||
|
if placeholder_age > stale_threshold:
|
||||||
|
logger.debug(
|
||||||
|
"Found stale placeholder for note %s (age=%ss), requeuing",
|
||||||
|
doc_id,
|
||||||
|
format(placeholder_age, ".1f"),
|
||||||
|
)
|
||||||
|
needs_indexing = True
|
||||||
|
else:
|
||||||
|
logger.debug(
|
||||||
|
"Skipping note %s with recent placeholder (age=%ss < %ss)",
|
||||||
|
doc_id,
|
||||||
|
format(placeholder_age, ".1f"),
|
||||||
|
format(stale_threshold, ".1f"),
|
||||||
|
)
|
||||||
|
|
||||||
|
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,
|
||||||
|
doc_id=doc_id,
|
||||||
|
doc_type="note",
|
||||||
|
operation="index",
|
||||||
|
modified_at=modified_at,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
queued += 1
|
||||||
|
|
||||||
|
# Log and record metrics after streaming
|
||||||
|
logger.info("[SCAN-%s] Found %s notes for %s", scan_id, note_count, user_id)
|
||||||
|
record_vector_sync_scan(note_count)
|
||||||
|
|
||||||
|
if initial_sync:
|
||||||
|
return queued
|
||||||
|
|
||||||
|
# Check for deleted documents (in Qdrant but not in Nextcloud)
|
||||||
|
# Use grace period: only delete after 2 consecutive scans confirm absence
|
||||||
|
for doc_id in indexed_doc_ids:
|
||||||
|
if doc_id not in nextcloud_doc_ids:
|
||||||
|
doc_key = (user_id, doc_id)
|
||||||
|
|
||||||
|
if doc_key in _potentially_deleted:
|
||||||
|
# Already marked as potentially deleted, check if grace period elapsed
|
||||||
|
first_missing_time = _potentially_deleted[doc_key]
|
||||||
|
time_missing = current_time - first_missing_time
|
||||||
|
|
||||||
|
if time_missing >= grace_period:
|
||||||
|
# Grace period elapsed, send for deletion
|
||||||
|
logger.info(
|
||||||
|
"Document %s missing for %ss (>%ss grace period), sending deletion",
|
||||||
|
doc_id,
|
||||||
|
format(time_missing, ".1f"),
|
||||||
|
format(grace_period, ".1f"),
|
||||||
|
)
|
||||||
|
await send_stream.send(
|
||||||
|
DocumentTask(
|
||||||
|
user_id=user_id,
|
||||||
|
doc_id=doc_id,
|
||||||
|
doc_type="note",
|
||||||
|
operation="delete",
|
||||||
|
modified_at=0,
|
||||||
|
)
|
||||||
|
)
|
||||||
|
queued += 1
|
||||||
|
# Remove from tracking after sending deletion
|
||||||
|
del _potentially_deleted[doc_key]
|
||||||
|
else:
|
||||||
|
logger.debug(
|
||||||
|
"Document %s still missing (%ss/%ss grace period)",
|
||||||
|
doc_id,
|
||||||
|
format(time_missing, ".1f"),
|
||||||
|
format(grace_period, ".1f"),
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
# First time missing, add to grace period tracking
|
||||||
|
logger.debug(
|
||||||
|
"Document %s missing for first time, starting grace period",
|
||||||
|
doc_id,
|
||||||
|
)
|
||||||
|
_potentially_deleted[doc_key] = current_time
|
||||||
|
|
||||||
|
return queued
|
||||||
|
|
||||||
|
|
||||||
async def scan_news_items(
|
async def scan_news_items(
|
||||||
user_id: str,
|
user_id: str,
|
||||||
send_stream: MemoryObjectSendStream[DocumentTask],
|
send_stream: MemoryObjectSendStream[DocumentTask],
|
||||||
|
|||||||
Reference in New Issue
Block a user