From 2d845cb70fc43001c53e48266a38a5f846fd92c8 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Sun, 31 May 2026 19:42:09 +0200 Subject: [PATCH] fix: address PR #814 reviewer follow-ups - gateway_client: guard token cache with a lazy anyio.Lock so concurrent embed calls share one M2M token request instead of racing - status subscriber: distinguish idle fetch timeouts from real broker errors (log + 5s backoff) instead of swallowing all and spinning - nats: warn when the bus URL uses unencrypted transport (non-tls://) - collection_metadata: accept an optional shared httpx client, make TLS verify explicit, document the unauthenticated control-plane contract - replace python -O-stripped asserts with explicit ValueError in the bus status builder and the api metadata source - document why the nil-UUID sentinel point can't collide with content ids Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/app.py | 9 ++- .../embedding/gateway_client.py | 64 +++++++++++-------- .../vector/collection_metadata.py | 53 +++++++++++++-- nextcloud_mcp_server/vector/queue/nats.py | 17 +++++ nextcloud_mcp_server/vector/queue/status.py | 17 ++++- tests/unit/providers/test_gateway_provider.py | 44 +++++++++++++ tests/unit/vector/test_collection_metadata.py | 23 +++++++ tests/unit/vector/test_nats_producer.py | 19 ++++++ 8 files changed, 209 insertions(+), 37 deletions(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index a7de80f8..52e89fa9 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -361,8 +361,13 @@ async def _build_status_subscriber( """ if not (settings.ingest_mode == "external" and settings.status_backend == "bus"): return None, None - assert settings.ingest_bus_url is not None - assert settings.tenant_id is not None + # Defence-in-depth (robust under ``python -O``, which strips asserts): + # __post_init__ already guarantees these when status_backend == "bus". + if settings.ingest_bus_url is None or settings.tenant_id is None: + raise ValueError( + "STATUS_BACKEND=bus requires INGEST_BUS_URL and TENANT_ID " + "(guaranteed by Settings validation)" + ) store = StatusStore(max_size=settings.vector_sync_queue_max_size) subscriber = await NatsStatusSubscriber.connect( url=settings.ingest_bus_url, diff --git a/nextcloud_mcp_server/embedding/gateway_client.py b/nextcloud_mcp_server/embedding/gateway_client.py index 1a4b9d44..1b593498 100644 --- a/nextcloud_mcp_server/embedding/gateway_client.py +++ b/nextcloud_mcp_server/embedding/gateway_client.py @@ -25,6 +25,7 @@ from __future__ import annotations import logging import time +import anyio import httpx from ..providers.openai import OpenAIProvider @@ -63,37 +64,48 @@ class GatewayTokenProvider: self.scope = scope self.timeout = timeout self._cache: tuple[str, float] | None = None # (token, expires_at) + # Serialises the check-then-fetch cycle so concurrent embed calls don't + # each issue a token request (and silently discard all-but-one token). + # Lazy-init: anyio primitives must not be created at import time (trio). + self._lock: anyio.Lock | None = None async def get_token(self, *, force_refresh: bool = False) -> str: - if ( - self._cache is not None - and not force_refresh - and time.time() < self._cache[1] - ): - return self._cache[0] + if self._lock is None: + self._lock = anyio.Lock() + async with self._lock: + # Re-check inside the lock: a concurrent caller may have just + # refreshed the cache while we waited to acquire it. + if ( + self._cache is not None + and not force_refresh + and time.time() < self._cache[1] + ): + return self._cache[0] - data = {"grant_type": "client_credentials"} - if self.scope: - data["scope"] = self.scope + data = {"grant_type": "client_credentials"} + if self.scope: + data["scope"] = self.scope - async with httpx.AsyncClient( - timeout=httpx.Timeout(self.timeout, connect=5.0) - ) as client: - resp = await client.post( - self.token_url, - data=data, - auth=(self.client_id, self.client_secret), + async with httpx.AsyncClient( + timeout=httpx.Timeout(self.timeout, connect=5.0) + ) as client: + resp = await client.post( + self.token_url, + data=data, + auth=(self.client_id, self.client_secret), + ) + resp.raise_for_status() + body = resp.json() + + expires_in = body.get("expires_in", 3600) + self._cache = ( + body["access_token"], + time.time() + expires_in - _EARLY_REFRESH_SECONDS, ) - resp.raise_for_status() - body = resp.json() - - expires_in = body.get("expires_in", 3600) - self._cache = ( - body["access_token"], - time.time() + expires_in - _EARLY_REFRESH_SECONDS, - ) - logger.info("Obtained embedding-gateway M2M token (expires in %ss)", expires_in) - return self._cache[0] + logger.info( + "Obtained embedding-gateway M2M token (expires in %ss)", expires_in + ) + return self._cache[0] class GatewayProvider(OpenAIProvider): diff --git a/nextcloud_mcp_server/vector/collection_metadata.py b/nextcloud_mcp_server/vector/collection_metadata.py index 9d8f7f4c..d8acb835 100644 --- a/nextcloud_mcp_server/vector/collection_metadata.py +++ b/nextcloud_mcp_server/vector/collection_metadata.py @@ -31,6 +31,9 @@ logger = logging.getLogger(__name__) # Deterministic sentinel point id (design §10.1). Carries collection metadata # and never matches a search (no user_id/doc_id/doc_type payload to match). +# The nil UUID is used deliberately: real chunk point ids are UUID5s derived +# from (namespace, doc_id, chunk_index), so the all-zero id can never collide +# with a content point — and it is trivially recognisable in Qdrant's UI. SENTINEL_POINT_ID = "00000000-0000-0000-0000-000000000000" # Sentinel payload keys. @@ -112,29 +115,67 @@ async def _read_from_qdrant( } -async def _read_from_api(api_url: str, collection_name: str) -> dict[str, Any] | None: +async def _read_from_api( + api_url: str, + collection_name: str, + *, + client: httpx.AsyncClient | None = None, +) -> dict[str, Any] | None: + """GET the control-plane metadata for a collection. + + ``client`` lets the caller pass a shared, lifespan-managed + :class:`httpx.AsyncClient`; when omitted we create a short-lived one. + + Auth note: the control-plane metadata endpoint is currently unauthenticated + (read-only collection identity, no tenant secrets), matching the gateway's + present state. When the control plane gains M2M auth this should reuse the + same OIDC credentials as ``GatewayTokenProvider`` (design §10.2). + """ url = f"{api_url.rstrip('/')}/v1/qdrant-collections/{collection_name}/metadata" - async with httpx.AsyncClient(timeout=httpx.Timeout(10.0, connect=5.0)) as client: - resp = await client.get(url) + + async def _do(c: httpx.AsyncClient) -> dict[str, Any] | None: + resp = await c.get(url) if resp.status_code == 404: return None resp.raise_for_status() return resp.json() + if client is not None: + return await _do(client) + # httpx verifies TLS by default (verify=True); stated here to be explicit. + async with httpx.AsyncClient( + timeout=httpx.Timeout(10.0, connect=5.0), verify=True + ) as owned: + return await _do(owned) + async def read_collection_metadata( client: AsyncQdrantClient, collection_name: str, settings: Settings | None = None, + *, + http_client: httpx.AsyncClient | None = None, ) -> dict[str, Any]: """Read collection metadata from the configured source, falling back to env - defaults on any miss/error (preserves query availability — §10.1).""" + defaults on any miss/error (preserves query availability — §10.1). + + ``http_client`` is forwarded to the ``api`` source so callers on the query + path can share a lifespan-managed client instead of opening one per read. + """ s = settings or get_settings() meta: dict[str, Any] | None = None try: if s.collection_metadata_source == "api": - assert s.collection_metadata_api_url is not None - meta = await _read_from_api(s.collection_metadata_api_url, collection_name) + # Defence-in-depth (robust under ``python -O``): __post_init__ + # guarantees the URL when COLLECTION_METADATA_SOURCE=api. + if s.collection_metadata_api_url is None: + raise ValueError( + "COLLECTION_METADATA_SOURCE=api requires " + "COLLECTION_METADATA_API_URL" + ) + meta = await _read_from_api( + s.collection_metadata_api_url, collection_name, client=http_client + ) else: meta = await _read_from_qdrant(client, collection_name) except Exception: diff --git a/nextcloud_mcp_server/vector/queue/nats.py b/nextcloud_mcp_server/vector/queue/nats.py index 7a5ecafe..b096f3cf 100644 --- a/nextcloud_mcp_server/vector/queue/nats.py +++ b/nextcloud_mcp_server/vector/queue/nats.py @@ -29,6 +29,22 @@ STREAM_NAME = "mcp" INGEST_SUBJECT_PREFIX = "mcp.ingest.requested" +def warn_if_insecure_nats_url(url: str) -> None: + """Log a warning when the bus URL is not TLS-encrypted. + + ``nats://`` (and ``ws://``) carry tenant document metadata in cleartext; + production deployments should use ``tls://`` (or ``wss://``). We connect + regardless — this is an operator alert, not a hard failure. + """ + scheme = url.split("://", 1)[0].lower() + if scheme not in ("tls", "wss"): + logger.warning( + "NATS bus URL uses unencrypted transport (scheme=%s://); " + "use tls:// in production to protect document metadata in transit", + scheme, + ) + + def _modified_at_rfc3339(modified_at: int) -> str: """DocumentTask.modified_at is an epoch int (0 for deletes).""" return datetime.fromtimestamp(int(modified_at), tz=timezone.utc).isoformat() @@ -74,6 +90,7 @@ class NatsTaskProducer: ) -> NatsTaskProducer: import nats # noqa: PLC0415 (lazy: optional dependency for external mode) + warn_if_insecure_nats_url(url) nc = await nats.connect(url) js = nc.jetstream() await cls._ensure_stream(js, num_replicas) diff --git a/nextcloud_mcp_server/vector/queue/status.py b/nextcloud_mcp_server/vector/queue/status.py index 724e805f..ede66f23 100644 --- a/nextcloud_mcp_server/vector/queue/status.py +++ b/nextcloud_mcp_server/vector/queue/status.py @@ -126,6 +126,9 @@ class NatsStatusSubscriber: ) -> NatsStatusSubscriber: import nats # noqa: PLC0415 + from .nats import warn_if_insecure_nats_url # noqa: PLC0415 + + warn_if_insecure_nats_url(url) nc = await nats.connect(url) js = nc.jetstream() return cls(nc, js, tenant_id, store) @@ -138,6 +141,7 @@ class NatsStatusSubscriber: ) -> None: """Durable pull-consumer loop. Requires a live broker (integration).""" import anyio # noqa: PLC0415 + import nats.errors # noqa: PLC0415 subject = f"mcp.document.*.{self.tenant_id}" sub = await self._js.pull_subscribe( @@ -148,10 +152,17 @@ class NatsStatusSubscriber: while not shutdown_event.is_set(): try: msgs = await sub.fetch(batch=16, timeout=5) + except nats.errors.TimeoutError: + # Expected when idle: no messages within the fetch window. Loop + # straight back to re-check shutdown — no log, no extra sleep. + continue except Exception: - # fetch timeout when idle — brief pause before re-checking - # shutdown, avoiding a tight check-and-sleep spin on quiet tenants. - await anyio.sleep(0.1) + # Real broker error (disconnect, auth failure, stream deleted). + # Log it and back off so we don't hot-spin against a dead broker. + logger.warning( + "NATS status subscriber fetch failed; retrying", exc_info=True + ) + await anyio.sleep(5) continue for msg in msgs: self.handle_message(msg.subject, msg.data) diff --git a/tests/unit/providers/test_gateway_provider.py b/tests/unit/providers/test_gateway_provider.py index fbefa0f7..76b53563 100644 --- a/tests/unit/providers/test_gateway_provider.py +++ b/tests/unit/providers/test_gateway_provider.py @@ -116,3 +116,47 @@ async def test_token_provider_caches_and_refreshes(monkeypatch): t3 = await tp.get_token() assert t3 == "tok2" assert calls["n"] == 2 + + +async def test_token_provider_concurrent_callers_issue_single_request(monkeypatch): + """Two concurrent get_token() calls must share one token request, not race.""" + import anyio + + calls = {"n": 0} + + async def handler(request: httpx.Request) -> httpx.Response: + calls["n"] += 1 + # Hold the "network" open so a second caller arrives mid-flight. + await anyio.sleep(0.05) + return httpx.Response( + 200, json={"access_token": f"tok{calls['n']}", "expires_in": 3600} + ) + + transport = httpx.MockTransport(handler) + orig_async_client = httpx.AsyncClient + + def _client(*args, **kwargs): + kwargs["transport"] = transport + return orig_async_client(*args, **kwargs) + + monkeypatch.setattr(httpx, "AsyncClient", _client) + + tp = GatewayTokenProvider( + token_url="https://idp.example/oauth2/token", + client_id="cid", + client_secret="sec", + ) + + results: list[str] = [] + + async def _fetch(): + results.append(await tp.get_token()) + + async with anyio.create_task_group() as tg: + tg.start_soon(_fetch) + tg.start_soon(_fetch) + + # The lock serialises the check-then-fetch cycle: only one HTTP request, + # and both callers observe the same cached token. + assert calls["n"] == 1 + assert results == ["tok1", "tok1"] diff --git a/tests/unit/vector/test_collection_metadata.py b/tests/unit/vector/test_collection_metadata.py index 3d7baabd..d3e8ebce 100644 --- a/tests/unit/vector/test_collection_metadata.py +++ b/tests/unit/vector/test_collection_metadata.py @@ -70,6 +70,29 @@ async def test_api_source(mocker): assert meta["embedding_identity"] == "amazon.titan-embed-text-v2:0" +async def test_api_source_uses_shared_http_client(): + """When a caller passes http_client, the read reuses it (no own client).""" + + def handler(request: httpx.Request) -> httpx.Response: + assert request.url.path == "/v1/qdrant-collections/col/metadata" + return httpx.Response(200, json={"embedding_identity": "mistral-embed"}) + + shared = httpx.AsyncClient(transport=httpx.MockTransport(handler)) + async with shared: + meta = await cm._read_from_api("https://cp", "col", client=shared) + assert meta["embedding_identity"] == "mistral-embed" + + +async def test_api_missing_url_falls_back_to_env(mocker): + """COLLECTION_METADATA_SOURCE=api with no URL must not crash the query path.""" + # Bypass Settings validation to simulate a -O / mutated-state edge case. + settings = Settings() + settings.collection_metadata_source = "api" + settings.collection_metadata_api_url = None + meta = await cm.read_collection_metadata(mocker.AsyncMock(), "col", settings) + assert meta == cm.env_default_metadata(settings) + + async def test_upsert_sentinel_builds_point(mocker): client = mocker.AsyncMock() await cm.upsert_sentinel( diff --git a/tests/unit/vector/test_nats_producer.py b/tests/unit/vector/test_nats_producer.py index 0f51f7f4..5c29cb46 100644 --- a/tests/unit/vector/test_nats_producer.py +++ b/tests/unit/vector/test_nats_producer.py @@ -12,6 +12,7 @@ from nextcloud_mcp_server.vector.queue.nats import ( NatsTaskProducer, _modified_at_rfc3339, msg_id, + warn_if_insecure_nats_url, ) from nextcloud_mcp_server.vector.queue.postgres import PostgresTaskProducer from nextcloud_mcp_server.vector.scanner import DocumentTask @@ -134,3 +135,21 @@ def test_transport_for(url, expected): async def test_postgres_producer_is_a_seam(): with pytest.raises(NotImplementedError, match="documented seam"): await PostgresTaskProducer.connect(object()) + + +@pytest.mark.parametrize( + "url,should_warn", + [ + ("nats://nats:4222", True), + ("ws://nats:8080", True), + ("tls://nats:4222", False), + ("wss://nats:8080", False), + ], +) +def test_warn_if_insecure_nats_url(url, should_warn, caplog): + import logging + + with caplog.at_level(logging.WARNING): + warn_if_insecure_nats_url(url) + warned = any("unencrypted transport" in r.getMessage() for r in caplog.records) + assert warned is should_warn