From 820b98dac1c9103e8332ae33e24456c7b8281b82 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 3 Jun 2026 15:21:50 +0200 Subject: [PATCH] fix: address PR #836 round-2 review (connect/timeout/observability) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 🟡 Document why ProcrastinateTaskProducer.connect() uses `await app.open_async()` (AwaitableContext: await opens a long-lived pool, closed by drain()) and add a connect()/drain() lifecycle unit test (InMemoryConnector) asserting the pool is opened by connect and closed by drain — previously untested. 🟡 get_procrastinate_conninfo: forward connect_timeout from DATABASE_URL or default 10s so an unreachable DB can't hang worker/API startup indefinitely; warn only on other dropped query params. + tests. 🟢 INGEST_DELETE_SUCCEEDED_JOBS (default true) makes the worker's succeeded-job deletion configurable for audit retention. 🟢 Worker startup logs via logger.info (structured/OTel) instead of click.echo. 🟢 INGEST_STALLED_JOB_SECONDS (default 300) makes the crash-reclaim threshold tunable for slow embedding backends; reclaim reads it per-run. The broad `except` in _apply_ingest_queue_schema_open is kept deliberately: procrastinate wraps psycopg errors, so narrowing to psycopg.errors.* would miss the wrapped DDL-conflict and turn a benign concurrent-apply race into a failure; the presence re-check re-raises genuine errors. Co-Authored-By: Claude Opus 4.8 (1M context) --- nextcloud_mcp_server/cli.py | 21 +++++++--- nextcloud_mcp_server/config.py | 38 +++++++++++++++---- .../vector/queue/procrastinate.py | 19 +++++++--- tests/unit/test_decomposition_config.py | 24 ++++++++++++ .../vector/test_procrastinate_producer.py | 16 ++++++++ 5 files changed, 99 insertions(+), 19 deletions(-) diff --git a/nextcloud_mcp_server/cli.py b/nextcloud_mcp_server/cli.py index 8bcaa096..da29f1c7 100644 --- a/nextcloud_mcp_server/cli.py +++ b/nextcloud_mcp_server/cli.py @@ -1,3 +1,4 @@ +import logging import os from importlib.metadata import version @@ -21,6 +22,8 @@ from nextcloud_mcp_server.server import AVAILABLE_APPS from .app import get_app +logger = logging.getLogger(__name__) + @click.command() @click.option( @@ -332,16 +335,24 @@ def worker(concurrency: int | None): # 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}" + # Structured log (not click.echo) so it lands in the JSON / OTel + # pipeline like every other startup message. + logger.info( + "Ingest worker started: queue=%s concurrency=%s delete_succeeded=%s", + INGEST_QUEUE_NAME, + workers, + settings.ingest_delete_succeeded_jobs, ) 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", + # Drop succeeded jobs (default) so the queue table stays lean and + # the KEDA queue-depth metric reflects only outstanding work; set + # INGEST_DELETE_SUCCEEDED_JOBS=false to retain them for audit. + delete_jobs="successful" + if settings.ingest_delete_succeeded_jobs + else "never", ) anyio.run(_run) diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index f372a312..3b44bd3d 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -178,6 +178,14 @@ _DEFAULTS: dict[str, Any] = { # `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 + # Reclaim an ingest job orphaned in ``doing`` by a crashed worker once its + # worker heartbeat is this many seconds stale (Deck #183). Default is well + # above the longest expected document; raise it for slow embedding backends. + "ingest_stalled_job_seconds": 300, + # Delete succeeded ingest jobs (keeps the queue table lean + the KEDA + # queue-depth metric clean). Set false to retain succeeded rows for audit + # (note: indexing success is also recorded in logs/metrics regardless). + "ingest_delete_succeeded_jobs": True, "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. @@ -255,6 +263,7 @@ _dynaconf = Dynaconf( # Port ranges Validator("METRICS_PORT", gte=1, lte=65535), # Positive integers + Validator("INGEST_STALLED_JOB_SECONDS", gte=1), Validator("VECTOR_SYNC_SCAN_INTERVAL", gte=1), Validator("VECTOR_SYNC_PROCESSOR_WORKERS", gte=1), Validator("VECTOR_SYNC_QUEUE_MAX_SIZE", gte=1), @@ -707,6 +716,8 @@ class Settings: # ``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) + ingest_stalled_job_seconds: int = 300 # crashed-worker reclaim threshold + ingest_delete_succeeded_jobs: bool = True # drop succeeded ingest jobs collection_metadata_source: str = "qdrant" # qdrant | api collection_metadata_api_url: str | None = None # CP URL when source=api embedding_gateway_url: str | None = None # required when provider=gateway @@ -1306,6 +1317,8 @@ def get_settings() -> Settings: "embedding_provider": "EMBEDDING_PROVIDER", "ingest_queue": "INGEST_QUEUE", "mcp_role": "MCP_ROLE", + "ingest_stalled_job_seconds": "INGEST_STALLED_JOB_SECONDS", + "ingest_delete_succeeded_jobs": "INGEST_DELETE_SUCCEEDED_JOBS", "collection_metadata_source": "COLLECTION_METADATA_SOURCE", "collection_metadata_api_url": "COLLECTION_METADATA_API_URL", "embedding_gateway_url": "EMBEDDING_GATEWAY_URL", @@ -1437,12 +1450,12 @@ 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. + Only the host/port/dbname/user/password components are forwarded, plus + ``connect_timeout`` (a libpq keyword) honored from the URL query string or + defaulted to 10s so a slow/unreachable DB can't hang worker/API startup + indefinitely. Any *other* ``?key=value`` query parameters are **dropped** + (TLS is set separately via :func:`_pg_ssl_params`, and SQLAlchemy-specific + options don't map cleanly to libpq keywords); a warning lists them. Raises ``ValueError`` for a non-Postgres URL — procrastinate is Postgres-only. """ @@ -1456,11 +1469,14 @@ def get_procrastinate_conninfo(database_url: str | None = None) -> str: f"got driver {url.drivername!r}" ) - if url.query: + # ``connect_timeout`` is forwarded (libpq keyword); everything else in the + # query string is dropped with a warning. + dropped = sorted(k for k in url.query if k != "connect_timeout") + if dropped: logging.getLogger(__name__).warning( "Dropping DATABASE_URL query parameters not forwarded to the " "procrastinate connector: %s", - ", ".join(sorted(url.query)), + ", ".join(dropped), ) params: dict[str, str] = {} @@ -1474,6 +1490,12 @@ def get_procrastinate_conninfo(database_url: str | None = None) -> str: params["user"] = url.username if url.password: params["password"] = url.password + # Honor an operator-supplied connect_timeout, else default to 10s. (make_url + # query values are str or a tuple of strs when repeated; take the last.) + _ct = url.query.get("connect_timeout") + if isinstance(_ct, (list, tuple)): + _ct = _ct[-1] if _ct else None + params["connect_timeout"] = str(_ct) if _ct else "10" params.update(_pg_ssl_params()) return make_conninfo(**params) diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index 456325c4..e25c5a2a 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/nextcloud_mcp_server/vector/queue/procrastinate.py @@ -56,11 +56,11 @@ _NAMESPACE = "ingest" INGEST_TASK_NAME = f"{_NAMESPACE}:process_document" # 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 +# heartbeat is this many seconds stale. The default is 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. Operators on slow embedding backends can tune +# it via INGEST_STALLED_JOB_SECONDS (read per-run in reclaim_stalled_ingest_jobs). # Tasks are defined as plain functions and registered onto a *fresh* Blueprint @@ -125,9 +125,10 @@ async def reclaim_stalled_ingest_jobs(context: JobContext, timestamp: int) -> No """ manager = context.app.job_manager retry_at = datetime.now(tz=timezone.utc) + stalled_after = get_settings().ingest_stalled_job_seconds reclaimed = 0 for job in await manager.get_stalled_jobs( - queue=INGEST_QUEUE_NAME, seconds_since_heartbeat=_STALLED_AFTER_SECONDS + queue=INGEST_QUEUE_NAME, seconds_since_heartbeat=stalled_after ): if job.id is None: continue @@ -307,6 +308,12 @@ class ProcrastinateTaskProducer: @classmethod async def connect(cls) -> ProcrastinateTaskProducer: app = get_procrastinate_app() + # ``App.open_async()`` returns procrastinate's dual-mode AwaitableContext: + # ``await``-ing it opens the connector pool and leaves it open (vs the + # ``async with`` form, which closes on block exit). The producer's pool is + # long-lived — owned by the server lifespan and torn down once in + # ``drain()`` (close_async) on shutdown — so the bare ``await`` is correct + # here, unlike the scoped ``async with`` used for one-shot schema apply. await app.open_async() return cls(app) diff --git a/tests/unit/test_decomposition_config.py b/tests/unit/test_decomposition_config.py index 3e18af0c..9a2869a0 100644 --- a/tests/unit/test_decomposition_config.py +++ b/tests/unit/test_decomposition_config.py @@ -116,6 +116,30 @@ class TestProcrastinateConninfo: assert parsed["dbname"] == "mcp" assert parsed.get("sslmode") == expected_sslmode + def test_conninfo_connect_timeout_defaults_to_10(self, monkeypatch): + from psycopg.conninfo import conninfo_to_dict + + monkeypatch.setattr( + config_module, + "get_database_url", + lambda: "postgresql+asyncpg://mcp:s@db/mcp", + ) + monkeypatch.setattr(config_module, "get_database_ssl", lambda: None) + parsed = conninfo_to_dict(config_module.get_procrastinate_conninfo()) + assert parsed["connect_timeout"] == "10" + + def test_conninfo_honors_url_connect_timeout(self, monkeypatch): + from psycopg.conninfo import conninfo_to_dict + + monkeypatch.setattr( + config_module, + "get_database_url", + lambda: "postgresql+asyncpg://mcp:s@db/mcp?connect_timeout=3", + ) + monkeypatch.setattr(config_module, "get_database_ssl", lambda: None) + parsed = conninfo_to_dict(config_module.get_procrastinate_conninfo()) + assert parsed["connect_timeout"] == "3" + def test_conninfo_ssl_mapping(self, monkeypatch): from psycopg.conninfo import conninfo_to_dict diff --git a/tests/unit/vector/test_procrastinate_producer.py b/tests/unit/vector/test_procrastinate_producer.py index f84b4b0c..8960d951 100644 --- a/tests/unit/vector/test_procrastinate_producer.py +++ b/tests/unit/vector/test_procrastinate_producer.py @@ -67,6 +67,22 @@ class TestProcrastinateTaskProducer: producer = pq.ProcrastinateTaskProducer(app) assert producer.clone() is producer + async def test_connect_opens_pool_and_drain_closes(self, app, monkeypatch): + # connect() resolves the process-wide app; point it at our in-memory one. + monkeypatch.setattr(pq, "get_procrastinate_app", lambda: app) + + producer = await pq.ProcrastinateTaskProducer.connect() + # `await app.open_async()` must actually open the connector (regression + # guard for the await-vs-`async with` form on the long-lived pool). + assert app.connector.states == ["open_async"] + + # An open pool means send() works end-to-end. + await producer.send(_task()) + assert len(app.connector.jobs) == 1 + + await producer.drain() + assert "closed_async" in app.connector.states + class TestProcessDocumentTask: async def test_runs_pipeline_and_closes_client(self, monkeypatch):