admin/mcp-nextcloud
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #849 from cbcoutinho/fix/vector-sync-pagination-churn-observability

Browse Source 
fix: paginate tagged-folder SEARCH so the vector-sync scanner discovers all files
This commit is contained in:
Chris Coutinho
2026-06-04 17:59:17 +02:00
committed by GitHub
parent 0919513f21 9452057570
commit aa0aa5dc6b
5 changed files with 469 additions and 27 deletions
Download Patch File Download Diff File Expand all files Collapse all files
nextcloud_mcp_server/client/__init__.py
+4 -1
View File
@@ -244,7 +244,10 @@ class NextcloudClient:
for dir_info in tagged_dirs:
dir_path = dir_info.get("path", "").strip("/")
try:
descendants = await self.webdav.find_by_type(
# find_all_by_type pages past Nextcloud's default ~100-result
# SEARCH page so every tagged-folder descendant is discovered;
# find_by_type would silently cap a large folder.
descendants = await self.webdav.find_all_by_type(
mime_type_filter, scope=dir_path
)
except Exception as e:
nextcloud_mcp_server/client/webdav.py
+234 -12
View File
@@ -6,13 +6,25 @@ import xml.etree.ElementTree as ET
from email.utils import parsedate_to_datetime
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import unquote
from xml.sax.saxutils import escape as xml_escape
from httpx import HTTPStatusError
from nextcloud_mcp_server.observability.metrics import document_scan_truncated_total
from .base import BaseNextcloudClient
logger = logging.getLogger(__name__)
# Paging defaults for WebDAV SEARCH. Nextcloud's SEARCH returns a server-default
# page (~100 results) when no ``<d:nresults>`` is sent, silently truncating large
# folders. ``search_files_all`` pages explicitly to fetch the complete result set.
WEBDAV_SEARCH_PAGE_SIZE = 500
# Hard ceiling so a pathologically large folder can't drive an unbounded crawl.
# Crossing it is logged as a truncation warning (and surfaced via a metric) so the
# cap can never again silently hide files.
WEBDAV_SEARCH_MAX_RESULTS = 50000
class WebDAVClient(BaseNextcloudClient):
"""Client for Nextcloud WebDAV operations."""
@@ -620,6 +632,7 @@ class WebDAVClient(BaseNextcloudClient):
properties: Optional[List[str]] = None,
order_by: Optional[List[Tuple[str, str]]] = None,
limit: Optional[int] = None,
offset: Optional[int] = None,
) -> List[Dict[str, Any]]:
"""Search for files using WebDAV SEARCH method (RFC 5323).
@@ -629,6 +642,10 @@ class WebDAVClient(BaseNextcloudClient):
properties: List of property names to retrieve (defaults to basic set)
order_by: List of (property, direction) tuples for sorting, e.g. [("getlastmodified", "descending")]
limit: Maximum number of results to return
offset: Number of leading results to skip (``<d:firstresult>``). Note
that not every Nextcloud release honours offset paging; callers
that need guaranteed completeness should use ``search_files_all``,
which detects an ignored offset and falls back.
Returns:
List of file/directory dictionaries with requested properties
@@ -651,6 +668,7 @@ class WebDAVClient(BaseNextcloudClient):
properties=properties,
order_by=order_by,
limit=limit,
offset=offset,
)
# The SEARCH endpoint is at the dav root
@@ -679,6 +697,164 @@ class WebDAVClient(BaseNextcloudClient):
logger.error("Unexpected error during search: %s", e)
raise e
async def search_files_all(
self,
scope: str = "",
where_conditions: Optional[str] = None,
properties: Optional[List[str]] = None,
order_by: Optional[List[Tuple[str, str]]] = None,
page_size: int = WEBDAV_SEARCH_PAGE_SIZE,
max_results: int = WEBDAV_SEARCH_MAX_RESULTS,
) -> List[Dict[str, Any]]:
"""Fetch the *complete* SEARCH result set, paging past the server default.
A plain ``search_files`` with no ``limit`` returns only Nextcloud's default
page (~100), silently dropping the rest of a large folder. This method pages
with ``<d:firstresult>`` until a short page signals the end. If the server
ignores the offset (a page repeats results already seen), it falls back to a
single fetch with an explicit large ``nresults`` so completeness never depends
on offset support.
Args:
scope: Directory path to search in (empty string for user root)
where_conditions: XML where-clause conditions
properties: Properties to retrieve (must include ``fileid`` for dedup)
order_by: Optional sort order
page_size: Results requested per page
max_results: Hard ceiling; crossing it logs a truncation warning and
increments ``webdav_search_truncated_total``
Returns:
All matching file/directory dicts, de-duplicated by file id / path.
"""
paged = await self._search_offset_paged(
scope, where_conditions, properties, order_by, page_size, max_results
)
# ``None`` signals the server ignored the offset (or an offset page
# failed) -- fetch everything in one bounded request instead.
if paged is None:
return await self._single_fetch_fallback(
scope, where_conditions, properties, order_by, max_results
)
self._warn_if_truncated(len(paged), scope, max_results)
return paged[:max_results]
async def _search_offset_paged(
self,
scope: str,
where_conditions: Optional[str],
properties: Optional[List[str]],
order_by: Optional[List[Tuple[str, str]]],
page_size: int,
max_results: int,
) -> Optional[List[Dict[str, Any]]]:
"""Page the SEARCH with ``<d:firstresult>`` until exhausted.
Returns the accumulated rows, or ``None`` when the server ignores the
offset (a page repeats already-seen rows, or an offset page errors) and
the caller should fall back to a single bounded fetch.
"""
def _key(item: Dict[str, Any]) -> Any:
# file_id is globally unique; path is the stable fallback when a
# producer omits fileid. ``id(item)`` is a last resort so an item
# missing both never collapses into another under a shared ``None``
# key (which would silently drop rows from the result set).
# ``is not None`` rather than truthiness so a (hypothetical)
# file_id of 0 isn't treated as absent.
file_id = item.get("file_id")
if file_id is not None:
return file_id
return item.get("path") or id(item)
results: List[Dict[str, Any]] = []
seen: set[Any] = set()
offset = 0
while len(results) < max_results:
try:
page = await self.search_files(
scope=scope,
where_conditions=where_conditions,
properties=properties,
order_by=order_by,
limit=page_size,
offset=offset,
)
except Exception:
# A failure on the very first page is a real error, not a
# paging quirk -- surface it. A later page failing means offset
# paging is unusable; signal a fallback rather than lose the tail.
if offset == 0:
raise
logger.warning(
"WebDAV SEARCH offset page failed for scope %r; "
"falling back to single fetch",
scope,
)
return None
if not page:
break
fresh = [item for item in page if _key(item) not in seen]
# Server ignored the offset (returned an already-seen page); signal
# the caller to re-fetch in one bounded request. The accumulated
# ``results`` are intentionally discarded -- the single fetch is
# authoritative and re-returns them, so nothing is lost.
if offset > 0 and not fresh:
logger.warning(
"WebDAV SEARCH ignored offset for scope %r; "
"falling back to single fetch (limit=%d)",
scope,
max_results,
)
return None
for item in fresh:
seen.add(_key(item))
results.append(item)
# A short page means we've reached the end of the result set.
if len(page) < page_size:
break
offset += page_size
return results
async def _single_fetch_fallback(
self,
scope: str,
where_conditions: Optional[str],
properties: Optional[List[str]],
order_by: Optional[List[Tuple[str, str]]],
max_results: int,
) -> List[Dict[str, Any]]:
"""Single SEARCH with a large explicit ``nresults`` (offset-free fallback)."""
results = await self.search_files(
scope=scope,
where_conditions=where_conditions,
properties=properties,
order_by=order_by,
limit=max_results,
)
self._warn_if_truncated(len(results), scope, max_results)
return results
@staticmethod
def _warn_if_truncated(count: int, scope: str, max_results: int) -> None:
"""Warn + count when a SEARCH hit the ceiling, so a cap is never silent."""
if count >= max_results:
document_scan_truncated_total.inc()
logger.warning(
"WebDAV SEARCH reached max_results=%d for scope %r; "
"results may be truncated -- raise WEBDAV_SEARCH_MAX_RESULTS",
max_results,
scope,
)
def _build_search_xml(
self,
scope: str,
@@ -686,6 +862,7 @@ class WebDAVClient(BaseNextcloudClient):
properties: List[str],
order_by: Optional[List[Tuple[str, str]]],
limit: Optional[int],
offset: Optional[int] = None,
) -> str:
"""Build the XML body for a SEARCH request."""
# Construct the scope path
@@ -716,10 +893,18 @@ class WebDAVClient(BaseNextcloudClient):
else:
orderby_xml = ""
# Build limit clause
limit_xml = (
f"<d:limit><d:nresults>{limit}</d:nresults></d:limit>" if limit else ""
)
# Build limit clause. ``<d:nresults>`` caps the page size; ``<d:firstresult>``
# is the paging offset. Nextcloud silently ignores an unsupported offset
# (returns the first page again) rather than erroring -- ``search_files_all``
# detects that non-progress and falls back to a single bounded fetch.
limit_parts = []
if limit:
limit_parts.append(f"<d:nresults>{limit}</d:nresults>")
# ``is not None`` (not truthiness) so a future explicit offset=0 is
# emitted rather than silently dropped.
if offset is not None:
limit_parts.append(f"<d:firstresult>{offset}</d:firstresult>")
limit_xml = f"<d:limit>{''.join(limit_parts)}</d:limit>" if limit_parts else ""
# Construct the full SEARCH XML
search_xml = f"""<?xml version="1.0" encoding="UTF-8"?>
@@ -942,13 +1127,56 @@ class WebDAVClient(BaseNextcloudClient):
# Find all PDFs
results = await find_by_type("application/pdf")
Note:
With ``limit=None`` this returns only Nextcloud's default SEARCH page
(~100 results), so it truncates large folders. Use ``find_all_by_type``
when complete coverage matters (e.g. building an indexing work-list).
"""
where_conditions, properties = self._type_search_args(mime_type)
return await self.search_files(
scope=scope,
where_conditions=where_conditions,
properties=properties,
limit=limit,
)
async def find_all_by_type(
self, mime_type: str, scope: str = ""
) -> List[Dict[str, Any]]:
"""Find *all* files of a MIME type, paging past the SEARCH default page.
Unlike ``find_by_type`` (single default-capped page), this pages the SEARCH
to completion so a large tagged folder is fully discovered. Used by the
vector-sync scanner's tagged-folder expansion, where a missed file means a
document that is never indexed.
Args:
mime_type: MIME type to search for (supports % wildcard)
scope: Directory path to search in (empty string for user root)
Returns:
All matching files (bounded by ``WEBDAV_SEARCH_MAX_RESULTS``).
"""
where_conditions, properties = self._type_search_args(mime_type)
return await self.search_files_all(
scope=scope,
where_conditions=where_conditions,
properties=properties,
)
@staticmethod
def _type_search_args(mime_type: str) -> Tuple[str, List[str]]:
"""Build the where-clause + property list for a MIME-type SEARCH."""
# Escape so a caller-supplied MIME type can't break the SEARCH XML or
# inject elements. All current callers pass literal strings, but this
# keeps the boundary safe for any future user-supplied value.
where_conditions = f"""
<d:like>
<d:prop>
<d:getcontenttype/>
</d:prop>
<d:literal>{mime_type}</d:literal>
<d:literal>{xml_escape(mime_type)}</d:literal>
</d:like>
"""
@@ -964,13 +1192,7 @@ class WebDAVClient(BaseNextcloudClient):
"getetag",
"fileid",
]
return await self.search_files(
scope=scope,
where_conditions=where_conditions,
properties=properties,
limit=limit,
)
return where_conditions, properties
async def list_favorites(
self, scope: str = "", limit: Optional[int] = None
nextcloud_mcp_server/observability/metrics.py
+12
View File
@@ -278,6 +278,18 @@ documents_indexed_total = Counter(
["source", "status"], # source: note | file | deck_card | news_item
)
# --- Document discovery / coverage ------------------------------------------
#
# Fires when a paged WebDAV SEARCH (folder-expansion during a scan) hits the
# WEBDAV_SEARCH_MAX_RESULTS ceiling, meaning the discovered file set was capped
# and some tagged documents may never be queued for indexing. This is the
# alertable signal that prevents the old *silent* 100-result truncation from
# recurring. Tenant is the Kubernetes ``namespace`` label, as elsewhere.
document_scan_truncated_total = Counter(
"astrolabe_document_scan_truncated_total",
"Times a folder-expansion SEARCH hit the result ceiling (coverage truncated)",
)
# =============================================================================
# Database Metrics
# =============================================================================
tests/unit/client/test_nextcloud_client.py
+19 -14
View File
@@ -2,10 +2,11 @@
Currently covers ``find_files_by_tag``: the wrapper that combines
``WebDAVClient.get_tag_by_name``, ``WebDAVClient.get_files_by_tag``, and
``WebDAVClient.find_by_type`` to resolve a system tag (and any tagged
``WebDAVClient.find_all_by_type`` to resolve a system tag (and any tagged
folders) into a flat list of files.
"""
from typing import Any
from unittest.mock import AsyncMock
import pytest
@@ -13,13 +14,15 @@ import pytest
from nextcloud_mcp_server.client import NextcloudClient, _normalise_search_result
def _make_client() -> NextcloudClient:
def _make_client() -> Any:
"""Build a NextcloudClient with mocked sub-clients.
The client constructor opens an httpx session; we don't need it, just
a stub instance whose ``webdav`` attribute we can replace.
a stub instance whose ``webdav`` attribute we can replace. Returned as
``Any`` so tests can freely reassign mocked methods on the sub-clients
without fighting the real ``WebDAVClient`` signatures.
"""
client = NextcloudClient.__new__(NextcloudClient)
client: Any = NextcloudClient.__new__(NextcloudClient)
client.username = "alice"
client.webdav = AsyncMock()
return client
@@ -99,7 +102,7 @@ class TestFindFilesByTag:
result = await client.find_files_by_tag("vector-index")
assert result == []
client.webdav.find_by_type.assert_not_called()
client.webdav.find_all_by_type.assert_not_called()
async def test_directly_tagged_files_pass_through_with_mime_filter(self):
client = _make_client()
@@ -127,7 +130,7 @@ class TestFindFilesByTag:
assert {f["id"] for f in result} == {1}
# No tagged dirs → no SEARCH walk.
client.webdav.find_by_type.assert_not_called()
client.webdav.find_all_by_type.assert_not_called()
async def test_expands_tagged_directory_into_pdf_descendants(self):
client = _make_client()
@@ -144,7 +147,7 @@ class TestFindFilesByTag:
]
)
# Search inside the folder returns two PDFs.
client.webdav.find_by_type = AsyncMock(
client.webdav.find_all_by_type = AsyncMock(
return_value=[
{
"file_id": 11,
@@ -174,8 +177,8 @@ class TestFindFilesByTag:
assert f["last_modified_timestamp"] is not None
# SEARCH was scoped to the tagged folder (no leading slash) and
# forwarded the requested MIME type as the positional first arg.
client.webdav.find_by_type.assert_awaited_once()
call_args = client.webdav.find_by_type.await_args
client.webdav.find_all_by_type.assert_awaited_once()
call_args = client.webdav.find_all_by_type.await_args
assert call_args.args[0] == "application/pdf"
assert call_args.kwargs["scope"] == "corpus"
@@ -199,7 +202,7 @@ class TestFindFilesByTag:
},
]
)
client.webdav.find_by_type = AsyncMock(
client.webdav.find_all_by_type = AsyncMock(
return_value=[
{
"file_id": 11,
@@ -244,7 +247,9 @@ class TestFindFilesByTag:
},
]
)
client.webdav.find_by_type = AsyncMock(side_effect=RuntimeError("REPORT 500"))
client.webdav.find_all_by_type = AsyncMock(
side_effect=RuntimeError("REPORT 500")
)
import logging
@@ -282,10 +287,10 @@ class TestFindFilesByTag:
# Without a MIME filter, directory expansion would fan out
# uncontrollably — the helper deliberately skips it.
assert {f["id"] for f in result} == {7}
client.webdav.find_by_type.assert_not_called()
client.webdav.find_all_by_type.assert_not_called()
async def test_skips_descendant_directories_in_search_results(self):
"""find_by_type can return collections too (e.g. when the SEARCH
"""find_all_by_type can return collections too (e.g. when the SEARCH
backend treats a folder's mime type as matching). Those must not
slip through and clobber file IDs."""
client = _make_client()
@@ -300,7 +305,7 @@ class TestFindFilesByTag:
}
]
)
client.webdav.find_by_type = AsyncMock(
client.webdav.find_all_by_type = AsyncMock(
return_value=[
{
"file_id": 50,
tests/unit/test_webdav_search_paging.py
+200
View File
@@ -0,0 +1,200 @@
"""Unit tests for paged WebDAV SEARCH (complete folder discovery).
These cover ``WebDAVClient.search_files_all`` -- the helper the vector-sync
scanner uses to expand a tagged folder into *all* its descendants, rather than
just Nextcloud's default ~100-result SEARCH page (which silently truncated large
folders and left documents unindexed).
The behaviour we pin:
* offset paging when the server honours ``<d:firstresult>``;
* automatic fallback to a single bounded fetch when the server *ignores*
offset (the real Nextcloud 31 behaviour -- a page repeats already-seen rows);
* a single short page terminates immediately;
* crossing ``max_results`` warns + increments the truncation metric;
* ``_build_search_xml`` emits the offset element only when asked.
"""
from typing import Any
from unittest.mock import AsyncMock
import httpx
import pytest
from nextcloud_mcp_server.client.webdav import WebDAVClient
pytestmark = pytest.mark.unit
def _make_client(mocker) -> Any:
# Returned as Any so tests can reassign mocked search methods without
# tripping ty's invalid-assignment on the real WebDAVClient signatures.
client: Any = WebDAVClient(mocker.AsyncMock(spec=httpx.AsyncClient), "alice")
return client
def _corpus(n: int) -> list[dict]:
return [{"file_id": i, "path": f"/dir/f{i}.pdf"} for i in range(n)]
async def test_single_short_page_returns_all_in_one_call(mocker):
"""A folder smaller than the page size resolves in a single SEARCH."""
client = _make_client(mocker)
client.search_files = AsyncMock(return_value=_corpus(10))
results = await client.search_files_all(scope="dir", page_size=500)
assert [r["file_id"] for r in results] == list(range(10))
client.search_files.assert_awaited_once()
async def test_offset_honored_pages_through_entire_corpus(mocker):
"""When the server honours offset, every page is fetched until exhausted."""
client = _make_client(mocker)
corpus = _corpus(250)
def fake(*, limit, offset=0, **_):
return corpus[offset : offset + limit]
client.search_files = AsyncMock(side_effect=fake)
results = await client.search_files_all(scope="dir", page_size=100)
assert [r["file_id"] for r in results] == list(range(250))
# 100, 100, 50 -> three pages, no fallback needed
assert client.search_files.await_count == 3
async def test_offset_ignored_falls_back_to_single_fetch(mocker):
"""Real Nextcloud ignores offset; we must still return the full corpus."""
client = _make_client(mocker)
corpus = _corpus(250)
def fake(*, limit, offset=0, **_):
# offset IGNORED: always return the first ``limit`` rows.
return corpus[:limit]
client.search_files = AsyncMock(side_effect=fake)
results = await client.search_files_all(scope="dir", page_size=100)
# page0 (0..99) -> page1(offset=100) repeats 0..99 -> detected ->
# single fetch with the large ceiling returns everything.
assert [r["file_id"] for r in results] == list(range(250))
async def test_truncation_warns_and_increments_metric(mocker):
"""Hitting max_results must surface (warn + metric), never silently drop."""
client = _make_client(mocker)
corpus = _corpus(20)
def fake(*, limit, offset=0, **_):
return corpus[offset : offset + limit]
client.search_files = AsyncMock(side_effect=fake)
metric = mocker.patch(
"nextcloud_mcp_server.client.webdav.document_scan_truncated_total"
)
results = await client.search_files_all(scope="dir", page_size=5, max_results=5)
assert len(results) == 5
metric.inc.assert_called_once()
async def test_offset_ignored_fallback_truncation_metric(mocker):
"""The fallback path also reports truncation when it hits the ceiling."""
client = _make_client(mocker)
corpus = _corpus(40)
def fake(*, limit, offset=0, **_):
return corpus[:limit] # offset ignored
client.search_files = AsyncMock(side_effect=fake)
metric = mocker.patch(
"nextcloud_mcp_server.client.webdav.document_scan_truncated_total"
)
results = await client.search_files_all(scope="dir", page_size=10, max_results=10)
assert len(results) == 10
metric.inc.assert_called_once()
async def test_offset_page_exception_falls_back_to_single_fetch(mocker):
"""If an offset page *raises* (e.g. server rejects firstresult), the
exception fallback must still return the full corpus -- and it must be
awaited (regression guard for a missing ``await``)."""
client = _make_client(mocker)
corpus = _corpus(80)
def fake(*, limit, offset=0, **_):
if offset > 0:
raise RuntimeError("server rejected <d:firstresult>")
return corpus[:limit]
client.search_files = AsyncMock(side_effect=fake)
results = await client.search_files_all(scope="dir", page_size=50)
# page0 (0..49) fills; page1(offset=50) raises -> fallback single fetch
# returns the whole corpus. A non-awaited coroutine would fail these.
assert isinstance(results, list)
assert [r["file_id"] for r in results] == list(range(80))
async def test_offset_page_exception_at_offset_zero_propagates(mocker):
"""A failure on the very first page is a real error, not a paging quirk."""
client = _make_client(mocker)
client.search_files = AsyncMock(side_effect=RuntimeError("boom"))
with pytest.raises(RuntimeError, match="boom"):
await client.search_files_all(scope="dir", page_size=50)
async def test_find_all_by_type_delegates_to_search_files_all(mocker):
client = _make_client(mocker)
client.search_files_all = AsyncMock(return_value=_corpus(3))
results = await client.find_all_by_type("application/pdf", scope="dir")
assert len(results) == 3
kwargs = client.search_files_all.await_args.kwargs
assert kwargs["scope"] == "dir"
assert "fileid" in kwargs["properties"]
assert "application/pdf" in kwargs["where_conditions"]
def test_type_search_args_escapes_mime_type(mocker):
"""A MIME value with XML metacharacters must not break / inject into the SEARCH."""
client = _make_client(mocker)
where, properties = client._type_search_args("application/pdf<&>")
# The injected metacharacters are escaped inside the <d:literal>, so they
# can't break the SEARCH XML or introduce new elements.
assert "pdf&lt;&amp;&gt;" in where
assert "pdf<&>" not in where
assert "fileid" in properties
def test_build_search_xml_emits_offset_only_when_set(mocker):
client = _make_client(mocker)
paged = client._build_search_xml(
scope="dir",
where_conditions="",
properties=["fileid"],
order_by=None,
limit=100,
offset=200,
)
assert "<d:nresults>100</d:nresults>" in paged
assert "<d:firstresult>200</d:firstresult>" in paged
unlimited = client._build_search_xml(
scope="dir",
where_conditions="",
properties=["fileid"],
order_by=None,
limit=None,
offset=None,
)
assert "<d:limit>" not in unlimited