From 21b7922bac45b74bf469a946bde211f651c2ceee Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 3 Jun 2026 04:11:11 +0200 Subject: [PATCH] feat: replace NATS ingest with procrastinate Postgres queue (#183) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Re-architect document ingest from the shared NATS-glued document-processor to a per-tenant, in-process model owned by nextcloud-mcp-server (Deck #183). The MCP server now owns both sides of ingest: - Producer (api role): the scanner defers one job per changed document into the app's Postgres via procrastinate (queueing_lock dedup; no execution lock, so a crashed worker can't deadlock a doc — Qdrant upserts are idempotent). - Consumer (worker role): `nextcloud-mcp-server worker` drains the queue and runs the existing process_document pipeline; a periodic task reclaims jobs orphaned in `doing` by a crash. INGEST_QUEUE selects the transport (auto: postgres when DATABASE_URL is Postgres, else the in-process anyio queue for SQLite/dev). procrastinate manages its own tables (applied on a fresh DB at startup and by `db upgrade`). The vector-sync status surface reads job counts from Postgres in postgres mode. procrastinate + psycopg3 ship in the [postgres] extra; the app's own engine still uses asyncpg (driver unification is a follow-up handled in the rendered Helm chart). NATS JetStream, the Postgres-queue stub, the bus status subscriber, and nats-py are removed. BREAKING CHANGE: the external-NATS-ingest env vars are removed (INGEST_MODE, STATUS_BACKEND, INGEST_BUS_URL, INGEST_BUS_NUM_REPLICAS, FACT_EVENT_EMITTER). Use INGEST_QUEUE (memory|postgres) and the `worker` command instead. TENANT_ID is retained (no longer NATS-subject-charset-validated). Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/configuration.md | 50 ++- nextcloud_mcp_server/api/management.py | 74 ++-- nextcloud_mcp_server/app.py | 151 +++----- nextcloud_mcp_server/auth/userinfo_routes.py | 25 +- nextcloud_mcp_server/canonical.py | 11 +- nextcloud_mcp_server/cli.py | 79 +++++ nextcloud_mcp_server/config.py | 158 ++++++--- nextcloud_mcp_server/models/semantic.py | 11 + nextcloud_mcp_server/server/semantic.py | 36 +- nextcloud_mcp_server/vector/ingest_status.py | 56 +++ nextcloud_mcp_server/vector/processor.py | 9 +- nextcloud_mcp_server/vector/queue/__init__.py | 6 +- nextcloud_mcp_server/vector/queue/factory.py | 59 +-- nextcloud_mcp_server/vector/queue/nats.py | 173 --------- nextcloud_mcp_server/vector/queue/ports.py | 16 +- nextcloud_mcp_server/vector/queue/postgres.py | 52 --- .../vector/queue/procrastinate.py | 335 ++++++++++++++++++ nextcloud_mcp_server/vector/queue/status.py | 195 ---------- .../vector/webhook_receiver.py | 7 +- pyproject.toml | 3 +- .../integration/test_ingest_queue_postgres.py | 139 ++++++++ tests/unit/test_decomposition_config.py | 145 ++++---- tests/unit/vector/test_ingest_status.py | 57 +++ tests/unit/vector/test_nats_producer.py | 155 -------- .../vector/test_procrastinate_producer.py | 191 ++++++++++ tests/unit/vector/test_status_store.py | 167 --------- uv.lock | 129 ++++++- 27 files changed, 1369 insertions(+), 1120 deletions(-) create mode 100644 nextcloud_mcp_server/vector/ingest_status.py delete mode 100644 nextcloud_mcp_server/vector/queue/nats.py delete mode 100644 nextcloud_mcp_server/vector/queue/postgres.py create mode 100644 nextcloud_mcp_server/vector/queue/procrastinate.py delete mode 100644 nextcloud_mcp_server/vector/queue/status.py create mode 100644 tests/integration/test_ingest_queue_postgres.py create mode 100644 tests/unit/vector/test_ingest_status.py delete mode 100644 tests/unit/vector/test_nats_producer.py create mode 100644 tests/unit/vector/test_procrastinate_producer.py delete mode 100644 tests/unit/vector/test_status_store.py diff --git a/docs/configuration.md b/docs/configuration.md index 1b4dd23c..fb0a38bc 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -785,9 +785,9 @@ docker-compose up ## Decomposition Hook Points (Optional, Advanced) -The server can optionally offload document processing and embeddings to external -services (the Astrolabe Cloud document-processor and embedding-gateway). These -are **opt-in**; every default reproduces the in-process monolith behavior, so +The server can optionally offload embeddings to an external gateway and split +ingest into a separate scale-to-zero worker process (Deck #183). These are +**opt-in**; every default reproduces the in-process monolith behavior, so self-hosters can ignore this section. ```bash @@ -799,21 +799,41 @@ EMBEDDING_GATEWAY_TOKEN_URL=... EMBEDDING_GATEWAY_CLIENT_ID=... EMBEDDING_GATEWAY_CLIENT_SECRET=... -# External ingest: publish to NATS instead of the in-process processor pool -INGEST_MODE=external # local (default) | external -STATUS_BACKEND=bus # local (default) | bus — REQUIRED with external -INGEST_BUS_URL=nats://nats:4222 -TENANT_ID= # NATS per-tenant subject token +# Ingest queue backend. Default (unset) auto-derives from DATABASE_URL: +# - PostgreSQL DATABASE_URL → "postgres" (the procrastinate queue) +# - SQLite / unset → "memory" (the in-process anyio queue) +INGEST_QUEUE=postgres # memory | postgres +# Process role (informational; the worker is launched via the `worker` command): +MCP_ROLE=all # api | worker | all (default) +TENANT_ID= # per-tenant identity (used in collection naming) +``` + +### Postgres ingest queue + worker (api/worker split) + +When `INGEST_QUEUE=postgres` (a PostgreSQL `DATABASE_URL`), the scanner **defers** +one job per changed document into the app's Postgres via +[procrastinate](https://procrastinate.readthedocs.io); a separate **worker** +process drains the queue (fetch → chunk → embed → upsert Qdrant). Run the two +roles as separate Deployments from the same image: + +```bash +# API pod (always-on): serves MCP/query + runs the scanner (defers jobs) +nextcloud-mcp-server run + +# Ingest worker (scale-to-zero on queue depth via KEDA): drains the queue +nextcloud-mcp-server worker -c 4 ``` Notes: -- `STATUS_BACKEND=local` with `INGEST_MODE=external` is rejected at startup - (the in-process job state is empty for externally-dispatched work). -- **`nats-py` ships as a core dependency** (small, pure-Python) and is imported - lazily — only when `INGEST_MODE=external`. Self-hosters who never enable - external ingest pay no runtime cost. -- `INGEST_MODE=external` + `STATUS_BACKEND=bus` opens **two** NATS connections - per pod (the ingest producer and the status subscriber are separate roles). + +- **procrastinate manages its own tables** (`procrastinate_jobs`, …) in the same + database. They are created on a fresh DB by the API pod at startup and by + `nextcloud-mcp-server db upgrade` — a migration lineage independent of the + app's Alembic schema. procrastinate is Postgres-only (psycopg3); it ships in + the `[postgres]` extra and is imported lazily. +- KEDA scales the worker on + `SELECT count(*) FROM procrastinate_jobs WHERE queue_name='ingest' AND status='todo'`. +- `INGEST_QUEUE=postgres` with a SQLite `DATABASE_URL` is rejected at startup. --- diff --git a/nextcloud_mcp_server/api/management.py b/nextcloud_mcp_server/api/management.py index 9dde161f..1ebb757d 100644 --- a/nextcloud_mcp_server/api/management.py +++ b/nextcloud_mcp_server/api/management.py @@ -293,52 +293,21 @@ async def get_vector_sync_status(request: Request) -> JSONResponse: ) try: - # Bus status backend (INGEST_MODE=external): there is no in-process - # queue; pending/terminal state comes from the NATS status subscriber's - # store. indexed_documents stays the mode-independent Qdrant count. - if settings.status_backend == "bus": - store = getattr(request.app.state, "status_store", None) - indexed_count = 0 - try: - qdrant_client = await get_qdrant_client() - count_result = await qdrant_client.count( - collection_name=settings.get_collection_name(), - count_filter=Filter(must=[get_placeholder_filter()]), - ) - indexed_count = count_result.count - except Exception as e: - logger.warning("Failed to query Qdrant for indexed count: %s", e) - return JSONResponse( - { - "status": "idle", - "indexed_documents": indexed_count, - "pending_documents": 0, - "status_backend": "bus", - "recent_states": store.counts() if store is not None else {}, - } - ) - - # Get document receive stream from app state (set by starlette_lifespan in app.py) - document_receive_stream = getattr( - request.app.state, "document_receive_stream", None + # Outstanding-work view depends on the queue backend (Deck #183): + # memory → stream buffer depth; postgres → procrastinate job counts. + from nextcloud_mcp_server.vector.ingest_status import ( # noqa: PLC0415 + get_ingest_pending, ) - if document_receive_stream is None: - logger.debug("document_receive_stream not available in app state") - return JSONResponse( - { - "status": "unknown", - "indexed_documents": 0, - "pending_documents": 0, - "message": "Vector sync stream not initialized", - } - ) + pending = await get_ingest_pending( + task_producer=getattr(request.app.state, "task_producer", None), + document_receive_stream=getattr( + request.app.state, "document_receive_stream", None + ), + ingest_queue=settings.ingest_queue, + ) - # Get pending count from stream statistics - stream_stats = document_receive_stream.statistics() - pending_count = stream_stats.current_buffer_used - - # Get Qdrant client and query indexed count + # Get Qdrant client and query indexed count (backend-independent) indexed_count = 0 try: qdrant_client = await get_qdrant_client() @@ -355,15 +324,18 @@ async def get_vector_sync_status(request: Request) -> JSONResponse: # Continue with indexed_count = 0 # Determine status - status = "syncing" if pending_count > 0 else "idle" + status = "syncing" if pending.pending > 0 else "idle" - return JSONResponse( - { - "status": status, - "indexed_documents": indexed_count, - "pending_documents": pending_count, - } - ) + body: dict[str, object] = { + "status": status, + "indexed_documents": indexed_count, + "pending_documents": pending.pending, + "ingest_queue": settings.ingest_queue, + } + if pending.job_counts is not None: + # Per-status breakdown (todo/doing/failed/…) on the postgres backend. + body["job_counts"] = pending.job_counts + return JSONResponse(body) except Exception as e: error_msg = _sanitize_error_for_client(e, "get_vector_sync_status") diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 52e89fa9..9bbf38b3 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -136,9 +136,8 @@ from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client from nextcloud_mcp_server.vector.queue import ( MemoryTaskProducer, TaskProducer, - build_external_producer, + build_producer, ) -from nextcloud_mcp_server.vector.queue.status import NatsStatusSubscriber, StatusStore from nextcloud_mcp_server.vector.scanner import DocumentTask, scanner_task from nextcloud_mcp_server.vector.webhook_receiver import handle_nextcloud_webhook @@ -332,12 +331,10 @@ class VectorSyncState: document_send_stream: MemoryObjectSendStream | None = None document_receive_stream: MemoryObjectReceiveStream | None = None # Ingest producer the scanner/webhook send to: the in-memory send stream - # (local mode) or the NATS bus producer (external mode). The webhook reads - # this; in local mode it is the same object as document_send_stream. + # (INGEST_QUEUE=memory) or the procrastinate producer (INGEST_QUEUE=postgres, + # Deck #183). The webhook reads this; in memory mode it is the same object as + # document_send_stream. task_producer: "TaskProducer | None" = None - # Bus status store (STATUS_BACKEND=bus): populated by the NATS status - # subscriber, read by the vector-sync status endpoint. None in local mode. - status_store: "StatusStore | None" = None shutdown_event: anyio.Event | None = None scanner_wake_event: anyio.Event | None = None # Long-lived task group used for fire-and-forget background work spawned @@ -350,33 +347,6 @@ class VectorSyncState: _vector_sync_state = VectorSyncState() -async def _build_status_subscriber( - settings: "Settings", -) -> "tuple[StatusStore | None, NatsStatusSubscriber | None]": - """Build the bus status store + subscriber when STATUS_BACKEND=bus. - - Returns ``(None, None)`` for local status (the status endpoint reads the - in-memory stream buffer instead). __post_init__ guarantees that bus status - only pairs with external ingest, so ingest_bus_url/tenant_id are set. - """ - if not (settings.ingest_mode == "external" and settings.status_backend == "bus"): - return None, None - # Defence-in-depth (robust under ``python -O``, which strips asserts): - # __post_init__ already guarantees these when status_backend == "bus". - if settings.ingest_bus_url is None or settings.tenant_id is None: - raise ValueError( - "STATUS_BACKEND=bus requires INGEST_BUS_URL and TENANT_ID " - "(guaranteed by Settings validation)" - ) - store = StatusStore(max_size=settings.vector_sync_queue_max_size) - subscriber = await NatsStatusSubscriber.connect( - url=settings.ingest_bus_url, - tenant_id=settings.tenant_id, - store=store, - ) - return store, subscriber - - @dataclass class AppContext: """Application context for BasicAuth mode.""" @@ -1681,39 +1651,40 @@ 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_MODE selects the transport - # (design §10.1): local uses the in-memory anyio stream + the - # in-process processor pool; external publishes to NATS and runs no - # in-process consumer (the document-processor consumes). - external = settings.ingest_mode == "external" + # 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" shutdown_event = anyio.Event() scanner_wake_event = anyio.Event() send_stream = None receive_stream = None task_producer: TaskProducer - if external: - task_producer = await build_external_producer(settings) - logger.info( - "Ingest mode external: publishing to %s", settings.ingest_bus_url + if use_postgres: + # Create procrastinate's tables before the scanner can defer. + # Lazy import: the procrastinate lib is a Postgres-only extra. + from nextcloud_mcp_server.vector.queue.procrastinate import ( # noqa: PLC0415 + apply_ingest_queue_schema, ) + + await apply_ingest_queue_schema() + task_producer = await build_producer(settings) + 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) - # Bus status backend: subscribe to mcp.document.* into a store the - # status endpoint reads (STATUS_BACKEND=bus; external mode only). - status_store, status_subscriber = await _build_status_subscriber(settings) - - # Store in app state for access from routes (ADR-007). In external + # 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 bus producer. + # 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.status_store = status_store app.state.shutdown_event = shutdown_event app.state.scanner_wake_event = scanner_wake_event @@ -1721,7 +1692,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = _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.status_store = status_store _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") @@ -1733,7 +1703,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = 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.status_store = status_store 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") @@ -1751,9 +1720,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = username, ) - # The in-process processor pool runs only in local mode; in - # external mode the document-processor service is the consumer. - if not external: + # 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( @@ -1765,10 +1734,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = username, ) - # Bus status subscriber (STATUS_BACKEND=bus). - if status_subscriber is not None: - await tg.start(status_subscriber.run, shutdown_event) - # Expose this long-lived task group to request-path code that # wants to spawn background work (e.g. ADR-019 verify-on-read # eviction). Eviction coroutines have their own try/except, so @@ -1776,9 +1741,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = _vector_sync_state.eviction_task_group = tg logger.info( - "Background sync tasks started: 1 scanner + %s processors (ingest=%s)", - 0 if external else settings.vector_sync_processor_workers, - settings.ingest_mode, + "Background sync tasks started: 1 scanner + %s processors (queue=%s)", + 0 if use_postgres else settings.vector_sync_processor_workers, + settings.ingest_queue, ) # Run MCP session manager and yield @@ -1791,12 +1756,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 - # Drain the shared bus connection (external mode only). + # Close the procrastinate connector pool (postgres mode). _drain = getattr(task_producer, "drain", None) - if external and _drain is not None: + if use_postgres and _drain is not None: await _drain() - if status_subscriber is not None: - await status_subscriber.aclose() await client.close() # TaskGroup automatically cancels all tasks on exit @@ -1904,11 +1867,12 @@ 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_MODE selects the transport - # (design §10.1): local uses the in-memory anyio stream + the - # in-process processor pool; external publishes to NATS and runs - # no in-process consumer (the document-processor consumes). - external = settings.ingest_mode == "external" + # 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" shutdown_event = anyio.Event() scanner_wake_event = anyio.Event() @@ -1918,11 +1882,17 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = send_stream = None receive_stream = None task_producer: TaskProducer - if external: - task_producer = await build_external_producer(settings) + if use_postgres: + # Create procrastinate's tables before any scanner defers. + # Lazy import: procrastinate is a Postgres-only extra. + from nextcloud_mcp_server.vector.queue.procrastinate import ( # noqa: PLC0415 + apply_ingest_queue_schema, + ) + + await apply_ingest_queue_schema() + task_producer = await build_producer(settings) logger.info( - "Ingest mode external: publishing to %s", - settings.ingest_bus_url, + "Ingest queue: postgres (procrastinate); worker drains it" ) else: send_stream, receive_stream = anyio.create_memory_object_stream[ @@ -1930,17 +1900,10 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = ](max_buffer_size=settings.vector_sync_queue_max_size) task_producer = MemoryTaskProducer(send_stream) - # Bus status backend: subscribe to mcp.document.* into a store - # the status endpoint reads (STATUS_BACKEND=bus; external only). - status_store, status_subscriber = await _build_status_subscriber( - 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.status_store = status_store app.state.shutdown_event = shutdown_event app.state.scanner_wake_event = scanner_wake_event @@ -1948,7 +1911,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = _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.status_store = status_store _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") @@ -1960,7 +1922,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = 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.status_store = status_store browser_app.state.shutdown_event = shutdown_event browser_app.state.scanner_wake_event = scanner_wake_event logger.info( @@ -1991,9 +1952,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = tg, ) - # In-process processor pool runs only in local mode; in - # external mode the document-processor service consumes. - if not external: + # 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( @@ -2004,10 +1965,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = nextcloud_host_for_sync, ) - # Bus status subscriber (STATUS_BACKEND=bus). - if status_subscriber is not None: - await tg.start(status_subscriber.run, shutdown_event) - # Expose this long-lived task group to request-path code # that wants to spawn background work (e.g. ADR-019 # verify-on-read eviction). Eviction coroutines have their @@ -2015,9 +1972,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = _vector_sync_state.eviction_task_group = tg logger.info( - "Background sync tasks started: 1 user manager + %s processors (ingest=%s)", - 0 if external else settings.vector_sync_processor_workers, - settings.ingest_mode, + "Background sync tasks started: 1 user manager + %s processors (queue=%s)", + 0 if use_postgres else settings.vector_sync_processor_workers, + settings.ingest_queue, ) # Run MCP session manager and yield @@ -2030,12 +1987,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 - # Drain the shared bus connection (external only). + # Close the procrastinate connector pool (postgres). _drain = getattr(task_producer, "drain", None) - if external and _drain is not None: + if use_postgres and _drain is not None: await _drain() - if status_subscriber is not None: - await status_subscriber.aclose() # Close token broker HTTP client if token_broker._http_client: await token_broker._http_client.aclose() diff --git a/nextcloud_mcp_server/auth/userinfo_routes.py b/nextcloud_mcp_server/auth/userinfo_routes.py index a2e56d51..85bc9771 100644 --- a/nextcloud_mcp_server/auth/userinfo_routes.py +++ b/nextcloud_mcp_server/auth/userinfo_routes.py @@ -115,17 +115,20 @@ async def _get_processing_status(request: Request) -> dict[str, Any] | None: return None try: - # Get document receive stream from app state - document_receive_stream = getattr( - request.app.state, "document_receive_stream", None + # Outstanding-work view depends on the queue backend (Deck #183): + # memory → stream buffer depth; postgres → procrastinate job counts (the + # in-memory stream is absent in postgres mode, so don't early-return on it). + from nextcloud_mcp_server.vector.ingest_status import ( # noqa: PLC0415 + get_ingest_pending, ) - if document_receive_stream is None: - logger.debug("document_receive_stream not available in app state") - return None - # Get pending count from stream statistics - stats = document_receive_stream.statistics() - pending_count = stats.current_buffer_used + pending = await get_ingest_pending( + task_producer=getattr(request.app.state, "task_producer", None), + document_receive_stream=getattr( + request.app.state, "document_receive_stream", None + ), + ingest_queue=settings.ingest_queue, + ) # Get Qdrant client and query indexed count indexed_count = 0 @@ -147,11 +150,11 @@ async def _get_processing_status(request: Request) -> dict[str, Any] | None: # Continue with indexed_count = 0 # Determine status - status = "syncing" if pending_count > 0 else "idle" + status = "syncing" if pending.pending > 0 else "idle" return { "indexed_count": indexed_count, - "pending_count": pending_count, + "pending_count": pending.pending, "status": status, } diff --git a/nextcloud_mcp_server/canonical.py b/nextcloud_mcp_server/canonical.py index 6872edfa..acdf26f7 100644 --- a/nextcloud_mcp_server/canonical.py +++ b/nextcloud_mcp_server/canonical.py @@ -2,9 +2,9 @@ The Astrolabe Cloud decomposition (design §2.3) fixes a single canonical JSON encoding so hashes computed here match those computed independently by the -external document-processor and embedding-gateway services. Any drift in -separators, key ordering, or unicode handling would break NATS dedup keys, -Qdrant point-ID idempotency, and ACL-hash compatibility. +external embedding-gateway service. Any drift in separators, key ordering, or +unicode handling would break Qdrant point-ID idempotency and ACL-hash +compatibility. """ from __future__ import annotations @@ -17,9 +17,8 @@ def canonical_json(obj: Any) -> bytes: """Encode ``obj`` to canonical JSON bytes. Deterministic across implementations: sorted keys, no inter-token - whitespace, non-ASCII preserved (UTF-8). Consumers: the NATS - ``Nats-Msg-Id`` dedup header (vector/queue/nats.py), Qdrant point IDs - (vector/payload_keys.py), and ACL hashes (acl_hash.py). + whitespace, non-ASCII preserved (UTF-8). Consumers: Qdrant point IDs + (vector/payload_keys.py) and ACL hashes (acl_hash.py). """ return json.dumps( obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False diff --git a/nextcloud_mcp_server/cli.py b/nextcloud_mcp_server/cli.py index 15e7de1f..fef400ca 100644 --- a/nextcloud_mcp_server/cli.py +++ b/nextcloud_mcp_server/cli.py @@ -281,6 +281,69 @@ def run( ) +@click.command() +@click.option( + "--concurrency", + "-c", + type=int, + default=None, + help="Max concurrent jobs. Defaults to VECTOR_SYNC_PROCESSOR_WORKERS.", +) +def worker(concurrency: int | None): + """Run the ingest worker (Deck #183). + + \b + Drains the per-tenant Postgres ingest queue (procrastinate): for each + deferred document it fetches the content as the owning user, parses, chunks, + embeds, and upserts into Qdrant. This is the scale-to-zero ``worker`` role of + the api/worker split; run it as a separate Deployment from the API pod. + + \b + Requires INGEST_QUEUE=postgres (a PostgreSQL DATABASE_URL); procrastinate is + Postgres-only. + + \b + Example: + $ export DATABASE_URL=postgresql+asyncpg://mcp:mcp@db/mcp + $ nextcloud-mcp-server worker -c 4 + """ + import anyio # noqa: PLC0415 + + settings = get_settings() + if settings.ingest_queue != "postgres": + raise click.ClickException( + "worker requires INGEST_QUEUE=postgres (a PostgreSQL DATABASE_URL); " + f"resolved INGEST_QUEUE={settings.ingest_queue!r}" + ) + + from nextcloud_mcp_server.vector.queue.procrastinate import ( # noqa: PLC0415 + INGEST_QUEUE_NAME, + apply_ingest_queue_schema, + get_procrastinate_app, + ) + + workers = concurrency or settings.vector_sync_processor_workers + app = get_procrastinate_app() + + async def _run() -> None: + # Defensive apply (the always-on API pod is the authoritative applier). + await apply_ingest_queue_schema(app) + async with app.open_async(): + click.echo( + f"Ingest worker started: queue={INGEST_QUEUE_NAME} concurrency={workers}" + ) + await app.run_worker_async( + queues=[INGEST_QUEUE_NAME], + concurrency=workers, + install_signal_handlers=True, + # Drop succeeded jobs so the queue table stays lean and the KEDA + # queue-depth metric reflects only outstanding work. + delete_jobs="successful", + ) + + anyio.run(_run) + + @click.group() def db(): """Database migration management commands.""" @@ -374,6 +437,21 @@ def upgrade(database_url: str | None, database_path: str | None, revision: str): try: click.echo(f"Upgrading database to revision: {revision}") upgrade_database(url, revision) + # Apply procrastinate's ingest-queue schema on Postgres so a one-shot + # migration/init job provisions everything the api + worker roles need + # (Deck #183). Idempotent + lazy import (Postgres-only extra). + from nextcloud_mcp_server.config import is_sqlite_url # noqa: PLC0415 + + if not is_sqlite_url(url): + import anyio # noqa: PLC0415 + + from nextcloud_mcp_server.vector.queue.procrastinate import ( # noqa: PLC0415 + apply_ingest_queue_schema, + build_app_for_url, + ) + + anyio.run(apply_ingest_queue_schema, build_app_for_url(url)) + click.echo(click.style("✓ Ingest queue schema applied", fg="green")) click.echo(click.style("✓ Database upgraded successfully", fg="green")) except Exception as e: click.echo(click.style(f"✗ Upgrade failed: {e}", fg="red"), err=True) @@ -483,6 +561,7 @@ def cli(): cli.add_command(run) +cli.add_command(worker) cli.add_command(db) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index a33220ee..189ae1fa 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -169,14 +169,19 @@ _DEFAULTS: dict[str, Any] = { # the current monolithic behavior; self-hosters who set none are # unaffected. See docs/architecture/mcp-decomposition.md (sibling repo). "embedding_provider": "autodetect", # autodetect | gateway - "ingest_mode": "local", # local | external - "status_backend": "local", # local | bus + # Ingest queue backend (Deck #183). None → auto: ``postgres`` (procrastinate) + # when DATABASE_URL is Postgres, else ``memory`` (the in-process anyio queue + # for SQLite/dev). procrastinate requires PostgreSQL. + "ingest_queue": None, # memory | postgres + # Process role for the per-tenant two-pod model (Deck #183). ``api`` runs the + # MCP/query server + scanner (defers jobs); the ``worker`` role is the + # `nextcloud-mcp-server worker` process that drains the queue. ``all`` keeps + # the monolithic behaviour (API + in-process SQLite pool). + "mcp_role": "all", # api | worker | all "collection_metadata_source": "qdrant", # qdrant | api # CP base URL for COLLECTION_METADATA_SOURCE=api (e.g. http://control-plane). # Required only when the source is api. "collection_metadata_api_url": None, - "fact_event_emitter": "none", # none | nats | stdout - "ingest_bus_url": None, # required when ingest_mode=external "embedding_gateway_url": None, # required when embedding_provider=gateway # Provider-namespaced model the gateway serves, "/" # (the gateway routes on the "/"-prefix; mistral/mistral-embed → Mistral @@ -191,8 +196,7 @@ _DEFAULTS: dict[str, Any] = { "embedding_gateway_client_id": None, "embedding_gateway_client_secret": None, "embedding_gateway_scope": None, # e.g. astrolabe-embedding-gateway/embed - "tenant_id": None, # NATS per-tenant subject token (UUID form) - "ingest_bus_num_replicas": 1, # JetStream stream replicas (prod: 3) + "tenant_id": None, # per-tenant identity (UUID form); see vector/payload_keys # Query-side ACL pre-filter (design §11). OFF by default: a Qdrant # `match any` on `acl_hash` excludes points missing the key, so enabling # this before a real ACL backfill would silently drop legacy results. @@ -699,12 +703,12 @@ class Settings: # MCP decomposition hook points (design §10, opt-in). All defaults # reproduce the current monolith; validated in __post_init__. embedding_provider: str = "autodetect" # autodetect | gateway - ingest_mode: str = "local" # local | external - status_backend: str = "local" # local | bus + # Ingest queue backend (Deck #183). None → resolved in __post_init__ to + # ``postgres`` when DATABASE_URL is Postgres, else ``memory``. + ingest_queue: str | None = None # memory | postgres + mcp_role: str = "all" # api | worker | all (Deck #183 two-pod model) collection_metadata_source: str = "qdrant" # qdrant | api collection_metadata_api_url: str | None = None # CP URL when source=api - fact_event_emitter: str = "none" # none | nats | stdout - ingest_bus_url: str | None = None # required when ingest_mode=external embedding_gateway_url: str | None = None # required when provider=gateway embedding_gateway_model: str = ( "mistral/mistral-embed" # provider-namespaced id the gateway routes on @@ -714,8 +718,7 @@ class Settings: embedding_gateway_client_id: str | None = None embedding_gateway_client_secret: str | None = None embedding_gateway_scope: str | None = None - tenant_id: str | None = None # NATS per-tenant subject token (UUID form) - ingest_bus_num_replicas: int = 1 # JetStream stream replicas (prod: 3) + tenant_id: str | None = None # per-tenant identity (UUID form) acl_prefilter_enabled: bool = False # query-side ACL pre-filter (§11); OFF def __post_init__(self): @@ -803,10 +806,8 @@ class Settings: # the monolith, so deployments that set none of these pass through. _enum_fields = { "embedding_provider": {"autodetect", "gateway"}, - "ingest_mode": {"local", "external"}, - "status_backend": {"local", "bus"}, + "mcp_role": {"api", "worker", "all"}, "collection_metadata_source": {"qdrant", "api"}, - "fact_event_emitter": {"none", "nats", "stdout"}, } for _field, _allowed in _enum_fields.items(): _val = (getattr(self, _field) or "").strip().lower() @@ -816,21 +817,25 @@ class Settings: f"{_field.upper()} must be one of {sorted(_allowed)}; got {_val!r}" ) - # Fail-fast: external ingest sources its status from the bus. With the - # in-process state machine empty, STATUS_BACKEND=local would leave - # status streams silently empty — crash loudly instead (design §10.1). - if self.status_backend == "local" and self.ingest_mode == "external": - raise RuntimeError( - "STATUS_BACKEND=local is incompatible with INGEST_MODE=external; " - "set STATUS_BACKEND=bus" + # Ingest queue backend (Deck #183). Unset → auto-derive from the + # database backend: procrastinate needs PostgreSQL, so SQLite/dev falls + # back to the in-process anyio queue. An explicit ``postgres`` against a + # SQLite DATABASE_URL is a misconfiguration — fail loudly. + _queue = (self.ingest_queue or "").strip().lower() + if not _queue: + _queue = "memory" if is_sqlite_url(get_database_url()) else "postgres" + if _queue not in {"memory", "postgres"}: + raise ValueError( + f"INGEST_QUEUE must be one of ['memory', 'postgres']; got {_queue!r}" + ) + self.ingest_queue = _queue + if self.ingest_queue == "postgres" and is_sqlite_url(get_database_url()): + raise ValueError( + "INGEST_QUEUE=postgres requires a PostgreSQL DATABASE_URL " + "(procrastinate is Postgres-only); use INGEST_QUEUE=memory for " + "SQLite/dev" ) - # Conditional-required settings for the active hook points. - if self.ingest_mode == "external": - if not self.ingest_bus_url: - raise ValueError("INGEST_BUS_URL is required when INGEST_MODE=external") - if not self.tenant_id: - raise ValueError("TENANT_ID is required when INGEST_MODE=external") if self.embedding_provider == "gateway" and not self.embedding_gateway_url: raise ValueError( "EMBEDDING_GATEWAY_URL is required when EMBEDDING_PROVIDER=gateway" @@ -859,23 +864,6 @@ class Settings: "client-credentials) or all left unset (unauthenticated gateway)" ) - # TENANT_ID is a NATS subject token; '.', '*', '>', and whitespace are - # reserved/illegal there and would silently break subscriptions (§3.4). - if self.tenant_id and ( - any(c in self.tenant_id for c in ".*>") - or any(c.isspace() for c in self.tenant_id) - ): - raise ValueError( - "TENANT_ID must not contain '.', '*', '>', or whitespace " - "(it is used as a NATS subject token)" - ) - - if self.ingest_bus_num_replicas < 1: - raise ValueError( - f"INGEST_BUS_NUM_REPLICAS must be >= 1; " - f"got {self.ingest_bus_num_replicas}" - ) - # --- ADR-022 follow-up: deployment mode is the single source of truth --- # The ENABLE_MULTI_USER_BASIC_AUTH and ENABLE_LOGIN_FLOW env vars were # removed in favour of MCP_DEPLOYMENT_MODE. We do TWO things here: @@ -1316,12 +1304,10 @@ def get_settings() -> Settings: "excluded_tags": "EXCLUDED_TAGS", # MCP decomposition hook points (design §10) "embedding_provider": "EMBEDDING_PROVIDER", - "ingest_mode": "INGEST_MODE", - "status_backend": "STATUS_BACKEND", + "ingest_queue": "INGEST_QUEUE", + "mcp_role": "MCP_ROLE", "collection_metadata_source": "COLLECTION_METADATA_SOURCE", "collection_metadata_api_url": "COLLECTION_METADATA_API_URL", - "fact_event_emitter": "FACT_EVENT_EMITTER", - "ingest_bus_url": "INGEST_BUS_URL", "embedding_gateway_url": "EMBEDDING_GATEWAY_URL", "embedding_gateway_model": "EMBEDDING_GATEWAY_MODEL", "embedding_gateway_token_url": "EMBEDDING_GATEWAY_TOKEN_URL", @@ -1329,7 +1315,6 @@ def get_settings() -> Settings: "embedding_gateway_client_secret": "EMBEDDING_GATEWAY_CLIENT_SECRET", "embedding_gateway_scope": "EMBEDDING_GATEWAY_SCOPE", "tenant_id": "TENANT_ID", - "ingest_bus_num_replicas": "INGEST_BUS_NUM_REPLICAS", "acl_prefilter_enabled": "ACL_PREFILTER_ENABLED", } @@ -1405,3 +1390,76 @@ def get_database_ssl() -> bool | ssl.SSLContext | None: if settings.database_verify_ssl is True: return True return None + + +def _pg_ssl_params() -> dict[str, str]: + """Map the DATABASE_VERIFY_SSL / DATABASE_CA_BUNDLE settings to libpq + keyword params for psycopg3 (used by procrastinate, Deck #183). + + psycopg/libpq takes ``sslmode`` (and ``sslrootcert``) rather than an + ``ssl.SSLContext`` like asyncpg, so we translate :func:`get_database_ssl`'s + intent into the equivalent libpq settings: + + - ``None`` (both unset) → ``{}`` (omit; libpq default ``prefer``, + matching the asyncpg default for cluster-local Postgres without TLS). + - ``False`` (DATABASE_VERIFY_SSL=false) → ``sslmode=require`` (encrypt but + do not verify the certificate). + - CA bundle set → ``sslmode=verify-full`` + ``sslrootcert``. + - ``True`` (verify, no bundle) → ``sslmode=verify-full`` (system trust). + """ + ssl_setting = get_database_ssl() + if ssl_setting is None: + return {} + if ssl_setting is False: + return {"sslmode": "require"} + settings = get_settings() + if settings.database_ca_bundle: + return {"sslmode": "verify-full", "sslrootcert": settings.database_ca_bundle} + return {"sslmode": "verify-full"} + + +def get_procrastinate_conninfo(database_url: str | None = None) -> str: + """Build a libpq conninfo string for procrastinate's psycopg3 connector. + + Derives the connection from ``DATABASE_URL`` (a SQLAlchemy URL such as + ``postgresql+asyncpg://user:pass@host/db``): the SQLAlchemy driver suffix + (``+asyncpg``/``+psycopg``) is stripped and the parts are rendered via + :func:`psycopg.conninfo.make_conninfo`, which quotes values correctly (never + f-string the password). TLS settings are appended from :func:`_pg_ssl_params`. + + This is driver-agnostic on purpose: procrastinate uses psycopg3 regardless of + which SQLAlchemy driver the app's own engine uses, so it works whether + ``DATABASE_URL`` carries ``+asyncpg`` or ``+psycopg``. + + TODO(Deck #183 follow-up, out-of-tree): unify the app's SQLAlchemy engine on + psycopg3 too (``postgresql+psycopg://``) and drop asyncpg, so the deployment + ships a single Postgres driver. This belongs in the rendered Helm chart + (set ``DATABASE_URL`` to a ``+psycopg`` URL) rather than rewriting the driver + in code — see charts repo, not this repo. + + Raises ``ValueError`` for a non-Postgres URL — procrastinate is Postgres-only. + """ + from psycopg.conninfo import make_conninfo # noqa: PLC0415 + from sqlalchemy.engine.url import make_url # noqa: PLC0415 + + url = make_url(database_url or get_database_url()) + if not url.drivername.startswith("postgresql"): + raise ValueError( + "get_procrastinate_conninfo requires a PostgreSQL DATABASE_URL; " + f"got driver {url.drivername!r}" + ) + + params: dict[str, str] = {} + if url.host: + params["host"] = url.host + if url.port: + params["port"] = str(url.port) + if url.database: + params["dbname"] = url.database + if url.username: + params["user"] = url.username + if url.password: + params["password"] = url.password + params.update(_pg_ssl_params()) + + return make_conninfo(**params) diff --git a/nextcloud_mcp_server/models/semantic.py b/nextcloud_mcp_server/models/semantic.py index bb0091dc..ae586ee8 100644 --- a/nextcloud_mcp_server/models/semantic.py +++ b/nextcloud_mcp_server/models/semantic.py @@ -170,6 +170,17 @@ class VectorSyncStatusResponse(BaseResponse): description='Sync status: "idle", "syncing", or "disabled"', ) enabled: bool = Field(default=False, description="Whether vector sync is enabled") + ingest_queue: str | None = Field( + default=None, + description='Ingest queue backend: "memory" or "postgres" (Deck #183)', + ) + job_counts: dict[str, int] | None = Field( + default=None, + description=( + "Per-status ingest job counts (todo/doing/failed/…) on the postgres " + "queue backend; None on the in-memory backend" + ), + ) __all__ = [ diff --git a/nextcloud_mcp_server/server/semantic.py b/nextcloud_mcp_server/server/semantic.py index 26e61967..f3f29cb2 100644 --- a/nextcloud_mcp_server/server/semantic.py +++ b/nextcloud_mcp_server/server/semantic.py @@ -945,23 +945,21 @@ def configure_semantic_tools(mcp: FastMCP): # missing attribute is a typo that should fail loudly. The # value itself can legitimately be ``None`` before sync starts, # which the check below handles. + # Outstanding-work view depends on the queue backend (Deck #183): + # memory → stream buffer depth; postgres → procrastinate job counts. + # Direct attribute access matches the eviction_task_group pattern at + # ``nc_semantic_search``: both AppContext and OAuthAppContext define + # these, so a missing attribute is a typo that should fail loudly. + from nextcloud_mcp_server.vector.ingest_status import ( # noqa: PLC0415 + get_ingest_pending, + ) + lifespan_ctx = ctx.request_context.lifespan_context - document_receive_stream = lifespan_ctx.document_receive_stream - - if document_receive_stream is None: - logger.debug( - "document_receive_stream not available in lifespan context" - ) - return VectorSyncStatusResponse( - indexed_count=0, - pending_count=0, - status="unknown", - enabled=True, - ) - - # Get pending count from stream statistics - stream_stats = document_receive_stream.statistics() - pending_count = stream_stats.current_buffer_used + pending = await get_ingest_pending( + task_producer=lifespan_ctx.task_producer, + document_receive_stream=lifespan_ctx.document_receive_stream, + ingest_queue=settings.ingest_queue, + ) # Get Qdrant client and query indexed count indexed_count = 0 @@ -981,13 +979,15 @@ def configure_semantic_tools(mcp: FastMCP): # Continue with indexed_count = 0 # Determine status - status = "syncing" if pending_count > 0 else "idle" + status = "syncing" if pending.pending > 0 else "idle" return VectorSyncStatusResponse( indexed_count=indexed_count, - pending_count=pending_count, + pending_count=pending.pending, status=status, enabled=True, + ingest_queue=settings.ingest_queue, + job_counts=pending.job_counts, ) except Exception as e: diff --git a/nextcloud_mcp_server/vector/ingest_status.py b/nextcloud_mcp_server/vector/ingest_status.py new file mode 100644 index 00000000..ae90aa48 --- /dev/null +++ b/nextcloud_mcp_server/vector/ingest_status.py @@ -0,0 +1,56 @@ +"""Shared read model for the vector-sync status surface (Deck #183). + +The status endpoints (``/api/v1/vector-sync/status``, the userinfo route, and +the ``nc_get_vector_sync_status`` MCP tool) all need the same "how much work is +outstanding" figure, computed differently per ``INGEST_QUEUE`` backend: + +- ``memory`` — the in-process anyio stream's buffer depth (today's behavior). +- ``postgres`` — procrastinate job counts read from the per-tenant Postgres + (``todo`` + ``doing``), plus the per-status breakdown for observability. + +``indexed_documents`` (the Qdrant placeholder count) is backend-independent and +stays at each call site. +""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from typing import Any + +logger = logging.getLogger(__name__) + + +@dataclass +class IngestPending: + """Outstanding-work view for the active ingest queue backend.""" + + pending: int + # Per-status counts (todo/doing/failed/…) on the postgres backend; None on + # the memory backend, which has no durable per-status breakdown. + job_counts: dict[str, int] | None = None + + +async def get_ingest_pending( + *, task_producer: Any, document_receive_stream: Any, ingest_queue: str | None +) -> IngestPending: + """Compute outstanding ingest work for the configured queue backend. + + Never raises — a status surface must stay available even if the queue is + unreachable; failures degrade to ``pending=0``. + """ + if ingest_queue == "postgres": + counts: dict[str, int] = {} + if task_producer is not None and hasattr(task_producer, "job_counts"): + try: + counts = await task_producer.job_counts() + except Exception as e: + logger.warning("Failed to read ingest job counts: %s", e) + pending = counts.get("todo", 0) + counts.get("doing", 0) + return IngestPending(pending=pending, job_counts=counts) + + if document_receive_stream is None: + return IngestPending(pending=0) + return IngestPending( + pending=document_receive_stream.statistics().current_buffer_used + ) diff --git a/nextcloud_mcp_server/vector/processor.py b/nextcloud_mcp_server/vector/processor.py index cfd17567..d6d7c475 100644 --- a/nextcloud_mcp_server/vector/processor.py +++ b/nextcloud_mcp_server/vector/processor.py @@ -154,7 +154,9 @@ async def processor_task( logger.info("Processor %s stopped", worker_id) -async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): +async def process_document( + doc_task: DocumentTask, nc_client: NextcloudClient, *, max_retries: int = 3 +): """ Process a single document: fetch, tokenize, embed, store in Qdrant. @@ -163,6 +165,10 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): Args: doc_task: Document task to process nc_client: Authenticated Nextcloud client + max_retries: In-process indexing attempts before re-raising. The default + (3) suits the in-process SQLite pool, which has no durable retry. The + procrastinate worker passes ``1`` so durable retry is owned by the + queue (and survives worker crashes), avoiding compounding 3×N retries. """ start_time = time.time() @@ -230,7 +236,6 @@ async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient): return # Handle indexing with retry - max_retries = 3 retry_delay = 1.0 for attempt in range(max_retries): diff --git a/nextcloud_mcp_server/vector/queue/__init__.py b/nextcloud_mcp_server/vector/queue/__init__.py index dcc55e2b..a0799e50 100644 --- a/nextcloud_mcp_server/vector/queue/__init__.py +++ b/nextcloud_mcp_server/vector/queue/__init__.py @@ -1,7 +1,7 @@ -"""Ingest-path ports & adapters (design §10, hexagonal).""" +"""Ingest-path ports & adapters (design §10, hexagonal; Deck #183).""" -from .factory import build_external_producer +from .factory import build_producer from .memory import MemoryTaskProducer from .ports import TaskProducer -__all__ = ["MemoryTaskProducer", "TaskProducer", "build_external_producer"] +__all__ = ["MemoryTaskProducer", "TaskProducer", "build_producer"] diff --git a/nextcloud_mcp_server/vector/queue/factory.py b/nextcloud_mcp_server/vector/queue/factory.py index cc673941..4628f959 100644 --- a/nextcloud_mcp_server/vector/queue/factory.py +++ b/nextcloud_mcp_server/vector/queue/factory.py @@ -1,16 +1,17 @@ -"""Composition root for the ingest producer (design §10). +"""Composition root for the ingest producer (Deck #183). -``build_external_producer`` is called by the lifespan only when -``INGEST_MODE=external``; local mode uses the in-memory stream directly (it -already satisfies :class:`TaskProducer`). The transport under ``external`` is -selected from the ``INGEST_BUS_URL`` scheme (``nats://`` now, ``postgres://`` -later) so moving the external processor to Postgres needs no new INGEST_MODE. +The transport is selected from ``INGEST_QUEUE``: + +- ``postgres`` → :class:`ProcrastinateTaskProducer`, which defers jobs into the + per-tenant Postgres for the out-of-process ``worker`` role to drain. +- ``memory`` (SQLite/dev default) → the in-process anyio stream, built inline by + the server lifespan (it owns both the send and receive ends), so it is not + produced here. """ from __future__ import annotations import logging -from urllib.parse import urlsplit from ...config import Settings from .ports import TaskProducer @@ -18,43 +19,19 @@ from .ports import TaskProducer logger = logging.getLogger(__name__) -def _transport_for(url: str) -> str: - scheme = urlsplit(url).scheme.lower() - if scheme.startswith("postgres"): - return "postgres" - if not scheme.startswith("nats"): - logger.warning( - "INGEST_BUS_URL scheme %r is neither nats:// nor postgres://; " - "defaulting to the NATS transport", - scheme, - ) - return "nats" +async def build_producer(settings: Settings) -> TaskProducer: + """Build the Postgres (procrastinate) ingest producer. - -async def build_external_producer(settings: Settings) -> TaskProducer: - """Build the external-ingest producer for the configured transport. - - Precondition: ``settings.ingest_mode == "external"`` (so __post_init__ has - guaranteed ``ingest_bus_url`` and ``tenant_id`` are set). + Precondition: ``settings.ingest_queue == "postgres"`` (the memory transport + is constructed inline by the lifespan because it needs the paired receive + stream for the in-process processor pool). """ - # Defence-in-depth (robust under ``python -O``, which strips asserts): - # __post_init__ already guarantees these when ingest_mode == external. - if settings.ingest_bus_url is None or settings.tenant_id is None: + if settings.ingest_queue != "postgres": raise ValueError( - "build_external_producer requires INGEST_BUS_URL and TENANT_ID " - "(guaranteed by Settings validation when INGEST_MODE=external)" + "build_producer is only for INGEST_QUEUE=postgres; the memory " + f"transport is built inline by the lifespan (got {settings.ingest_queue!r})" ) - transport = _transport_for(settings.ingest_bus_url) - if transport == "postgres": - from .postgres import PostgresTaskProducer # noqa: PLC0415 + from .procrastinate import ProcrastinateTaskProducer # noqa: PLC0415 - return await PostgresTaskProducer.connect(settings) - - from .nats import NatsTaskProducer # noqa: PLC0415 - - return await NatsTaskProducer.connect( - url=settings.ingest_bus_url, - tenant_id=settings.tenant_id, - num_replicas=settings.ingest_bus_num_replicas, - ) + return await ProcrastinateTaskProducer.connect() diff --git a/nextcloud_mcp_server/vector/queue/nats.py b/nextcloud_mcp_server/vector/queue/nats.py deleted file mode 100644 index 58e38f55..00000000 --- a/nextcloud_mcp_server/vector/queue/nats.py +++ /dev/null @@ -1,173 +0,0 @@ -"""NATS JetStream ``TaskProducer`` — external ingest transport (design §3.4). - -Publishes ``mcp.ingest.requested.{tenant_id}`` for the external -document-processor to consume. Translates the in-process ``DocumentTask`` into -the wire ``IngestMessage`` schema (mirrored in astrolabe-cloud-website's -``bus/messages.py``), with the JetStream ``Nats-Msg-Id`` dedup header per §3.4. - -This server is only the *producer* on this transport; the document-processor -owns the consumer. ``nats-py`` is imported lazily so deployments that never -enable external ingest don't pay the import. -""" - -from __future__ import annotations - -import hashlib -import logging -from datetime import datetime, timezone -from types import TracebackType -from typing import TYPE_CHECKING, Any - -from ...canonical import canonical_json - -if TYPE_CHECKING: - from ..scanner import DocumentTask - -logger = logging.getLogger(__name__) - -STREAM_NAME = "mcp" -INGEST_SUBJECT_PREFIX = "mcp.ingest.requested" - - -def warn_if_insecure_nats_url(url: str) -> None: - """Log a warning when the bus URL is not TLS-encrypted. - - ``nats://`` (and ``ws://``) carry tenant document metadata in cleartext; - production deployments should use ``tls://`` (or ``wss://``). We connect - regardless — this is an operator alert, not a hard failure. - """ - scheme = url.split("://", 1)[0].lower() - if scheme not in ("tls", "wss"): - logger.warning( - "NATS bus URL uses unencrypted transport (scheme=%s://); " - "use tls:// in production to protect document metadata in transit", - scheme, - ) - - -def _modified_at_rfc3339(modified_at: int) -> str: - """DocumentTask.modified_at is an epoch int (0 for deletes).""" - return datetime.fromtimestamp(int(modified_at), tz=timezone.utc).isoformat() - - -def _content_hash(task: DocumentTask) -> str: - """etag is the change-detection token; fall back to modified_at when it is - absent (e.g. deletes, or sources whose etag we don't thread through). - - TODO(follow-up, PR #814 review): thread etags for file / deck_card / - news_item scans too (only note scans pass etag today). Until then their - JetStream Nats-Msg-Id dedup keys off modified_at, which misses content - changes that leave modified_at unchanged (e.g. a file move/rename). - """ - return task.etag or str(task.modified_at) - - -def msg_id(tenant_id: str, doc_id: str, modified_at_rfc3339: str) -> str: - """JetStream dedup header per §3.4. SHA-256 over canonical JSON (NOT the - BLAKE2b helper) — it is an opaque external header, not a stored field.""" - return hashlib.sha256( - canonical_json( - { - "tenant_id": tenant_id, - "doc_id": doc_id, - "modified_at": modified_at_rfc3339, - } - ) - ).hexdigest() - - -class NatsTaskProducer: - """Publishes ingest requests to NATS JetStream.""" - - def __init__(self, nc: Any, js: Any, tenant_id: str): - self._nc = nc - self._js = js - self.tenant_id = tenant_id - - @classmethod - async def connect( - cls, *, url: str, tenant_id: str, num_replicas: int = 1 - ) -> NatsTaskProducer: - import nats # noqa: PLC0415 (lazy: optional dependency for external mode) - - warn_if_insecure_nats_url(url) - nc = await nats.connect(url) - js = nc.jetstream() - await cls._ensure_stream(js, num_replicas) - logger.info("Connected NATS ingest producer: url=%s, tenant=%s", url, tenant_id) - return cls(nc, js, tenant_id) - - @staticmethod - async def _ensure_stream(js: Any, num_replicas: int) -> None: - # noqa: PLC0415 — nats.js types are only importable once nats-py is present. - from nats.js.api import RetentionPolicy, StreamConfig # noqa: PLC0415 - - config = StreamConfig( - name=STREAM_NAME, - subjects=["mcp.>"], - retention=RetentionPolicy.LIMITS, - num_replicas=num_replicas, - ) - try: - await js.add_stream(config=config) - logger.info("nats.stream_created stream=%s", STREAM_NAME) - except Exception as exc: - # add_stream is idempotent in spirit but errors when the stream - # already exists; treat as benign (mirrors the processor's - # ensure_stream). A genuinely broken broker surfaces on publish. - logger.info("nats.stream_exists_or_unavailable detail=%s", exc) - - def ingest_message(self, task: DocumentTask) -> dict[str, Any]: - """DocumentTask → wire IngestMessage dict (mirrors the sibling schema).""" - return { - "tenant_id": self.tenant_id, - "doc_id": task.doc_id, - "content_hash": _content_hash(task), - "modified_at": _modified_at_rfc3339(task.modified_at), - "doc_type": task.doc_type, - "operation": task.operation, - "user_id": task.user_id, - "file_path": task.file_path, - } - - async def send(self, task: DocumentTask) -> None: - message = self.ingest_message(task) - subject = f"{INGEST_SUBJECT_PREFIX}.{self.tenant_id}" - headers = { - "Nats-Msg-Id": msg_id(self.tenant_id, task.doc_id, message["modified_at"]) - } - await self._js.publish(subject, canonical_json(message), headers=headers) - - # The scanner/oauth_sync use the producer as a clone-able async context - # manager (memory-stream semantics). The bus connection is owned by the - # lifespan, so cloning shares it and __aexit__ is a no-op (close happens via - # aclose() on shutdown). - def clone(self) -> NatsTaskProducer: - return self - - async def __aenter__(self) -> NatsTaskProducer: - return self - - async def __aexit__( - self, - exc_type: type[BaseException] | None, - exc: BaseException | None, - tb: TracebackType | None, - ) -> None: - return None - - # The bare suppression marker silences S7503 (async method without await): - # ``async def`` is required by the TaskProducer protocol, but this handle - # close is a genuine no-op. - async def aclose(self) -> None: # NOSONAR - # Per-handle close (e.g. a per-user scanner clone exiting). The bus - # connection is shared and owned by the lifespan, so this is a no-op; - # the connection is torn down once via ``drain()`` on shutdown. - return None - - async def drain(self) -> None: - """Drain + close the shared NATS connection (lifespan shutdown only).""" - try: - await self._nc.drain() - except Exception: - logger.warning("NATS drain on shutdown failed", exc_info=True) diff --git a/nextcloud_mcp_server/vector/queue/ports.py b/nextcloud_mcp_server/vector/queue/ports.py index 978bad4c..39855e96 100644 --- a/nextcloud_mcp_server/vector/queue/ports.py +++ b/nextcloud_mcp_server/vector/queue/ports.py @@ -1,18 +1,18 @@ -"""Ingest-path ports (design §10, hexagonal). +"""Ingest-path ports (design §10, hexagonal; Deck #183). A ``TaskProducer`` is where the scanner + webhook receiver send a ``DocumentTask``. The transport behind it is swappable: -- the in-process anyio ``MemoryObjectSendStream`` (local ingest — the default), -- ``NatsTaskProducer`` (external ingest → the document-processor), and -- a future Postgres-queue producer (seam only; the *external* processor owns the - consume side — see ``postgres.py``). +- ``MemoryTaskProducer`` over the in-process anyio ``MemoryObjectSendStream`` + (``INGEST_QUEUE=memory`` — the SQLite/dev default), and +- ``ProcrastinateTaskProducer`` (``INGEST_QUEUE=postgres``), which defers jobs + into the per-tenant Postgres for the out-of-process ``worker`` role to drain. The protocol is exactly the surface the scanner/oauth_sync already use on the memory stream (``send`` + ``clone`` + ``async with``), so both adapters drop in -with only a type-annotation change at the call sites. There is intentionally NO -consumer port: the MCP server's only in-process consumer is the memory stream; -when ingest is external the document-processor is the consumer, not this server. +with only a type-annotation change at the call sites. There is no consumer port: +in memory mode the in-process processor pool is the consumer; in postgres mode +the procrastinate worker is. """ from __future__ import annotations diff --git a/nextcloud_mcp_server/vector/queue/postgres.py b/nextcloud_mcp_server/vector/queue/postgres.py deleted file mode 100644 index dc15fc27..00000000 --- a/nextcloud_mcp_server/vector/queue/postgres.py +++ /dev/null @@ -1,52 +0,0 @@ -"""Postgres-queue ``TaskProducer`` — documented seam, not implemented. - -The external document-processor may later drain a Postgres-backed queue instead -of NATS to limit NATS operational overhead. Processing stays *external*; only -the transport changes — so on this server it would be a drop-in producer swap. -The consume side + the queue-table migration belong to that processor-side -refactor (cross-repo), NOT here. This stub exists so the transport value and the -``TaskProducer`` Protocol conformance are testable today. -""" - -from __future__ import annotations - -from types import TracebackType -from typing import TYPE_CHECKING, Any - -if TYPE_CHECKING: - from ..scanner import DocumentTask - -_NOT_IMPLEMENTED = ( - "Postgres ingest transport is a documented seam. The external " - "document-processor owns the Postgres-drain refactor (transport swap only; " - "processing stays external). Use INGEST_BUS_URL=nats://… for now." -) - - -class PostgresTaskProducer: - @classmethod - async def connect(cls, settings: Any) -> PostgresTaskProducer: - raise NotImplementedError(_NOT_IMPLEMENTED) - - async def send(self, task: DocumentTask) -> None: # pragma: no cover - raise NotImplementedError(_NOT_IMPLEMENTED) - - def clone(self) -> PostgresTaskProducer: # pragma: no cover - return self - - async def __aenter__(self) -> PostgresTaskProducer: # pragma: no cover - return self - - async def __aexit__( - self, - exc_type: type[BaseException] | None, - exc: BaseException | None, - tb: TracebackType | None, - ) -> None: # pragma: no cover - return None - - # The bare suppression marker silences S7503 (async method without await): - # ``async def`` is required by the TaskProducer protocol; this stub is a - # no-op until the Postgres transport lands. - async def aclose(self) -> None: # NOSONAR # pragma: no cover - return None diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py new file mode 100644 index 00000000..a68d75b2 --- /dev/null +++ b/nextcloud_mcp_server/vector/queue/procrastinate.py @@ -0,0 +1,335 @@ +"""Procrastinate-backed ingest queue — the Postgres ``TaskProducer`` + worker +(Deck #183). + +This replaces NATS JetStream and the old Postgres-queue stub. The MCP server now +owns *both* sides of ingest: + +- **Producer** (API role / scanner) — :class:`ProcrastinateTaskProducer.send` + *defers* one ``ingest:process_document`` job per changed document into the + per-tenant Postgres (the same app DB; procrastinate manages its own tables). +- **Consumer** (worker role) — ``nextcloud-mcp-server worker`` runs + :func:`procrastinate.App.run_worker`, which drains the ``ingest`` queue and + invokes the existing :func:`process_document` pipeline. + +Design notes: + +- **No execution ``lock``, only ``queueing_lock``.** procrastinate does NOT + auto-reclaim ``doing`` jobs, so a per-doc execution lock would permanently + deadlock a document if a worker crashed mid-job. The Qdrant upsert is + idempotent (deterministic ``uuid5`` point IDs), so a concurrent/re-run is + harmless; ``queueing_lock`` (partial-unique on ``status='todo'``) is enough to + dedupe enqueues, and :func:`reclaim_stalled_ingest_jobs` retries jobs orphaned + in ``doing`` by a crash. +- procrastinate is Postgres-only and uses asyncio; ``anyio`` runs natively on the + asyncio backend, so the worker can call the anyio-based pipeline directly. +- Tasks are defined on a :class:`procrastinate.Blueprint` so the connector is + decoupled from the task registry: production binds a real + :class:`PsycopgConnector`; unit tests bind ``testing.InMemoryConnector``. +""" + +from __future__ import annotations + +import logging +from dataclasses import asdict +from datetime import datetime, timezone +from types import TracebackType +from typing import TYPE_CHECKING + +from procrastinate import App, Blueprint, JobContext, PsycopgConnector, RetryStrategy +from procrastinate.connector import BaseConnector +from procrastinate.exceptions import AlreadyEnqueued + +from ...config import get_procrastinate_conninfo, get_settings +from ..scanner import DocumentTask + +if TYPE_CHECKING: + from ...client import NextcloudClient + +logger = logging.getLogger(__name__) + +# Single queue for document ingest. KEDA scales the worker Deployment on the +# depth of this queue (``SELECT count(*) FROM procrastinate_jobs WHERE +# queue_name='ingest' AND status='todo'``). +INGEST_QUEUE_NAME = "ingest" +# Blueprint namespace → registered task names are prefixed ``ingest:``. +_NAMESPACE = "ingest" +INGEST_TASK_NAME = f"{_NAMESPACE}:process_document" +_RECLAIM_TASK_NAME = f"{_NAMESPACE}:reclaim_stalled_jobs" + +# A crashed worker leaves its job in ``doing``; reclaim it once its (per-worker) +# heartbeat is this many seconds stale. Sized well above the longest expected +# ``process_document`` (PDF render + embedding) so a slow-but-live worker — whose +# heartbeat stays current during a long job — is never reclaimed out from under +# itself. +_STALLED_AFTER_SECONDS = 300 + + +# Tasks are defined as plain functions and registered onto a *fresh* Blueprint +# per app (see _build_ingest_blueprint). procrastinate's add_tasks_from mutates +# the blueprint's task names in place (namespace prefixing), so a single shared +# Blueprint cannot be added to more than one App — which the tests (in-memory + +# real Postgres) and any re-init path require. +async def process_document_task( + *, + user_id: str, + doc_id: str, + doc_type: str, + operation: str, + modified_at: int, + file_path: str | None = None, + metadata: dict[str, int | str] | None = None, + etag: str | None = None, + owner_id: str | None = None, +) -> None: + """Worker entry: rebuild the DocumentTask, resolve creds, run the pipeline.""" + # Local imports avoid a heavy import chain at blueprint-definition time + # (this module is also imported by the API pod just to defer jobs). + from ..oauth_sync import NotProvisionedError # noqa: PLC0415 + from ..processor import process_document # noqa: PLC0415 + + task = DocumentTask( + user_id=user_id, + doc_id=doc_id, + doc_type=doc_type, + operation=operation, + modified_at=modified_at, + file_path=file_path, + metadata=metadata, + etag=etag, + owner_id=owner_id, + ) + try: + nc_client = await _resolve_client(user_id) + except NotProvisionedError: + # A deprovisioned user must not pin a worker slot retrying forever. + # Finish the job as a no-op; the next scan re-enqueues once the user + # re-provisions an app password. Other errors (transient DB/network) + # propagate so procrastinate's retry strategy handles them. + logger.warning( + "ingest.skip_no_credentials user=%s doc=%s:%s", user_id, doc_type, doc_id + ) + return + + try: + # Durable retry is procrastinate's job; disable the in-process loop. + await process_document(task, nc_client, max_retries=1) + finally: + await nc_client.close() + + +async def reclaim_stalled_ingest_jobs(context: JobContext, timestamp: int) -> None: + """Re-queue ingest jobs orphaned in ``doing`` by a crashed worker. + + procrastinate prunes dead *workers* but does not reset their in-flight jobs; + without this they'd sit in ``doing`` forever. ``timestamp`` is procrastinate's + periodic-run marker (unused). + """ + manager = context.app.job_manager + retry_at = datetime.now(tz=timezone.utc) + reclaimed = 0 + for job in await manager.get_stalled_jobs( + queue=INGEST_QUEUE_NAME, seconds_since_heartbeat=_STALLED_AFTER_SECONDS + ): + if job.id is None: + continue + await manager.retry_job_by_id_async(job_id=job.id, retry_at=retry_at) + reclaimed += 1 + if reclaimed: + logger.warning("ingest.reclaimed_stalled_jobs count=%d", reclaimed) + + +async def _resolve_client(user_id: str) -> NextcloudClient: + """Build an authenticated NextcloudClient for ``user_id`` in the worker. + + Single-user BasicAuth uses the shared env credentials; every multi-user mode + resolves the user's locally-stored app password (BasicAuth). + """ + from ...client import NextcloudClient # noqa: PLC0415 + from ...config_validators import AuthMode, detect_auth_mode # noqa: PLC0415 + + settings = get_settings() + if detect_auth_mode(settings) == AuthMode.SINGLE_USER_BASIC: + return NextcloudClient.from_env() + + from ..oauth_sync import get_user_client_basic_auth # noqa: PLC0415 + + host = settings.nextcloud_host + if not host: + raise ValueError("NEXTCLOUD_HOST is required for multi-user ingest") + return await get_user_client_basic_auth(user_id, host) + + +def _build_ingest_blueprint() -> Blueprint: + """Create a fresh Blueprint with the ingest tasks registered. + + Fresh per call because ``add_tasks_from`` mutates the blueprint's task names + (namespace prefixing), so the same Blueprint cannot be reused across Apps. + """ + bp = Blueprint() + # Durable retry owned by the queue (survives worker crashes); the in-process + # retry loop in process_document is disabled on this path via max_retries=1. + bp.task( + name="process_document", + queue=INGEST_QUEUE_NAME, + retry=RetryStrategy(max_attempts=5, exponential_wait=4), + )(process_document_task) + reclaim = bp.task( + name="reclaim_stalled_jobs", queue=INGEST_QUEUE_NAME, pass_context=True + )(reclaim_stalled_ingest_jobs) + bp.periodic(cron="*/5 * * * *", periodic_id="reclaim_stalled_ingest")(reclaim) + return bp + + +def build_app(connector: BaseConnector) -> App: + """Build an App for the given connector with the ingest tasks registered. + + Shared by production (:func:`get_procrastinate_app`) and tests (which pass a + ``testing.InMemoryConnector``). + """ + app = App(connector=connector) + app.add_tasks_from(_build_ingest_blueprint(), namespace=_NAMESPACE) + return app + + +def build_app_for_url(database_url: str) -> App: + """Build an App bound to an explicit Postgres URL (for the CLI, which may + target a ``--database-url`` that differs from the ``DATABASE_URL`` env).""" + return build_app( + PsycopgConnector(conninfo=get_procrastinate_conninfo(database_url)) + ) + + +_app: App | None = None + + +def get_procrastinate_app() -> App: + """Process-wide procrastinate App bound to the Postgres app database.""" + global _app + if _app is None: + _app = build_app(PsycopgConnector(conninfo=get_procrastinate_conninfo())) + return _app + + +async def _ingest_schema_present(app: App) -> bool: + row = await app.connector.execute_query_one_async( + "SELECT to_regclass('procrastinate_jobs') IS NOT NULL AS present" + ) + return bool(row["present"]) + + +async def apply_ingest_queue_schema(app: App | None = None) -> None: + """Create procrastinate's tables on a fresh database (apply-if-absent). + + procrastinate's ``schema.sql`` uses bare ``CREATE TYPE``/``CREATE TABLE`` + (not ``IF NOT EXISTS``), so it errors if re-applied — it is meant to run + once on a fresh DB. We skip when ``procrastinate_jobs`` already exists; + *version* upgrades use procrastinate's own migration files (operator-run, a + lineage independent of the app's Alembic schema). + + Safe to call concurrently across rolling-update pods without an advisory + lock: Postgres DDL is transactional and procrastinate applies the whole + schema in one transaction, so a pod that loses the race rolls back cleanly + and we treat the resulting error as benign once the schema is present. + + Opens a short-lived connection, so it is safe to call from the CLI + (``db upgrade`` / worker startup). + """ + app = app or get_procrastinate_app() + async with app.open_async(): + if await _ingest_schema_present(app): + logger.debug("ingest queue schema already present; skipping apply") + return + try: + await app.schema_manager.apply_schema_async() + logger.info("Applied procrastinate ingest queue schema") + except Exception: + # A racing pod likely committed the schema while our transaction + # rolled back atomically. Benign iff the schema is now present. + if await _ingest_schema_present(app): + logger.info("Ingest queue schema applied concurrently by another pod") + return + raise + + +# Job-status keys procrastinate flattens into each list_queues row (alongside +# ``name`` and ``jobs_count``). ``aborting`` is legacy/unused since v3.0.0. +_JOB_STATUSES = ("todo", "doing", "succeeded", "failed", "cancelled", "aborted") + + +async def get_ingest_job_counts(app: App | None = None) -> dict[str, int]: + """Return ingest job counts by status (``todo``/``doing``/``failed``/…). + + Reads procrastinate's per-queue stats via the manager API (not hand-written + SQL) so a future schema bump doesn't silently break the status surface. The + manager flattens its per-status ``stats`` into top-level row keys, so we read + the known status keys directly. Assumes the app's connector is already open. + """ + app = app or get_procrastinate_app() + counts: dict[str, int] = {} + for row in await app.job_manager.list_queues_async(queue=INGEST_QUEUE_NAME): + for status in _JOB_STATUSES: + if status in row: + counts[status] = counts.get(status, 0) + int(row[status]) + return counts + + +def _doc_queueing_lock(task: DocumentTask) -> str: + """Per-document enqueue-dedup key (partial-unique on ``status='todo'``).""" + return f"{task.user_id}:{task.doc_type}:{task.doc_id}" + + +class ProcrastinateTaskProducer: + """``TaskProducer`` that defers ingest jobs into Postgres via procrastinate. + + The App's connector pool is owned by the server lifespan (opened once, + closed on shutdown), so ``clone``/``aenter``/``aexit``/``aclose`` are no-ops + — there is no per-handle resource like the memory stream's clones. + """ + + def __init__(self, app: App): + self._app = app + + @classmethod + async def connect(cls) -> ProcrastinateTaskProducer: + app = get_procrastinate_app() + await app.open_async() + return cls(app) + + async def send(self, task: DocumentTask, /) -> None: + key = _doc_queueing_lock(task) + deferrer = self._app.configure_task(INGEST_TASK_NAME, queueing_lock=key) + try: + await deferrer.defer_async(**asdict(task)) + except AlreadyEnqueued: + # A todo job already exists for this doc; the next periodic scan + # re-evaluates freshness (placeholder/Qdrant modified_at only + # advances after a successful index), so this is not a lost update. + logger.debug("ingest.already_enqueued key=%s", key) + + async def job_counts(self) -> dict[str, int]: + """Ingest job counts by status (for the vector-sync status surface).""" + return await get_ingest_job_counts(self._app) + + def clone(self) -> ProcrastinateTaskProducer: + return self + + async def __aenter__(self) -> ProcrastinateTaskProducer: + return self + + async def __aexit__( + self, + exc_type: type[BaseException] | None, + exc: BaseException | None, + tb: TracebackType | None, + ) -> None: + return None + + # The bare suppression marker silences S7503 (async method without await): + # ``async def`` is required by the TaskProducer protocol; per-handle close is + # a no-op (the pool is owned by the lifespan, drained once on shutdown). + async def aclose(self) -> None: # NOSONAR + return None + + async def drain(self) -> None: + """Close the shared connector pool (lifespan shutdown only).""" + await self._app.close_async() diff --git a/nextcloud_mcp_server/vector/queue/status.py b/nextcloud_mcp_server/vector/queue/status.py deleted file mode 100644 index d759d5e5..00000000 --- a/nextcloud_mcp_server/vector/queue/status.py +++ /dev/null @@ -1,195 +0,0 @@ -"""Status surface for ingest jobs (design §10.1, ``STATUS_BACKEND``). - -- ``local``: in-process job state — the memory-stream buffer (today's behavior, - read directly by the status endpoint). -- ``bus``: a background subscriber consumes - ``mcp.document.{ready,failed,reparsed}.{tenant_id}`` into a bounded in-process - :class:`StatusStore` that the status endpoint / ``nc_get_vector_sync_status`` - read. - -**Honest constraint (design §10.2 / decision):** MCP progress notifications -(``ctx.report_progress``) can only be emitted inside an *active tool-call -request*; a background subscriber has no ``ctx`` and the MCP SDK exposes no -out-of-band push. So "surface events as MCP progress notifications" is delivered -via this store (polled by the status endpoint / a tool), not an unsolicited -server push. True server-initiated progress / SSE is a follow-up — the -``on_event`` callback seam is left in place for it. -""" - -from __future__ import annotations - -import logging -from collections import OrderedDict -from typing import TYPE_CHECKING, Any, Callable - -if TYPE_CHECKING: - import anyio - from anyio.abc import TaskStatus - -logger = logging.getLogger(__name__) - -# Terminal/intermediate document states carried on mcp.document.* subjects. -_VALID_STATES = {"ready", "failed", "reparsed"} - - -class StatusStore: - """Bounded LRU of recent document states keyed by ``doc_id``.""" - - def __init__(self, max_size: int = 10_000): - self._entries: OrderedDict[str, dict[str, Any]] = OrderedDict() - self._max = max_size - - def record( - self, - doc_id: str, - state: str, - *, - content_hash: str | None = None, - transitioned_at: str | None = None, - ) -> None: - self._entries[doc_id] = { - "state": state, - "content_hash": content_hash, - "transitioned_at": transitioned_at, - } - self._entries.move_to_end(doc_id) - while len(self._entries) > self._max: - self._entries.popitem(last=False) - - def get(self, doc_id: str) -> dict[str, Any] | None: - return self._entries.get(doc_id) - - def counts(self) -> dict[str, int]: - out: dict[str, int] = {} - for entry in self._entries.values(): - out[entry["state"]] = out.get(entry["state"], 0) + 1 - return out - - def __len__(self) -> int: - return len(self._entries) - - -def state_from_subject(subject: str) -> str | None: - """``mcp.document..`` → ```` (or None if unknown).""" - parts = subject.split(".") - if len(parts) >= 4 and parts[0] == "mcp" and parts[1] == "document": - state = parts[2] - if state in _VALID_STATES: - return state - return None - - -class NatsStatusSubscriber: - """Consumes ``mcp.document.*.{tenant_id}`` into a :class:`StatusStore`.""" - - def __init__( - self, - nc: Any, - js: Any, - tenant_id: str, - store: StatusStore, - on_event: Callable[[str, str], None] | None = None, - ): - self._nc = nc - self._js = js - self.tenant_id = tenant_id - self.store = store - # on_event(doc_id, state) — seam for a future SSE / progress bridge. - self._on_event = on_event - - def handle_message(self, subject: str, data: bytes) -> None: - """Parse one status message into the store. Unit-testable without NATS.""" - import json # noqa: PLC0415 - - state = state_from_subject(subject) - if state is None: - logger.warning("status.unknown_subject subject=%s", subject) - return - try: - payload = json.loads(data) - doc_id = payload["doc_id"] - except Exception: - logger.warning("status.bad_message subject=%s", subject, exc_info=True) - return - self.store.record( - doc_id, - state, - content_hash=payload.get("content_hash"), - transitioned_at=payload.get("transitioned_at"), - ) - if self._on_event is not None: - self._on_event(doc_id, state) - - @classmethod - async def connect( - cls, *, url: str, tenant_id: str, store: StatusStore - ) -> NatsStatusSubscriber: - import nats # noqa: PLC0415 - - from .nats import warn_if_insecure_nats_url # noqa: PLC0415 - - warn_if_insecure_nats_url(url) - nc = await nats.connect(url) - js = nc.jetstream() - return cls(nc, js, tenant_id, store) - - async def run( - self, - shutdown_event: anyio.Event, - *, - task_status: TaskStatus | None = None, - ) -> None: - """Durable pull-consumer loop. Requires a live broker (integration).""" - import anyio # noqa: PLC0415 - import nats.errors # noqa: PLC0415 - - subject = f"mcp.document.*.{self.tenant_id}" - # Signal "task running" *before* the first (fallible) subscribe: bus - # status is a non-critical observability path, so a broker that isn't - # ready at startup should retry below rather than crash the lifespan. - # ``started()`` therefore means "the subscriber loop is running", not - # "the subscription succeeded". - if task_status is not None: - task_status.started() - - sub = None - while not shutdown_event.is_set(): - if sub is None: - try: - sub = await self._js.pull_subscribe( - subject, durable=f"mcp-status-{self.tenant_id}" - ) - except Exception: - # Broker not ready / transient connect error: back off and - # retry the subscribe instead of giving up. - logger.warning( - "NATS status subscribe failed; retrying", exc_info=True - ) - await anyio.sleep(5) - continue - try: - msgs = await sub.fetch(batch=16, timeout=5) - except nats.errors.TimeoutError: - # Expected when idle: no messages within the fetch window. Loop - # straight back to re-check shutdown — no log, no extra sleep. - continue - except Exception: - # Real broker error (disconnect, auth failure, stream deleted): - # drop the (possibly dead) subscription, back off, and - # re-subscribe on the next iteration rather than hot-spinning. - logger.warning( - "NATS status subscriber fetch failed; re-subscribing", - exc_info=True, - ) - sub = None - await anyio.sleep(5) - continue - for msg in msgs: - self.handle_message(msg.subject, msg.data) - await msg.ack() - - async def aclose(self) -> None: - try: - await self._nc.drain() - except Exception: - logger.warning("NATS status subscriber drain failed", exc_info=True) diff --git a/nextcloud_mcp_server/vector/webhook_receiver.py b/nextcloud_mcp_server/vector/webhook_receiver.py index c1cb50fa..d2a95df6 100644 --- a/nextcloud_mcp_server/vector/webhook_receiver.py +++ b/nextcloud_mcp_server/vector/webhook_receiver.py @@ -42,9 +42,10 @@ async def handle_nextcloud_webhook(request: Request) -> JSONResponse: """Receive a Nextcloud webhook and queue a DocumentTask for vector sync. Returns quickly so NC's webhook worker is not blocked. The task producer is - read from ``request.app.state.task_producer`` (the in-memory send stream in - local mode, or the NATS bus producer in external mode); when vector sync - isn't running we return 503 so NC retries delivery. + read from ``request.app.state.task_producer`` (the in-memory send stream when + ``INGEST_QUEUE=memory``, or the procrastinate producer when + ``INGEST_QUEUE=postgres``); when vector sync isn't running we return 503 so + NC retries delivery. When ``WEBHOOK_SECRET`` is set, the request must carry ``Authorization: Bearer `` (registered via ``authData`` so NC diff --git a/pyproject.toml b/pyproject.toml index 8d6319cf..03ef34a9 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,7 +46,6 @@ dependencies = [ "dynaconf>=3.2.13,<4.0", "mistralai>=2.4.5", "sqlalchemy[asyncio]>=2.0", - "nats-py>=2.14.0", ] classifiers = [ "Development Status :: 4 - Beta", @@ -134,6 +133,8 @@ nextcloud-mcp-server = "nextcloud_mcp_server.cli:cli" [project.optional-dependencies] postgres = [ "asyncpg>=0.29", + "procrastinate>=3.8", + "psycopg[binary,pool]>=3.2", ] [[tool.uv.index]] diff --git a/tests/integration/test_ingest_queue_postgres.py b/tests/integration/test_ingest_queue_postgres.py new file mode 100644 index 00000000..bc84189c --- /dev/null +++ b/tests/integration/test_ingest_queue_postgres.py @@ -0,0 +1,139 @@ +"""End-to-end Postgres smoke for the procrastinate ingest queue (Deck #183). + +Validates the queue mechanics the in-memory connector can't: real +``queueing_lock`` partial-unique dedup, idempotent schema apply, and the +``list_queues`` stats the status surface reads. Opt-in like +``test_storage_postgres.py``:: + + docker compose --profile postgres up -d postgres-test + export TEST_DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp + uv run pytest tests/integration/test_ingest_queue_postgres.py -v -m postgres + +Skipped when ``TEST_DATABASE_URL`` is unset or the service is unreachable. +""" + +from __future__ import annotations + +import os +import socket +from urllib.parse import urlparse + +import pytest + +import nextcloud_mcp_server.config as config_module +from nextcloud_mcp_server.vector.queue.procrastinate import ( + INGEST_QUEUE_NAME, + ProcrastinateTaskProducer, + apply_ingest_queue_schema, + build_app_for_url, + get_ingest_job_counts, +) +from nextcloud_mcp_server.vector.scanner import DocumentTask + +pytestmark = [pytest.mark.integration, pytest.mark.postgres] + + +def _postgres_url() -> str | None: + return os.environ.get("TEST_DATABASE_URL") or None + + +def _reachable(url: str) -> bool: + parsed = urlparse(url) + try: + with socket.create_connection( + (parsed.hostname or "localhost", parsed.port or 5432), timeout=1.0 + ): + return True + except OSError: + return False + + +@pytest.fixture +def postgres_url() -> str: + url = _postgres_url() + if not url: + pytest.skip( + "TEST_DATABASE_URL not set — run " + "`docker compose --profile postgres up -d postgres-test` and export " + "TEST_DATABASE_URL=postgresql+asyncpg://mcp:mcp@localhost:5433/mcp" + ) + if not _reachable(url): + pytest.skip(f"Postgres at {url} is not reachable") + return url + + +@pytest.fixture +async def fresh_app(postgres_url: str, monkeypatch: pytest.MonkeyPatch): + """Drop+recreate the public schema, then apply procrastinate's schema.""" + from sqlalchemy import text + from sqlalchemy.ext.asyncio import create_async_engine + + engine = create_async_engine(postgres_url, future=True) + try: + async with engine.begin() as conn: + await conn.execute(text("DROP SCHEMA public CASCADE")) + await conn.execute(text("CREATE SCHEMA public")) + finally: + await engine.dispose() + + # get_procrastinate_conninfo derives ssl from settings; point it at the URL. + monkeypatch.setattr(config_module, "get_database_url", lambda: postgres_url) + monkeypatch.setattr(config_module, "get_database_ssl", lambda: None) + + app = build_app_for_url(postgres_url) + await apply_ingest_queue_schema(app) + return app + + +def _task(doc_id: str, doc_type: str = "note") -> DocumentTask: + return DocumentTask( + user_id="alice", + doc_id=doc_id, + doc_type=doc_type, + operation="index", + modified_at=100, + etag=f"etag-{doc_id}", + ) + + +async def test_ingest_queue_end_to_end(fresh_app): + """One self-contained smoke against real Postgres. + + Kept as a single test so each assertion runs against the same freshly-applied + schema — splitting across functions reintroduces the inter-test ``DROP + SCHEMA`` that confuses pooled psycopg connections' cached prepared statements + (a test-harness artifact, not a production path: prod never drops the schema). + """ + # 1. Schema is present and a second apply is a no-op (idempotent). + await apply_ingest_queue_schema(fresh_app) + + async with fresh_app.open_async(): + present = await fresh_app.connector.execute_query_one_async( + "SELECT to_regclass('procrastinate_jobs') IS NOT NULL AS present" + ) + assert present["present"] is True + + # 2. Defer + real queueing_lock dedup (one todo per doc). + producer = ProcrastinateTaskProducer(fresh_app) + await producer.send(_task("1")) + await producer.send(_task("1")) # deduped by queueing_lock + await producer.send(_task("2")) + + rows = await fresh_app.connector.execute_query_all_async( + "SELECT count(*) AS n FROM procrastinate_jobs " + "WHERE queue_name = %(q)s AND status = 'todo'", + q=INGEST_QUEUE_NAME, + ) + assert rows[0]["n"] == 2 + + # 3. The status-surface counts read agrees. + counts = await get_ingest_job_counts(fresh_app) + assert counts.get("todo") == 2 + + # 4. Fresh todo jobs are not "doing", so none are stalled. + stalled = list( + await fresh_app.job_manager.get_stalled_jobs( + queue=INGEST_QUEUE_NAME, seconds_since_heartbeat=0 + ) + ) + assert stalled == [] diff --git a/tests/unit/test_decomposition_config.py b/tests/unit/test_decomposition_config.py index 4240a978..3e18af0c 100644 --- a/tests/unit/test_decomposition_config.py +++ b/tests/unit/test_decomposition_config.py @@ -1,4 +1,4 @@ -"""Tests for the MCP decomposition hook-point settings (design §10). +"""Tests for the MCP decomposition hook-point settings (design §10, Deck #183). Every default must reproduce the monolith; the opt-in settings are validated in ``Settings.__post_init__``. @@ -6,6 +6,7 @@ in ``Settings.__post_init__``. import pytest +import nextcloud_mcp_server.config as config_module from nextcloud_mcp_server.canonical import canonical_json from nextcloud_mcp_server.config import Settings @@ -16,23 +17,21 @@ class TestDecompositionDefaults: def test_defaults_are_monolith(self): s = Settings() assert s.embedding_provider == "autodetect" - assert s.ingest_mode == "local" - assert s.status_backend == "local" + # SQLite/dev default → the in-process memory queue. + assert s.ingest_queue == "memory" + assert s.mcp_role == "all" assert s.collection_metadata_source == "qdrant" - assert s.fact_event_emitter == "none" - assert s.ingest_bus_url is None assert s.embedding_gateway_url is None assert s.tenant_id is None - assert s.ingest_bus_num_replicas == 1 def test_enum_values_normalized(self): # Mixed case / surrounding whitespace is normalized before validation. s = Settings( collection_metadata_source=" QDRANT ", - fact_event_emitter="NONE", + mcp_role=" API ", ) assert s.collection_metadata_source == "qdrant" - assert s.fact_event_emitter == "none" + assert s.mcp_role == "api" class TestEnumValidation: @@ -40,58 +39,48 @@ class TestEnumValidation: "field,value", [ ("embedding_provider", "openai"), - ("ingest_mode", "remote"), - ("status_backend", "redis"), - ("collection_metadata_source", "postgres"), - ("fact_event_emitter", "kafka"), + ("mcp_role", "leader"), + ("collection_metadata_source", "redis"), ], ) def test_invalid_enum_rejected(self, field, value): with pytest.raises(ValueError, match=field.upper()): Settings(**{field: value}) + def test_invalid_ingest_queue_rejected(self): + with pytest.raises(ValueError, match="INGEST_QUEUE"): + Settings(ingest_queue="kafka") -class TestFailFast: - def test_external_with_local_status_crashes(self): - with pytest.raises( - RuntimeError, - match="STATUS_BACKEND=local is incompatible with INGEST_MODE=external", - ): - Settings( - ingest_mode="external", - status_backend="local", - ingest_bus_url="nats://nats:4222", - tenant_id="tenant-uuid", - ) + +class TestIngestQueueResolution: + def test_postgres_requires_postgres_url(self): + # Explicit postgres against the default SQLite DATABASE_URL is a + # misconfiguration (procrastinate is Postgres-only). + with pytest.raises(ValueError, match="INGEST_QUEUE=postgres requires"): + Settings(ingest_queue="postgres") + + def test_auto_postgres_when_database_url_is_postgres(self, monkeypatch): + monkeypatch.setattr( + config_module, + "get_database_url", + lambda: "postgresql+asyncpg://mcp:mcp@db/mcp", + ) + assert Settings().ingest_queue == "postgres" + + def test_explicit_memory_on_postgres_url(self, monkeypatch): + monkeypatch.setattr( + config_module, + "get_database_url", + lambda: "postgresql+asyncpg://mcp:mcp@db/mcp", + ) + assert Settings(ingest_queue="memory").ingest_queue == "memory" class TestConditionalRequired: - def test_external_requires_bus_url(self): - with pytest.raises(ValueError, match="INGEST_BUS_URL is required"): - Settings(ingest_mode="external", status_backend="bus", tenant_id="t1") - - def test_external_requires_tenant_id(self): - with pytest.raises(ValueError, match="TENANT_ID is required"): - Settings( - ingest_mode="external", - status_backend="bus", - ingest_bus_url="nats://nats:4222", - ) - def test_gateway_requires_gateway_url(self): with pytest.raises(ValueError, match="EMBEDDING_GATEWAY_URL is required"): Settings(embedding_provider="gateway") - def test_external_happy_path(self): - s = Settings( - ingest_mode="external", - status_backend="bus", - ingest_bus_url="nats://nats:4222", - tenant_id="0a1b2c3d-0000-0000-0000-000000000000", - ) - assert s.ingest_mode == "external" - assert s.status_backend == "bus" - def test_gateway_happy_path(self): s = Settings( embedding_provider="gateway", @@ -100,24 +89,60 @@ class TestConditionalRequired: assert s.embedding_provider == "gateway" -class TestTenantIdSubjectToken: - @pytest.mark.parametrize( - "tenant_id", - ["a.b", "a*b", "a>b", "a b", "a\tb"], - ) - def test_illegal_subject_chars_rejected(self, tenant_id): - with pytest.raises(ValueError, match="TENANT_ID must not contain"): - Settings(tenant_id=tenant_id) - - def test_uuid_form_accepted(self): +class TestTenantId: + def test_arbitrary_tenant_id_accepted(self): + # The old NATS-subject charset restriction was dropped with NATS + # (Deck #183); tenant_id is now just an opaque per-tenant identity. s = Settings(tenant_id="0a1b2c3d-0000-0000-0000-000000000000") assert s.tenant_id == "0a1b2c3d-0000-0000-0000-000000000000" -class TestReplicas: - def test_zero_replicas_rejected(self): - with pytest.raises(ValueError, match="INGEST_BUS_NUM_REPLICAS must be >= 1"): - Settings(ingest_bus_num_replicas=0) +class TestProcrastinateConninfo: + @pytest.mark.parametrize( + "url,expected_sslmode", + [ + ("postgresql+asyncpg://mcp:p%40ss@db:5432/mcp", None), + ], + ) + def test_conninfo_round_trips_password(self, monkeypatch, url, expected_sslmode): + from psycopg.conninfo import conninfo_to_dict + + monkeypatch.setattr(config_module, "get_database_url", lambda: url) + # No SSL settings → sslmode omitted (libpq default ``prefer``). + monkeypatch.setattr(config_module, "get_database_ssl", lambda: None) + parsed = conninfo_to_dict(config_module.get_procrastinate_conninfo()) + assert parsed["password"] == "p@ss" + assert parsed["host"] == "db" + assert parsed["dbname"] == "mcp" + assert parsed.get("sslmode") == expected_sslmode + + def test_conninfo_ssl_mapping(self, monkeypatch): + from psycopg.conninfo import conninfo_to_dict + + monkeypatch.setattr( + config_module, + "get_database_url", + lambda: "postgresql+asyncpg://mcp:s@db/mcp", + ) + # verify off → encrypt without verifying. + monkeypatch.setattr(config_module, "get_database_ssl", lambda: False) + assert ( + conninfo_to_dict(config_module.get_procrastinate_conninfo())["sslmode"] + == "require" + ) + # verify on → verify-full. + monkeypatch.setattr(config_module, "get_database_ssl", lambda: True) + assert ( + conninfo_to_dict(config_module.get_procrastinate_conninfo())["sslmode"] + == "verify-full" + ) + + def test_conninfo_rejects_non_postgres(self, monkeypatch): + monkeypatch.setattr( + config_module, "get_database_url", lambda: "sqlite+aiosqlite:///x.db" + ) + with pytest.raises(ValueError, match="requires a PostgreSQL DATABASE_URL"): + config_module.get_procrastinate_conninfo() class TestCanonicalJson: diff --git a/tests/unit/vector/test_ingest_status.py b/tests/unit/vector/test_ingest_status.py new file mode 100644 index 00000000..27923cd2 --- /dev/null +++ b/tests/unit/vector/test_ingest_status.py @@ -0,0 +1,57 @@ +"""Unit tests for the shared ingest-status read model (Deck #183).""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from nextcloud_mcp_server.vector.ingest_status import get_ingest_pending + +pytestmark = pytest.mark.unit + + +class TestGetIngestPending: + async def test_postgres_reads_job_counts(self): + producer = AsyncMock() + producer.job_counts.return_value = {"todo": 5, "doing": 2, "failed": 1} + + result = await get_ingest_pending( + task_producer=producer, + document_receive_stream=None, + ingest_queue="postgres", + ) + assert result.pending == 7 # todo + doing + assert result.job_counts == {"todo": 5, "doing": 2, "failed": 1} + + async def test_postgres_degrades_to_zero_on_error(self): + producer = AsyncMock() + producer.job_counts.side_effect = RuntimeError("db down") + + result = await get_ingest_pending( + task_producer=producer, + document_receive_stream=None, + ingest_queue="postgres", + ) + assert result.pending == 0 + assert result.job_counts == {} + + async def test_memory_reads_stream_buffer(self): + stream = SimpleNamespace( + statistics=lambda: SimpleNamespace(current_buffer_used=3) + ) + result = await get_ingest_pending( + task_producer=None, + document_receive_stream=stream, + ingest_queue="memory", + ) + assert result.pending == 3 + assert result.job_counts is None + + async def test_memory_without_stream_is_zero(self): + result = await get_ingest_pending( + task_producer=None, + document_receive_stream=None, + ingest_queue="memory", + ) + assert result.pending == 0 + assert result.job_counts is None diff --git a/tests/unit/vector/test_nats_producer.py b/tests/unit/vector/test_nats_producer.py deleted file mode 100644 index 5c29cb46..00000000 --- a/tests/unit/vector/test_nats_producer.py +++ /dev/null @@ -1,155 +0,0 @@ -"""NATS ingest producer: DocumentTask → IngestMessage + dedup header (§3.4).""" - -import hashlib -import json -from pathlib import Path - -import pytest - -from nextcloud_mcp_server.canonical import canonical_json -from nextcloud_mcp_server.vector.queue.factory import _transport_for -from nextcloud_mcp_server.vector.queue.nats import ( - NatsTaskProducer, - _modified_at_rfc3339, - msg_id, - warn_if_insecure_nats_url, -) -from nextcloud_mcp_server.vector.queue.postgres import PostgresTaskProducer -from nextcloud_mcp_server.vector.scanner import DocumentTask - -FIXTURE = Path(__file__).parents[2] / "fixtures" / "ingest_message_example.json" -TENANT = "00000000-0000-0000-0000-000000000001" - - -def _producer(mocker, tenant_id=TENANT): - return NatsTaskProducer( - nc=mocker.MagicMock(), js=mocker.AsyncMock(), tenant_id=tenant_id - ) - - -def test_ingest_message_translation(mocker): - p = _producer(mocker) - task = DocumentTask( - user_id="alice", - doc_id="12345", - doc_type="file", - operation="index", - modified_at=1700000000, - file_path="/Documents/report.pdf", - etag="etag-abc123", - ) - msg = p.ingest_message(task) - assert msg["tenant_id"] == TENANT # from settings, not the task - assert msg["content_hash"] == "etag-abc123" # etag wins - assert msg["user_id"] == "alice" - assert msg["doc_type"] == "file" - assert msg["operation"] == "index" - assert msg["file_path"] == "/Documents/report.pdf" - - -def test_content_hash_falls_back_to_modified_at(mocker): - p = _producer(mocker) - task = DocumentTask( - user_id="u", doc_id="d", doc_type="note", operation="delete", modified_at=0 - ) - assert p.ingest_message(task)["content_hash"] == "0" - - -async def test_send_publishes_with_dedup_header(mocker): - p = _producer(mocker) - task = DocumentTask( - user_id="alice", - doc_id="12345", - doc_type="file", - operation="index", - modified_at=1700000000, - etag="e", - ) - await p.send(task) - p._js.publish.assert_awaited_once() - args = p._js.publish.await_args.args - kwargs = p._js.publish.await_args.kwargs - assert args[0] == f"mcp.ingest.requested.{TENANT}" - expected_mid = msg_id(TENANT, "12345", _modified_at_rfc3339(1700000000)) - assert kwargs["headers"]["Nats-Msg-Id"] == expected_mid - assert json.loads(args[1])["doc_id"] == "12345" - - -def test_msg_id_known_vector(): - mid = msg_id("t", "d", "2026-01-01T00:00:00+00:00") - expected = hashlib.sha256( - canonical_json( - { - "tenant_id": "t", - "doc_id": "d", - "modified_at": "2026-01-01T00:00:00+00:00", - } - ) - ).hexdigest() - assert mid == expected - - -def test_publisher_matches_shared_fixture(mocker): - # The same fixture is validated as an IngestMessage in the processor repo. - # Here we assert the publisher emits exactly the fixture's key set + stable - # field values (modified_at format is allowed to differ — epoch→ISO). - fixture = json.loads(FIXTURE.read_text(encoding="utf-8")) - p = _producer(mocker, tenant_id=fixture["tenant_id"]) - task = DocumentTask( - user_id=fixture["user_id"], - doc_id=fixture["doc_id"], - doc_type=fixture["doc_type"], - operation=fixture["operation"], - modified_at=1764201600, - file_path=fixture["file_path"], - etag=fixture["content_hash"], - ) - msg = p.ingest_message(task) - assert set(msg.keys()) == set(fixture.keys()) - for key in ( - "tenant_id", - "doc_id", - "content_hash", - "doc_type", - "operation", - "user_id", - "file_path", - ): - assert msg[key] == fixture[key] - assert msg["modified_at"] # non-empty ISO timestamp - - -@pytest.mark.parametrize( - "url,expected", - [ - ("nats://nats:4222", "nats"), - ("postgres://h/db", "postgres"), - ("postgresql://h/db", "postgres"), - ("https://elsewhere", "nats"), - ], -) -def test_transport_for(url, expected): - assert _transport_for(url) == expected - - -async def test_postgres_producer_is_a_seam(): - with pytest.raises(NotImplementedError, match="documented seam"): - await PostgresTaskProducer.connect(object()) - - -@pytest.mark.parametrize( - "url,should_warn", - [ - ("nats://nats:4222", True), - ("ws://nats:8080", True), - ("tls://nats:4222", False), - ("wss://nats:8080", False), - ], -) -def test_warn_if_insecure_nats_url(url, should_warn, caplog): - import logging - - with caplog.at_level(logging.WARNING): - warn_if_insecure_nats_url(url) - warned = any("unencrypted transport" in r.getMessage() for r in caplog.records) - assert warned is should_warn diff --git a/tests/unit/vector/test_procrastinate_producer.py b/tests/unit/vector/test_procrastinate_producer.py new file mode 100644 index 00000000..d182969a --- /dev/null +++ b/tests/unit/vector/test_procrastinate_producer.py @@ -0,0 +1,191 @@ +"""Unit tests for the procrastinate ingest producer + task (Deck #183). + +Uses procrastinate's in-memory connector so no live Postgres is required. +""" + +from unittest.mock import AsyncMock + +import pytest +from procrastinate import testing + +import nextcloud_mcp_server.vector.queue.procrastinate as pq +from nextcloud_mcp_server.vector.scanner import DocumentTask + +pytestmark = pytest.mark.unit + + +@pytest.fixture +def app(): + """An App bound to the in-memory connector with the ingest tasks.""" + return pq.build_app(testing.InMemoryConnector()) + + +def _task(doc_id="42", doc_type="note", operation="index"): + return DocumentTask( + user_id="alice", + doc_id=doc_id, + doc_type=doc_type, + operation=operation, + modified_at=100, + etag="etag-abc", + ) + + +class TestProcrastinateTaskProducer: + async def test_send_defers_with_correct_job_shape(self, app): + async with app.open_async(): + producer = pq.ProcrastinateTaskProducer(app) + await producer.send(_task()) + + jobs = list(app.connector.jobs.values()) + assert len(jobs) == 1 + job = jobs[0] + assert job["task_name"] == pq.INGEST_TASK_NAME + assert job["queue_name"] == pq.INGEST_QUEUE_NAME + assert job["queueing_lock"] == "alice:note:42" + assert job["lock"] is None # no execution lock (crash-deadlock guard) + assert job["args"]["doc_id"] == "42" + assert job["args"]["etag"] == "etag-abc" + + async def test_duplicate_send_is_deduped(self, app): + async with app.open_async(): + producer = pq.ProcrastinateTaskProducer(app) + await producer.send(_task()) + # Same doc again → AlreadyEnqueued, swallowed; still one job. + await producer.send(_task()) + assert len(app.connector.jobs) == 1 + + async def test_distinct_docs_create_separate_jobs(self, app): + async with app.open_async(): + producer = pq.ProcrastinateTaskProducer(app) + await producer.send(_task(doc_id="1")) + await producer.send(_task(doc_id="2")) + assert len(app.connector.jobs) == 2 + + def test_clone_returns_self(self, app): + producer = pq.ProcrastinateTaskProducer(app) + assert producer.clone() is producer + + +class TestProcessDocumentTask: + async def test_runs_pipeline_and_closes_client(self, monkeypatch): + captured = {} + + fake_client = AsyncMock() + + async def fake_resolve(user_id): + captured["user_id"] = user_id + return fake_client + + async def fake_process(task, nc_client, *, max_retries): + captured["task"] = task + captured["nc_client"] = nc_client + captured["max_retries"] = max_retries + + monkeypatch.setattr(pq, "_resolve_client", fake_resolve) + monkeypatch.setattr( + "nextcloud_mcp_server.vector.processor.process_document", fake_process + ) + + # Calling the Task runs its wrapped function in-process. + await pq.process_document_task( + user_id="alice", + doc_id="42", + doc_type="note", + operation="index", + modified_at=100, + etag="e1", + ) + + assert captured["user_id"] == "alice" + assert isinstance(captured["task"], DocumentTask) + assert captured["task"].doc_id == "42" + assert captured["task"].etag == "e1" + # Worker disables the in-process retry loop; durable retry is the queue's. + assert captured["max_retries"] == 1 + fake_client.close.assert_awaited_once() + + async def test_skips_on_missing_credentials(self, monkeypatch): + from nextcloud_mcp_server.vector.oauth_sync import NotProvisionedError + + async def fake_resolve(user_id): + raise NotProvisionedError("no app password") + + called = False + + async def fake_process(*args, **kwargs): + nonlocal called + called = True + + monkeypatch.setattr(pq, "_resolve_client", fake_resolve) + monkeypatch.setattr( + "nextcloud_mcp_server.vector.processor.process_document", fake_process + ) + + # Returns cleanly (job succeeds as a no-op); pipeline never runs. + await pq.process_document_task( + user_id="ghost", + doc_id="9", + doc_type="note", + operation="index", + modified_at=0, + ) + assert called is False + + +class TestReclaimStalledJobs: + async def test_reclaims_each_stalled_job(self): + from datetime import datetime + + retried: list[int] = [] + + class Job: + def __init__(self, id): + self.id = id + + class FakeManager: + async def get_stalled_jobs(self, queue=None, seconds_since_heartbeat=0): + assert queue == pq.INGEST_QUEUE_NAME + return [Job(1), Job(2), Job(None)] # None id is skipped + + async def retry_job_by_id_async(self, job_id, retry_at): + assert isinstance(retry_at, datetime) + retried.append(job_id) + + class FakeApp: + job_manager = FakeManager() + + class Ctx: + app = FakeApp() + + await pq.reclaim_stalled_ingest_jobs(Ctx(), timestamp=0) + assert retried == [1, 2] + + +class TestGetIngestJobCounts: + async def test_aggregates_stats_rows(self): + class FakeManager: + async def list_queues_async(self, queue=None): + assert queue == pq.INGEST_QUEUE_NAME + # procrastinate flattens per-status stats into top-level keys. + return [ + { + "name": "ingest", + "jobs_count": 6, + "todo": 3, + "doing": 1, + "succeeded": 0, + "failed": 2, + "cancelled": 0, + "aborted": 0, + } + ] + + class FakeApp: + job_manager = FakeManager() + + counts = await pq.get_ingest_job_counts(FakeApp()) + assert counts["todo"] == 3 + assert counts["doing"] == 1 + assert counts["failed"] == 2 + assert counts["succeeded"] == 0 diff --git a/tests/unit/vector/test_status_store.py b/tests/unit/vector/test_status_store.py deleted file mode 100644 index ed0d556b..00000000 --- a/tests/unit/vector/test_status_store.py +++ /dev/null @@ -1,167 +0,0 @@ -"""StatusStore + NATS status message handling (design §10.1, STATUS_BACKEND=bus).""" - -import json - -from nextcloud_mcp_server.vector.queue.status import ( - NatsStatusSubscriber, - StatusStore, - state_from_subject, -) - - -def test_store_records_and_counts(): - store = StatusStore() - store.record("d1", "ready", content_hash="h1") - store.record("d2", "failed") - store.record("d1", "ready", content_hash="h1") # idempotent overwrite - assert len(store) == 2 - assert store.counts() == {"ready": 1, "failed": 1} - assert store.get("d1")["content_hash"] == "h1" - - -def test_store_is_bounded_lru(): - store = StatusStore(max_size=2) - store.record("d1", "ready") - store.record("d2", "ready") - store.record("d3", "ready") # evicts d1 - assert len(store) == 2 - assert store.get("d1") is None - assert store.get("d3") is not None - - -def test_state_from_subject(): - assert state_from_subject("mcp.document.ready.tenant-1") == "ready" - assert state_from_subject("mcp.document.failed.tenant-1") == "failed" - assert state_from_subject("mcp.document.reparsed.tenant-1") == "reparsed" - assert state_from_subject("mcp.document.bogus.tenant-1") is None - assert state_from_subject("mcp.ingest.requested.tenant-1") is None - - -def test_handle_message_records_state(): - store = StatusStore() - events = [] - sub = NatsStatusSubscriber( - nc=None, - js=None, - tenant_id="t1", - store=store, - on_event=lambda d, s: events.append((d, s)), - ) - payload = json.dumps( - { - "tenant_id": "t1", - "doc_id": "doc-9", - "content_hash": "abc", - "transitioned_at": "2026-05-27T00:00:00Z", - } - ).encode() - sub.handle_message("mcp.document.ready.t1", payload) - entry = store.get("doc-9") - assert entry["state"] == "ready" - assert entry["content_hash"] == "abc" - assert events == [("doc-9", "ready")] - - -def test_handle_message_ignores_bad_payload_and_subject(): - store = StatusStore() - sub = NatsStatusSubscriber(nc=None, js=None, tenant_id="t1", store=store) - sub.handle_message("mcp.document.ready.t1", b"not json") - sub.handle_message("mcp.ingest.requested.t1", b'{"doc_id":"x"}') - assert len(store) == 0 - - -async def test_run_signals_started_then_retries_subscribe(mocker, monkeypatch): - """run() signals started before subscribing, retries a failed subscribe, - and consumes messages once subscribed.""" - import anyio - - # Make backoff sleeps instant so the retry path doesn't stall the test. - async def _no_sleep(*_a, **_k): - return None - - monkeypatch.setattr(anyio, "sleep", _no_sleep) - - store = StatusStore() - js = mocker.AsyncMock() - - # First subscribe attempt fails (broker not ready), second succeeds. - fake_sub = mocker.AsyncMock() - js.pull_subscribe.side_effect = [ConnectionError("broker not ready"), fake_sub] - - shutdown = anyio.Event() - msg = mocker.Mock() - msg.subject = "mcp.document.ready.t1" - msg.data = json.dumps({"doc_id": "d1", "content_hash": "h1"}).encode() - msg.ack = mocker.AsyncMock() - - fetches = {"n": 0} - - async def _fetch(*_a, **_k): - fetches["n"] += 1 - if fetches["n"] == 1: - return [msg] - shutdown.set() # stop the loop after the first batch is handled - return [] - - fake_sub.fetch.side_effect = _fetch - - task_status = mocker.Mock() - subscriber = NatsStatusSubscriber( - nc=mocker.AsyncMock(), js=js, tenant_id="t1", store=store - ) - - await subscriber.run(shutdown, task_status=task_status) - - # started() fires before any subscribe attempt and exactly once. - task_status.started.assert_called_once() - # The failed first subscribe was retried (two attempts total). - assert js.pull_subscribe.call_count == 2 - # The message from the successful subscription was recorded + acked. - assert store.get("d1") == { - "state": "ready", - "content_hash": "h1", - "transitioned_at": None, - } - msg.ack.assert_awaited_once() - - -async def test_run_resubscribes_after_fetch_error(mocker, monkeypatch): - """A non-timeout fetch error drops the subscription and re-subscribes.""" - import anyio - import nats.errors - - async def _no_sleep(*_a, **_k): - return None - - monkeypatch.setattr(anyio, "sleep", _no_sleep) - - store = StatusStore() - js = mocker.AsyncMock() - first_sub = mocker.AsyncMock() - second_sub = mocker.AsyncMock() - js.pull_subscribe.side_effect = [first_sub, second_sub] - - shutdown = anyio.Event() - - # first_sub.fetch raises a real broker error → re-subscribe. - first_sub.fetch.side_effect = ConnectionResetError("broker dropped") - - # second_sub.fetch idles once (timeout) then stops the loop. - fetches = {"n": 0} - - async def _second_fetch(*_a, **_k): - fetches["n"] += 1 - if fetches["n"] == 1: - raise nats.errors.TimeoutError - shutdown.set() - return [] - - second_sub.fetch.side_effect = _second_fetch - - subscriber = NatsStatusSubscriber( - nc=mocker.AsyncMock(), js=js, tenant_id="t1", store=store - ) - await subscriber.run(shutdown) - - # Re-subscribed after the fetch error (two subscriptions used). - assert js.pull_subscribe.call_count == 2 diff --git a/uv.lock b/uv.lock index fb7a4239..8ba2433f 100644 --- a/uv.lock +++ b/uv.lock @@ -655,6 +655,18 @@ toml = [ { name = "tomli", marker = "python_full_version <= '3.11'" }, ] +[[package]] +name = "croniter" +version = "6.2.2" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "python-dateutil" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/df/de/5832661ed55107b8a09af3f0a2e71e0957226a59eb1dcf0a445cce6daf20/croniter-6.2.2.tar.gz", hash = "sha256:ba60832a5ec8e12e51b8691c3309a113d1cf6526bdf1a48150ce8ec7a532d0ab", size = 113762, upload-time = "2026-03-15T08:43:48.112Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/d0/39/783980e78cb92c2d7bdb1fc7dbc86e94ccc6d58224d76a7f1f51b6c51e30/croniter-6.2.2-py3-none-any.whl", hash = "sha256:a5d17b1060974d36251ea4faf388233eca8acf0d09cbd92d35f4c4ac8f279960", size = 45422, upload-time = "2026-03-15T08:43:46.626Z" }, +] + [[package]] name = "cryptography" version = "46.0.3" @@ -2169,15 +2181,6 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/6c/28/dd72947e59a6a8c856448a5e74da6201cb5502ddff644fbc790e4bd40b9a/multiprocess-0.70.18-py39-none-any.whl", hash = "sha256:e78ca805a72b1b810c690b6b4cc32579eba34f403094bbbae962b7b5bf9dfcb8", size = 133478, upload-time = "2025-04-17T03:11:26.253Z" }, ] -[[package]] -name = "nats-py" -version = "2.14.0" -source = { registry = "https://pypi.org/simple" } -sdist = { url = "https://files.pythonhosted.org/packages/c3/f8/b956c4621ba88748ed707c52e69f95b7a50c8914e750edca59a5bef84a76/nats_py-2.14.0.tar.gz", hash = "sha256:4ed02cb8e3b55c68074a063aa2687087115d805d1513297da90cb2068fb07bed", size = 120751, upload-time = "2026-02-23T22:44:58.988Z" } -wheels = [ - { url = "https://files.pythonhosted.org/packages/f9/39/0e87753df1072254bac190b33ed34b264f28f6aa9bea0f01b7e818071756/nats_py-2.14.0-py3-none-any.whl", hash = "sha256:4116f5d2233ce16e63c3d5538fa40a5e207f75fcf42a741773929ddf1e29d19d", size = 82259, upload-time = "2026-02-23T22:45:00.152Z" }, -] - [[package]] name = "nextcloud-mcp-server" version = "0.96.0" @@ -2199,7 +2202,6 @@ dependencies = [ { name = "markdownify" }, { name = "mcp", extra = ["cli"] }, { name = "mistralai" }, - { name = "nats-py" }, { name = "openai" }, { name = "opentelemetry-api" }, { name = "opentelemetry-exporter-otlp-proto-grpc" }, @@ -2224,6 +2226,8 @@ dependencies = [ [package.optional-dependencies] postgres = [ { name = "asyncpg" }, + { name = "procrastinate" }, + { name = "psycopg", extra = ["binary", "pool"] }, ] [package.dev-dependencies] @@ -2262,7 +2266,6 @@ requires-dist = [ { name = "markdownify", specifier = ">=0.14.1" }, { name = "mcp", extras = ["cli"], specifier = ">=1.27,<1.28" }, { name = "mistralai", specifier = ">=2.4.5" }, - { name = "nats-py", specifier = ">=2.14.0" }, { name = "openai", specifier = ">=2.8.1" }, { name = "opentelemetry-api", specifier = ">=1.28.2" }, { name = "opentelemetry-exporter-otlp-proto-grpc", specifier = ">=1.28.2" }, @@ -2271,7 +2274,9 @@ requires-dist = [ { name = "opentelemetry-instrumentation-logging", specifier = ">=0.49b2" }, { name = "opentelemetry-sdk", specifier = ">=1.28.2" }, { name = "pillow", specifier = ">=10.3.0,<12.0.0" }, + { name = "procrastinate", marker = "extra == 'postgres'", specifier = ">=3.8" }, { name = "prometheus-client", specifier = ">=0.21.0" }, + { name = "psycopg", extras = ["binary", "pool"], marker = "extra == 'postgres'", specifier = ">=3.2" }, { name = "pydantic", specifier = ">=2.11.4" }, { name = "pyjwt", extras = ["crypto"], specifier = ">=2.8.0" }, { name = "pymupdf", specifier = ">=1.26.6" }, @@ -2906,6 +2911,24 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/4b/a6/38c8e2f318bf67d338f4d629e93b0b4b9af331f455f0390ea8ce4a099b26/portalocker-3.2.0-py3-none-any.whl", hash = "sha256:3cdc5f565312224bc570c49337bd21428bba0ef363bbcf58b9ef4a9f11779968", size = 22424, upload-time = "2025-06-14T13:20:38.083Z" }, ] +[[package]] +name = "procrastinate" +version = "3.8.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "asgiref" }, + { name = "attrs" }, + { name = "croniter" }, + { name = "packaging" }, + { name = "psycopg", extra = ["pool"] }, + { name = "python-dateutil" }, + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/8f/cd/cbb88b0f19fa94e8a610af2fd3844e96b70591f4263ef4c36f10e4ebe4e2/procrastinate-3.8.1.tar.gz", hash = "sha256:cf7f11dfd4247daa166e9b61a211f9d5b70512d86eccc2bf4298f6ad182a32fa", size = 85343, upload-time = "2026-04-08T06:24:21.385Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/c5/ef/05a54e7ef9328d3d91a1a3b84ccf08a578128a48c57cd1117d1fbd8e6f17/procrastinate-3.8.1-py3-none-any.whl", hash = "sha256:67db4e9f0243c45775c02a0090fb3bfc7877d496e6b279d960d9ad4b1fa2f185", size = 148736, upload-time = "2026-04-08T06:24:19.754Z" }, +] + [[package]] name = "prometheus-client" version = "0.23.1" @@ -3041,6 +3064,90 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/0e/15/4f02896cc3df04fc465010a4c6a0cd89810f54617a32a70ef531ed75d61c/protobuf-6.33.2-py3-none-any.whl", hash = "sha256:7636aad9bb01768870266de5dc009de2d1b936771b38a793f73cbbf279c91c5c", size = 170501, upload-time = "2025-12-06T00:17:52.211Z" }, ] +[[package]] +name = "psycopg" +version = "3.3.4" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions", marker = "python_full_version < '3.13'" }, + { name = "tzdata", marker = "sys_platform == 'win32'" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/db/2f/cb91e5502ec9de1de6f1b76cfbf69531932725361168bb06963620c77e2e/psycopg-3.3.4.tar.gz", hash = "sha256:e21207764952cff81b6b8bdacad9a3939f2793367fdac2987b3aac36a651b5bc", size = 165799, upload-time = "2026-05-01T23:31:55.179Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/5c/e0/7b3dee031daae7743609ce3c746565d4a3ed7c2c186479eb48e34e838c64/psycopg-3.3.4-py3-none-any.whl", hash = "sha256:b6bbc25ccf05c8fad3b061d9db2ef0909a555171b84b07f29458a447253d679a", size = 213001, upload-time = "2026-05-01T23:20:50.816Z" }, +] + +[package.optional-dependencies] +binary = [ + { name = "psycopg-binary", marker = "implementation_name != 'pypy'" }, +] +pool = [ + { name = "psycopg-pool" }, +] + +[[package]] +name = "psycopg-binary" +version = "3.3.4" +source = { registry = "https://pypi.org/simple" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b6/82/df3312c0ca083d5b43b352f27d4dd8b1e614bd334473074715d9e0000da4/psycopg_binary-3.3.4-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:612a627d733f695b1de1f9b4bd511c15f999a5d8b915d444bbd7dd71cf3370da", size = 4609813, upload-time = "2026-05-01T23:26:30.612Z" }, + { url = "https://files.pythonhosted.org/packages/1f/b5/d74d542458d3e8ac0571d8a88f57ca369999b9a82f4fa528052d0d7d3e4c/psycopg_binary-3.3.4-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:13a7f380824c35896dcac7fe0f61440f7ca49d6dc73f3c13a9a4471e6a3b302e", size = 4676799, upload-time = "2026-05-01T23:26:38.475Z" }, + { url = "https://files.pythonhosted.org/packages/09/67/06bab9c60671999f4c6ceff1b334f3ac1f9fc5789eb467c714623ea21de9/psycopg_binary-3.3.4-cp311-cp311-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:276904e3452d6a23d474ef9a21eee19f20eed3d53ddd2576af033827e0ba0992", size = 5497050, upload-time = "2026-05-01T23:26:47.061Z" }, + { url = "https://files.pythonhosted.org/packages/72/9b/023433e2b20f970de1e22d29132a95281277646da0b2e2879dd4ee94b8c1/psycopg_binary-3.3.4-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:ab8cca8ef8fb1ccf5b048ae5bd78ba55b9e4b5d472e3ce5ca39ff4d2a9c249e4", size = 5172428, upload-time = "2026-05-01T23:26:56.708Z" }, + { url = "https://files.pythonhosted.org/packages/08/cd/ae16da8fde228a38b2fe9269bbc13cf89e0186173f2265600f02d6a71e64/psycopg_binary-3.3.4-cp311-cp311-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:7465bfe6087d2d5b42d4c53b9b11ca9f218e477317a4a162a10e3c19e984ba8e", size = 6762746, upload-time = "2026-05-01T23:27:07.023Z" }, + { url = "https://files.pythonhosted.org/packages/4f/81/0ba09fa5f5f88779093a2541a8e02489825721f258ab88058b11d68b3eb5/psycopg_binary-3.3.4-cp311-cp311-manylinux_2_38_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:22cdbf5f91ef7bb91fe0c5757e1962d3127a8010256eefd9c61fcaf441802097", size = 5006033, upload-time = "2026-05-01T23:27:12.221Z" }, + { url = "https://files.pythonhosted.org/packages/73/6a/629136040cc3497adb442a305710b5913f2a754d4630fc3d3717c4c0df65/psycopg_binary-3.3.4-cp311-cp311-musllinux_1_2_aarch64.whl", hash = "sha256:e2631da29253a98bd496e6c4813b24e09a4fe3fb2a9e88513305d6f8747cce95", size = 4534175, upload-time = "2026-05-01T23:27:18.248Z" }, + { url = "https://files.pythonhosted.org/packages/7c/32/1027f843c6dc2d5d51960ee62cc0c2cf755a4c39455aff1371173edbef7d/psycopg_binary-3.3.4-cp311-cp311-musllinux_1_2_ppc64le.whl", hash = "sha256:7f7668f30b9dd5163197e5cbf4e0efd54e00f0a859cc566ce56cfc31f4054839", size = 4224203, upload-time = "2026-05-01T23:27:24.3Z" }, + { url = "https://files.pythonhosted.org/packages/0b/e1/380a724d9093c74adb14d4fce920ea8327838abb61f760b1448586b14a8e/psycopg_binary-3.3.4-cp311-cp311-musllinux_1_2_riscv64.whl", hash = "sha256:cffc3408d77a27973f33e5d909b624cce683db5fc25964b02fe0aae7886c1007", size = 3954509, upload-time = "2026-05-01T23:27:30.815Z" }, + { url = "https://files.pythonhosted.org/packages/db/cd/895893ae575a09c97ccfd5def070d88993d955ef34df45a881fd5ff506d6/psycopg_binary-3.3.4-cp311-cp311-musllinux_1_2_x86_64.whl", hash = "sha256:0579252a1202cd73e4da137a1426e2dae993ae44e757605344282af3a082848c", size = 4259551, upload-time = "2026-05-01T23:27:38.828Z" }, + { url = "https://files.pythonhosted.org/packages/dd/c6/2330a20794e37a3ec609ef2fd8522919ec7a4395a1abf979a8e2d1775cd5/psycopg_binary-3.3.4-cp311-cp311-win_amd64.whl", hash = "sha256:41f2ec0fea529832982bcb6c9415de3c86264ebe562b77a467c0fbcd7efbba8d", size = 3572054, upload-time = "2026-05-01T23:27:45.455Z" }, + { url = "https://files.pythonhosted.org/packages/95/7d/03818e13ba7f36de93573c93ee3482006d3dfa8b0f8d28df511bad0a1a92/psycopg_binary-3.3.4-cp312-cp312-macosx_10_13_x86_64.whl", hash = "sha256:5ab28a2a7649df3b72e6b674b4c190e448e8e77cf496a65bd846472048de2089", size = 4591122, upload-time = "2026-05-01T23:27:56.162Z" }, + { url = "https://files.pythonhosted.org/packages/a5/b9/11b341edf8d54e2694726b273fe9652b254d989f4f63e3ac6816ad6b55f4/psycopg_binary-3.3.4-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:6402a9d8146cf4b3974ded3fd28a971e83dc6a0333eb7822524a3aa20b546578", size = 4669943, upload-time = "2026-05-01T23:28:04.522Z" }, + { url = "https://files.pythonhosted.org/packages/8b/18/4665bacd65e7865b4372fcd8abb8b9186ada4b0025f8c2ca691b364a556c/psycopg_binary-3.3.4-cp312-cp312-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:580ae30a5f95ccd90008ec697d3ed6a4a2047a516407ad904283fa42086936e9", size = 5469697, upload-time = "2026-05-01T23:28:11.337Z" }, + { url = "https://files.pythonhosted.org/packages/7c/b1/b83136c6e510593d9b0c759ba5384337bc4ad82d19fda675adc4b2703c84/psycopg_binary-3.3.4-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:e7510c37550f91a187e3660a8cc50d4b760f8c3b8b2f89ebc5698cd2c7f2c85d", size = 5152995, upload-time = "2026-05-01T23:28:20.529Z" }, + { url = "https://files.pythonhosted.org/packages/67/8d/a9821e2a648afe6091989929982a3b0f00b2631a859cb81379728f08fb75/psycopg_binary-3.3.4-cp312-cp312-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:77df19583501ea288eaf15ac0fe7ad01e6d8091a91d5c41df5c718f307d8e31b", size = 6738180, upload-time = "2026-05-01T23:28:30.654Z" }, + { url = "https://files.pythonhosted.org/packages/7e/58/2e349e8d23905dc2317b80ac65f48fb6f821a4777a4e994a60da91c4850f/psycopg_binary-3.3.4-cp312-cp312-manylinux_2_38_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:018fbed325936da502feb546642c982dcc4b9ffdea32dfef78dbf3b7f7ad4070", size = 4978828, upload-time = "2026-05-01T23:28:37.277Z" }, + { url = "https://files.pythonhosted.org/packages/45/48/57b00d03b4721878326122a1f1e6b0a90b85bcaec56b5b2f8ea6cfa45235/psycopg_binary-3.3.4-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:17a21953a9e5ff3a16dab692625a3676e2f101db5e40072f39dbee2250194d68", size = 4509757, upload-time = "2026-05-01T23:28:43.078Z" }, + { url = "https://files.pythonhosted.org/packages/25/37/33b47d8c007df69aec500df5889767c4d313748e8e9e27a2fef8a6dabcee/psycopg_binary-3.3.4-cp312-cp312-musllinux_1_2_ppc64le.whl", hash = "sha256:eb05ee1c2b817d27c537333224c9e83c7afb86fe7296ba970990068baf819b16", size = 4190546, upload-time = "2026-05-01T23:28:50.016Z" }, + { url = "https://files.pythonhosted.org/packages/ca/c6/32b0835dbc2122617902b649d76a91c1e75406e76bf3d595b0c3bb5ffad6/psycopg_binary-3.3.4-cp312-cp312-musllinux_1_2_riscv64.whl", hash = "sha256:773d573e11f437ce0bdb95b7c18dc58390494f96d43f8b45b9760436114f7652", size = 3926197, upload-time = "2026-05-01T23:28:55.55Z" }, + { url = "https://files.pythonhosted.org/packages/cd/68/d190ef0c0c5b16ded07831dabc8ddd412f4cdab07ec6e30ed38d9bda0e1f/psycopg_binary-3.3.4-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:71e55ccbdfae79a2ed9c6369c3008a3025817ff9d7e27b32a2d84e2a4267e66e", size = 4236627, upload-time = "2026-05-01T23:29:05.336Z" }, + { url = "https://files.pythonhosted.org/packages/25/8f/81dcbc2e8454b74d14881275ea45f00791052dac531a9fa8be1730d1685b/psycopg_binary-3.3.4-cp312-cp312-win_amd64.whl", hash = "sha256:494ca54901be8cf9eb7e02c25b731f2317c378efa44f43e8f9bd0e1184ae7be4", size = 3560782, upload-time = "2026-05-01T23:29:11.967Z" }, + { url = "https://files.pythonhosted.org/packages/09/43/13e9c406fbbf354580476e248a16b64802a376873ebe6339e30bb655572d/psycopg_binary-3.3.4-cp313-cp313-macosx_10_13_x86_64.whl", hash = "sha256:fbd1d4ed566895ad2d3bf4ddfd8bae90026930ddf29df3b9d91d32c8c47866a7", size = 4590377, upload-time = "2026-05-01T23:29:18.782Z" }, + { url = "https://files.pythonhosted.org/packages/22/be/2923cd7c3683e7afdecf4f10796a18de02f5c5ddc0969aa2ad0a8cdd3bbd/psycopg_binary-3.3.4-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:75a9067e236f9b9ae3535b66fe99bddb33d39c0de10112e49b9ab11eee53dc31", size = 4669023, upload-time = "2026-05-01T23:29:25.884Z" }, + { url = "https://files.pythonhosted.org/packages/96/a0/2c913d6fe13d6a8bd13597d36739bf47af063ad9399e402cfecab16f3c1e/psycopg_binary-3.3.4-cp313-cp313-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:b56b603ebcea8aa10b46228b8410ba7f13e7c2ee54389d4d9be0927fd8ce2a70", size = 5467423, upload-time = "2026-05-01T23:29:33.416Z" }, + { url = "https://files.pythonhosted.org/packages/e7/38/205d10bc1ad0df4a21c5c51659126bd3ea0ef98fcad1e852f78c249bb9c3/psycopg_binary-3.3.4-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:c677c4ad433cb7150c8cd304a0769ae3bcfbe5ea0676eb53faa7b1443b16d0d3", size = 5151137, upload-time = "2026-05-01T23:29:42.013Z" }, + { url = "https://files.pythonhosted.org/packages/36/fc/f0381ddcd45eff3bb70dbca6823a996048d7f507b2ec3fc92c6fabc0fe87/psycopg_binary-3.3.4-cp313-cp313-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:26df2717e59c0473e4465a97dfb1b7afebaa479277870fd5784d1436470db47c", size = 6736671, upload-time = "2026-05-01T23:29:51.626Z" }, + { url = "https://files.pythonhosted.org/packages/95/40/fa545ae152c24327651e5624e4902121e808270be36c10b12e9939be09bc/psycopg_binary-3.3.4-cp313-cp313-manylinux_2_38_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:1dc1f79fd16bb1f3f4421417a514607539f17804d95c7ed617265369d1981cae", size = 4979601, upload-time = "2026-05-01T23:29:56.961Z" }, + { url = "https://files.pythonhosted.org/packages/86/e4/2f8a47ee97f90cd2b933d0463081d35631ff419de2b8c984a5f369857de0/psycopg_binary-3.3.4-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:136f199a407b5348b9b857c504aff60c77622a28482e7195839ce1b51238c4cc", size = 4510513, upload-time = "2026-05-01T23:30:07.243Z" }, + { url = "https://files.pythonhosted.org/packages/0e/0e/94e842ff4a7f98ed162580ca2e8b8864b28c1e0350f2443f8ee47f821167/psycopg_binary-3.3.4-cp313-cp313-musllinux_1_2_ppc64le.whl", hash = "sha256:b6f5a29e9c775b9f12a1a717aa7a2c80f9e1db6f27ba44a5b59c80ac61d2ffcf", size = 4187243, upload-time = "2026-05-01T23:30:15.352Z" }, + { url = "https://files.pythonhosted.org/packages/d0/83/fc6c174b672e29b7de996ea77b6cbddf46c891751c3355f6974292baa6b4/psycopg_binary-3.3.4-cp313-cp313-musllinux_1_2_riscv64.whl", hash = "sha256:ee17a2cf4943cde261adfad1bbc5bf38d6b3776d7afff74c7cabcbeaeb08c260", size = 3927347, upload-time = "2026-05-01T23:30:21.186Z" }, + { url = "https://files.pythonhosted.org/packages/e9/65/768364d4a97a15b1a7f47ba52688c1686f22941d8332a8398cefc468e25f/psycopg_binary-3.3.4-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:5c4ab71be17bdca30cb34c34c4e1496e2f5d6f20c199c12bad226070b22ef9bf", size = 4236393, upload-time = "2026-05-01T23:30:26.211Z" }, + { url = "https://files.pythonhosted.org/packages/bd/3b/218efbc9e645becd80cdf651acda05f85cfe546b7a9c0458c7cbc8fe1f74/psycopg_binary-3.3.4-cp313-cp313-win_amd64.whl", hash = "sha256:dbfdb9b6cc79f31104a7b162a2b921b765fcc62af6c00540a167a8de47e4ed38", size = 3564592, upload-time = "2026-05-01T23:30:31.764Z" }, + { url = "https://files.pythonhosted.org/packages/48/a6/828c9185701dab71b234c2a76c38a08b098ebfec5020716b4e93807492b5/psycopg_binary-3.3.4-cp314-cp314-macosx_10_15_x86_64.whl", hash = "sha256:28b7398fdd19db3232c884fb24550bdfe951221f510e195e233299e4c9b78f97", size = 4607292, upload-time = "2026-05-01T23:30:38.962Z" }, + { url = "https://files.pythonhosted.org/packages/92/58/5b40dbc9d839045c9dae956960e4fb6d20bcabe6c59a2aa34fc3a371913f/psycopg_binary-3.3.4-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:1fbaa292a3c8bb61b45df1ad3da1908ccee7cb889db9425e3557d9e34e2a4829", size = 4687023, upload-time = "2026-05-01T23:30:47.227Z" }, + { url = "https://files.pythonhosted.org/packages/85/a9/793f0ac107a9003b48441d0d1f9f616d96e0f37458dd8dc12528ceff55fb/psycopg_binary-3.3.4-cp314-cp314-manylinux2014_ppc64le.manylinux_2_17_ppc64le.whl", hash = "sha256:94596f9e7633ee3f6440711d43bb70aa31cc0a46a900ab8b4201a366ace5c9e7", size = 5486985, upload-time = "2026-05-01T23:30:55.517Z" }, + { url = "https://files.pythonhosted.org/packages/8f/26/42e8533497e2592334f68ec529cf5f840f7fa4e99575a4bb61aa184dbfbf/psycopg_binary-3.3.4-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:8c0056529e68dbe9184cd4019a1f3d8f3a4ead2f6fc7a5afcf27d3314edd1277", size = 5168745, upload-time = "2026-05-01T23:31:01.904Z" }, + { url = "https://files.pythonhosted.org/packages/15/af/b7151776cc08d5935d45c833ec818a9beb417cf7c08239af1aafbdae78ee/psycopg_binary-3.3.4-cp314-cp314-manylinux_2_27_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:2c09aad7051326e7603c14e50636db9c01f78272dc54b3accff03d46370461e6", size = 6761486, upload-time = "2026-05-01T23:31:14.511Z" }, + { url = "https://files.pythonhosted.org/packages/d0/ed/c92533b9124712d592cbf1cd6c76da933a2e0acea81dfe1fbe7e735f0cff/psycopg_binary-3.3.4-cp314-cp314-manylinux_2_38_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:514404ed543efd620c85602b747df2a23cf1241b4067199e1a66f2d2757aaa41", size = 4997427, upload-time = "2026-05-01T23:31:20.901Z" }, + { url = "https://files.pythonhosted.org/packages/a2/23/ccadfd0de416aa188356daa199453af24087b042e296088706d190ae0295/psycopg_binary-3.3.4-cp314-cp314-musllinux_1_2_aarch64.whl", hash = "sha256:46893c26858be12cc49ca4226ed6a60b4bfccadd946b3bebb783a60b38788228", size = 4533549, upload-time = "2026-05-01T23:31:26.204Z" }, + { url = "https://files.pythonhosted.org/packages/fd/a0/c8f43cee36386f7bc891ab41a9d31ea07cf9826038e732da79f26b1e5f34/psycopg_binary-3.3.4-cp314-cp314-musllinux_1_2_ppc64le.whl", hash = "sha256:df1d567fc430f6df15c9fcf67d87685fc49bdb325adc0db5af1adfb2f44eb5c9", size = 4210256, upload-time = "2026-05-01T23:31:33.884Z" }, + { url = "https://files.pythonhosted.org/packages/4e/2c/c1547871be3790676e8868b38655496422f94f0978dfb66b74bdba2f1676/psycopg_binary-3.3.4-cp314-cp314-musllinux_1_2_riscv64.whl", hash = "sha256:6b9016b1714da4dd5ecaaa75b82098aa5a0b87854ce9b092e21c27c4ae23e014", size = 3946204, upload-time = "2026-05-01T23:31:39.626Z" }, + { url = "https://files.pythonhosted.org/packages/c4/b1/f6670f00fa7ea601584623f6c11602ab92117d83eaff885e0210f6de7418/psycopg_binary-3.3.4-cp314-cp314-musllinux_1_2_x86_64.whl", hash = "sha256:47c656a8a7ba6eb0cff1801a4caaa9c8bdc12d03080e273aff1c8ac39971a77e", size = 4255811, upload-time = "2026-05-01T23:31:44.986Z" }, + { url = "https://files.pythonhosted.org/packages/eb/e6/5fff07a70d1f945ed90ae131c3bd76cab32beff7c58c6db15ad5820b6d1f/psycopg_binary-3.3.4-cp314-cp314-win_amd64.whl", hash = "sha256:c37e024c07308cd06cf3ec51bfd0e7f6157585a4d84d1bce4a7f5f7913719bf8", size = 3666849, upload-time = "2026-05-01T23:31:51.165Z" }, +] + +[[package]] +name = "psycopg-pool" +version = "3.3.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/90/82/7a23d26039827ecd4ebe93905651029ddd307c5182ad59296dfb6f67b528/psycopg_pool-3.3.1.tar.gz", hash = "sha256:b10b10b7a175d5cc1592147dc5b7eec8a9e0834eb3ed2c4a92c858e2f51eb63c", size = 31661, upload-time = "2026-05-01T23:31:59.809Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/37/ed/89c2c620af0e1660354cd8aabf9f5b21f911597ce22acb37c805d6c86bc8/psycopg_pool-3.3.1-py3-none-any.whl", hash = "sha256:2af5b432941c4c9ad5c87b3fa410aec910ec8f7c122855897983a06c45f2e4b5", size = 40023, upload-time = "2026-05-01T23:31:53.136Z" }, +] + [[package]] name = "ptyprocess" version = "0.7.0"