diff --git a/nextcloud_mcp_server/vector/queue/procrastinate.py b/nextcloud_mcp_server/vector/queue/procrastinate.py index a7bd9edb..41994917 100644 --- a/nextcloud_mcp_server/vector/queue/procrastinate.py +++ b/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 diff --git a/tests/unit/vector/test_procrastinate_producer.py b/tests/unit/vector/test_procrastinate_producer.py index 225081e0..c5a703e2 100644 --- a/tests/unit/vector/test_procrastinate_producer.py +++ b/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}