diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index f5d0d8c2..2a92cafe 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -129,6 +129,7 @@ from nextcloud_mcp_server.vector.oauth_sync import ( oauth_processor_task, user_manager_task, ) +from nextcloud_mcp_server.vector.placeholder import sweep_orphan_placeholders from nextcloud_mcp_server.vector.processor import processor_task from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client from nextcloud_mcp_server.vector.scanner import scanner_task @@ -1437,6 +1438,34 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = await stack.enter_async_context(_maybe_login_flow_cleanup(app)) yield + async def _sweep_orphan_placeholders_if_enabled() -> None: + """One-shot Pod-startup sweep of cross-restart placeholder orphans. + + See ``vector.placeholder.sweep_orphan_placeholders`` and Deck + card #101. Both lifespan branches (single-user BasicAuth and + OAuth / multi-user BasicAuth) call this after the qdrant + client is initialised and before the scanner / user-manager + tasks spawn. Failures are non-fatal — the existing staleness + gate will eventually re-queue orphans on the slow ~5h path. + """ + if not settings.vector_sync_orphan_sweep_enabled: + return + try: + qdrant_client = await get_qdrant_client() + swept, kept = await sweep_orphan_placeholders( + qdrant_client, settings.qdrant_collection + ) + logger.info( + "vector_sync.orphan_sweep", + extra={ + "swept": swept, + "kept": kept, + "collection": settings.qdrant_collection, + }, + ) + except Exception: + logger.exception("vector_sync.orphan_sweep_failed") + @asynccontextmanager async def starlette_lifespan(app: Starlette): # Set OAuth context for OAuth login routes (ADR-004) @@ -1612,6 +1641,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = f"Cannot start vector sync - Qdrant initialization failed: {e}" ) from e + # Orphan-sweep before scanner starts — card #101. + await _sweep_orphan_placeholders_if_enabled() + # Initialize shared state send_stream, receive_stream = anyio.create_memory_object_stream( max_buffer_size=settings.vector_sync_queue_max_size @@ -1774,6 +1806,11 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = f"Cannot start vector sync - Qdrant initialization failed: {e}" ) from e + # Orphan-sweep before scanners spawn — card #101. Runs once + # across the shared (per-tenant) collection regardless of + # how many per-user scanners the user-manager later starts. + await _sweep_orphan_placeholders_if_enabled() + # Clean up stale app passwords at startup (BasicAuth mode only) if not oauth_enabled: try: diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index ca841636..b37c19d4 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -87,6 +87,12 @@ _DEFAULTS: dict[str, Any] = { "vector_sync_processor_workers": 3, "vector_sync_queue_max_size": 10000, "vector_sync_user_poll_interval": 60, + # Orphan-sweep at Pod startup (card #101). When True, delete any + # placeholders carrying a different / absent ``instance_id`` before + # the scanner's first cycle, so a Pod restart mid-batch doesn't + # leave work stuck behind the 5x-scan-interval staleness gate. + # Escape hatch only — leave on by default. + "vector_sync_orphan_sweep_enabled": True, # Verify-on-read concurrency cap (ADR-019) "verification_concurrency": 20, # Qdrant @@ -585,6 +591,7 @@ class Settings: vector_sync_processor_workers: int = 3 vector_sync_queue_max_size: int = 10000 vector_sync_user_poll_interval: int = 60 # seconds - OAuth mode user discovery + vector_sync_orphan_sweep_enabled: bool = True # card #101 # Verify-on-read concurrency (ADR-019). Cap on parallel Nextcloud # round-trips during search-result verification fan-out. Lower this if the @@ -1076,6 +1083,7 @@ def get_settings() -> Settings: "vector_sync_processor_workers": "VECTOR_SYNC_PROCESSOR_WORKERS", "vector_sync_queue_max_size": "VECTOR_SYNC_QUEUE_MAX_SIZE", "vector_sync_user_poll_interval": "VECTOR_SYNC_USER_POLL_INTERVAL", + "vector_sync_orphan_sweep_enabled": "VECTOR_SYNC_ORPHAN_SWEEP_ENABLED", # Verify-on-read (ADR-019) "verification_concurrency": "VERIFICATION_CONCURRENCY", # Qdrant settings diff --git a/nextcloud_mcp_server/vector/placeholder.py b/nextcloud_mcp_server/vector/placeholder.py index 3b23bdd1..e31ae871 100644 --- a/nextcloud_mcp_server/vector/placeholder.py +++ b/nextcloud_mcp_server/vector/placeholder.py @@ -30,6 +30,13 @@ from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client logger = logging.getLogger(__name__) +# Stamped on every placeholder this Pod-process writes. A fresh UUID per +# process means a restarted Pod sees its predecessor's placeholders as +# "not mine" and deletes them in ``sweep_orphan_placeholders`` at startup, +# instead of waiting out the ``5 × VECTOR_SYNC_SCAN_INTERVAL`` staleness +# gate (~5h with the deployed 1h scan interval). Card #101. +_INSTANCE_ID = str(uuid.uuid4()) + def _generate_placeholder_id(doc_type: str, doc_id: str) -> str: """Generate deterministic UUID for placeholder point. @@ -98,6 +105,10 @@ async def write_placeholder_point( "modified_at": modified_at, "etag": etag, "queued_at": int(time.time()), + # Pod-process identity. ``sweep_orphan_placeholders`` uses + # the (placeholder.instance_id != _INSTANCE_ID) predicate to + # detect orphans from a crashed predecessor Pod. + "instance_id": _INSTANCE_ID, } # Add file_path for files @@ -319,3 +330,88 @@ def get_placeholder_filter() -> FieldCondition: key="is_placeholder", match=MatchValue(value=False), ) + + +# Batch size for the orphan-sweep scroll + delete loop. Big enough to +# keep round-trip count down, small enough that a single delete payload +# isn't unreasonable. Matches the order of magnitude of a per-tenant +# placeholder count (108 in the originating incident). +_ORPHAN_SWEEP_BATCH_SIZE = 100 + + +async def sweep_orphan_placeholders( + qdrant_client, + collection_name: str, + *, + batch_size: int = _ORPHAN_SWEEP_BATCH_SIZE, +) -> tuple[int, int]: + """Delete placeholder points written by a previous Pod-process. + + Scrolls all ``is_placeholder=true`` points in the collection, + paginated. For each batch, partitions points by whether their + ``instance_id`` payload field matches the current Pod's + ``_INSTANCE_ID``. Orphans (different ``instance_id`` OR field + absent — back-compat for placeholders written by pre-fix Pod + versions) are deleted by point ID; own-Pod placeholders are + left alone for the scanner's staleness gate to handle normally. + + Called once at Pod startup from ``app.starlette_lifespan`` — + NOT periodically. The own-Pod path relies on the existing + ``5 × VECTOR_SYNC_SCAN_INTERVAL`` gate; this helper only + addresses the cross-Pod-restart gap. Card #101. + + Args: + qdrant_client: Async Qdrant client. + collection_name: Target collection. + batch_size: Scroll page size. Default 100 — small enough that + a single delete payload is reasonable, large enough that + round-trip count stays bounded for typical placeholder + counts (~hundreds per tenant). + + Returns: + ``(swept, kept)`` — number of placeholders deleted as orphans, + and number left in place as belonging to the current Pod. + """ + placeholder_filter = Filter( + must=[ + FieldCondition(key="is_placeholder", match=MatchValue(value=True)), + ] + ) + swept = 0 + kept = 0 + offset = None + + while True: + points, offset = await qdrant_client.scroll( + collection_name=collection_name, + scroll_filter=placeholder_filter, + limit=batch_size, + offset=offset, + with_payload=True, + with_vectors=False, + ) + if not points: + break + + orphan_ids = [] + for point in points: + payload = point.payload or {} + point_instance = payload.get("instance_id") + if point_instance == _INSTANCE_ID: + kept += 1 + else: + orphan_ids.append(point.id) + + if orphan_ids: + await qdrant_client.delete( + collection_name=collection_name, + points_selector=orphan_ids, + ) + swept += len(orphan_ids) + + # ``offset is None`` signals the scroll cursor has been + # exhausted — Qdrant's contract for paginated scroll. + if offset is None: + break + + return swept, kept diff --git a/tests/unit/vector/test_placeholder.py b/tests/unit/vector/test_placeholder.py new file mode 100644 index 00000000..e7212b2e --- /dev/null +++ b/tests/unit/vector/test_placeholder.py @@ -0,0 +1,196 @@ +"""Unit tests for placeholder orphan-sweep (card #101). + +When the per-tenant Pod OOMKills mid-batch the in-memory processor +queue is lost. Placeholder points written to Qdrant survive, and the +next Pod's scanner would skip them under the +``5 × VECTOR_SYNC_SCAN_INTERVAL`` staleness gate (~5h with the deployed +1h scan interval). ``sweep_orphan_placeholders`` runs once at Pod +startup and deletes any placeholder whose ``instance_id`` payload +doesn't match the current Pod-process, restoring throughput within +one scan cycle. + +The sweep contract this file pins: + +* placeholder with ``instance_id != _INSTANCE_ID`` → deleted +* placeholder with **absent** ``instance_id`` → deleted + (back-compat for placeholders written by pre-fix Pod versions) +* placeholder with ``instance_id == _INSTANCE_ID`` → kept +* empty / no-results scroll → no-op (no spurious delete call) +* multi-page scroll → all pages visited, deletes batched per page +""" + +from __future__ import annotations + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from nextcloud_mcp_server.vector import placeholder as placeholder_module +from nextcloud_mcp_server.vector.placeholder import sweep_orphan_placeholders + + +def _scrolled_point(point_id, instance_id=...): + """Stand-in for a qdrant_client Record. Sweep reads only + ``id`` and ``payload``; passing ``instance_id=...`` (Ellipsis) + omits the field entirely (the back-compat case).""" + payload: dict = {"is_placeholder": True} + if instance_id is not ...: + payload["instance_id"] = instance_id + return SimpleNamespace(id=point_id, payload=payload) + + +def _make_client(scroll_pages): + """Build an AsyncMock qdrant client whose ``scroll`` returns the + given ``(points, next_offset)`` pages in order, then mimics the + real client's exhausted-cursor return of ``([], None)``.""" + client = AsyncMock() + pages = list(scroll_pages) + [([], None)] + client.scroll.side_effect = pages + return client + + +@pytest.mark.unit +async def test_sweeps_orphan_with_different_instance_id(monkeypatch): + monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-new") + client = _make_client( + [ + ([_scrolled_point("p1", instance_id="pod-old")], None), + ] + ) + + swept, kept = await sweep_orphan_placeholders(client, "nextcloud_content") + + assert (swept, kept) == (1, 0) + client.delete.assert_awaited_once() + # Selector is the orphan's point id — own-Pod placeholders never + # reach the delete call. + assert client.delete.await_args.kwargs["points_selector"] == ["p1"] + + +@pytest.mark.unit +async def test_sweeps_placeholder_with_absent_instance_id(monkeypatch): + """Back-compat: placeholders written by pre-fix Pod versions have + no ``instance_id`` field. They MUST be treated as orphans so the + first deploy of this fix doesn't leave old placeholders behind.""" + monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-new") + client = _make_client( + [ + ([_scrolled_point("legacy", instance_id=...)], None), + ] + ) + + swept, kept = await sweep_orphan_placeholders(client, "nextcloud_content") + + assert (swept, kept) == (1, 0) + assert client.delete.await_args.kwargs["points_selector"] == ["legacy"] + + +@pytest.mark.unit +async def test_keeps_own_pod_placeholders(monkeypatch): + """A surviving Pod's own placeholders must NOT be deleted — + that's what the staleness gate is for, and the processor may + still be working through them.""" + monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-current") + client = _make_client( + [ + ( + [ + _scrolled_point("mine-1", instance_id="pod-current"), + _scrolled_point("mine-2", instance_id="pod-current"), + ], + None, + ), + ] + ) + + swept, kept = await sweep_orphan_placeholders(client, "nextcloud_content") + + assert (swept, kept) == (0, 2) + client.delete.assert_not_awaited() + + +@pytest.mark.unit +async def test_noop_when_no_placeholders_exist(monkeypatch): + """Cold-boot tenant with an empty collection — sweep does NOT + issue a delete request, so a trivially-empty batch can't trip + Qdrant validation on an empty selector list.""" + monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-fresh") + client = _make_client([([], None)]) + + swept, kept = await sweep_orphan_placeholders(client, "nextcloud_content") + + assert (swept, kept) == (0, 0) + client.delete.assert_not_awaited() + + +@pytest.mark.unit +async def test_walks_multiple_scroll_pages(monkeypatch): + """Scroll cursor exhaustion is signalled by ``offset is None`` per + Qdrant's contract. Sweep must keep paging until the cursor is + exhausted, batching the delete per page.""" + monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-current") + client = _make_client( + [ + ( + [ + _scrolled_point("orphan-1", instance_id="pod-prev"), + _scrolled_point("mine", instance_id="pod-current"), + ], + "next-cursor", + ), + ( + [_scrolled_point("orphan-2", instance_id=...)], + None, + ), + ] + ) + + swept, kept = await sweep_orphan_placeholders( + client, "nextcloud_content", batch_size=2 + ) + + assert (swept, kept) == (2, 1) + # One delete per page (each page had at least one orphan). + assert client.delete.await_count == 2 + delete_selectors = [ + call.kwargs["points_selector"] for call in client.delete.await_args_list + ] + assert delete_selectors == [["orphan-1"], ["orphan-2"]] + + +@pytest.mark.unit +async def test_write_placeholder_payload_includes_instance_id(monkeypatch): + """Pin the new payload contract: ``write_placeholder_point`` MUST + stamp the current Pod's ``_INSTANCE_ID`` onto the payload it + upserts, so the next Pod's sweep can identify these placeholders + as belonging to a different process.""" + monkeypatch.setattr(placeholder_module, "_INSTANCE_ID", "pod-pinned") + + fake_qdrant = AsyncMock() + fake_settings = SimpleNamespace(get_collection_name=lambda: "nextcloud_content") + fake_embedding = SimpleNamespace(get_dimension=lambda: 4) + + async def fake_get_qdrant_client(): + return fake_qdrant + + monkeypatch.setattr(placeholder_module, "get_qdrant_client", fake_get_qdrant_client) + monkeypatch.setattr(placeholder_module, "get_settings", lambda: fake_settings) + monkeypatch.setattr( + placeholder_module, "get_embedding_service", lambda: fake_embedding + ) + + await placeholder_module.write_placeholder_point( + doc_id="d-42", + doc_type="note", + user_id="alice", + modified_at=1700000000, + etag="abc", + ) + + fake_qdrant.upsert.assert_awaited_once() + upserted_point = fake_qdrant.upsert.await_args.kwargs["points"][0] + assert upserted_point.payload["instance_id"] == "pod-pinned" + # Sanity: the other contract-pinning fields are still emitted. + assert upserted_point.payload["is_placeholder"] is True + assert upserted_point.payload["status"] == "pending"