diff --git a/docs/ADR-028-ingest-transport-port.md b/docs/ADR-028-ingest-transport-port.md new file mode 100644 index 00000000..c5b87af0 --- /dev/null +++ b/docs/ADR-028-ingest-transport-port.md @@ -0,0 +1,126 @@ +# 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. + `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, +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..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 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 ( - MemoryTaskProducer, - TaskProducer, - build_producer, -) -from nextcloud_mcp_server.vector.scanner import DocumentTask, scanner_task +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() @@ -347,6 +348,70 @@ 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). + + ``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 + 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 published (app.state + 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 + + +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 + # 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 class AppContext: """Application context for BasicAuth mode.""" @@ -1663,86 +1728,59 @@ 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) + # Named ingest_transport (not transport) to avoid shadowing the + # get_app(transport=...) HTTP-transport parameter. + ingest_transport = await build_transport(settings) - # 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, ingest_transport, shutdown_event, scanner_wake_event + ) # 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, 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 ingest_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,8 +1790,8 @@ 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, - settings.ingest_queue, + ingest_transport.active_consumer_count, + ingest_transport.backend_name, ) # Run MCP session manager and yield @@ -1766,10 +1804,13 @@ 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 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 @@ -1877,65 +1918,27 @@ 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) + # Named ingest_transport (not transport) to avoid shadowing the + # get_app(transport=...) HTTP-transport parameter. + ingest_transport = await build_transport(settings) - # 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, ingest_transport, shutdown_event, scanner_wake_event + ) # Background sync authenticates as each provisioned user via # locally-stored Nextcloud app passwords (Login Flow v2 / @@ -1947,11 +1950,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, @@ -1960,18 +1963,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 ingest_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,8 +1997,8 @@ 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, - settings.ingest_queue, + ingest_transport.active_consumer_count, + ingest_transport.backend_name, ) # Run MCP session manager and yield @@ -1995,10 +2011,12 @@ 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 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/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..aed1e72c --- /dev/null +++ b/nextcloud_mcp_server/vector/queue/transport.py @@ -0,0 +1,264 @@ +"""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 + @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. + + 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 + + @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: + """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; subclasses that own resources (a connector pool, stream + handles) override this to release them — see + :meth:`DistributedTransport.aclose` and :meth:`LocalTransport.aclose`. + """ + 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): + # "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 + ) + 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: + return self._producer + + @property + def backend_name(self) -> str: + return "memory" + + @property + def send_stream(self) -> MemoryObjectSendStream[DocumentTask]: + return self._send_stream + + @property + 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: + # 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()) + # 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): + """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. + + 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): + # 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: + return self._producer + + @property + def backend_name(self) -> str: + return "postgres" + + 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) + + logger.info("Ingest queue: memory (in-process anyio stream + processor pool)") + 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..afb3b606 --- /dev/null +++ b/tests/unit/vector/test_ingest_transport.py @@ -0,0 +1,145 @@ +"""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, +) +from nextcloud_mcp_server.vector.scanner import DocumentTask + +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) + 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 + + 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 + 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. + 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) + # Not yet started → no active consumers. + assert transport.active_consumer_count == 0 + 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] + 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 + + 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): + 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 + # 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() + assert transport.active_consumer_count == 0 + + async def test_aclose_drains_producer(self): + producer = AsyncMock() + transport = DistributedTransport(producer) + + await transport.aclose() + + producer.drain.assert_awaited_once()