On Wed, 28 Oct 2026 at 22:33:16 +0100, Jelte Fennema-Nio
<[email protected]> wrote:
> --tb=short, only the last 30 server log
> lines, pg_ctl --silent, framework internals hidden from tracebacks,
> and a clearer error for unconsumed background query results.
Thanks, Jelte, for working on this. I'm happy to see this move to Python
happening and would like to help. Two follow-ups:
1. result(timeout=...) and exception(timeout=...) mark a future consumed
before waiting. A caught timeout can therefore allow connection reuse
while the query still runs. The first patch fixes this: eight baseline
failures become passes (20/20 total).
2. The bulk pg_visibility conversion (8a8f008f7596) misses
close_portal("") after begin / select txid_current(), retaining a snapshot
that psql releases. The second patch adds two passing comparison tests
for this documented behavior.
Apply order: the timeout patch, then the snapshot patch, to v2 (1d2adb67c551).
On macOS arm64, all 22 new cases, core regression and isolation pass.
The known Python 3.9 timestamp failures and optional-feature skips remain.
Reproduction steps, exact revisions and detailed results are in [1].
My Postgres hacking AI harness did the legwork here. I haven't yet
reviewed the findings or patches by hand, but I trust it enough to share
them for review. If anything is off, please tell me and I'll adjust or
stop this work.
[1]
https://github.com/NikolayS/postgres/blob/d972a089cf7b640ee235489c71daac100d9f4bad/review/pytest-followup/README.md
Nik
From 74d33c28b721db9355ad5030ffa3c57c9d062fb2 Mon Sep 17 00:00:00 2001
From: Nik Samokhvalov <[email protected]>
Date: Wed, 28 Oct 2026 21:17:55 -0700
Subject: [PATCH 1/2] Keep background query outcomes unconsumed after a wait
timeout
---
src/test/Makefile | 1 +
src/test/meson.build | 2 +
src/test/pytest/Makefile | 14 ++++
src/test/pytest/libpq/_core.py | 20 +++--
src/test/pytest/meson.build | 12 +++
src/test/pytest/pyt/test_future.py | 127 +++++++++++++++++++++++++++++
6 files changed, 170 insertions(+), 6 deletions(-)
create mode 100644 src/test/pytest/Makefile
create mode 100644 src/test/pytest/meson.build
create mode 100644 src/test/pytest/pyt/test_future.py
diff --git a/src/test/Makefile b/src/test/Makefile
index 3eb0a06abb..0be9771d71 100644
--- a/src/test/Makefile
+++ b/src/test/Makefile
@@ -18,6 +18,7 @@ SUBDIRS = \
modules \
perl \
postmaster \
+ pytest \
recovery \
regress \
subscription
diff --git a/src/test/meson.build b/src/test/meson.build
index cd45cbf57f..98b602ec97 100644
--- a/src/test/meson.build
+++ b/src/test/meson.build
@@ -26,3 +26,5 @@ if icu.found()
endif
subdir('perl')
+
+subdir('pytest')
diff --git a/src/test/pytest/Makefile b/src/test/pytest/Makefile
new file mode 100644
index 0000000000..c5d5a0637f
--- /dev/null
+++ b/src/test/pytest/Makefile
@@ -0,0 +1,14 @@
+# src/test/pytest/Makefile
+
+subdir = src/test/pytest
+top_builddir = ../../..
+include $(top_builddir)/src/Makefile.global
+
+check:
+ $(pytest_check)
+
+installcheck:
+ $(pytest_installcheck)
+
+clean distclean maintainer-clean:
+ rm -rf tmp_check
diff --git a/src/test/pytest/libpq/_core.py b/src/test/pytest/libpq/_core.py
index d68562726f..5f22448d08 100644
--- a/src/test/pytest/libpq/_core.py
+++ b/src/test/pytest/libpq/_core.py
@@ -16,7 +16,7 @@ import ctypes
import os
import sys
import warnings
-from concurrent.futures import Future, ThreadPoolExecutor
+from concurrent.futures import CancelledError, Future, ThreadPoolExecutor
from typing import Any, NamedTuple, NoReturn
from ._bindings import (
@@ -121,8 +121,8 @@ class _MustConsumeFuture(Future):
A plain ``concurrent.futures.Future`` silently drops an exception that
nobody retrieves, which would let a background query fail unnoticed. This
- subclass records whether ``result()``/``exception()`` was called, so that
- ``PGconn.close()`` can turn a forgotten ``.result()`` into a visible test
+ subclass records whether ``result()``/``exception()`` retrieved an outcome,
+ so that ``PGconn.close()`` can turn a forgotten ``.result()`` into a visible test
failure instead of a silent pass.
``origin`` are exception notes saying what the query is and where the
@@ -147,9 +147,11 @@ class _MustConsumeFuture(Future):
def result(self, timeout: float | None = None) -> Any:
__tracebackhide__ = True # Don't show in pytest stack traces
- self.consumed = True
try:
- return super().result(timeout)
+ # exception() distinguishes a wait timeout from a TimeoutError
+ # raised by the query. Only retrieving an outcome consumes it.
+ self.exception(timeout)
+ return super().result(timeout=0)
except BaseException as e:
# Its traceback runs through concurrent.futures internals and the
# worker thread's query code, none of which tells anything about
@@ -158,8 +160,14 @@ class _MustConsumeFuture(Future):
raise e.with_traceback(None)
def exception(self, timeout: float | None = None) -> BaseException | None:
+ try:
+ exc = super().exception(timeout)
+ except CancelledError:
+ if self.cancelled():
+ self.consumed = True
+ raise
self.consumed = True
- return super().exception(timeout)
+ return exc
class PGconn(contextlib.AbstractContextManager):
diff --git a/src/test/pytest/meson.build b/src/test/pytest/meson.build
new file mode 100644
index 0000000000..5255c5a17f
--- /dev/null
+++ b/src/test/pytest/meson.build
@@ -0,0 +1,12 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+tests += {
+ 'name': 'pytest_framework',
+ 'sd': meson.current_source_dir(),
+ 'bd': meson.current_build_dir(),
+ 'pytest': {
+ 'tests': [
+ 'pyt/test_future.py',
+ ],
+ },
+}
diff --git a/src/test/pytest/pyt/test_future.py b/src/test/pytest/pyt/test_future.py
new file mode 100644
index 0000000000..62f23dcae1
--- /dev/null
+++ b/src/test/pytest/pyt/test_future.py
@@ -0,0 +1,127 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+"""Retrieving a background outcome, rather than attempting to wait, consumes it."""
+
+from concurrent.futures import CancelledError, Future, TimeoutError
+import signal
+
+import pytest
+
+from libpq import UnconsumedResultError
+from libpq._core import PGconn, _MustConsumeFuture
+
+
[email protected]("method", ["result", "exception"])
+def test_pending_timeout(method):
+ future = _MustConsumeFuture(["pending query"])
+ with pytest.raises(TimeoutError):
+ getattr(future, method)(timeout=0)
+ assert not future.consumed
+
+ # Finishing after a timeout must not silently discard a later query error.
+ future.set_exception(ValueError("query failed later"))
+ conn = object.__new__(PGconn)
+ conn._pending = future
+ with pytest.raises(UnconsumedResultError, match="never consumed"):
+ conn._check_pending()
+ assert isinstance(future.exception(), ValueError)
+ conn._check_pending()
+ assert conn._pending is None
+
+
[email protected]("method", ["result", "exception"])
+def test_completion_during_timeout_unwind(method, monkeypatch):
+ future = _MustConsumeFuture(["query completing after timeout"])
+
+ # Force completion after the stdlib wait times out, before our wrapper
+ # handles the timeout. Checking done() in a finally block loses the error.
+ def wrap(wait):
+ def timed_wait(self, timeout=None):
+ try:
+ return wait(self, timeout)
+ except TimeoutError:
+ self.set_exception(ValueError("query failed after wait"))
+ raise
+ return timed_wait
+
+ monkeypatch.setattr(Future, "result", wrap(Future.result))
+ monkeypatch.setattr(Future, "exception", wrap(Future.exception))
+ with pytest.raises(TimeoutError):
+ getattr(future, method)(timeout=0)
+ assert future.done()
+ assert not future.consumed
+
+
[email protected]("method", ["result", "exception"])
[email protected]("error", [None, ValueError, TimeoutError, CancelledError, KeyboardInterrupt])
+def test_completed_outcome(method, error):
+ future = _MustConsumeFuture(["finished query"])
+ if error is None:
+ future.set_result(42)
+ assert getattr(future, method)() == (42 if method == "result" else None)
+ else:
+ payload = error("query outcome")
+ future.set_exception(payload)
+ if method == "result":
+ with pytest.raises(error) as caught:
+ future.result()
+ assert caught.value is payload
+ else:
+ assert future.exception() is payload
+ assert future.consumed
+
+
[email protected]("method", ["result", "exception"])
+def test_cancelled_outcome(method):
+ future = _MustConsumeFuture(["cancelled query"])
+ assert future.cancel()
+ with pytest.raises(CancelledError):
+ getattr(future, method)()
+ assert future.consumed
+
+
[email protected](not hasattr(signal, "setitimer"), reason="requires interval timers")
[email protected]("method", ["result", "exception"])
+def test_interrupted_wait(method):
+ future = _MustConsumeFuture(["interrupted wait"])
+
+ def interrupt(signum, frame):
+ raise KeyboardInterrupt
+
+ old_handler = signal.signal(signal.SIGALRM, interrupt)
+ old_timer = signal.getitimer(signal.ITIMER_REAL)
+ try:
+ signal.setitimer(signal.ITIMER_REAL, 0.02)
+ with pytest.raises(KeyboardInterrupt):
+ getattr(future, method)(timeout=2)
+ finally:
+ signal.setitimer(signal.ITIMER_REAL, *old_timer)
+ signal.signal(signal.SIGALRM, old_handler)
+ assert not future.consumed
+
+
[email protected]("method", ["result", "exception"])
+def test_connection_stays_guarded_after_timeout(pg, method):
+ blocker = pg.connect()
+ waiter = pg.connect()
+ pid = waiter.sql("SELECT pg_backend_pid()")
+ # Limit cleanup time if the test itself fails to release its lock.
+ waiter.sql("SET statement_timeout = '30s'")
+ blocker.sql("SELECT pg_advisory_lock(12345)")
+ future = waiter.background_sql("SELECT pg_advisory_lock(12345)")
+ try:
+ pg.poll_query_until(
+ "SELECT wait_event = 'advisory' FROM pg_stat_activity WHERE pid = $1",
+ pid,
+ )
+ with pytest.raises(TimeoutError):
+ getattr(future, method)(timeout=0)
+ # Exercise the real connection guard without issuing a second libpq
+ # command if the guard is broken on the unfixed framework.
+ with pytest.raises(UnconsumedResultError, match="still busy"):
+ waiter._check_pending()
+ finally:
+ blocker.sql("SELECT pg_advisory_unlock(12345)")
+ future.result(timeout=30)
+ assert waiter.sql("SELECT 42") == 42
+ assert waiter.sql("SELECT pg_advisory_unlock(12345)")
--
2.50.1 (Apple Git-155)
From 467b0809935751c8864bcc6391d5aa1fcfc393d9 Mon Sep 17 00:00:00 2001
From: Nik Samokhvalov <[email protected]>
Date: Wed, 28 Oct 2026 21:17:55 -0700
Subject: [PATCH 2/2] Compare simple and extended query snapshot lifetimes
---
src/test/pytest/meson.build | 1 +
src/test/pytest/pyt/test_portal_snapshot.py | 70 +++++++++++++++++++++
2 files changed, 71 insertions(+)
create mode 100644 src/test/pytest/pyt/test_portal_snapshot.py
diff --git a/src/test/pytest/meson.build b/src/test/pytest/meson.build
index 5255c5a17f..3f038dd103 100644
--- a/src/test/pytest/meson.build
+++ b/src/test/pytest/meson.build
@@ -7,6 +7,7 @@ tests += {
'pytest': {
'tests': [
'pyt/test_future.py',
+ 'pyt/test_portal_snapshot.py',
],
},
}
diff --git a/src/test/pytest/pyt/test_portal_snapshot.py b/src/test/pytest/pyt/test_portal_snapshot.py
new file mode 100644
index 0000000000..c1f0cb5ebd
--- /dev/null
+++ b/src/test/pytest/pyt/test_portal_snapshot.py
@@ -0,0 +1,70 @@
+# Copyright (c) 2026, PostgreSQL Global Development Group
+
+"""Compare idle transaction snapshots in extended and simple query protocols."""
+
+import os
+from pathlib import Path
+import select
+import subprocess
+
+import pytest
+
+from pypg.paths import BINDIR
+
+
+def has_snapshot(pg, pid):
+ return pg.sql(
+ "SELECT backend_xmin IS NOT NULL FROM pg_stat_activity WHERE pid = $1",
+ pid,
+ )
+
+
+def test_extended_query_portal_snapshot(pg):
+ conn = pg.connect()
+ pid = conn.sql("SELECT pg_backend_pid()")
+ conn.sql_batch("BEGIN", "SELECT txid_current()")
+ try:
+ assert has_snapshot(pg, pid)
+ conn.close_portal("")
+ assert not has_snapshot(pg, pid)
+ assert pg.sql(
+ "SELECT backend_xid IS NOT NULL AND state = 'idle in transaction' "
+ "FROM pg_stat_activity WHERE pid = $1", pid,
+ )
+ finally:
+ conn.sql("ROLLBACK")
+
+
[email protected](os.name == "nt", reason="select() on a pipe requires POSIX")
+def test_psql_releases_statement_snapshot(pg):
+ proc = subprocess.Popen(
+ [str(Path(BINDIR) / "psql"), "-X", "-qAt", "-v", "ON_ERROR_STOP=1",
+ "-f", "-", "-d", pg.connstr()],
+ stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+ text=True,
+ )
+ try:
+ proc.stdin.write("BEGIN;\nSELECT pg_backend_pid(), txid_current();\n")
+ proc.stdin.flush()
+ ready, _, _ = select.select([proc.stdout], [], [], 10)
+ assert ready, "psql did not complete the statement"
+ pid, xid = map(int, proc.stdout.readline().strip().split("|"))
+ assert xid > 0
+ # Query output can arrive before ReadyForQuery. Wait until the server
+ # is idle in the transaction before inspecting its released snapshot.
+ pg.poll_query_until(
+ "SELECT state = 'idle in transaction' "
+ "FROM pg_stat_activity WHERE pid = $1", pid,
+ )
+ assert not has_snapshot(pg, pid)
+ assert pg.sql(
+ "SELECT backend_xid IS NOT NULL FROM pg_stat_activity WHERE pid = $1",
+ pid,
+ )
+ finally:
+ try:
+ proc.communicate("ROLLBACK;\n\\q\n", timeout=10)
+ except subprocess.TimeoutExpired:
+ proc.kill()
+ proc.communicate()
+ assert proc.returncode == 0
--
2.50.1 (Apple Git-155)
