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)

Reply via email to