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) <noreply@anthropic.com>

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
+    # Defence-in-depth (robust under ``python -O``, which strips asserts):
-    assert settings.tenant_id is not None
+    # __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,

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,8 +64,17 @@ 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._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
@@ -92,7 +102,9 @@ class GatewayTokenProvider:
                 body["access_token"],
                 time.time() + expires_in - _EARLY_REFRESH_SECONDS,
             )
-        logger.info("Obtained embedding-gateway M2M token (expires in %ss)", expires_in)
+            logger.info(
+                "Obtained embedding-gateway M2M token (expires in %ss)", expires_in
+            )
             return self._cache[0]
 
 

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
+            # Defence-in-depth (robust under ``python -O``): __post_init__
-            meta = await _read_from_api(s.collection_metadata_api_url, collection_name)
+            # 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:

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)

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
+                # Real broker error (disconnect, auth failure, stream deleted).
-                # shutdown, avoiding a tight check-and-sleep spin on quiet tenants.
+                # Log it and back off so we don't hot-spin against a dead broker.
-                await anyio.sleep(0.1)
+                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)

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"]

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(

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