From b10ce150322d20cae71e2de8a2db063b0cf0d000 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 3 Jun 2026 15:32:16 +0200 Subject: [PATCH] fix: address PR #836 round-3 review (lock-key invariant, single open) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🟡 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) --- nextcloud_mcp_server/app.py | 28 ++++++++----------- nextcloud_mcp_server/vector/queue/factory.py | 15 ++++++---- nextcloud_mcp_server/vector/queue/ports.py | 12 ++++++-- .../vector/queue/procrastinate.py | 19 ++++++++++++- .../vector/test_procrastinate_producer.py | 26 +++++++++++++++++ 5 files changed, 75 insertions(+), 25 deletions(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 92e64467..09d6568d 100644 --- a/nextcloud_mcp_server/app.py +++ b/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. - # Lazy import: the procrastinate lib is a Postgres-only extra. - from nextcloud_mcp_server.vector.queue.procrastinate import ( # noqa: PLC0415 - apply_ingest_queue_schema, - ) - - await apply_ingest_queue_schema() - task_producer = await build_producer(settings) + # Open the connector once (build_producer) and reuse it to create + # procrastinate's tables before the scanner can defer — a single + # open/close cycle, matching the worker command. + producer = await build_producer(settings) + await producer.ensure_schema() + task_producer = producer 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. - # Lazy import: procrastinate is a Postgres-only extra. - from nextcloud_mcp_server.vector.queue.procrastinate import ( # noqa: PLC0415 - apply_ingest_queue_schema, - ) - - await apply_ingest_queue_schema() - task_producer = await build_producer(settings) + # Single open/close cycle: build_producer opens the connector + # and ensure_schema reuses it to create procrastinate's tables + # before any scanner defers (matches the worker command). + producer = await build_producer(settings) + await producer.ensure_schema() + task_producer = producer logger.info( "Ingest queue: postgres (procrastinate); worker drains it" ) diff --git a/nextcloud_mcp_server/vector/queue/factory.py b/nextcloud_mcp_server/vector/queue/factory.py index 4628f959..6a0cab91 100644 --- a/nextcloud_mcp_server/vector/queue/factory.py +++ b/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 - is constructed inline by the lifespan because it needs the paired receive - stream for the in-process processor pool). + Returns the concrete :class:`ProcrastinateTaskProducer` (not just the + ``TaskProducer`` protocol) so the lifespan can call ``ensure_schema()`` on + 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( diff --git a/nextcloud_mcp_server/vector/queue/ports.py b/nextcloud_mcp_server/vector/queue/ports.py index 39855e96..09bfd932 100644 --- a/nextcloud_mcp_server/vector/queue/ports.py +++ b/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 - connection it is a no-op (the connection is owned by the lifespan, - which drains it once on shutdown). + For the memory stream this closes the clone; for a shared connection it + is a no-op (the connection is owned by the lifespan, which tears it down + 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. diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index e25c5a2a..049f3eaa 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/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) diff --git a/tests/unit/vector/test_procrastinate_producer.py b/tests/unit/vector/test_procrastinate_producer.py index 8960d951..f472044a 100644 --- a/tests/unit/vector/test_procrastinate_producer.py +++ b/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