From e7bcdb1950ff31d7204b18d88aa59a8cff778698 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 19:43:31 +0200 Subject: [PATCH 1/7] feat: add IngestTransport port for local/distributed ingest backends Finish the hexagonal ports-&-adapters split started in #183. The producer side already had a TaskProducer port + adapters, but the consumer side was unabstracted and the INGEST_QUEUE selection leaked into a duplicated `if use_postgres:` branch across both app.py lifespan paths. Introduce an IngestTransport ABC (vector/queue/transport.py) that bundles the producer with running (or not running) the in-process consumer pool, built by a single build_transport() factory: - LocalTransport (INGEST_QUEUE=memory): in-process anyio stream drained by an N-worker pool that run_consumers starts. - DistributedTransport (INGEST_QUEUE=postgres): wraps ProcrastinateTaskProducer; run_consumers is a no-op because the consumer is the external `worker` role. Both lifespan paths now call build_transport + _wire_vector_sync_state (new helper that centralizes the app.state / module-singleton / browser-app writes) + transport.run_consumers + transport.aclose(), with no INGEST_QUEUE branching and no getattr drain probe. Adding a future backend (Redis/NATS/SQS) is one new adapter + one build_transport arm, with no app.py or scanner change. Preserves the single-tenant parallelism invariant (one shared multiplexed queue + N-worker pool, per-document not per-user dispatch) and documents it in ADR-028. The worker CLI is unchanged (it is the external consumer). Refs: Deck #196 (Deck #197 tracks the explicit parallelism regression test) Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/ADR-028-ingest-transport-port.md | 124 ++++++++ nextcloud_mcp_server/app.py | 272 +++++++++--------- nextcloud_mcp_server/cli.py | 5 + nextcloud_mcp_server/vector/queue/__init__.py | 18 +- .../vector/queue/transport.py | 202 +++++++++++++ tests/unit/vector/test_ingest_transport.py | 114 ++++++++ 6 files changed, 593 insertions(+), 142 deletions(-) create mode 100644 docs/ADR-028-ingest-transport-port.md create mode 100644 nextcloud_mcp_server/vector/queue/transport.py create mode 100644 tests/unit/vector/test_ingest_transport.py diff --git a/docs/ADR-028-ingest-transport-port.md b/docs/ADR-028-ingest-transport-port.md new file mode 100644 index 00000000..544b22c2 --- /dev/null +++ b/docs/ADR-028-ingest-transport-port.md @@ -0,0 +1,124 @@ +# ADR-028: Ingest transport port (local anyio vs distributed procrastinate) + +## Status + +Accepted — 2026-06-04 + +## Context + +Document ingest (scanner/webhook → fetch → chunk → embed → upsert to Qdrant) +runs in one of two modes, selected by `INGEST_QUEUE`: + +- `memory` (the SQLite/dev default): an in-process anyio + `MemoryObjectStream` drained by a pool of in-process worker tasks. +- `postgres`: jobs are deferred into the per-tenant Postgres via + [procrastinate](https://procrastinate.readthedocs.io/) and drained by a + *separate* `nextcloud-mcp-server worker` process (the scale-to-zero + api/worker split — KEDA scales the worker Deployment on queue depth). + +ADR-007 introduced the in-process model; Deck #183 / PR #836 added the +postgres backend and a `TaskProducer` **Protocol** (`vector/queue/ports.py`) +so the scanner and webhook receiver send a `DocumentTask` to a backend-agnostic +sink — `MemoryTaskProducer` (anyio) or `ProcrastinateTaskProducer` (Postgres). + +That abstracted the **producer** side, but left two gaps: + +1. **The consumer side was not abstracted.** Memory mode spun up an in-process + pool inside the server lifespan; postgres mode relied on the external worker + CLI. Nothing tied these together. +2. **Mode selection leaked into the lifespan.** `app.py` carried a duplicated + `if use_postgres: build producer + ensure_schema … else: create stream …` + branch, plus a conditional N-worker startup and a + `getattr(task_producer, "drain", None)` shutdown probe — repeated across the + two near-identical lifespan paths (single-user BasicAuth and multi-user + OAuth/BasicAuth). Adding a third backend (Redis/NATS/SQS) would have meant + editing both blocks. + +## Decision + +Introduce an `IngestTransport` abstraction that owns **both** sides of one +ingest backend — the producer to wire into `app.state`/the scanner, and how (or +whether) the in-process consumer pool runs — built by a single +`build_transport(settings)` factory. + +``` +build_transport(settings) -> + INGEST_QUEUE=postgres -> DistributedTransport(build_producer(settings)) + INGEST_QUEUE=memory -> LocalTransport(vector_sync_queue_max_size) +``` + +`IngestTransport` (`vector/queue/transport.py`) exposes: + +- `producer` — the `TaskProducer` to wire into `app.state` / hand to the scanner. +- `send_stream` / `receive_stream` — the raw anyio stream ends in memory mode, + `None` for distributed backends (the latter keeps `ingest_status` queue-depth + and the integration conftest's stream-singleton handling working unchanged). +- `run_consumers(task_group, spawn_worker, count)` — start the in-process pool; + a **no-op** for distributed backends, whose consumer is the external worker. +- `aclose()` — tear down backend-owned resources once on shutdown (closes the + procrastinate connector pool; a no-op for the memory stream, which task-group + cancellation closes). + +The lifespan supplies a `spawn_worker` closure so the transport never learns +about auth modes — the single-user closure binds a shared `nc_client`+username, +the multi-user closure binds the Nextcloud host for per-document credential +resolution. Both forward anyio's injected `task_status` so `tg.start` observes +each worker's readiness. + +### Why an ABC for the transport but a Protocol for the producer + +`TaskProducer` is a `Protocol` specifically so anyio's third-party +`MemoryObjectSendStream` satisfies it structurally. The transport has exactly +two in-house adapters that share the `producer` storage and the +`receive_stream`/`run_consumers`/`aclose` defaults, so a concrete `abc.ABC` is +simpler, gives shared default implementations, and checks more cleanly under +`ty`. We keep `TaskProducer`/`build_producer` unchanged; the transport *wraps* a +producer. + +### No consumer port + +Deliberately, there is no consumer *port* (mirroring `ports.py`): in memory mode +the in-process pool is the consumer; in postgres mode the external worker is. +The worker CLI (`cli.py worker`) talks to procrastinate's `App` directly +(`run_worker_async`) — a different control surface from the in-process pool — so +it does not route through `IngestTransport`; `DistributedTransport.run_consumers` +is a no-op precisely because that separate process is the consumer. + +### Single-tenant parallelism invariant + +A single tenant must process its users' files **in parallel**, never one user +fully then the next. This holds by construction and is documented here as a +contract: + +- **Local backend:** `LocalTransport.run_consumers` hands each of N workers + (`VECTOR_SYNC_PROCESSOR_WORKERS`, default 3) an independent `clone()` of *one* + shared receive stream. All users' `DocumentTask`s are multiplexed onto that + single queue and dispatched **per-document**, so N documents — from any mix of + users — are in flight at once. +- **Distributed backend:** the worker runs `run_worker_async(concurrency=N)` + (default N = `VECTOR_SYNC_PROCESSOR_WORKERS`) over the single `ingest` queue, + and procrastinate's only lock is a per-**document** `queueing_lock` + (`user_id:doc_type:doc_id`) — there is no per-user lock — so different users' + jobs run concurrently across worker slots and pods. + +An explicit anyio overlap test for this invariant is tracked as a follow-up +(Deck #197). + +## Consequences + +- The two `app.py` lifespan paths are backend-agnostic: `build_transport()` + + `_wire_vector_sync_state()` + `transport.run_consumers()` + + `transport.aclose()`, with no `INGEST_QUEUE` branching and no `getattr` drain + probe. +- Adding a future queue backend is one new `IngestTransport` adapter + one + `build_transport` arm — no change to `app.py`, the scanner, or the webhook + receiver. +- `app.state.task_producer` / `_vector_sync_state.task_producer` and the queue + depth surface (`ingest_status.py`) keep their existing contracts. + +## References + +- ADR-007 — Background vector database synchronization (in-process anyio model) +- ADR-010 — Webhook-based vector database synchronization +- Deck #183 / PR #836 — procrastinate Postgres ingest queue + `TaskProducer` port +- Deck #196 — this work; Deck #197 — explicit parallelism test follow-up diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 09d6568d..4145acbb 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -9,7 +9,7 @@ import traceback from collections.abc import AsyncIterator from contextlib import AsyncExitStack, asynccontextmanager from dataclasses import dataclass -from typing import cast +from typing import Any, cast from urllib.parse import urlparse import anyio @@ -134,11 +134,11 @@ from nextcloud_mcp_server.vector.placeholder import sweep_orphan_placeholders from nextcloud_mcp_server.vector.processor import processor_task from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client from nextcloud_mcp_server.vector.queue import ( - MemoryTaskProducer, + IngestTransport, TaskProducer, - build_producer, + build_transport, ) -from nextcloud_mcp_server.vector.scanner import DocumentTask, scanner_task +from nextcloud_mcp_server.vector.scanner import scanner_task from nextcloud_mcp_server.vector.webhook_receiver import handle_nextcloud_webhook logger = logging.getLogger(__name__) @@ -347,6 +347,47 @@ class VectorSyncState: _vector_sync_state = VectorSyncState() +def _wire_vector_sync_state( + app: Starlette, + transport: IngestTransport, + shutdown_event: anyio.Event, + scanner_wake_event: anyio.Event, +) -> None: + """Publish the ingest transport + sync events to every state surface. + + Both lifespan paths (single-user and multi-user) previously duplicated these + writes three times each: ``app.state``, the module singleton + ``_vector_sync_state`` (read by FastMCP session lifespans), and the mounted + ``/app`` browser sub-app. ``document_send_stream``/``document_receive_stream`` + come from the transport — ``None`` in postgres mode (no in-process stream) — + and ``task_producer`` is the transport's producer in both modes (Deck #183, + ADR-028). + """ + send_stream = transport.send_stream + receive_stream = transport.receive_stream + task_producer = transport.producer + + def _apply(state: Any) -> None: + state.document_send_stream = send_stream + state.document_receive_stream = receive_stream + state.task_producer = task_producer + state.shutdown_event = shutdown_event + state.scanner_wake_event = scanner_wake_event + + # app.state (Starlette) + the module singleton share the same attribute names. + _apply(app.state) + _apply(_vector_sync_state) + logger.info("Vector sync state stored in module singleton") + + # Also share with the mounted /app browser sub-app, if present. + for route in app.routes: + if isinstance(route, Mount) and route.path == "/app": + browser_app = cast(Starlette, route.app) + _apply(browser_app.state) + logger.info("Vector sync state shared with browser_app for /app") + break + + @dataclass class AppContext: """Application context for BasicAuth mode.""" @@ -1663,60 +1704,21 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Orphan-sweep before scanner starts — card #101. await _sweep_orphan_placeholders_if_enabled() - # Initialize shared state. INGEST_QUEUE selects the transport - # (Deck #183): ``memory`` uses the in-process anyio stream + the - # in-process processor pool (SQLite/dev); ``postgres`` defers jobs to - # the per-tenant Postgres via procrastinate and runs no in-process - # consumer (the separate ``worker`` role drains the queue). - use_postgres = settings.ingest_queue == "postgres" + # Initialize the ingest transport. INGEST_QUEUE selects the backend + # (Deck #183, ADR-028): ``memory`` builds an in-process anyio stream + # drained by an in-process pool (SQLite/dev); ``postgres`` defers jobs + # to the per-tenant Postgres via procrastinate and runs no in-process + # consumer (the separate ``worker`` role drains the queue). The + # transport hides that choice — this path is now backend-agnostic. shutdown_event = anyio.Event() scanner_wake_event = anyio.Event() - send_stream = None - receive_stream = None - task_producer: TaskProducer - if use_postgres: - # 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[ - DocumentTask - ](max_buffer_size=settings.vector_sync_queue_max_size) - task_producer = MemoryTaskProducer(send_stream) + transport = await build_transport(settings) + task_producer = transport.producer - # Store in app state for access from routes (ADR-007). In postgres - # mode there is no in-memory stream, so document_send/receive_stream - # stay None; task_producer is the procrastinate producer. - app.state.document_send_stream = send_stream - app.state.document_receive_stream = receive_stream - app.state.task_producer = task_producer - app.state.shutdown_event = shutdown_event - app.state.scanner_wake_event = scanner_wake_event - - # Also store in module singleton for FastMCP session lifespans - _vector_sync_state.document_send_stream = send_stream - _vector_sync_state.document_receive_stream = receive_stream - _vector_sync_state.task_producer = task_producer - _vector_sync_state.shutdown_event = shutdown_event - _vector_sync_state.scanner_wake_event = scanner_wake_event - logger.info("Vector sync state stored in module singleton") - - # Also share with browser_app for /app route - for route in app.routes: - if isinstance(route, Mount) and route.path == "/app": - browser_app = cast(Starlette, route.app) - browser_app.state.document_send_stream = send_stream - browser_app.state.document_receive_stream = receive_stream - browser_app.state.task_producer = task_producer - browser_app.state.shutdown_event = shutdown_event - browser_app.state.scanner_wake_event = scanner_wake_event - logger.info("Vector sync state shared with browser_app for /app") - break + # Publish to app.state (ADR-007), the module singleton (FastMCP + # session lifespans), and the /app browser sub-app in one place. + _wire_vector_sync_state(app, transport, shutdown_event, scanner_wake_event) # Start background tasks using anyio TaskGroup async with anyio.create_task_group() as tg: @@ -1730,19 +1732,28 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = username, ) - # The in-process processor pool runs only in memory mode; in - # postgres mode the out-of-process worker is the consumer. - if not use_postgres: - assert receive_stream is not None - for i in range(settings.vector_sync_processor_workers): - await tg.start( - processor_task, - i, - receive_stream.clone(), - shutdown_event, - client, - username, - ) + # Start the in-process consumer pool. ``run_consumers`` is a + # no-op for the distributed (postgres) backend — its consumer is + # the out-of-process ``worker`` role. The closure binds this + # mode's shared client+username and forwards anyio's injected + # ``task_status`` so ``tg.start`` observes each worker's + # readiness. One shared receive stream + N workers ⇒ a single + # multiplexed queue processed with N-way parallelism (ADR-028). + async def spawn_worker( + worker_id, receive_stream, *, task_status=anyio.TASK_STATUS_IGNORED + ): + await processor_task( + worker_id, + receive_stream, + shutdown_event, + client, + username, + task_status=task_status, + ) + + await transport.run_consumers( + tg, spawn_worker, settings.vector_sync_processor_workers + ) # Expose this long-lived task group to request-path code that # wants to spawn background work (e.g. ADR-019 verify-on-read @@ -1752,7 +1763,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info( "Background sync tasks started: 1 scanner + %s processors (queue=%s)", - 0 if use_postgres else settings.vector_sync_processor_workers, + 0 + if settings.ingest_queue == "postgres" + else settings.vector_sync_processor_workers, settings.ingest_queue, ) @@ -1766,10 +1779,11 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = shutdown_event.set() # Request path must not spawn into a cancelling group. _vector_sync_state.eviction_task_group = None - # Close the procrastinate connector pool (postgres mode). - _drain = getattr(task_producer, "drain", None) - if use_postgres and _drain is not None: - await _drain() + # Tear down backend-owned resources (closes the + # procrastinate connector pool in postgres mode; no-op + # for the memory stream, which task-group cancellation + # closes). + await transport.aclose() await client.close() # TaskGroup automatically cancels all tasks on exit @@ -1877,65 +1891,26 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = except Exception as e: logger.warning("App password cleanup failed (non-fatal): %s", e) - # Initialize shared state. INGEST_QUEUE selects the transport - # (Deck #183): ``memory`` uses the in-process anyio stream + the - # in-process processor pool; ``postgres`` defers jobs via - # procrastinate and runs no in-process consumer (the separate - # ``worker`` role drains the queue). - use_postgres = settings.ingest_queue == "postgres" + # Initialize the ingest transport. INGEST_QUEUE selects the + # backend (Deck #183, ADR-028): ``memory`` builds an in-process + # anyio stream drained by an in-process pool; ``postgres`` defers + # jobs via procrastinate and runs no in-process consumer (the + # separate ``worker`` role drains the queue). The transport hides + # that choice — this path is now backend-agnostic. shutdown_event = anyio.Event() scanner_wake_event = anyio.Event() # User state tracking for user manager user_states: dict = {} - send_stream = None - receive_stream = None - task_producer: TaskProducer - if use_postgres: - # 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" - ) - else: - send_stream, receive_stream = anyio.create_memory_object_stream[ - DocumentTask - ](max_buffer_size=settings.vector_sync_queue_max_size) - task_producer = MemoryTaskProducer(send_stream) + transport = await build_transport(settings) + task_producer = transport.producer - # Store in app state for access from routes (ADR-007) - app.state.document_send_stream = send_stream - app.state.document_receive_stream = receive_stream - app.state.task_producer = task_producer - app.state.shutdown_event = shutdown_event - app.state.scanner_wake_event = scanner_wake_event - - # Also store in module singleton for FastMCP session lifespans - _vector_sync_state.document_send_stream = send_stream - _vector_sync_state.document_receive_stream = receive_stream - _vector_sync_state.task_producer = task_producer - _vector_sync_state.shutdown_event = shutdown_event - _vector_sync_state.scanner_wake_event = scanner_wake_event - logger.info("Vector sync state stored in module singleton") - - # Also share with browser_app for /app route - for route in app.routes: - if isinstance(route, Mount) and route.path == "/app": - browser_app = cast(Starlette, route.app) - browser_app.state.document_send_stream = send_stream - browser_app.state.document_receive_stream = receive_stream - browser_app.state.task_producer = task_producer - browser_app.state.shutdown_event = shutdown_event - browser_app.state.scanner_wake_event = scanner_wake_event - logger.info( - "Vector sync state shared with browser_app for /app" - ) - break + # Publish to app.state (ADR-007), the module singleton (FastMCP + # session lifespans), and the /app browser sub-app in one place. + _wire_vector_sync_state( + app, transport, shutdown_event, scanner_wake_event + ) # Background sync authenticates as each provisioned user via # locally-stored Nextcloud app passwords (Login Flow v2 / @@ -1960,18 +1935,31 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = tg, ) - # In-process processor pool runs only in memory mode; in - # postgres mode the out-of-process worker consumes. - if not use_postgres: - assert receive_stream is not None - for i in range(settings.vector_sync_processor_workers): - await tg.start( - oauth_processor_task, - i, - receive_stream.clone(), - shutdown_event, - nextcloud_host_for_sync, - ) + # Start the in-process consumer pool. ``run_consumers`` is a + # no-op for the distributed (postgres) backend — the + # out-of-process ``worker`` role consumes there. The closure + # binds this mode's nextcloud_host (per-document credential + # resolution) and forwards anyio's injected ``task_status``. + # One shared receive stream + N workers ⇒ a single + # multiplexed queue draining every user's documents with + # N-way parallelism, never one user at a time (ADR-028). + async def spawn_worker( + worker_id, + receive_stream, + *, + task_status=anyio.TASK_STATUS_IGNORED, + ): + await oauth_processor_task( + worker_id, + receive_stream, + shutdown_event, + nextcloud_host_for_sync, + task_status=task_status, + ) + + await transport.run_consumers( + tg, spawn_worker, settings.vector_sync_processor_workers + ) # Expose this long-lived task group to request-path code # that wants to spawn background work (e.g. ADR-019 @@ -1981,7 +1969,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info( "Background sync tasks started: 1 user manager + %s processors (queue=%s)", - 0 if use_postgres else settings.vector_sync_processor_workers, + 0 + if settings.ingest_queue == "postgres" + else settings.vector_sync_processor_workers, settings.ingest_queue, ) @@ -1995,10 +1985,10 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = shutdown_event.set() # Request path must not spawn into a cancelling group. _vector_sync_state.eviction_task_group = None - # Close the procrastinate connector pool (postgres). - _drain = getattr(task_producer, "drain", None) - if use_postgres and _drain is not None: - await _drain() + # Tear down backend-owned resources (closes the + # procrastinate connector pool in postgres mode; + # no-op for the memory stream). + await transport.aclose() # Close token broker HTTP client if token_broker._http_client: await token_broker._http_client.aclose() diff --git a/nextcloud_mcp_server/cli.py b/nextcloud_mcp_server/cli.py index a5b17257..aa714930 100644 --- a/nextcloud_mcp_server/cli.py +++ b/nextcloud_mcp_server/cli.py @@ -325,6 +325,11 @@ def worker(concurrency: int | None): get_procrastinate_app, ) + # This is the consumer side of the distributed (postgres) ingest backend. + # Unlike the in-process anyio pool, the worker talks to procrastinate's App + # directly (run_worker_async), so it does NOT go through IngestTransport — + # DistributedTransport.run_consumers is a deliberate no-op precisely because + # this separate process is the consumer (see vector/queue/transport.py). workers = concurrency or settings.vector_sync_processor_workers app = get_procrastinate_app() diff --git a/nextcloud_mcp_server/vector/queue/__init__.py b/nextcloud_mcp_server/vector/queue/__init__.py index a0799e50..eac2cde8 100644 --- a/nextcloud_mcp_server/vector/queue/__init__.py +++ b/nextcloud_mcp_server/vector/queue/__init__.py @@ -3,5 +3,21 @@ from .factory import build_producer from .memory import MemoryTaskProducer from .ports import TaskProducer +from .transport import ( + DistributedTransport, + IngestTransport, + LocalTransport, + SpawnWorker, + build_transport, +) -__all__ = ["MemoryTaskProducer", "TaskProducer", "build_producer"] +__all__ = [ + "DistributedTransport", + "IngestTransport", + "LocalTransport", + "MemoryTaskProducer", + "SpawnWorker", + "TaskProducer", + "build_producer", + "build_transport", +] diff --git a/nextcloud_mcp_server/vector/queue/transport.py b/nextcloud_mcp_server/vector/queue/transport.py new file mode 100644 index 00000000..9ce6f62d --- /dev/null +++ b/nextcloud_mcp_server/vector/queue/transport.py @@ -0,0 +1,202 @@ +"""Ingest-path transport (design §10, hexagonal; Deck #183 follow-up, ADR-028). + +The :class:`TaskProducer` port (see ``ports.py``) is *one* side of ingest — the +sink the scanner/webhook send a ``DocumentTask`` to. An :class:`IngestTransport` +is the composition object that owns *both* sides of one backend: + +- the ``producer`` to wire into ``app.state`` / hand to the scanner, and +- how the **consumer** side runs for *this* process. + +Two adapters, selected by ``INGEST_QUEUE`` via :func:`build_transport`: + +- :class:`LocalTransport` (``memory`` — the SQLite/dev default): an in-process + anyio ``MemoryObjectStream`` drained by a pool of in-process workers that + :meth:`run_consumers` starts. +- :class:`DistributedTransport` (``postgres``): wraps the + :class:`ProcrastinateTaskProducer`; :meth:`run_consumers` is a no-op because + the consumer is a *separate* process — the ``nextcloud-mcp-server worker`` + role drains the queue (see ``cli.py``). + +There is deliberately no consumer *port* (mirroring ``ports.py``): in memory +mode the in-process pool is the consumer, in postgres mode the external worker +is. The transport just encapsulates "build the producer + run (or don't run) +the in-process consumers" so the server lifespan has a single branch-free shape +and a new backend (Redis/NATS/SQS) drops in as one more adapter + one +:func:`build_transport` arm — no ``app.py`` or scanner change. + +Why an ABC here but a ``Protocol`` for ``TaskProducer``: the producer port is a +Protocol so anyio's third-party ``MemoryObjectSendStream`` satisfies it +structurally; the transport has exactly two in-house adapters that share the +``producer`` storage and the ``receive_stream``/``run_consumers``/``aclose`` +defaults, so a concrete ABC is simpler and checks more cleanly under ``ty``. + +The single-tenant parallelism invariant lives here: :class:`LocalTransport` +hands every worker a ``clone()`` of *one* shared receive stream, so a tenant's +users are processed by an N-worker pool off a single multiplexed queue +(per-document, not per-user, dispatch) — never one user fully then the next. +""" + +from __future__ import annotations + +import abc +import logging +from collections.abc import Awaitable, Callable +from typing import TYPE_CHECKING + +import anyio +from anyio.abc import TaskGroup +from anyio.streams.memory import ( + MemoryObjectReceiveStream, + MemoryObjectSendStream, +) + +from .factory import build_producer +from .memory import MemoryTaskProducer + +if TYPE_CHECKING: + from ...config import Settings + from ..scanner import DocumentTask + from .ports import TaskProducer + from .procrastinate import ProcrastinateTaskProducer + +logger = logging.getLogger(__name__) + + +# A worker-spawn callback supplied by the lifespan: given a worker index and a +# *fresh* receive handle, start one in-process consumer in the lifespan's task +# group. The lifespan owns this closure so the transport never learns about auth +# modes (single shared nc_client+username vs per-document credential resolution +# by host). anyio's ``TaskGroup.start`` injects a ``task_status`` keyword, which +# the closure must forward to the underlying ``processor_task`` / +# ``multi_user_processor_task`` (else ``start`` blocks forever); hence ``...``. +SpawnWorker = Callable[..., Awaitable[None]] + + +class IngestTransport(abc.ABC): + """Owns one ingest backend: the producer + running its in-process consumers. + + Built once per server lifespan by :func:`build_transport`. The + :class:`TaskProducer` port is unchanged; this is a higher-level composition + object owned only by the lifespan (the worker CLI talks to procrastinate + directly — see module docstring). + """ + + @property + @abc.abstractmethod + def producer(self) -> TaskProducer: + """The :class:`TaskProducer` to wire into ``app.state`` / the scanner.""" + + @property + def send_stream(self) -> MemoryObjectSendStream[DocumentTask] | None: + """Memory backend's raw send end; ``None`` for distributed backends. + + Exposed only so the lifespan can keep populating + ``_vector_sync_state.document_send_stream`` (which the integration + conftest saves/closes as a singleton). Producers send via + :attr:`producer`, not this. + """ + return None + + @property + def receive_stream(self) -> MemoryObjectReceiveStream[DocumentTask] | None: + """Memory backend's receive end (queue-depth surface); ``None`` for + distributed backends, which have no in-process stream (``ingest_status`` + reads procrastinate job counts via the producer instead).""" + return None + + async def run_consumers( + self, task_group: TaskGroup, spawn_worker: SpawnWorker, count: int + ) -> None: + """Start the in-process consumer pool in ``task_group``. + + No-op by default: distributed backends are drained by the external + ``worker`` role, so there is nothing to start in the API process. + """ + return None + + async def aclose(self) -> None: + """Tear down backend-owned resources once on lifespan shutdown. + + No-op by default (the memory stream is closed by task-group cancellation, + exactly as before); :class:`DistributedTransport` closes its connector + pool here. + """ + return None + + +class LocalTransport(IngestTransport): + """In-process anyio memory stream + processor pool (``INGEST_QUEUE=memory``). + + Builds the paired send/receive streams up front and owns the receive end; + :meth:`run_consumers` hands each worker an independent ``clone()`` so every + receiver observes end-of-stream when the scanner's send handles all close. + """ + + def __init__(self, max_buffer_size: float): + send_stream, receive_stream = anyio.create_memory_object_stream["DocumentTask"]( + max_buffer_size=max_buffer_size + ) + self._send_stream = send_stream + self._receive_stream = receive_stream + self._producer = MemoryTaskProducer(send_stream) + + @property + def producer(self) -> TaskProducer: + return self._producer + + @property + def send_stream(self) -> MemoryObjectSendStream[DocumentTask]: + return self._send_stream + + @property + def receive_stream(self) -> MemoryObjectReceiveStream[DocumentTask]: + return self._receive_stream + + async def run_consumers( + self, task_group: TaskGroup, spawn_worker: SpawnWorker, count: int + ) -> None: + # One shared receive stream, N workers each draining a clone → a single + # multiplexed queue processed with N-way parallelism across all users in + # the tenant (per-document dispatch). ``start`` (not ``start_soon``) + # waits for each worker's ``task_status.started()`` readiness, matching + # the prior inline lifespan behaviour. + for i in range(count): + await task_group.start(spawn_worker, i, self._receive_stream.clone()) + + +class DistributedTransport(IngestTransport): + """Postgres/procrastinate producer; consumers are the external worker role. + + The producer's connector pool is opened by :func:`build_producer` and owned + by the lifespan; :meth:`run_consumers` is the inherited no-op (the + ``nextcloud-mcp-server worker`` process drains the queue) and :meth:`aclose` + closes the pool once on shutdown. + """ + + def __init__(self, producer: ProcrastinateTaskProducer): + self._producer = producer + + @property + def producer(self) -> TaskProducer: + return self._producer + + async def aclose(self) -> None: + await self._producer.drain() + + +async def build_transport(settings: Settings) -> IngestTransport: + """Build the ingest transport for the configured ``INGEST_QUEUE`` backend. + + - ``postgres`` → :class:`DistributedTransport`. Reuses :func:`build_producer` + (which opens the connector pool) and applies procrastinate's schema once on + that same open pool before any scanner can defer — a single open/close + cycle, matching the ``worker`` command. + - ``memory`` (SQLite/dev default) → :class:`LocalTransport`. + """ + if settings.ingest_queue == "postgres": + producer = await build_producer(settings) + await producer.ensure_schema() + logger.info("Ingest queue: postgres (procrastinate); worker drains it") + return DistributedTransport(producer) + + return LocalTransport(max_buffer_size=settings.vector_sync_queue_max_size) diff --git a/tests/unit/vector/test_ingest_transport.py b/tests/unit/vector/test_ingest_transport.py new file mode 100644 index 00000000..93242874 --- /dev/null +++ b/tests/unit/vector/test_ingest_transport.py @@ -0,0 +1,114 @@ +"""Unit tests for the ingest transport port (ADR-028; Deck #196). + +Covers the factory's backend selection and each adapter's contract. The +single-tenant parallelism invariant (cross-user overlap) has its own follow-up +(Deck #197); here we only assert that ``LocalTransport.run_consumers`` starts the +requested number of workers off the shared stream. +""" + +from types import SimpleNamespace +from typing import cast +from unittest.mock import AsyncMock + +import anyio +import pytest +from anyio.abc import TaskGroup + +import nextcloud_mcp_server.vector.queue.transport as transport_mod +from nextcloud_mcp_server.config import Settings +from nextcloud_mcp_server.vector.queue import ( + DistributedTransport, + LocalTransport, + MemoryTaskProducer, + SpawnWorker, + build_transport, +) + +pytestmark = pytest.mark.unit + + +def _settings(**kwargs) -> Settings: + """A duck-typed Settings carrying only the fields build_transport reads. + + cast keeps ``ty`` honest about the real signature while avoiding the cost of + constructing a full Settings (dynaconf + validators) for a two-field read. + """ + return cast(Settings, SimpleNamespace(**kwargs)) + + +class TestBuildTransport: + async def test_memory_returns_local_transport(self): + settings = _settings(ingest_queue="memory", vector_sync_queue_max_size=7) + transport = await build_transport(settings) + + assert isinstance(transport, LocalTransport) + assert isinstance(transport.producer, MemoryTaskProducer) + # Memory backend exposes both raw stream ends. + assert transport.send_stream is not None + assert transport.receive_stream is not None + + async def test_postgres_returns_distributed_transport(self, monkeypatch): + producer = AsyncMock() + + async def fake_build_producer(settings): + return producer + + monkeypatch.setattr(transport_mod, "build_producer", fake_build_producer) + + settings = _settings(ingest_queue="postgres") + transport = await build_transport(settings) + + assert isinstance(transport, DistributedTransport) + assert transport.producer is producer + # Schema applied once on the open pool before any defer. + producer.ensure_schema.assert_awaited_once() + # No in-process stream for the distributed backend. + assert transport.send_stream is None + assert transport.receive_stream is None + + +class TestLocalTransport: + async def test_run_consumers_starts_count_workers_off_shared_stream(self): + transport = LocalTransport(max_buffer_size=5) + started: list[int] = [] + received_streams: list[object] = [] + + async def fake_worker( + worker_id, receive_stream, *, task_status=anyio.TASK_STATUS_IGNORED + ): + started.append(worker_id) + received_streams.append(receive_stream) + # Must signal readiness or tg.start blocks forever. + task_status.started() + + async with anyio.create_task_group() as tg: + await transport.run_consumers(tg, fake_worker, 3) + + assert sorted(started) == [0, 1, 2] + # Each worker gets its own (cloned) receive handle, none None. + assert len(received_streams) == 3 + assert all(s is not None for s in received_streams) + + +class TestDistributedTransport: + async def test_run_consumers_is_noop(self): + producer = AsyncMock() + transport = DistributedTransport(producer) + + # Passing sentinel task group / spawn callback proves the no-op never + # touches them (the external worker is the consumer). The casts satisfy + # the signature; the values are deliberately unusable to catch any + # accidental use. + sentinel_tg = cast(TaskGroup, object()) + sentinel_spawn = cast(SpawnWorker, None) + await transport.run_consumers(sentinel_tg, sentinel_spawn, count=3) + + producer.assert_not_awaited() + + async def test_aclose_drains_producer(self): + producer = AsyncMock() + transport = DistributedTransport(producer) + + await transport.aclose() + + producer.drain.assert_awaited_once() From 655d608fb73d801629997cf022e56ca54b03e243 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 20:07:19 +0200 Subject: [PATCH 2/7] refactor: address PR #851 review round 1 (ingest transport) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add IngestTransport.active_consumer_count (0 by default; LocalTransport stores the started count) so app.py logs the worker count without re-checking INGEST_QUEUE — the last backend-knowledge leak in the lifespan is gone. - Document that DistributedTransport is postgres/procrastinate-specific by design (aclose() calls ProcrastinateTaskProducer.drain()); other distributed backends would be separate IngestTransport subclasses. - Clarify the _wire_vector_sync_state log line (writes app.state + singleton, not only the singleton). - Strengthen the LocalTransport test: assert active_consumer_count transitions 0→N and that each worker receives a distinct cloned receive stream. Refs: Deck #196 Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/app.py | 10 +++----- .../vector/queue/transport.py | 23 +++++++++++++++++++ tests/unit/vector/test_ingest_transport.py | 11 ++++++++- 3 files changed, 36 insertions(+), 8 deletions(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 4145acbb..f514a29e 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -377,7 +377,7 @@ def _wire_vector_sync_state( # app.state (Starlette) + the module singleton share the same attribute names. _apply(app.state) _apply(_vector_sync_state) - logger.info("Vector sync state stored in module singleton") + logger.info("Vector sync state published (app.state + module singleton)") # Also share with the mounted /app browser sub-app, if present. for route in app.routes: @@ -1763,9 +1763,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info( "Background sync tasks started: 1 scanner + %s processors (queue=%s)", - 0 - if settings.ingest_queue == "postgres" - else settings.vector_sync_processor_workers, + transport.active_consumer_count, settings.ingest_queue, ) @@ -1969,9 +1967,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info( "Background sync tasks started: 1 user manager + %s processors (queue=%s)", - 0 - if settings.ingest_queue == "postgres" - else settings.vector_sync_processor_workers, + transport.active_consumer_count, settings.ingest_queue, ) diff --git a/nextcloud_mcp_server/vector/queue/transport.py b/nextcloud_mcp_server/vector/queue/transport.py index 9ce6f62d..fb4fcbbf 100644 --- a/nextcloud_mcp_server/vector/queue/transport.py +++ b/nextcloud_mcp_server/vector/queue/transport.py @@ -104,6 +104,17 @@ class IngestTransport(abc.ABC): reads procrastinate job counts via the producer instead).""" return None + @property + def active_consumer_count(self) -> int: + """In-process consumers started by :meth:`run_consumers` for this process. + + ``0`` by default — distributed backends run their consumers as a separate + ``worker`` process. Lets the lifespan log the worker count without + re-inspecting ``INGEST_QUEUE`` (keeping the backend choice inside the + transport). + """ + return 0 + async def run_consumers( self, task_group: TaskGroup, spawn_worker: SpawnWorker, count: int ) -> None: @@ -139,6 +150,7 @@ class LocalTransport(IngestTransport): self._send_stream = send_stream self._receive_stream = receive_stream self._producer = MemoryTaskProducer(send_stream) + self._active_consumer_count = 0 @property def producer(self) -> TaskProducer: @@ -152,6 +164,10 @@ class LocalTransport(IngestTransport): def receive_stream(self) -> MemoryObjectReceiveStream[DocumentTask]: return self._receive_stream + @property + def active_consumer_count(self) -> int: + return self._active_consumer_count + async def run_consumers( self, task_group: TaskGroup, spawn_worker: SpawnWorker, count: int ) -> None: @@ -162,6 +178,7 @@ class LocalTransport(IngestTransport): # the prior inline lifespan behaviour. for i in range(count): await task_group.start(spawn_worker, i, self._receive_stream.clone()) + self._active_consumer_count = count class DistributedTransport(IngestTransport): @@ -171,6 +188,12 @@ class DistributedTransport(IngestTransport): by the lifespan; :meth:`run_consumers` is the inherited no-op (the ``nextcloud-mcp-server worker`` process drains the queue) and :meth:`aclose` closes the pool once on shutdown. + + This adapter is postgres/procrastinate-specific by design: :meth:`aclose` + calls ``ProcrastinateTaskProducer.drain()`` (the narrow ``_producer`` type + confirms it). A different distributed backend (Redis/NATS/SQS) with its own + shutdown semantics would be a separate :class:`IngestTransport` subclass, not + a reconfiguration of this one. """ def __init__(self, producer: ProcrastinateTaskProducer): diff --git a/tests/unit/vector/test_ingest_transport.py b/tests/unit/vector/test_ingest_transport.py index 93242874..d7fb4b40 100644 --- a/tests/unit/vector/test_ingest_transport.py +++ b/tests/unit/vector/test_ingest_transport.py @@ -70,6 +70,8 @@ class TestBuildTransport: class TestLocalTransport: async def test_run_consumers_starts_count_workers_off_shared_stream(self): transport = LocalTransport(max_buffer_size=5) + # Not yet started → no active consumers. + assert transport.active_consumer_count == 0 started: list[int] = [] received_streams: list[object] = [] @@ -85,9 +87,12 @@ class TestLocalTransport: await transport.run_consumers(tg, fake_worker, 3) assert sorted(started) == [0, 1, 2] - # Each worker gets its own (cloned) receive handle, none None. + assert transport.active_consumer_count == 3 + # Each worker gets its own distinct cloned receive handle (so each + # observes end-of-stream when the senders all close), none None. assert len(received_streams) == 3 assert all(s is not None for s in received_streams) + assert len({id(s) for s in received_streams}) == 3 class TestDistributedTransport: @@ -95,6 +100,9 @@ class TestDistributedTransport: producer = AsyncMock() transport = DistributedTransport(producer) + # No in-process consumers for the distributed backend. + assert transport.active_consumer_count == 0 + # Passing sentinel task group / spawn callback proves the no-op never # touches them (the external worker is the consumer). The casts satisfy # the signature; the values are deliberately unusable to catch any @@ -104,6 +112,7 @@ class TestDistributedTransport: await transport.run_consumers(sentinel_tg, sentinel_spawn, count=3) producer.assert_not_awaited() + assert transport.active_consumer_count == 0 async def test_aclose_drains_producer(self): producer = AsyncMock() From 2179e9ddb0bbae8567230eea65d94d703baf518e Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 20:27:40 +0200 Subject: [PATCH 3/7] refactor: address PR #851 review round 2 (ingest transport) - Rename the lifespan-local `transport` to `ingest_transport` in both paths so it no longer shadows the get_app(transport=...) HTTP-transport parameter. - Log the memory backend selection in build_transport, symmetric with the postgres branch, so startup logs name the chosen ingest backend either way. - Note in _wire_vector_sync_state why eviction_task_group is intentionally not set there (it only exists once the lifespan's task group is running). Refs: Deck #196 Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/app.py | 34 ++++++++++++------- .../vector/queue/transport.py | 1 + 2 files changed, 23 insertions(+), 12 deletions(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index f514a29e..5aa61ea4 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -362,6 +362,10 @@ def _wire_vector_sync_state( come from the transport — ``None`` in postgres mode (no in-process stream) — and ``task_producer`` is the transport's producer in both modes (Deck #183, ADR-028). + + ``eviction_task_group`` is deliberately not set here: it only exists once the + lifespan has entered its ``anyio.create_task_group()`` (after this call), so + the lifespan assigns it on the singleton directly at that point. """ send_stream = transport.send_stream receive_stream = transport.receive_stream @@ -1713,12 +1717,16 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = shutdown_event = anyio.Event() scanner_wake_event = anyio.Event() - transport = await build_transport(settings) - task_producer = transport.producer + # Named ingest_transport (not transport) to avoid shadowing the + # get_app(transport=...) HTTP-transport parameter. + ingest_transport = await build_transport(settings) + task_producer = ingest_transport.producer # Publish to app.state (ADR-007), the module singleton (FastMCP # session lifespans), and the /app browser sub-app in one place. - _wire_vector_sync_state(app, transport, shutdown_event, scanner_wake_event) + _wire_vector_sync_state( + app, ingest_transport, shutdown_event, scanner_wake_event + ) # Start background tasks using anyio TaskGroup async with anyio.create_task_group() as tg: @@ -1751,7 +1759,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = task_status=task_status, ) - await transport.run_consumers( + await ingest_transport.run_consumers( tg, spawn_worker, settings.vector_sync_processor_workers ) @@ -1763,7 +1771,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info( "Background sync tasks started: 1 scanner + %s processors (queue=%s)", - transport.active_consumer_count, + ingest_transport.active_consumer_count, settings.ingest_queue, ) @@ -1781,7 +1789,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # procrastinate connector pool in postgres mode; no-op # for the memory stream, which task-group cancellation # closes). - await transport.aclose() + await ingest_transport.aclose() await client.close() # TaskGroup automatically cancels all tasks on exit @@ -1901,13 +1909,15 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # User state tracking for user manager user_states: dict = {} - transport = await build_transport(settings) - task_producer = transport.producer + # Named ingest_transport (not transport) to avoid shadowing the + # get_app(transport=...) HTTP-transport parameter. + ingest_transport = await build_transport(settings) + task_producer = ingest_transport.producer # Publish to app.state (ADR-007), the module singleton (FastMCP # session lifespans), and the /app browser sub-app in one place. _wire_vector_sync_state( - app, transport, shutdown_event, scanner_wake_event + app, ingest_transport, shutdown_event, scanner_wake_event ) # Background sync authenticates as each provisioned user via @@ -1955,7 +1965,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = task_status=task_status, ) - await transport.run_consumers( + await ingest_transport.run_consumers( tg, spawn_worker, settings.vector_sync_processor_workers ) @@ -1967,7 +1977,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info( "Background sync tasks started: 1 user manager + %s processors (queue=%s)", - transport.active_consumer_count, + ingest_transport.active_consumer_count, settings.ingest_queue, ) @@ -1984,7 +1994,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Tear down backend-owned resources (closes the # procrastinate connector pool in postgres mode; # no-op for the memory stream). - await transport.aclose() + await ingest_transport.aclose() # Close token broker HTTP client if token_broker._http_client: await token_broker._http_client.aclose() diff --git a/nextcloud_mcp_server/vector/queue/transport.py b/nextcloud_mcp_server/vector/queue/transport.py index fb4fcbbf..4e2dcf49 100644 --- a/nextcloud_mcp_server/vector/queue/transport.py +++ b/nextcloud_mcp_server/vector/queue/transport.py @@ -222,4 +222,5 @@ async def build_transport(settings: Settings) -> IngestTransport: logger.info("Ingest queue: postgres (procrastinate); worker drains it") return DistributedTransport(producer) + logger.info("Ingest queue: memory (in-process anyio stream + processor pool)") return LocalTransport(max_buffer_size=settings.vector_sync_queue_max_size) From c0c52c1b34ef687bf7768ccd81491a3703787690 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 20:36:58 +0200 Subject: [PATCH 4/7] refactor: address PR #851 review round 3 (ingest transport) - LocalTransport.run_consumers increments active_consumer_count per worker (instead of once after the loop) so the count is accurate if a later tg.start() raises mid-pool. - Add LocalTransport.aclose() to explicitly close its owned send/receive stream ends (belt-and-suspenders against unclosed-resource warnings; anyio aclose is idempotent, and by shutdown the scanner is already winding down). Reworded the base IngestTransport.aclose() docstring to point at the overrides. - Inline ingest_transport.producer at the scanner/user_manager call sites, dropping the single-use task_producer alias in both lifespan paths. - Annotate DistributedTransport._producer explicitly as ProcrastinateTaskProducer so the drain() coupling is visible and ty catches drift. - Add a unit test for LocalTransport.aclose() (closes the owned streams, idempotent). Refs: Deck #196 Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/app.py | 10 +++---- .../vector/queue/transport.py | 28 +++++++++++++++---- tests/unit/vector/test_ingest_transport.py | 20 +++++++++++++ 3 files changed, 47 insertions(+), 11 deletions(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 5aa61ea4..9c7149a1 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -1720,7 +1720,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Named ingest_transport (not transport) to avoid shadowing the # get_app(transport=...) HTTP-transport parameter. ingest_transport = await build_transport(settings) - task_producer = ingest_transport.producer # Publish to app.state (ADR-007), the module singleton (FastMCP # session lifespans), and the /app browser sub-app in one place. @@ -1730,10 +1729,10 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Start background tasks using anyio TaskGroup async with anyio.create_task_group() as tg: - # Start scanner task (publishes to task_producer) + # Start scanner task (publishes to the transport's producer) await tg.start( scanner_task, - task_producer, + ingest_transport.producer, shutdown_event, scanner_wake_event, client, @@ -1912,7 +1911,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Named ingest_transport (not transport) to avoid shadowing the # get_app(transport=...) HTTP-transport parameter. ingest_transport = await build_transport(settings) - task_producer = ingest_transport.producer # Publish to app.state (ADR-007), the module singleton (FastMCP # session lifespans), and the /app browser sub-app in one place. @@ -1930,11 +1928,11 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # management API revoke endpoint (via app.state.oauth_context). async with anyio.create_task_group() as tg: # Start user manager task (supervises per-user scanners). - # Each per-user scanner clones task_producer; for the bus + # Each per-user scanner clones the producer; for the bus # producer clone() returns the shared connection. await tg.start( user_manager_task, - task_producer, + ingest_transport.producer, shutdown_event, scanner_wake_event, token_storage, diff --git a/nextcloud_mcp_server/vector/queue/transport.py b/nextcloud_mcp_server/vector/queue/transport.py index 4e2dcf49..fb2fd4c4 100644 --- a/nextcloud_mcp_server/vector/queue/transport.py +++ b/nextcloud_mcp_server/vector/queue/transport.py @@ -128,9 +128,9 @@ class IngestTransport(abc.ABC): async def aclose(self) -> None: """Tear down backend-owned resources once on lifespan shutdown. - No-op by default (the memory stream is closed by task-group cancellation, - exactly as before); :class:`DistributedTransport` closes its connector - pool here. + No-op by default; subclasses that own resources (a connector pool, stream + handles) override this to release them — see + :meth:`DistributedTransport.aclose` and :meth:`LocalTransport.aclose`. """ return None @@ -178,7 +178,22 @@ class LocalTransport(IngestTransport): # the prior inline lifespan behaviour. for i in range(count): await task_group.start(spawn_worker, i, self._receive_stream.clone()) - self._active_consumer_count = count + # Increment per-worker (not once after the loop) so the count is + # accurate even if a later start() raises — a crash log then reflects + # how many workers were actually live. + self._active_consumer_count += 1 + + async def aclose(self) -> None: + # Belt-and-suspenders cleanup of the two stream ends this transport owns, + # so they don't linger until GC (which can emit unclosed-resource + # warnings under the test runner / alternative runtimes). anyio's aclose + # is idempotent, so the scanner's own ``async with`` on the send side + # (single-user) closing it first is harmless; worker receive *clones* are + # independent handles, closed by task-group cancellation. By shutdown the + # ``shutdown_event`` is already set, so the scanner is winding down rather + # than issuing fresh sends. + await self._send_stream.aclose() + await self._receive_stream.aclose() class DistributedTransport(IngestTransport): @@ -197,7 +212,10 @@ class DistributedTransport(IngestTransport): """ def __init__(self, producer: ProcrastinateTaskProducer): - self._producer = producer + # Explicit (not inferred): aclose() calls drain(), which lives on the + # concrete ProcrastinateTaskProducer, not the TaskProducer protocol — + # the annotation keeps that coupling visible and lets ty catch drift. + self._producer: ProcrastinateTaskProducer = producer @property def producer(self) -> TaskProducer: diff --git a/tests/unit/vector/test_ingest_transport.py b/tests/unit/vector/test_ingest_transport.py index d7fb4b40..c5ddbadc 100644 --- a/tests/unit/vector/test_ingest_transport.py +++ b/tests/unit/vector/test_ingest_transport.py @@ -23,6 +23,7 @@ from nextcloud_mcp_server.vector.queue import ( SpawnWorker, build_transport, ) +from nextcloud_mcp_server.vector.scanner import DocumentTask pytestmark = pytest.mark.unit @@ -94,6 +95,25 @@ class TestLocalTransport: assert all(s is not None for s in received_streams) assert len({id(s) for s in received_streams}) == 3 + async def test_aclose_closes_owned_streams_idempotently(self): + transport = LocalTransport(max_buffer_size=5) + await transport.aclose() + + # The send end is closed → the producer raises rather than silently + # dropping (the producer wraps the same stream aclose() closed). + with pytest.raises(anyio.ClosedResourceError): + await transport.producer.send( + DocumentTask( + user_id="u", + doc_id="1", + doc_type="note", + operation="index", + modified_at=0, + ) + ) + # Idempotent: closing again is a no-op, not an error. + await transport.aclose() + class TestDistributedTransport: async def test_run_consumers_is_noop(self): From c4401af9c619596715875f034410e17d6e010f9b Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 20:42:57 +0200 Subject: [PATCH 5/7] refactor: address PR #851 review round 4 (ingest transport) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Clear the module-singleton ingest references (task_producer, document_send_stream, document_receive_stream) on lifespan shutdown via a new _clear_vector_sync_state() helper, mirroring the eviction_task_group cleanup. Defense-in-depth so a late webhook (or a module-singleton integration test) can't touch a producer/stream backed by an already-closed resource. - Add IngestTransport.backend_name ("memory"/"postgres") and use it in both lifespan log lines, removing the last settings.ingest_queue read from the background-sync setup — the lifespan no longer inspects the backend at all. - Cover backend_name in the build_transport adapter-selection tests. Refs: Deck #196 Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/app.py | 23 +++++++++++++++++-- .../vector/queue/transport.py | 17 ++++++++++++++ tests/unit/vector/test_ingest_transport.py | 2 ++ 3 files changed, 40 insertions(+), 2 deletions(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 9c7149a1..329ec51b 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -392,6 +392,21 @@ def _wire_vector_sync_state( break +def _clear_vector_sync_state() -> None: + """Drop the module-singleton ingest references on lifespan shutdown. + + Mirrors the ``eviction_task_group = None`` cleanup so that any code reaching + the singleton in the narrow window between transport teardown and process + exit (e.g. a late webhook) sees ``None`` rather than a producer/stream backed + by an already-closed resource. The per-request ``shutdown_event`` gate is the + primary guard; this is defense-in-depth. Integration tests with module-level + singletons also benefit (no stale closed producer leaks between runs). + """ + _vector_sync_state.task_producer = None + _vector_sync_state.document_send_stream = None + _vector_sync_state.document_receive_stream = None + + @dataclass class AppContext: """Application context for BasicAuth mode.""" @@ -1771,7 +1786,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info( "Background sync tasks started: 1 scanner + %s processors (queue=%s)", ingest_transport.active_consumer_count, - settings.ingest_queue, + ingest_transport.backend_name, ) # Run MCP session manager and yield @@ -1789,6 +1804,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # for the memory stream, which task-group cancellation # closes). await ingest_transport.aclose() + # Drop stale singleton refs to the now-closed transport. + _clear_vector_sync_state() await client.close() # TaskGroup automatically cancels all tasks on exit @@ -1976,7 +1993,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.info( "Background sync tasks started: 1 user manager + %s processors (queue=%s)", ingest_transport.active_consumer_count, - settings.ingest_queue, + ingest_transport.backend_name, ) # Run MCP session manager and yield @@ -1993,6 +2010,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # procrastinate connector pool in postgres mode; # no-op for the memory stream). await ingest_transport.aclose() + # Drop stale singleton refs to the now-closed transport. + _clear_vector_sync_state() # Close token broker HTTP client if token_broker._http_client: await token_broker._http_client.aclose() diff --git a/nextcloud_mcp_server/vector/queue/transport.py b/nextcloud_mcp_server/vector/queue/transport.py index fb2fd4c4..6b27e36d 100644 --- a/nextcloud_mcp_server/vector/queue/transport.py +++ b/nextcloud_mcp_server/vector/queue/transport.py @@ -86,6 +86,15 @@ class IngestTransport(abc.ABC): def producer(self) -> TaskProducer: """The :class:`TaskProducer` to wire into ``app.state`` / the scanner.""" + @property + @abc.abstractmethod + def backend_name(self) -> str: + """Short backend identifier for logs/metrics (``memory``/``postgres``). + + Lets the lifespan log which ingest backend is active without reading + ``settings.ingest_queue`` — the backend choice stays inside the transport. + """ + @property def send_stream(self) -> MemoryObjectSendStream[DocumentTask] | None: """Memory backend's raw send end; ``None`` for distributed backends. @@ -156,6 +165,10 @@ class LocalTransport(IngestTransport): def producer(self) -> TaskProducer: return self._producer + @property + def backend_name(self) -> str: + return "memory" + @property def send_stream(self) -> MemoryObjectSendStream[DocumentTask]: return self._send_stream @@ -221,6 +234,10 @@ class DistributedTransport(IngestTransport): def producer(self) -> TaskProducer: return self._producer + @property + def backend_name(self) -> str: + return "postgres" + async def aclose(self) -> None: await self._producer.drain() diff --git a/tests/unit/vector/test_ingest_transport.py b/tests/unit/vector/test_ingest_transport.py index c5ddbadc..afb3b606 100644 --- a/tests/unit/vector/test_ingest_transport.py +++ b/tests/unit/vector/test_ingest_transport.py @@ -44,6 +44,7 @@ class TestBuildTransport: assert isinstance(transport, LocalTransport) assert isinstance(transport.producer, MemoryTaskProducer) + assert transport.backend_name == "memory" # Memory backend exposes both raw stream ends. assert transport.send_stream is not None assert transport.receive_stream is not None @@ -61,6 +62,7 @@ class TestBuildTransport: assert isinstance(transport, DistributedTransport) assert transport.producer is producer + assert transport.backend_name == "postgres" # Schema applied once on the open pool before any defer. producer.ensure_schema.assert_awaited_once() # No in-process stream for the distributed backend. From bf84db35b4d9efe1ae9ac70d9bd2f4cd9e789dd4 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 20:49:17 +0200 Subject: [PATCH 6/7] refactor: address PR #851 review round 5 (ingest transport) - _clear_vector_sync_state also nulls shutdown_event / scanner_wake_event on shutdown, symmetric with the stream/producer fields (the next startup rebinds them via _wire_vector_sync_state). - Comment that the "DocumentTask" string subscript in LocalTransport is intentional (TYPE_CHECKING-only class; anyio ignores the runtime type arg). - Move app.py's annotation-only IngestTransport / TaskProducer imports under TYPE_CHECKING (the module uses `from __future__ import annotations`), keeping only build_transport at runtime. Refs: Deck #196 Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/app.py | 17 +++++++++++------ nextcloud_mcp_server/vector/queue/transport.py | 3 +++ 2 files changed, 14 insertions(+), 6 deletions(-) diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 329ec51b..dfb6b377 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -9,7 +9,7 @@ import traceback from collections.abc import AsyncIterator from contextlib import AsyncExitStack, asynccontextmanager from dataclasses import dataclass -from typing import Any, cast +from typing import TYPE_CHECKING, Any, cast from urllib.parse import urlparse import anyio @@ -133,14 +133,15 @@ from nextcloud_mcp_server.vector.oauth_sync import ( from nextcloud_mcp_server.vector.placeholder import sweep_orphan_placeholders from nextcloud_mcp_server.vector.processor import processor_task from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client -from nextcloud_mcp_server.vector.queue import ( - IngestTransport, - TaskProducer, - build_transport, -) +from nextcloud_mcp_server.vector.queue import build_transport from nextcloud_mcp_server.vector.scanner import scanner_task from nextcloud_mcp_server.vector.webhook_receiver import handle_nextcloud_webhook +if TYPE_CHECKING: + # Annotation-only in this module (the file uses `from __future__ import + # annotations`, so these are never evaluated at runtime). + from nextcloud_mcp_server.vector.queue import IngestTransport, TaskProducer + logger = logging.getLogger(__name__) HTTPXClientInstrumentor().instrument() @@ -405,6 +406,10 @@ def _clear_vector_sync_state() -> None: _vector_sync_state.task_producer = None _vector_sync_state.document_send_stream = None _vector_sync_state.document_receive_stream = None + # Symmetric with the fields above: the just-fired events belong to the + # closed lifespan; the next startup's _wire_vector_sync_state rebinds them. + _vector_sync_state.shutdown_event = None + _vector_sync_state.scanner_wake_event = None @dataclass diff --git a/nextcloud_mcp_server/vector/queue/transport.py b/nextcloud_mcp_server/vector/queue/transport.py index 6b27e36d..aed1e72c 100644 --- a/nextcloud_mcp_server/vector/queue/transport.py +++ b/nextcloud_mcp_server/vector/queue/transport.py @@ -153,6 +153,9 @@ class LocalTransport(IngestTransport): """ def __init__(self, max_buffer_size: float): + # "DocumentTask" as a string (not the symbol): the class is + # TYPE_CHECKING-only here, and anyio ignores the runtime value of the + # type argument — so the string is intentional, not a typo. send_stream, receive_stream = anyio.create_memory_object_stream["DocumentTask"]( max_buffer_size=max_buffer_size ) From 6c7b679b52fc0e55d5274e7cfd163efbf1bb7a3d Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Thu, 4 Jun 2026 20:53:36 +0200 Subject: [PATCH 7/7] docs: correct ADR-028 aclose() description (PR #851 round 6 nit) LocalTransport.aclose() (added in round 3) closes its owned stream ends; the ADR still described aclose() as a no-op for the memory stream. Update the prose to match the shipped behaviour. Doc-only. Refs: Deck #196 Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/ADR-028-ingest-transport-port.md | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/docs/ADR-028-ingest-transport-port.md b/docs/ADR-028-ingest-transport-port.md index 544b22c2..c5b87af0 100644 --- a/docs/ADR-028-ingest-transport-port.md +++ b/docs/ADR-028-ingest-transport-port.md @@ -55,9 +55,11 @@ build_transport(settings) -> and the integration conftest's stream-singleton handling working unchanged). - `run_consumers(task_group, spawn_worker, count)` — start the in-process pool; a **no-op** for distributed backends, whose consumer is the external worker. -- `aclose()` — tear down backend-owned resources once on shutdown (closes the - procrastinate connector pool; a no-op for the memory stream, which task-group - cancellation closes). +- `aclose()` — tear down backend-owned resources once on shutdown. + `DistributedTransport` closes the procrastinate connector pool; + `LocalTransport` closes its owned send/receive stream ends (belt-and-suspenders + — task-group cancellation already closes the per-worker clones, and anyio + `aclose` is idempotent). The base default is a no-op. The lifespan supplies a `spawn_worker` closure so the transport never learns about auth modes — the single-user closure binds a shared `nc_client`+username,