diff --git a/docs/configuration.md b/docs/configuration.md index b16443ec..6438f819 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -489,10 +489,11 @@ aware of: 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. +- **Concurrency**: all verifications fan out under a shared semaphore. + Tunable via the `VERIFICATION_CONCURRENCY` env var (settings field + `verification_concurrency`, default 20) — lower it if your Nextcloud + backend struggles with the parallel fan-out, or raise it on a healthy + connection to speed up large result pages. - **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 diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 86a93565..5ef8a47f 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -323,7 +323,7 @@ class VectorSyncState: # 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 + eviction_task_group: TaskGroup | None = None # Module-level singleton for vector sync state @@ -340,7 +340,15 @@ 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 + + @property + def eviction_task_group(self) -> TaskGroup | None: + # Read dynamically from the module-level singleton instead of + # snapshotting at lifespan-yield time. Snapshotting is order-sensitive: + # if the FastMCP server lifespan ever runs before the Starlette + # lifespan assigns the task group, every session for the life of the + # process would see ``None`` and fall back to inline eviction. + return _vector_sync_state.eviction_task_group @dataclass @@ -359,7 +367,11 @@ 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 + + @property + def eviction_task_group(self) -> TaskGroup | None: + # See AppContext.eviction_task_group for rationale. + return _vector_sync_state.eviction_task_group class BasicAuthMiddleware: @@ -576,7 +588,8 @@ 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, + # eviction_task_group is exposed via @property (reads + # _vector_sync_state at access time, not snapshot). ) finally: logger.info("Shutting down BasicAuth session") @@ -1197,7 +1210,8 @@ 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, + # eviction_task_group is exposed via @property (reads + # _vector_sync_state at access time, not snapshot). ) finally: logger.info("Shutting down MCP server") diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 13a829fe..eedc254c 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -65,6 +65,8 @@ _DEFAULTS: dict[str, Any] = { "vector_sync_processor_workers": 3, "vector_sync_queue_max_size": 10000, "vector_sync_user_poll_interval": 60, + # Verify-on-read concurrency cap (ADR-019) + "verification_concurrency": 20, # Qdrant "qdrant_url": None, "qdrant_location": None, @@ -170,6 +172,7 @@ _dynaconf = Dynaconf( Validator("VECTOR_SYNC_PROCESSOR_WORKERS", gte=1), Validator("VECTOR_SYNC_QUEUE_MAX_SIZE", gte=1), Validator("VECTOR_SYNC_USER_POLL_INTERVAL", gte=1), + Validator("VERIFICATION_CONCURRENCY", gte=1), Validator("DOCUMENT_CHUNK_SIZE", gte=1), # Non-negative Validator("DOCUMENT_CHUNK_OVERLAP", gte=0), @@ -455,6 +458,12 @@ class Settings: vector_sync_queue_max_size: int = 10000 vector_sync_user_poll_interval: int = 60 # seconds - OAuth mode user discovery + # Verify-on-read concurrency (ADR-019). Cap on parallel Nextcloud + # round-trips during search-result verification fan-out. Lower this if the + # Nextcloud backend struggles with the parallel load; raise it on a + # healthy connection to speed up large result pages. + verification_concurrency: int = 20 + # Qdrant settings (mutually exclusive modes) qdrant_url: str | None = None # Network mode: http://qdrant:6333 qdrant_location: str | None = None # Local mode: :memory: or /path/to/data @@ -793,6 +802,8 @@ 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", + # Verify-on-read (ADR-019) + "verification_concurrency": "VERIFICATION_CONCURRENCY", # Qdrant settings "qdrant_url": "QDRANT_URL", "qdrant_location": "QDRANT_LOCATION", diff --git a/nextcloud_mcp_server/search/verification.py b/nextcloud_mcp_server/search/verification.py index fb352222..581e8dcf 100644 --- a/nextcloud_mcp_server/search/verification.py +++ b/nextcloud_mcp_server/search/verification.py @@ -38,17 +38,13 @@ import anyio from anyio.abc import TaskGroup from httpx import HTTPStatusError +from nextcloud_mcp_server.config import get_settings from nextcloud_mcp_server.search.algorithms import SearchResult from nextcloud_mcp_server.vector.eviction import delete_document_points logger = logging.getLogger(__name__) -# Default cap on concurrent verification round-trips against Nextcloud. Matches -# the convention in ``server/semantic.py`` for context-expansion fan-out. -DEFAULT_VERIFICATION_CONCURRENCY = 20 - - BatchVerifier = Callable[ [Any, list[SearchResult], anyio.Semaphore], Awaitable[set[int | str]] ] @@ -73,10 +69,24 @@ async def _verify_notes( accessible: set[int | str] = set() async def check(result: SearchResult) -> None: + doc_id = result.id + # Parse defensively before the network call so a malformed payload + # produces a specific log line, not a generic "unexpected error" from + # the catch-all ``except Exception`` below. Mirrors ``_verify_deck_cards``. + try: + note_id_int = int(doc_id) + except (TypeError, ValueError) as e: + logger.warning( + "Non-numeric note id %r: %s; keeping result", + doc_id, + e, + ) + accessible.add(doc_id) + return + async with semaphore: - doc_id = result.id try: - await client.notes.get_note(int(doc_id)) + await client.notes.get_note(note_id_int) accessible.add(doc_id) except HTTPStatusError as e: if _is_definitive_404_or_403(e): @@ -260,6 +270,14 @@ async def _verify_news_items( # a per-item News API endpoint. The shared semaphore protects # against runaway concurrent fetches, but the payload itself can # be large (News auto-purge cap is in the thousands of items). + # + # NOTE: ``batch_size`` is intentionally unbounded (-1). A numeric + # ceiling here would silently *break correctness*: any item beyond + # the cap would be missing from ``present_ids`` and incorrectly + # dropped from the result set. The fail-open contract requires + # fetching every item the user has access to. See the news caveat + # in docs/configuration.md (Verify-on-Read) for the latency + # tradeoff and follow-up paths. items = await client.news.get_items(batch_size=-1, get_read=True) except HTTPStatusError as e: # If the News API itself is gone (app disabled, user lost access), @@ -335,7 +353,7 @@ async def verify_search_results( results: list[SearchResult], *, evict_on_missing: bool = True, - max_concurrent: int = DEFAULT_VERIFICATION_CONCURRENCY, + max_concurrent: int | None = None, eviction_task_group: TaskGroup | None = None, ) -> list[SearchResult]: """Filter search results to those the user can currently access. @@ -359,7 +377,9 @@ async def verify_search_results( multiple chunks per document). evict_on_missing: Schedule lazy eviction for inaccessible docs. max_concurrent: Cap on concurrent verification round-trips against - Nextcloud. Defaults to ``DEFAULT_VERIFICATION_CONCURRENCY``. + Nextcloud. When ``None`` (the default), resolved from + ``Settings.verification_concurrency`` (env var + ``VERIFICATION_CONCURRENCY``, default 20). 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`` @@ -373,6 +393,9 @@ async def verify_search_results( user_id: str = client.username + if max_concurrent is None: + max_concurrent = get_settings().verification_concurrency + # Group unique (doc_id, doc_type) by doc_type so each verifier sees a # deduplicated batch. We pick one SearchResult per (id, doc_type) to carry # metadata (path, board_id/stack_id) into the verifier — chunks of the diff --git a/tests/unit/search/test_verification.py b/tests/unit/search/test_verification.py index 25727131..916270ca 100644 --- a/tests/unit/search/test_verification.py +++ b/tests/unit/search/test_verification.py @@ -137,6 +137,24 @@ async def test_verify_notes_unexpected_exception_keeps(mocker): assert result == {7} +@pytest.mark.unit +async def test_verify_notes_non_numeric_id_keeps(mocker): + """Non-numeric note id must not surface as a generic 'unexpected error'. + + The defensive int() guard runs before the network call and produces a + type-specific log line; result is kept (fail-open). + """ + notes_client = SimpleNamespace( + get_note=mocker.AsyncMock(side_effect=AssertionError("must not be called")) + ) + client = SimpleNamespace(notes=notes_client, username="alice") + + result = await _verify_notes(client, [_make_result("not-a-number")], _sem()) + + assert result == {"not-a-number"} + notes_client.get_note.assert_not_awaited() + + @pytest.mark.unit async def test_verify_notes_mixed_outcomes(mocker): """Mix of accessible, deleted, and transient — only deleted is dropped.""" @@ -207,6 +225,27 @@ async def test_verify_news_items_api_404_drops_all(mocker): assert result == set() +@pytest.mark.unit +async def test_verify_news_items_api_403_drops_all(mocker): + """News API 403 (e.g. user lost access to the app) drops all items.""" + news_client = SimpleNamespace( + get_items=mocker.AsyncMock(side_effect=_http_error(403)) + ) + client = SimpleNamespace(news=news_client, username="alice") + + result = await _verify_news_items( + client, + [ + _make_result(1, doc_type="news_item"), + _make_result(2, doc_type="news_item"), + _make_result(3, doc_type="news_item"), + ], + _sem(), + ) + + assert result == set() + + @pytest.mark.unit async def test_verify_news_items_transient_keeps_all(mocker): news_client = SimpleNamespace( @@ -265,6 +304,23 @@ async def test_verify_files_404_via_get_file_info_drops(mocker): assert result == set() +@pytest.mark.unit +async def test_verify_files_403_drops(mocker): + """get_file_info raising HTTPStatusError(403) is a definitive drop.""" + webdav_client = SimpleNamespace( + get_file_info=mocker.AsyncMock(side_effect=_http_error(403)) + ) + client = SimpleNamespace(webdav=webdav_client, username="alice") + + result = await _verify_files( + client, + [_make_result(124, doc_type="file", metadata={"path": "forbidden.txt"})], + _sem(), + ) + + assert result == set() + + @pytest.mark.unit async def test_verify_files_missing_path_metadata_keeps_unverified(mocker): """Without a path in metadata we cannot verify — fail open, don't drop."""