diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 9bbf38b3..92e64467 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -355,10 +355,18 @@ class AppContext: storage: "RefreshTokenStorage | None" = None document_send_stream: MemoryObjectSendStream | None = None document_receive_stream: MemoryObjectReceiveStream | None = None - task_producer: "TaskProducer | None" = None shutdown_event: anyio.Event | None = None scanner_wake_event: anyio.Event | None = None + @property + def task_producer(self) -> "TaskProducer | None": + # Read dynamically from the module-level singleton (like + # eviction_task_group) rather than snapshotting at yield time — that way + # a session can't observe a stale ``None`` and the per-session yields + # can't forget to forward it (the bug this property replaces). The + # vector-sync status tool reads this for postgres-backend job counts. + return _vector_sync_state.task_producer + @property def eviction_task_group(self) -> TaskGroup | None: # Read dynamically from the module-level singleton instead of @@ -381,10 +389,14 @@ class OAuthAppContext: server_client_id: str | None = None # MCP server's OAuth client ID (static or DCR) document_send_stream: MemoryObjectSendStream | None = None document_receive_stream: MemoryObjectReceiveStream | None = None - task_producer: "TaskProducer | None" = None shutdown_event: anyio.Event | None = None scanner_wake_event: anyio.Event | None = None + @property + def task_producer(self) -> "TaskProducer | None": + # See AppContext.task_producer for rationale. + return _vector_sync_state.task_producer + @property def eviction_task_group(self) -> TaskGroup | None: # See AppContext.eviction_task_group for rationale. @@ -606,8 +618,8 @@ async def app_lifespan_basic(server: FastMCP) -> AsyncIterator[AppContext]: document_receive_stream=_vector_sync_state.document_receive_stream, shutdown_event=_vector_sync_state.shutdown_event, scanner_wake_event=_vector_sync_state.scanner_wake_event, - # eviction_task_group is exposed via @property (reads - # _vector_sync_state at access time, not snapshot). + # task_producer and eviction_task_group are exposed via @property + # (read _vector_sync_state at access time, not snapshot). ) finally: logger.info("Shutting down BasicAuth session") @@ -1239,8 +1251,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = document_receive_stream=_vector_sync_state.document_receive_stream, shutdown_event=_vector_sync_state.shutdown_event, scanner_wake_event=_vector_sync_state.scanner_wake_event, - # eviction_task_group is exposed via @property (reads - # _vector_sync_state at access time, not snapshot). + # task_producer and eviction_task_group are exposed via + # @property (read _vector_sync_state at access time). ) finally: logger.info("Shutting down MCP server") diff --git a/nextcloud_mcp_server/cli.py b/nextcloud_mcp_server/cli.py index fef400ca..8bcaa096 100644 --- a/nextcloud_mcp_server/cli.py +++ b/nextcloud_mcp_server/cli.py @@ -326,9 +326,12 @@ def worker(concurrency: int | None): 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) + # Open the connector pool once and reuse it for both the defensive + # schema apply (the always-on API pod is the authoritative applier) and + # the worker loop — manage_connection=False avoids a redundant + # open/close cycle on startup. async with app.open_async(): + await apply_ingest_queue_schema(app, manage_connection=False) click.echo( f"Ingest worker started: queue={INGEST_QUEUE_NAME} concurrency={workers}" ) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index 189ae1fa..f372a312 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -1437,6 +1437,13 @@ def get_procrastinate_conninfo(database_url: str | None = None) -> str: (set ``DATABASE_URL`` to a ``+psycopg`` URL) rather than rewriting the driver in code — see charts repo, not this repo. + Only the host/port/dbname/user/password components are forwarded; any + ``?key=value`` query parameters on the URL (e.g. ``application_name``, + ``connect_timeout``) are **dropped** — TLS is set separately via + :func:`_pg_ssl_params`, and SQLAlchemy-specific query options don't map + cleanly to libpq keywords. A warning is logged when params are dropped so + operators aren't surprised. + Raises ``ValueError`` for a non-Postgres URL — procrastinate is Postgres-only. """ from psycopg.conninfo import make_conninfo # noqa: PLC0415 @@ -1449,6 +1456,13 @@ def get_procrastinate_conninfo(database_url: str | None = None) -> str: f"got driver {url.drivername!r}" ) + if url.query: + logging.getLogger(__name__).warning( + "Dropping DATABASE_URL query parameters not forwarded to the " + "procrastinate connector: %s", + ", ".join(sorted(url.query)), + ) + params: dict[str, str] = {} if url.host: params["host"] = url.host diff --git a/nextcloud_mcp_server/vector/ingest_status.py b/nextcloud_mcp_server/vector/ingest_status.py index ae90aa48..6f470525 100644 --- a/nextcloud_mcp_server/vector/ingest_status.py +++ b/nextcloud_mcp_server/vector/ingest_status.py @@ -36,6 +36,12 @@ async def get_ingest_pending( ) -> IngestPending: """Compute outstanding ingest work for the configured queue backend. + ``task_producer`` and ``document_receive_stream`` are intentionally typed + ``Any``: they're duck-typed across backends. Only ``ProcrastinateTaskProducer`` + exposes ``job_counts`` (the ``TaskProducer`` protocol doesn't), and the memory + backend reads the anyio stream's ``statistics()`` — so no single concrete type + or Protocol fits both branches, and we probe with ``hasattr`` instead. + Never raises — a status surface must stay available even if the queue is unreachable; failures degrade to ``pending=0``. """ diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index a68d75b2..456325c4 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/nextcloud_mcp_server/vector/queue/procrastinate.py @@ -54,7 +54,6 @@ 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 @@ -217,8 +216,8 @@ async def _ingest_schema_present(app: App) -> bool: 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). +async def _apply_ingest_queue_schema_open(app: App) -> None: + """Apply the ingest-queue schema on an already-open connector (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 @@ -230,25 +229,41 @@ async def apply_ingest_queue_schema(app: App | None = None) -> None: 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. + """ + 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: + # The apply runs in a single transaction, so any failure rolls back + # atomically (no partial schema). The only benign case is losing the + # create race to another pod — confirmed by re-checking presence. Any + # other failure (network, auth, …) leaves the schema absent, so this + # branch re-raises it rather than masking it. + if await _ingest_schema_present(app): + logger.info("Ingest queue schema applied concurrently by another pod") + return + raise - Opens a short-lived connection, so it is safe to call from the CLI - (``db upgrade`` / worker startup). + +async def apply_ingest_queue_schema( + app: App | None = None, *, manage_connection: bool = True +) -> None: + """Create procrastinate's tables on a fresh database (apply-if-absent). + + By default opens a short-lived connection, so it is safe to call standalone + from the CLI ``db upgrade`` path. Pass ``manage_connection=False`` when the + caller already holds an open connector (the ``worker`` command opens the App + once and reuses it) to avoid a redundant open/close cycle. """ app = app or get_procrastinate_app() + if not manage_connection: + await _apply_ingest_queue_schema_open(app) + return 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 + await _apply_ingest_queue_schema_open(app) # Job-status keys procrastinate flattens into each list_queues row (alongside diff --git a/tests/unit/test_app_context_task_producer.py b/tests/unit/test_app_context_task_producer.py new file mode 100644 index 00000000..ab411eca --- /dev/null +++ b/tests/unit/test_app_context_task_producer.py @@ -0,0 +1,34 @@ +"""Regression test for the lifespan context `task_producer` exposure (Deck #183). + +`nc_get_vector_sync_status` reads `lifespan_ctx.task_producer` for postgres-backend +job counts. It was previously a snapshot dataclass field the per-session yields +forgot to populate, so the tool always reported `pending=0` on the postgres +backend. It is now a `@property` that reads the module singleton live (like +`eviction_task_group`); these tests pin that contract. +""" + +from typing import cast + +import pytest + +import nextcloud_mcp_server.app as app_module +from nextcloud_mcp_server.app import AppContext, OAuthAppContext +from nextcloud_mcp_server.client import NextcloudClient + +pytestmark = pytest.mark.unit + + +def test_app_context_task_producer_reads_vector_sync_state(monkeypatch): + sentinel = object() + monkeypatch.setattr(app_module._vector_sync_state, "task_producer", sentinel) + ctx = AppContext(client=cast(NextcloudClient, None)) + assert ctx.task_producer is sentinel + + +def test_oauth_app_context_task_producer_reads_vector_sync_state(monkeypatch): + sentinel = object() + monkeypatch.setattr(app_module._vector_sync_state, "task_producer", sentinel) + ctx = OAuthAppContext( + nextcloud_host="https://example.test", token_verifier=object() + ) + assert ctx.task_producer is sentinel