fix(ingest): disable psycopg prepared statements behind the pgbouncer pooler

nextcloud_mcp_server/vector/queue/procrastinate.py

@@ -501,12 +501,29 @@ def build_app(connector: BaseConnector) -> App:
     return app
 
 
+# psycopg3 auto-prepares statements as server-side NAMED statements (``_pg3_N``, a
+# per-client counter). Behind a pgbouncer TRANSACTION-mode pooler (Deck #424) many
+# clients share the same backend server connections, so those names collide across
+# clients (``DuplicatePreparedStatement``, or a param-count ``ProtocolViolation``).
+# The failed statement aborts its transaction, and in transaction pooling an
+# aborted-but-open transaction never releases its server connection — the tenant's
+# small server pool fills with stuck ``idle in transaction (aborted)`` backends and
+# a new worker's pool init then times out (``psycopg_pool.PoolTimeout``). Disabling
+# auto-prepare (``prepare_threshold=None`` → psycopg uses unnamed statements) is the
+# standard fix. Applied UNCONDITIONALLY: harmless connecting direct (a negligible
+# re-parse cost for this poll-heavy workload) and it removes the footgun of pointing
+# a worker at a transaction-mode pooler without it — exactly how #424 regressed
+# (LISTEN/NOTIFY was disabled for the pooler, but prepared statements were not).
+# Passed to psycopg via the pool's per-connection ``kwargs`` (PsycopgConnector
+# forwards ``**kwargs`` to ``psycopg_pool.AsyncConnectionPool``).
+def _psycopg_connector(conninfo: str) -> PsycopgConnector:
+    return PsycopgConnector(conninfo=conninfo, kwargs={"prepare_threshold": None})
+
+
 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))
-    )
+    return build_app(_psycopg_connector(get_procrastinate_conninfo(database_url)))
 
 
 _app: App | None = None
@@ -516,7 +533,7 @@ 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()))
+        _app = build_app(_psycopg_connector(get_procrastinate_conninfo()))
     return _app
 
 

tests/unit/vector/test_procrastinate_producer.py

@@ -5,7 +5,7 @@
 
 from types import SimpleNamespace
 from typing import cast
-from unittest.mock import AsyncMock
+from unittest.mock import AsyncMock, patch
 
 import pytest
 from procrastinate import App, JobContext, testing
@@ -394,3 +394,25 @@ class FakeApp:
         assert set(by_queue) == {"ingest-fast", "ingest-ocr"}
         assert by_queue["ingest-fast"]["todo"] == 3
         assert by_queue["ingest-ocr"]["failed"] == 2
+
+
+def test_build_app_for_url_disables_psycopg_prepared_statements():
+    """The procrastinate connector MUST disable psycopg auto-prepared statements
+    (``prepare_threshold=None``) so it is safe behind the pgbouncer transaction-mode
+    pooler (Deck #424): per-client ``_pg3_N`` statement names otherwise collide over
+    the shared server connections, abort transactions, and exhaust the server pool."""
+    with patch.object(pq, "PsycopgConnector") as mock_conn:
+        pq.build_app_for_url("postgresql+psycopg://u:p@h:5432/db")
+    mock_conn.assert_called_once()
+    assert mock_conn.call_args.kwargs["kwargs"] == {"prepare_threshold": None}
+
+
+def test_get_procrastinate_app_disables_psycopg_prepared_statements(monkeypatch):
+    """The process-wide App builds the same prepared-statement-safe connector."""
+    monkeypatch.setattr(pq, "_app", None)  # reset the module singleton before/after
+    monkeypatch.setattr(pq, "get_procrastinate_conninfo", lambda: "host=h dbname=db")
+    with patch.object(pq, "PsycopgConnector") as mock_conn:
+        pq.get_procrastinate_app()
+    monkeypatch.setattr(pq, "_app", None)  # don't leak a mock-backed singleton
+    mock_conn.assert_called_once()
+    assert mock_conn.call_args.kwargs["kwargs"] == {"prepare_threshold": None}