refactor(search): address PR #750 round 5 review feedback
Tightens verifier consistency, closes test gaps, hardens the fire-and-forget eviction snapshot, and routes the new concurrency knob through Settings. - Pre-flight ``int()`` guard in ``_verify_notes`` mirrors ``_verify_deck_cards``, so a non-numeric note id produces a type-specific log line instead of falling through to the generic "unexpected error" branch. - Adds explicit 403 tests for the file and news verifiers (symmetry with the existing notes/deck 403 tests) plus a ``non_numeric_id_keeps`` test. - ``AppContext`` and ``OAuthAppContext`` no longer snapshot ``_vector_sync_state.eviction_task_group`` at lifespan-yield time. Both expose it as a ``@property`` that reads the singleton dynamically, removing the order-sensitive race where a future startup-ordering change could silently degrade fire-and-forget eviction to inline forever. - Adds ``verification_concurrency`` (env var ``VERIFICATION_CONCURRENCY``, default 20) to ``Settings`` with a dynaconf validator; ``verify_search_results`` resolves the cap lazily from settings when the caller doesn't override it. - Enriches the news verifier TODO to call out that ``batch_size=-1`` is intentional — a numeric ceiling would silently break correctness because any item beyond the cap would be missing from ``present_ids`` and dropped. - Updates ``Optional[TaskGroup]`` to ``TaskGroup | None`` per project style. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.7
parent
926722b09d
commit
ffcca23a7b
@@ -489,10 +489,11 @@ aware of:
|
|||||||
unique documents, so verification adds 3-5 round-trips. With the default
|
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
|
20-way concurrency this is one parallel batch — usually under 100 ms on a
|
||||||
healthy connection.
|
healthy connection.
|
||||||
- **Concurrency**: all verifications fan out under a shared semaphore
|
- **Concurrency**: all verifications fan out under a shared semaphore.
|
||||||
(`DEFAULT_VERIFICATION_CONCURRENCY = 20` in `search/verification.py`). The
|
Tunable via the `VERIFICATION_CONCURRENCY` env var (settings field
|
||||||
limit is not currently exposed as an env var; if production workloads
|
`verification_concurrency`, default 20) — lower it if your Nextcloud
|
||||||
saturate Nextcloud, consider opening an issue to make it tunable.
|
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
|
- **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
|
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
|
per search that contains any news result, then intersects locally. The
|
||||||
|
|||||||
@@ -323,7 +323,7 @@ class VectorSyncState:
|
|||||||
# Long-lived task group used for fire-and-forget background work spawned
|
# 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
|
# 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.
|
# 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
|
# Module-level singleton for vector sync state
|
||||||
@@ -340,7 +340,15 @@ class AppContext:
|
|||||||
document_receive_stream: Optional[MemoryObjectReceiveStream] = None
|
document_receive_stream: Optional[MemoryObjectReceiveStream] = None
|
||||||
shutdown_event: Optional[anyio.Event] = None
|
shutdown_event: Optional[anyio.Event] = None
|
||||||
scanner_wake_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
|
@dataclass
|
||||||
@@ -359,7 +367,11 @@ class OAuthAppContext:
|
|||||||
document_receive_stream: Optional[MemoryObjectReceiveStream] = None
|
document_receive_stream: Optional[MemoryObjectReceiveStream] = None
|
||||||
shutdown_event: Optional[anyio.Event] = None
|
shutdown_event: Optional[anyio.Event] = None
|
||||||
scanner_wake_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:
|
class BasicAuthMiddleware:
|
||||||
@@ -576,7 +588,8 @@ async def app_lifespan_basic(server: FastMCP) -> AsyncIterator[AppContext]:
|
|||||||
document_receive_stream=_vector_sync_state.document_receive_stream,
|
document_receive_stream=_vector_sync_state.document_receive_stream,
|
||||||
shutdown_event=_vector_sync_state.shutdown_event,
|
shutdown_event=_vector_sync_state.shutdown_event,
|
||||||
scanner_wake_event=_vector_sync_state.scanner_wake_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:
|
finally:
|
||||||
logger.info("Shutting down BasicAuth session")
|
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,
|
document_receive_stream=_vector_sync_state.document_receive_stream,
|
||||||
shutdown_event=_vector_sync_state.shutdown_event,
|
shutdown_event=_vector_sync_state.shutdown_event,
|
||||||
scanner_wake_event=_vector_sync_state.scanner_wake_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:
|
finally:
|
||||||
logger.info("Shutting down MCP server")
|
logger.info("Shutting down MCP server")
|
||||||
|
|||||||
@@ -65,6 +65,8 @@ _DEFAULTS: dict[str, Any] = {
|
|||||||
"vector_sync_processor_workers": 3,
|
"vector_sync_processor_workers": 3,
|
||||||
"vector_sync_queue_max_size": 10000,
|
"vector_sync_queue_max_size": 10000,
|
||||||
"vector_sync_user_poll_interval": 60,
|
"vector_sync_user_poll_interval": 60,
|
||||||
|
# Verify-on-read concurrency cap (ADR-019)
|
||||||
|
"verification_concurrency": 20,
|
||||||
# Qdrant
|
# Qdrant
|
||||||
"qdrant_url": None,
|
"qdrant_url": None,
|
||||||
"qdrant_location": None,
|
"qdrant_location": None,
|
||||||
@@ -170,6 +172,7 @@ _dynaconf = Dynaconf(
|
|||||||
Validator("VECTOR_SYNC_PROCESSOR_WORKERS", gte=1),
|
Validator("VECTOR_SYNC_PROCESSOR_WORKERS", gte=1),
|
||||||
Validator("VECTOR_SYNC_QUEUE_MAX_SIZE", gte=1),
|
Validator("VECTOR_SYNC_QUEUE_MAX_SIZE", gte=1),
|
||||||
Validator("VECTOR_SYNC_USER_POLL_INTERVAL", gte=1),
|
Validator("VECTOR_SYNC_USER_POLL_INTERVAL", gte=1),
|
||||||
|
Validator("VERIFICATION_CONCURRENCY", gte=1),
|
||||||
Validator("DOCUMENT_CHUNK_SIZE", gte=1),
|
Validator("DOCUMENT_CHUNK_SIZE", gte=1),
|
||||||
# Non-negative
|
# Non-negative
|
||||||
Validator("DOCUMENT_CHUNK_OVERLAP", gte=0),
|
Validator("DOCUMENT_CHUNK_OVERLAP", gte=0),
|
||||||
@@ -455,6 +458,12 @@ class Settings:
|
|||||||
vector_sync_queue_max_size: int = 10000
|
vector_sync_queue_max_size: int = 10000
|
||||||
vector_sync_user_poll_interval: int = 60 # seconds - OAuth mode user discovery
|
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 settings (mutually exclusive modes)
|
||||||
qdrant_url: str | None = None # Network mode: http://qdrant:6333
|
qdrant_url: str | None = None # Network mode: http://qdrant:6333
|
||||||
qdrant_location: str | None = None # Local mode: :memory: or /path/to/data
|
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_processor_workers": "VECTOR_SYNC_PROCESSOR_WORKERS",
|
||||||
"vector_sync_queue_max_size": "VECTOR_SYNC_QUEUE_MAX_SIZE",
|
"vector_sync_queue_max_size": "VECTOR_SYNC_QUEUE_MAX_SIZE",
|
||||||
"vector_sync_user_poll_interval": "VECTOR_SYNC_USER_POLL_INTERVAL",
|
"vector_sync_user_poll_interval": "VECTOR_SYNC_USER_POLL_INTERVAL",
|
||||||
|
# Verify-on-read (ADR-019)
|
||||||
|
"verification_concurrency": "VERIFICATION_CONCURRENCY",
|
||||||
# Qdrant settings
|
# Qdrant settings
|
||||||
"qdrant_url": "QDRANT_URL",
|
"qdrant_url": "QDRANT_URL",
|
||||||
"qdrant_location": "QDRANT_LOCATION",
|
"qdrant_location": "QDRANT_LOCATION",
|
||||||
|
|||||||
@@ -38,17 +38,13 @@ import anyio
|
|||||||
from anyio.abc import TaskGroup
|
from anyio.abc import TaskGroup
|
||||||
from httpx import HTTPStatusError
|
from httpx import HTTPStatusError
|
||||||
|
|
||||||
|
from nextcloud_mcp_server.config import get_settings
|
||||||
from nextcloud_mcp_server.search.algorithms import SearchResult
|
from nextcloud_mcp_server.search.algorithms import SearchResult
|
||||||
from nextcloud_mcp_server.vector.eviction import delete_document_points
|
from nextcloud_mcp_server.vector.eviction import delete_document_points
|
||||||
|
|
||||||
logger = logging.getLogger(__name__)
|
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[
|
BatchVerifier = Callable[
|
||||||
[Any, list[SearchResult], anyio.Semaphore], Awaitable[set[int | str]]
|
[Any, list[SearchResult], anyio.Semaphore], Awaitable[set[int | str]]
|
||||||
]
|
]
|
||||||
@@ -73,10 +69,24 @@ async def _verify_notes(
|
|||||||
accessible: set[int | str] = set()
|
accessible: set[int | str] = set()
|
||||||
|
|
||||||
async def check(result: SearchResult) -> None:
|
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:
|
async with semaphore:
|
||||||
doc_id = result.id
|
|
||||||
try:
|
try:
|
||||||
await client.notes.get_note(int(doc_id))
|
await client.notes.get_note(note_id_int)
|
||||||
accessible.add(doc_id)
|
accessible.add(doc_id)
|
||||||
except HTTPStatusError as e:
|
except HTTPStatusError as e:
|
||||||
if _is_definitive_404_or_403(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
|
# a per-item News API endpoint. The shared semaphore protects
|
||||||
# against runaway concurrent fetches, but the payload itself can
|
# against runaway concurrent fetches, but the payload itself can
|
||||||
# be large (News auto-purge cap is in the thousands of items).
|
# 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)
|
items = await client.news.get_items(batch_size=-1, get_read=True)
|
||||||
except HTTPStatusError as e:
|
except HTTPStatusError as e:
|
||||||
# If the News API itself is gone (app disabled, user lost access),
|
# If the News API itself is gone (app disabled, user lost access),
|
||||||
@@ -335,7 +353,7 @@ async def verify_search_results(
|
|||||||
results: list[SearchResult],
|
results: list[SearchResult],
|
||||||
*,
|
*,
|
||||||
evict_on_missing: bool = True,
|
evict_on_missing: bool = True,
|
||||||
max_concurrent: int = DEFAULT_VERIFICATION_CONCURRENCY,
|
max_concurrent: int | None = None,
|
||||||
eviction_task_group: TaskGroup | None = None,
|
eviction_task_group: TaskGroup | None = None,
|
||||||
) -> list[SearchResult]:
|
) -> list[SearchResult]:
|
||||||
"""Filter search results to those the user can currently access.
|
"""Filter search results to those the user can currently access.
|
||||||
@@ -359,7 +377,9 @@ async def verify_search_results(
|
|||||||
multiple chunks per document).
|
multiple chunks per document).
|
||||||
evict_on_missing: Schedule lazy eviction for inaccessible docs.
|
evict_on_missing: Schedule lazy eviction for inaccessible docs.
|
||||||
max_concurrent: Cap on concurrent verification round-trips against
|
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
|
eviction_task_group: Optional long-lived task group on which to
|
||||||
spawn fire-and-forget eviction. Pass
|
spawn fire-and-forget eviction. Pass
|
||||||
``ctx.request_context.lifespan_context.eviction_task_group``
|
``ctx.request_context.lifespan_context.eviction_task_group``
|
||||||
@@ -373,6 +393,9 @@ async def verify_search_results(
|
|||||||
|
|
||||||
user_id: str = client.username
|
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
|
# 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
|
# deduplicated batch. We pick one SearchResult per (id, doc_type) to carry
|
||||||
# metadata (path, board_id/stack_id) into the verifier — chunks of the
|
# metadata (path, board_id/stack_id) into the verifier — chunks of the
|
||||||
|
|||||||
@@ -137,6 +137,24 @@ async def test_verify_notes_unexpected_exception_keeps(mocker):
|
|||||||
assert result == {7}
|
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
|
@pytest.mark.unit
|
||||||
async def test_verify_notes_mixed_outcomes(mocker):
|
async def test_verify_notes_mixed_outcomes(mocker):
|
||||||
"""Mix of accessible, deleted, and transient — only deleted is dropped."""
|
"""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()
|
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
|
@pytest.mark.unit
|
||||||
async def test_verify_news_items_transient_keeps_all(mocker):
|
async def test_verify_news_items_transient_keeps_all(mocker):
|
||||||
news_client = SimpleNamespace(
|
news_client = SimpleNamespace(
|
||||||
@@ -265,6 +304,23 @@ async def test_verify_files_404_via_get_file_info_drops(mocker):
|
|||||||
assert result == set()
|
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
|
@pytest.mark.unit
|
||||||
async def test_verify_files_missing_path_metadata_keeps_unverified(mocker):
|
async def test_verify_files_missing_path_metadata_keeps_unverified(mocker):
|
||||||
"""Without a path in metadata we cannot verify — fail open, don't drop."""
|
"""Without a path in metadata we cannot verify — fail open, don't drop."""
|
||||||
|
|||||||
Reference in New Issue
Block a user