fix: address PR #836 round-3 review (lock-key invariant, single open)

🟡 Document the _doc_queueing_lock ":" delimiter invariant (user_id and the
   controlled doc_type enum are colon-free, so the key is collision-safe; a
   future doc_type with ":" must not be added).
🟡 API pod no longer opens the procrastinate connector twice on startup: add
   ProcrastinateTaskProducer.ensure_schema() (applies the schema on the
   already-open pool) and have both lifespan branches build the producer then
   ensure_schema — one open/close cycle, matching the worker. build_producer now
   returns the concrete producer type.
🟢 Document in ports.py that a long-lived-connection producer may optionally
   provide drain() (lifespan probes via getattr).
🟢 Add a unit test that a non-credential pipeline error propagates (for
   procrastinate's RetryStrategy) and still closes the client via finally.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

nextcloud_mcp_server/app.py

@@ -1676,14 +1676,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
             receive_stream = None
             task_producer: TaskProducer
             if use_postgres:
-                # Create procrastinate's tables before the scanner can defer.
+                # Open the connector once (build_producer) and reuse it to create
-                # Lazy import: the procrastinate lib is a Postgres-only extra.
+                # procrastinate's tables before the scanner can defer — a single
-                from nextcloud_mcp_server.vector.queue.procrastinate import (  # noqa: PLC0415
+                # open/close cycle, matching the worker command.
-                    apply_ingest_queue_schema,
+                producer = await build_producer(settings)
-                )
+                await producer.ensure_schema()
-
+                task_producer = producer
-                await apply_ingest_queue_schema()
-                task_producer = await build_producer(settings)
                 logger.info("Ingest queue: postgres (procrastinate); worker drains it")
             else:
                 send_stream, receive_stream = anyio.create_memory_object_stream[
@@ -1895,14 +1893,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
                 receive_stream = None
                 task_producer: TaskProducer
                 if use_postgres:
-                    # Create procrastinate's tables before any scanner defers.
+                    # Single open/close cycle: build_producer opens the connector
-                    # Lazy import: procrastinate is a Postgres-only extra.
+                    # and ensure_schema reuses it to create procrastinate's tables
-                    from nextcloud_mcp_server.vector.queue.procrastinate import (  # noqa: PLC0415
+                    # before any scanner defers (matches the worker command).
-                        apply_ingest_queue_schema,
+                    producer = await build_producer(settings)
-                    )
+                    await producer.ensure_schema()
-
+                    task_producer = producer
-                    await apply_ingest_queue_schema()
-                    task_producer = await build_producer(settings)
                     logger.info(
                         "Ingest queue: postgres (procrastinate); worker drains it"
                     )

nextcloud_mcp_server/vector/queue/factory.py

@@ -12,19 +12,24 @@ The transport is selected from ``INGEST_QUEUE``:
 from __future__ import annotations
 
 import logging
+from typing import TYPE_CHECKING
 
 from ...config import Settings
-from .ports import TaskProducer
+
+if TYPE_CHECKING:
+    from .procrastinate import ProcrastinateTaskProducer
 
 logger = logging.getLogger(__name__)
 
 
-async def build_producer(settings: Settings) -> TaskProducer:
+async def build_producer(settings: Settings) -> ProcrastinateTaskProducer:
     """Build the Postgres (procrastinate) ingest producer.
 
-    Precondition: ``settings.ingest_queue == "postgres"`` (the memory transport
+    Returns the concrete :class:`ProcrastinateTaskProducer` (not just the
-    is constructed inline by the lifespan because it needs the paired receive
+    ``TaskProducer`` protocol) so the lifespan can call ``ensure_schema()`` on
-    stream for the in-process processor pool).
+    the open connector. Precondition: ``settings.ingest_queue == "postgres"``
+    (the memory transport is constructed inline by the lifespan because it needs
+    the paired receive stream for the in-process processor pool).
     """
     if settings.ingest_queue != "postgres":
         raise ValueError(

nextcloud_mcp_server/vector/queue/ports.py

@@ -52,8 +52,14 @@ class TaskProducer(Protocol):
     async def aclose(self) -> None:
         """Close *this* handle (e.g. a per-user clone when its scanner exits).
 
-        For the memory stream this closes the clone; for the shared bus
+        For the memory stream this closes the clone; for a shared connection it
-        connection it is a no-op (the connection is owned by the lifespan,
+        is a no-op (the connection is owned by the lifespan, which tears it down
-        which drains it once on shutdown).
+        once on shutdown).
         """
         ...
+
+    # Note: this protocol deliberately omits ``drain()``. An implementation that
+    # owns a long-lived shared connection (e.g. ProcrastinateTaskProducer's
+    # connector pool) may additionally provide ``async def drain()`` for the
+    # lifespan to close that pool once on shutdown; the lifespan probes for it
+    # with ``getattr(task_producer, "drain", None)``, so it stays optional.

nextcloud_mcp_server/vector/queue/procrastinate.py

@@ -290,7 +290,15 @@ async def get_ingest_job_counts(app: App | None = None) -> dict[str, int]:
 
 
 def _doc_queueing_lock(task: DocumentTask) -> str:
-    """Per-document enqueue-dedup key (partial-unique on ``status='todo'``)."""
+    """Per-document enqueue-dedup key (partial-unique on ``status='todo'``).
+
+    Collision-safe with a raw ``:`` delimiter because the first two segments can
+    never themselves contain ``:``: ``user_id`` is a Nextcloud username/UID (no
+    colons) and ``doc_type`` is a controlled enum (``note``/``file``/
+    ``deck_card``/``news_item``). The trailing ``doc_id`` may contain anything —
+    it's the final unambiguous segment. A future ``doc_type`` containing ``:``
+    would break this invariant, so keep doc_type colon-free.
+    """
     return f"{task.user_id}:{task.doc_type}:{task.doc_id}"
 
 
@@ -328,6 +336,15 @@ class ProcrastinateTaskProducer:
             # advances after a successful index), so this is not a lost update.
             logger.debug("ingest.already_enqueued key=%s", key)
 
+    async def ensure_schema(self) -> None:
+        """Apply the ingest-queue schema on the producer's already-open pool.
+
+        Lets the API lifespan provision the schema without a second open/close
+        cycle (it already opened the connector to build this producer) — the
+        ``worker`` command shares the same single-open pattern.
+        """
+        await _apply_ingest_queue_schema_open(self._app)
+
     async def job_counts(self) -> dict[str, int]:
         """Ingest job counts by status (for the vector-sync status surface)."""
         return await get_ingest_job_counts(self._app)

tests/unit/vector/test_procrastinate_producer.py

@@ -122,6 +122,32 @@ class TestProcessDocumentTask:
         assert captured["max_retries"] == 1
         fake_client.close.assert_awaited_once()
 
+    async def test_pipeline_error_propagates_and_closes_client(self, monkeypatch):
+        # A non-credential failure must propagate (so procrastinate's
+        # RetryStrategy picks it up) and still close the client via finally.
+        fake_client = AsyncMock()
+
+        async def fake_resolve(user_id):
+            return fake_client
+
+        async def fake_process(task, nc_client, *, max_retries):
+            raise RuntimeError("transient qdrant failure")
+
+        monkeypatch.setattr(pq, "_resolve_client", fake_resolve)
+        monkeypatch.setattr(
+            "nextcloud_mcp_server.vector.processor.process_document", fake_process
+        )
+
+        with pytest.raises(RuntimeError, match="transient qdrant failure"):
+            await pq.process_document_task(
+                user_id="alice",
+                doc_id="42",
+                doc_type="note",
+                operation="index",
+                modified_at=100,
+            )
+        fake_client.close.assert_awaited_once()
+
     async def test_skips_on_missing_credentials(self, monkeypatch):
         from nextcloud_mcp_server.vector.oauth_sync import NotProvisionedError