fix: address PR #836 round-2 review (connect/timeout/observability)

🟡 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) <noreply@anthropic.com>

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(
+            # Structured log (not click.echo) so it lands in the JSON / OTel
-                f"Ingest worker started: queue={INGEST_QUEUE_NAME} concurrency={workers}"
+            # 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
+                # Drop succeeded jobs (default) so the queue table stays lean and
-                # queue-depth metric reflects only outstanding work.
+                # the KEDA queue-depth metric reflects only outstanding work; set
-                delete_jobs="successful",
+                # INGEST_DELETE_SUCCEEDED_JOBS=false to retain them for audit.
+                delete_jobs="successful"
+                if settings.ingest_delete_succeeded_jobs
+                else "never",
             )
 
     anyio.run(_run)

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
+    Only the host/port/dbname/user/password components are forwarded, plus
-    ``?key=value`` query parameters on the URL (e.g. ``application_name``,
+    ``connect_timeout`` (a libpq keyword) honored from the URL query string or
-    ``connect_timeout``) are **dropped** — TLS is set separately via
+    defaulted to 10s so a slow/unreachable DB can't hang worker/API startup
-    :func:`_pg_ssl_params`, and SQLAlchemy-specific query options don't map
+    indefinitely. Any *other* ``?key=value`` query parameters are **dropped**
-    cleanly to libpq keywords. A warning is logged when params are dropped so
+    (TLS is set separately via :func:`_pg_ssl_params`, and SQLAlchemy-specific
-    operators aren't surprised.
+    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)

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
+# heartbeat is this many seconds stale. The default is sized well above the
-# ``process_document`` (PDF render + embedding) so a slow-but-live worker — whose
+# longest expected ``process_document`` (PDF render + embedding) so a slow-but-
-# heartbeat stays current during a long job — is never reclaimed out from under
+# live worker — whose heartbeat stays current during a long job — is never
-# itself.
+# reclaimed out from under itself. Operators on slow embedding backends can tune
-_STALLED_AFTER_SECONDS = 300
+# 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)
 

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
 

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):