refactor(search): address PR #750 round 9 review feedback
- Add concurrency-safety comments to per-verifier accessible sets in _verify_notes/_verify_files/_verify_deck_cards. Same rationale as accessible_by_type in verify_search_results: anyio is cooperative, set.add() is not an await point. - Document 401 exclusion in _is_definitive_404_or_403 (treated as transient because it usually signals expired credentials, not permanent denial). - Note multi-user compounding in the news verifier semaphore comment: N concurrent users hold N slots out of the shared budget. - Log inaccessible doc ids with a type tag (e.g. "int:42" vs "str:42") so ghost-record logs disambiguate id types. - Type the BatchVerifier alias and the four verifier function signatures with NextcloudClientProtocol instead of Any (algorithms.py exposes the right interface; the protocol is runtime_checkable). - Surface verified_chunk_count vs dropped_count semantics in the nc_semantic_search tool docstring Returns block (chunks vs unique documents). - Add comments to the two max_concurrent=20 sites in server/semantic.py noting they are intentionally distinct from settings.verification_concurrency (different request phases). 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
3153c9dac4
commit
852ffa3678
@@ -39,14 +39,18 @@ 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.config import get_settings
|
||||||
from nextcloud_mcp_server.search.algorithms import SearchResult
|
from nextcloud_mcp_server.search.algorithms import (
|
||||||
|
NextcloudClientProtocol,
|
||||||
|
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__)
|
||||||
|
|
||||||
|
|
||||||
BatchVerifier = Callable[
|
BatchVerifier = Callable[
|
||||||
[Any, list[SearchResult], anyio.Semaphore], Awaitable[set[int | str]]
|
[NextcloudClientProtocol, list[SearchResult], anyio.Semaphore],
|
||||||
|
Awaitable[set[int | str]],
|
||||||
]
|
]
|
||||||
"""(client, results, semaphore) -> set of doc_ids accessible to the user."""
|
"""(client, results, semaphore) -> set of doc_ids accessible to the user."""
|
||||||
|
|
||||||
@@ -57,15 +61,26 @@ BatchVerifier = Callable[
|
|||||||
|
|
||||||
|
|
||||||
def _is_definitive_404_or_403(exc: BaseException) -> bool:
|
def _is_definitive_404_or_403(exc: BaseException) -> bool:
|
||||||
"""Return True if exc indicates the document is definitively inaccessible."""
|
"""Return True if exc indicates the document is definitively inaccessible.
|
||||||
|
|
||||||
|
401 is intentionally excluded — it usually signals expired credentials
|
||||||
|
rather than permanent denial, so it is treated as transient (keep the
|
||||||
|
result; the next query will re-verify after the client refreshes).
|
||||||
|
"""
|
||||||
if isinstance(exc, HTTPStatusError):
|
if isinstance(exc, HTTPStatusError):
|
||||||
return exc.response.status_code in (403, 404)
|
return exc.response.status_code in (403, 404)
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
async def _verify_notes(
|
async def _verify_notes(
|
||||||
client: Any, results: list[SearchResult], semaphore: anyio.Semaphore
|
client: NextcloudClientProtocol,
|
||||||
|
results: list[SearchResult],
|
||||||
|
semaphore: anyio.Semaphore,
|
||||||
) -> set[int | str]:
|
) -> set[int | str]:
|
||||||
|
# Mutated by inner check() tasks under the task group below. Safe
|
||||||
|
# without a lock: anyio is cooperative, .add() is not an await
|
||||||
|
# point, so two tasks cannot race on the same write. Same rationale
|
||||||
|
# as accessible_by_type in verify_search_results.
|
||||||
accessible: set[int | str] = set()
|
accessible: set[int | str] = set()
|
||||||
|
|
||||||
async def check(result: SearchResult) -> None:
|
async def check(result: SearchResult) -> None:
|
||||||
@@ -116,8 +131,14 @@ async def _verify_notes(
|
|||||||
|
|
||||||
|
|
||||||
async def _verify_files(
|
async def _verify_files(
|
||||||
client: Any, results: list[SearchResult], semaphore: anyio.Semaphore
|
client: NextcloudClientProtocol,
|
||||||
|
results: list[SearchResult],
|
||||||
|
semaphore: anyio.Semaphore,
|
||||||
) -> set[int | str]:
|
) -> set[int | str]:
|
||||||
|
# Mutated by inner check() tasks under the task group below. Safe
|
||||||
|
# without a lock: anyio is cooperative, .add() is not an await
|
||||||
|
# point, so two tasks cannot race on the same write. Same rationale
|
||||||
|
# as accessible_by_type in verify_search_results.
|
||||||
accessible: set[int | str] = set()
|
accessible: set[int | str] = set()
|
||||||
|
|
||||||
async def check(result: SearchResult) -> None:
|
async def check(result: SearchResult) -> None:
|
||||||
@@ -184,8 +205,14 @@ async def _verify_files(
|
|||||||
|
|
||||||
|
|
||||||
async def _verify_deck_cards(
|
async def _verify_deck_cards(
|
||||||
client: Any, results: list[SearchResult], semaphore: anyio.Semaphore
|
client: NextcloudClientProtocol,
|
||||||
|
results: list[SearchResult],
|
||||||
|
semaphore: anyio.Semaphore,
|
||||||
) -> set[int | str]:
|
) -> set[int | str]:
|
||||||
|
# Mutated by inner check() tasks under the task group below. Safe
|
||||||
|
# without a lock: anyio is cooperative, .add() is not an await
|
||||||
|
# point, so two tasks cannot race on the same write. Same rationale
|
||||||
|
# as accessible_by_type in verify_search_results.
|
||||||
accessible: set[int | str] = set()
|
accessible: set[int | str] = set()
|
||||||
|
|
||||||
async def check(result: SearchResult) -> None:
|
async def check(result: SearchResult) -> None:
|
||||||
@@ -263,7 +290,9 @@ async def _verify_deck_cards(
|
|||||||
|
|
||||||
|
|
||||||
async def _verify_news_items(
|
async def _verify_news_items(
|
||||||
client: Any, results: list[SearchResult], semaphore: anyio.Semaphore
|
client: NextcloudClientProtocol,
|
||||||
|
results: list[SearchResult],
|
||||||
|
semaphore: anyio.Semaphore,
|
||||||
) -> set[int | str]:
|
) -> set[int | str]:
|
||||||
"""Batch-verify news items with a single fetch.
|
"""Batch-verify news items with a single fetch.
|
||||||
|
|
||||||
@@ -284,6 +313,12 @@ async def _verify_news_items(
|
|||||||
# the same semaphore. Latency of this fetch is proportional to the
|
# the same semaphore. Latency of this fetch is proportional to the
|
||||||
# user's full news corpus; see the News caveat in
|
# user's full news corpus; see the News caveat in
|
||||||
# docs/configuration.md for production guidance.
|
# docs/configuration.md for production guidance.
|
||||||
|
#
|
||||||
|
# Multi-user note: the "≤1 per search" bound is per-search, not
|
||||||
|
# per-process. If N users simultaneously search news content, all N
|
||||||
|
# hold a slot for the duration of their respective fetches, each
|
||||||
|
# consuming 1/max_concurrent of the shared verification budget. A
|
||||||
|
# single news-heavy user can therefore hold their slot for seconds.
|
||||||
async with semaphore:
|
async with semaphore:
|
||||||
try:
|
try:
|
||||||
# TODO(perf): if profiling shows this fetch dominates query latency
|
# TODO(perf): if profiling shows this fetch dominates query latency
|
||||||
@@ -493,10 +528,12 @@ async def verify_search_results(
|
|||||||
inaccessible.add((doc_id, doc_type))
|
inaccessible.add((doc_id, doc_type))
|
||||||
|
|
||||||
if inaccessible:
|
if inaccessible:
|
||||||
|
# Tag ids with their type (int vs str) so ghost-record logs are
|
||||||
|
# unambiguous: int 42 and str "42" both render as "42" otherwise.
|
||||||
logger.info(
|
logger.info(
|
||||||
"Verification dropped %d inaccessible document(s): %s",
|
"Verification dropped %d inaccessible document(s): %s",
|
||||||
len(inaccessible),
|
len(inaccessible),
|
||||||
sorted((str(d), t) for d, t in inaccessible),
|
sorted((f"{type(d).__name__}:{d}", t) for d, t in inaccessible),
|
||||||
)
|
)
|
||||||
|
|
||||||
# Filter results, preserving order. All chunks of an inaccessible document
|
# Filter results, preserving order. All chunks of an inaccessible document
|
||||||
|
|||||||
@@ -87,7 +87,16 @@ def configure_semantic_tools(mcp: FastMCP):
|
|||||||
context_chars: Number of characters to include before/after matched chunk (default: 300)
|
context_chars: Number of characters to include before/after matched chunk (default: 300)
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
SemanticSearchResponse with matching documents ranked by fusion scores
|
SemanticSearchResponse with matching documents ranked by fusion scores.
|
||||||
|
|
||||||
|
Verification fields (ADR-019 verify-on-read):
|
||||||
|
- verified_chunk_count: chunk rows that passed access checks
|
||||||
|
(sized in chunks; counted before trimming to ``limit``, so it
|
||||||
|
can exceed ``len(results)`` when a doc has multiple matching
|
||||||
|
chunks).
|
||||||
|
- dropped_count: unique ``(doc_id, doc_type)`` pairs evicted as
|
||||||
|
ghost records during this search (sized in documents, not
|
||||||
|
chunks).
|
||||||
"""
|
"""
|
||||||
settings = get_settings()
|
settings = get_settings()
|
||||||
client = await get_client(ctx)
|
client = await get_client(ctx)
|
||||||
@@ -248,8 +257,15 @@ def configure_semantic_tools(mcp: FastMCP):
|
|||||||
context_chars,
|
context_chars,
|
||||||
)
|
)
|
||||||
|
|
||||||
# Fetch context for all results in parallel
|
# Fetch context for all results in parallel.
|
||||||
# Limit concurrent requests to prevent connection pool exhaustion
|
# Limit concurrent requests to prevent connection pool exhaustion.
|
||||||
|
#
|
||||||
|
# Intentionally distinct from settings.verification_concurrency:
|
||||||
|
# that knob bounds Nextcloud round-trips during access
|
||||||
|
# verification (ADR-019); this one bounds context-expansion
|
||||||
|
# fetches that run only when ``include_context=True``. Operators
|
||||||
|
# tuning one rarely want the other in lockstep, so they share
|
||||||
|
# the default value (20) but not the env var.
|
||||||
max_concurrent = 20
|
max_concurrent = 20
|
||||||
semaphore = anyio.Semaphore(max_concurrent)
|
semaphore = anyio.Semaphore(max_concurrent)
|
||||||
expanded_results = [None] * len(results)
|
expanded_results = [None] * len(results)
|
||||||
@@ -507,7 +523,15 @@ def configure_semantic_tools(mcp: FastMCP):
|
|||||||
accessible_results = [None] * len(search_response.results)
|
accessible_results = [None] * len(search_response.results)
|
||||||
full_contents = [None] * len(search_response.results)
|
full_contents = [None] * len(search_response.results)
|
||||||
|
|
||||||
# Limit concurrent requests to prevent connection pool exhaustion
|
# Limit concurrent requests to prevent connection pool exhaustion.
|
||||||
|
#
|
||||||
|
# Intentionally distinct from settings.verification_concurrency:
|
||||||
|
# that knob bounds Nextcloud round-trips during access
|
||||||
|
# verification (ADR-019). This one bounds the answer tool's
|
||||||
|
# full-content fetch — a separate request phase tied to RAG
|
||||||
|
# answer generation. Operators tuning one rarely want the other
|
||||||
|
# in lockstep, so they share the default value (20) but not the
|
||||||
|
# env var.
|
||||||
max_concurrent = 20
|
max_concurrent = 20
|
||||||
semaphore = anyio.Semaphore(max_concurrent)
|
semaphore = anyio.Semaphore(max_concurrent)
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user