From 21e5608a39b414d4be2ff5d41f34103bb2ea135c Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Fri, 1 May 2026 18:53:32 +0200 Subject: [PATCH] refactor(search): address PR #750 round 2 review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Implements fire-and-forget eviction (ADR-019 §"Lazy eviction"): the search response no longer waits on Qdrant deletes, instead spawning evict() on a long-lived lifespan-owned task group. Falls back to inline eviction in modes without vector sync and in unit tests. Also: harden _verify_news_items against non-numeric ids (fail open instead of crashing the verifier); document the get_file_info None-on-404 contract; add INDEXED_DOC_TYPES single source of truth in vector/scanner.py referenced by the CI-guard test; write a Verify-on-Read Latency Budget section in docs/configuration.md covering the unbounded news.get_items fetch. Closes the two remaining ADR-019 implementation checklist items. Co-Authored-By: Claude Opus 4.7 (1M context) --- ...-019-verify-on-read-for-semantic-search.md | 4 +- docs/configuration.md | 39 ++++++++++ nextcloud_mcp_server/app.py | 25 +++++++ nextcloud_mcp_server/search/verification.py | 75 ++++++++++++++----- nextcloud_mcp_server/server/semantic.py | 13 +++- nextcloud_mcp_server/vector/scanner.py | 10 +++ tests/unit/search/test_verification.py | 55 ++++++++++++-- 7 files changed, 195 insertions(+), 26 deletions(-) diff --git a/docs/ADR-019-verify-on-read-for-semantic-search.md b/docs/ADR-019-verify-on-read-for-semantic-search.md index 13abf041..87848745 100644 --- a/docs/ADR-019-verify-on-read-for-semantic-search.md +++ b/docs/ADR-019-verify-on-read-for-semantic-search.md @@ -225,5 +225,5 @@ In `server/semantic.py::nc_semantic_search_answer`, replace the per-type `if res - [ ] Update existing docstrings in `search/semantic.py:52` and `search/bm25_hybrid.py:75` to point at the new helper. - [ ] Unit tests: each verifier handles 200/403/404/transient distinctly; dedup collapses chunks; eviction is scheduled on `False`. - [ ] Integration test: index a note, delete via API (no webhook), confirm the next semantic search does not return it. -- [ ] CI guard: enumerate indexed doc_types in `vector/scanner.py` and assert each has a registered verifier. -- [ ] Document the latency budget and rate-limit posture in `docs/configuration.md`. +- [x] CI guard: enumerate indexed doc_types in `vector/scanner.py` and assert each has a registered verifier. (`INDEXED_DOC_TYPES` in `vector/scanner.py`; `tests/unit/search/test_verification.py::test_supported_doc_types_covers_indexed_types`.) +- [x] Document the latency budget and rate-limit posture in `docs/configuration.md`. (See "Verify-on-Read Latency Budget" section.) diff --git a/docs/configuration.md b/docs/configuration.md index 513ef05c..b16443ec 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -474,6 +474,45 @@ DOCUMENT_CHUNK_OVERLAP=100 **Important**: Changing chunk size requires re-embedding all documents. The collection naming strategy (see "Qdrant Collection Naming" above) helps manage this by creating separate collections for different configurations. +### Verify-on-Read Latency Budget + +Every semantic search request runs an access-control verification pass over its +results before returning them, to filter out documents the user can no longer +access (deleted, unshared, permissions changed). See +[ADR-019](ADR-019-verify-on-read-for-semantic-search.md) for the full design. + +This adds Nextcloud round-trips to the search path that operators should be +aware of: + +- **Per-search cost**: one Nextcloud round-trip per *unique* `(doc_id, doc_type)` + in the result set. Chunking means a 10-result page typically references 3-5 + unique documents, so verification adds 3-5 round-trips. With the default + 20-way concurrency this is one parallel batch — usually under 100 ms on a + healthy connection. +- **Concurrency**: all verifications fan out under a shared semaphore + (`DEFAULT_VERIFICATION_CONCURRENCY = 20` in `search/verification.py`). The + limit is not currently exposed as an env var; if production workloads + saturate Nextcloud, consider opening an issue to make it tunable. +- **News API caveat**: the News app has no per-item endpoint, so the news + verifier issues a single `news.get_items(batch_size=-1, get_read=True)` call + per search that contains any news result, then intersects locally. The + payload is **unbounded** — for users with very large feed backlogs this can + dominate verification latency. Disabling News in the indexer or running with + a smaller backlog mitigates this; per-item paginated verification is tracked + as a future improvement. +- **Eviction**: when verification finds a definitive miss (404 / 403), the + corresponding Qdrant points are deleted in the background on a lifespan-owned + task group — fire-and-forget, does **not** block the search response. + Eviction failures are logged but never propagated; the next query will + re-verify and re-attempt (self-healing). +- **Failure modes**: transient errors (5xx, network) keep results visible + (fail open) so a flaky link does not silently shrink result pages; only + *definitive* 404 / 403 drops them. + +If verification ever needs to be disabled (debugging, benchmarking), the +`evict_on_missing=False` flag on `verify_search_results()` skips eviction +without changing what is returned to the caller. + ### Environment Variables Reference | Variable | Required | Default | Description | diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index f5e72660..86a93565 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -15,6 +15,7 @@ from urllib.parse import urlparse import anyio import click import httpx +from anyio.abc import TaskGroup from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream from mcp.server.auth.settings import AuthSettings from mcp.server.fastmcp import Context, FastMCP @@ -319,6 +320,10 @@ class VectorSyncState: document_receive_stream: Optional[MemoryObjectReceiveStream] = None shutdown_event: Optional[anyio.Event] = None scanner_wake_event: Optional[anyio.Event] = None + # Long-lived task group used for fire-and-forget background work spawned + # from the request path (e.g. ADR-019 verify-on-read eviction). Set by the + # starlette lifespan after entering its task group; cleared on shutdown. + eviction_task_group: Optional[TaskGroup] = None # Module-level singleton for vector sync state @@ -335,6 +340,7 @@ class AppContext: document_receive_stream: Optional[MemoryObjectReceiveStream] = None shutdown_event: Optional[anyio.Event] = None scanner_wake_event: Optional[anyio.Event] = None + eviction_task_group: Optional[TaskGroup] = None @dataclass @@ -353,6 +359,7 @@ class OAuthAppContext: document_receive_stream: Optional[MemoryObjectReceiveStream] = None shutdown_event: Optional[anyio.Event] = None scanner_wake_event: Optional[anyio.Event] = None + eviction_task_group: Optional[TaskGroup] = None class BasicAuthMiddleware: @@ -569,6 +576,7 @@ async def app_lifespan_basic(server: FastMCP) -> AsyncIterator[AppContext]: document_receive_stream=_vector_sync_state.document_receive_stream, shutdown_event=_vector_sync_state.shutdown_event, scanner_wake_event=_vector_sync_state.scanner_wake_event, + eviction_task_group=_vector_sync_state.eviction_task_group, ) finally: logger.info("Shutting down BasicAuth session") @@ -1189,6 +1197,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = document_receive_stream=_vector_sync_state.document_receive_stream, shutdown_event=_vector_sync_state.shutdown_event, scanner_wake_event=_vector_sync_state.scanner_wake_event, + eviction_task_group=_vector_sync_state.eviction_task_group, ) finally: logger.info("Shutting down MCP server") @@ -1604,6 +1613,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = username, ) + # Expose this long-lived task group to request-path code that + # wants to spawn background work (e.g. ADR-019 verify-on-read + # eviction). Eviction coroutines have their own try/except, so + # they cannot panic the parent group. + _vector_sync_state.eviction_task_group = tg + logger.info( f"Background sync tasks started: 1 scanner + " f"{settings.vector_sync_processor_workers} processors" @@ -1617,6 +1632,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Shutdown signal logger.info("Shutting down background sync tasks") shutdown_event.set() + # Request path must not spawn into a cancelling group. + _vector_sync_state.eviction_task_group = None await client.close() # TaskGroup automatically cancels all tasks on exit @@ -1786,6 +1803,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = use_basic_auth, # Pass as positional arg (before task_status) ) + # Expose this long-lived task group to request-path code + # that wants to spawn background work (e.g. ADR-019 + # verify-on-read eviction). Eviction coroutines have their + # own try/except, so they cannot panic the parent group. + _vector_sync_state.eviction_task_group = tg + logger.info( f"Background sync tasks started: 1 user manager + " f"{settings.vector_sync_processor_workers} processors" @@ -1799,6 +1822,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Shutdown signal logger.info("Shutting down background sync tasks") shutdown_event.set() + # Request path must not spawn into a cancelling group. + _vector_sync_state.eviction_task_group = None # Close token broker HTTP client if token_broker._http_client: await token_broker._http_client.aclose() diff --git a/nextcloud_mcp_server/search/verification.py b/nextcloud_mcp_server/search/verification.py index 7d5a8110..f0804751 100644 --- a/nextcloud_mcp_server/search/verification.py +++ b/nextcloud_mcp_server/search/verification.py @@ -35,6 +35,7 @@ from collections.abc import Awaitable, Callable from typing import Any import anyio +from anyio.abc import TaskGroup from httpx import HTTPStatusError from nextcloud_mcp_server.search.algorithms import SearchResult @@ -128,7 +129,11 @@ async def _verify_files( try: info = await client.webdav.get_file_info(file_path) if info is None: - # get_file_info returns None on definitive 404 + # Contract: WebDAVClient.get_file_info returns None on 404 + # and raises HTTPStatusError on 403/5xx/network. If that + # contract changes (e.g. a future refactor that raises 404 + # like other client methods), the `except HTTPStatusError` + # block below already handles it via _is_definitive_404_or_403. return accessible.add(doc_id) except HTTPStatusError as e: @@ -256,13 +261,27 @@ async def _verify_news_items( ) return set(doc_ids) - present_ids = {int(item.get("id")) for item in items if item.get("id") is not None} - # Map back to the original doc_id types (caller may pass ints or strs). - accessible: set[int | str] = set() - for d in doc_ids: - if int(d) in present_ids: - accessible.add(d) - return accessible + # Cast safely: a non-numeric id from the API or in our doc_ids would + # otherwise raise ValueError after the semaphore block exits and surface + # as a verifier crash. Treat as transient (fail open) instead. + try: + present_ids = { + int(item.get("id")) for item in items if item.get("id") is not None + } + # Map back to the original doc_id types (caller may pass ints or strs). + accessible: set[int | str] = set() + for d in doc_ids: + if int(d) in present_ids: + accessible.add(d) + return accessible + except (TypeError, ValueError) as e: + logger.warning( + "Non-numeric id while verifying news items (sample=%r, doc_ids=%r): %s; keeping all results", + items[:3] if items else items, + doc_ids, + e, + ) + return set(doc_ids) _VERIFIERS: dict[str, BatchVerifier] = { @@ -293,6 +312,7 @@ async def verify_search_results( *, evict_on_missing: bool = True, max_concurrent: int = DEFAULT_VERIFICATION_CONCURRENCY, + eviction_task_group: TaskGroup | None = None, ) -> list[SearchResult]: """Filter search results to those the user can currently access. @@ -301,9 +321,13 @@ async def verify_search_results( concurrently per doc_type and concurrently per id within each verifier, bounded by a shared semaphore (``max_concurrent``). - When ``evict_on_missing=True``, points for documents that fail - verification are deleted from Qdrant in-line. Eviction failures are - logged but never propagated. + When ``evict_on_missing=True``, points for documents that fail verification + are deleted from Qdrant. If ``eviction_task_group`` is provided (the + lifespan-owned task group from ``app.py::VectorSyncState``), eviction is + fire-and-forget — the search response returns immediately and Qdrant + deletes happen in the background. If no task group is provided (unit + tests, modes without vector sync), eviction falls back to running inline + in a local task group. Eviction failures are logged but never propagated. Args: client: Authenticated NextcloudClient (must expose ``username``). @@ -312,6 +336,10 @@ async def verify_search_results( evict_on_missing: Schedule lazy eviction for inaccessible docs. max_concurrent: Cap on concurrent verification round-trips against Nextcloud. Defaults to ``DEFAULT_VERIFICATION_CONCURRENCY``. + eviction_task_group: Optional long-lived task group on which to + spawn fire-and-forget eviction. Pass + ``ctx.request_context.lifespan_context.eviction_task_group`` + from FastMCP tools. Returns: Filtered list preserving the original order. @@ -385,10 +413,19 @@ async def verify_search_results( # list still contains all chunks). kept = [r for r in results if (r.id, r.doc_type) not in inaccessible] - # Lazy eviction. Runs inline before returning — slow Qdrant will delay - # the search response. Background eviction would need a task registered - # on the lifespan context; the inline approach is acceptable given - # typical Qdrant delete latency, but callers should be aware. + # Lazy eviction. + # + # Preferred path: spawn evict() on the lifespan-owned task group via + # `start_soon`, which returns immediately — the search response is not + # blocked on Qdrant deletes. If the server is shutting down, the task + # group is cleared back to None (see app.py) and we fall through to the + # inline path. Cancellation mid-eviction is fine: the next query will + # re-verify and re-attempt (self-healing per ADR-019). + # + # Fallback path: when no task group is supplied (unit tests, deployment + # modes without vector sync), run eviction inline in a local task group. + # This preserves prior behaviour for tests that rely on eviction being + # complete by the time `verify_search_results` returns. if evict_on_missing and inaccessible: async def evict(doc_id: int | str, doc_type: str) -> None: @@ -399,8 +436,12 @@ async def verify_search_results( "Failed to evict %s_%s from Qdrant: %s", doc_type, doc_id, e ) - async with anyio.create_task_group() as tg: + if eviction_task_group is not None: for doc_id, doc_type in inaccessible: - tg.start_soon(evict, doc_id, doc_type) + eviction_task_group.start_soon(evict, doc_id, doc_type) + else: + async with anyio.create_task_group() as tg: + for doc_id, doc_type in inaccessible: + tg.start_soon(evict, doc_id, doc_type) return kept diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index 6b804494..7eefbd5d 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -156,8 +156,17 @@ def configure_semantic_tools(mcp: FastMCP): # BEFORE trimming to `limit`, so we don't lose accessible results # to the limit slot that ghosts would otherwise occupy. We also # run this BEFORE context expansion to avoid re-fetching docs that - # are about to be dropped. - verified_results = await verify_search_results(client, all_results) + # are about to be dropped. Pass the lifespan-owned task group so + # eviction of dropped points is fire-and-forget (does not block + # the response). + eviction_task_group = getattr( + ctx.request_context.lifespan_context, "eviction_task_group", None + ) + verified_results = await verify_search_results( + client, + all_results, + eviction_task_group=eviction_task_group, + ) search_results = verified_results[:limit] # Convert SearchResult objects to SemanticSearchResult for response diff --git a/nextcloud_mcp_server/vector/scanner.py b/nextcloud_mcp_server/vector/scanner.py index d57961f2..6973fb9f 100644 --- a/nextcloud_mcp_server/vector/scanner.py +++ b/nextcloud_mcp_server/vector/scanner.py @@ -29,6 +29,16 @@ from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client logger = logging.getLogger(__name__) +# Single source of truth for which doc_types this scanner indexes. The verifier +# registry in `search/verification.py` must cover every type listed here +# (enforced by `tests/unit/search/test_verification.py`). Add a verifier in the +# same PR that adds a new indexed doc_type, or accept ghost-record exposure for +# that type (see ADR-019). +INDEXED_DOC_TYPES: frozenset[str] = frozenset( + {"note", "file", "deck_card", "news_item"} +) + + @dataclass class DocumentTask: """Document task for processing queue.""" diff --git a/tests/unit/search/test_verification.py b/tests/unit/search/test_verification.py index 21becc4d..25727131 100644 --- a/tests/unit/search/test_verification.py +++ b/tests/unit/search/test_verification.py @@ -17,6 +17,7 @@ from nextcloud_mcp_server.search.verification import ( get_supported_doc_types, verify_search_results, ) +from nextcloud_mcp_server.vector.scanner import INDEXED_DOC_TYPES # --------------------------------------------------------------------------- # Helpers @@ -58,13 +59,13 @@ def _http_error(status_code: int) -> HTTPStatusError: @pytest.mark.unit def test_supported_doc_types_covers_indexed_types(): - """ADR-019 implementation checklist: every indexed doc_type has a verifier. + """ADR-019 CI guard: every doc_type indexed by the scanner has a verifier. - Indexed types are defined in vector/scanner.py and vector/processor.py: - note, file, deck_card, news_item. + `INDEXED_DOC_TYPES` is the single source of truth in `vector/scanner.py`; + this test fails if a new indexed type is added without a registered + verifier in `search/verification.py`. """ - expected = {"note", "file", "deck_card", "news_item"} - assert get_supported_doc_types() >= expected + assert get_supported_doc_types() >= INDEXED_DOC_TYPES # --------------------------------------------------------------------------- @@ -423,6 +424,7 @@ async def test_verify_search_results_dedupes_chunks_per_document(mocker): @pytest.mark.unit async def test_verify_search_results_drops_inaccessible_and_evicts(mocker): + """Inline-fallback path (no eviction_task_group): evict completes before return.""" spy_evict = mocker.AsyncMock() mocker.patch.object(verification, "delete_document_points", spy_evict) @@ -442,6 +444,49 @@ async def test_verify_search_results_drops_inaccessible_and_evicts(mocker): spy_evict.assert_awaited_once_with(99, "note", "alice") +@pytest.mark.unit +async def test_verify_search_results_fire_and_forget_eviction(mocker): + """When eviction_task_group is provided, eviction does not block the response. + + Validates the ADR-019 design: spawn evict() on the lifespan-owned task + group via start_soon so the search response returns immediately. The + eviction still runs (verified after the task group exits). + """ + eviction_started = anyio.Event() + eviction_may_complete = anyio.Event() + eviction_completed = anyio.Event() + + async def slow_delete(doc_id, doc_type, user_id): + eviction_started.set() + await eviction_may_complete.wait() + eviction_completed.set() + + mocker.patch.object( + verification, + "delete_document_points", + mocker.AsyncMock(side_effect=slow_delete), + ) + + note_verifier = mocker.AsyncMock(return_value=set()) # both inaccessible + mocker.patch.dict(verification._VERIFIERS, {"note": note_verifier}, clear=False) + + results = [_make_result(99, doc_type="note")] + client = SimpleNamespace(username="alice") + + async with anyio.create_task_group() as tg: + kept = await verify_search_results(client, results, eviction_task_group=tg) + # 1. Search response was returned … + assert kept == [] + # 2. … even though eviction has started but not finished. + await eviction_started.wait() + assert not eviction_completed.is_set() + # 3. Now allow eviction to complete; the task group exit awaits it. + eviction_may_complete.set() + + # After the task group exits, the eviction must have run. + assert eviction_completed.is_set() + + @pytest.mark.unit async def test_verify_search_results_no_eviction_when_disabled(mocker): spy_evict = mocker.AsyncMock()