How can I know when Postgre is returning a idle_in_transaction_session_timeout?

[dvarrazzo]

Originally posted by @b0uh in #987

Hello,

I'm migrating some code from psycopg2 to psycopg3 and it is going well so far, thanks for the hard work.
I have some difficulties around the behavior of idle_in_transaction_session_timeout setting though.

With psycopg3, the error/exception is generic:

import psycopg
from time import sleep

with psycopg.connect("...") as conn:
    with conn.cursor() as cursor:
        cursor.execute("SET SESSION idle_in_transaction_session_timeout = 1000")
        sleep(2)
        cursor.execute("SELECT * from pg_tables")

# Exception raised:
# OperationalError: consuming input failed: server closed the connection unexpectedly
# 	This probably means the server terminated abnormally
# 	before or while processing the request.

With psycopg2, the error/exception is more explicit:

import psycopg2
conn = psycopg2.connect("....")
with conn.cursor() as cursor:
    cursor.execute("SET SESSION idle_in_transaction_session_timeout = 1000")
    sleep(2)
    cursor.execute("SELECT * from pg_tables")
conn.close()

# Exception raised:
# psycopg2.errors.IdleInTransactionSessionTimeout: terminating connection due to idle-in-transaction timeout
# server closed the connection unexpectedly
# 	This probably means the server terminated abnormally
# 	before or while processing the request.

While looking at the code of psycopg3 I noticed the IdleInTransactionSessionTimeout exception which makes me wonder if maybe I missed something.

Does someone has been able to have clear error for idle_in_transaction_session_timeout?

Thank you!

[dvarrazzo]

It seems a bug indeed. The database sends the right code, which can be seen by adding

        conn.pgconn.trace(2)
        conn.pgconn.set_trace_flags(
            psycopg.pq.Trace.SUPPRESS_TIMESTAMPS | psycopg.pq.Trace.REGRESS_MODE)

to the repro above. We can see:

F	10	Query	 "BEGIN"
B	10	CommandComplete	 "BEGIN"
B	5	ReadyForQuery	 T
F	58	Query	 "SET SESSION idle_in_transaction_session_timeout = 100"
B	8	CommandComplete	 "SET"
B	5	ReadyForQuery	 T
F	28	Query	 "SELECT * from pg_tables"
B	NN	ErrorResponse	 S "FATAL" V "FATAL" C "25P03" M "terminating connection due to idle-in-transaction timeout" F "SSSS" L "SSSS" R "SSSS" \x00

The code 25P03 is correctly reported. However we hit the error result and raise the exception in:

psycopg/psycopg/psycopg/pq/pq_ctypes.py

Lines 596 to 600 in 5cb5635

	def consume_input(self) -> None:
	if 1 != impl.PQconsumeInput(self._pgconn_ptr):
	raise e.OperationalError(
	f"consuming input failed: {self.get_error_message()}"
	)

So we raise the exception just getting a message from PQerrorMessage, without handling any result, therefore we don't have the sqlstate code available.

A quick test to try to obtain a result, using a pgconn.consume_input() wrapper:

def _consume_or_raise(pgconn: PGconn) -> None:
    # DRAFT: rought cut, should probably consult PQisBusy too
    try:
        pgconn.consume_input()
    except e.OperationalError:
        if res := pgconn.get_result():
            raise e.error_from_result(res)
        else:
            raise

doesn't get a good result: the error returned looks like:

psycopg.DatabaseError: server closed the connection unexpectedly
	This probably means the server terminated abnormally
	before or while processing the request.
invalid socket

which seems to signify that the get_result() request attempted to return a new result, but by now the connection has already been closed by the client too, hence the "invalid socket".

I wonder what path does psycopg2 follow, and what is the correct way to use the libpq in this case. Will ask the hackers.

[dvarrazzo]

Actually, @b0uh, I can't reproduce your result with psycopg2, I get an OperationalError too. Are you sure about your psycopg2 example? Which version are you using?

Adding PSYCOPG_DEBUG and enabling trace mode (which requires some ctypes trickery in psycopg2), I can see:

[179755] pq_execute: executing SYNC query: pgconn = 0x60df25d95af0
[179755]     SELECT * from pg_tables
2025-01-07 23:45:42.250610	F	28	Query	 "SELECT * from pg_tables"
2025-01-07 23:45:42.250762	B	122	ErrorResponse	 S "FATAL" V "FATAL" C "25P03" M "terminating connection due to idle-in-transaction timeout" F "postgres.c" L "3398" R "ProcessInterrupts" \x00
[179755] pq_execute: entering synchronous DBAPI compatibility mode
[179755] pq_fetch: pgstatus = PGRES_FATAL_ERROR
[179755] pq_fetch: uh-oh, something FAILED: status = 7 pgconn = 0x76147212ad40
[179755] pq_raise: PQresultErrorMessage: err=server closed the connection unexpectedly
	This probably means the server terminated abnormally
	before or while processing the request.

[179755] pq_raise: err2=server closed the connection unexpectedly
	This probably means the server terminated abnormally
	before or while processing the request.

[179755] curs_execute: res = -1, pgres = (nil)
[179755] curs_close: cursor at 0x7614721056c0 closed
Traceback (most recent call last):
  File "/home/piro/dev/psycopg2/test_close_conn.py", line 32, in <module>
    cursor.execute("SELECT * from pg_tables")
psycopg2.OperationalError: server closed the connection unexpectedly
	This probably means the server terminated abnormally
	before or while processing the request.

[179755] conn_close: PQfinish called

which is doubly-confusing because it seems that the error comes from PQresultErrorMessage, so 1) how did psycopg2 get the error and 2) why I can't see the right exception?

Regardless of the problem 2, probably the difference in 1 is that psycopg2 uses the sync path. By forcing the async path adding psycopg2.extensions.set_wait_callback(psycopg2.extras.wait_select) I can see a path more similar to psycopg 3:

[194685] conn_poll: async_status = ASYNC_READ
[194685] conn_poll: poll reading
[194685] pq_get_result_async: calling PQconsumeInput()
2025-01-07 23:58:12.684522	B	122	ErrorResponse	 S "FATAL" V "FATAL" C "25P03" M "terminating connection due to idle-in-transaction timeout" F "postgres.c" L "3398" R "ProcessInterrupts" \x00
[194685] pq_get_result_async: calling PQisBusy()
[194685] pq_get_result_async: got result PGRES_FATAL_ERROR
[194685] pq_get_result_async: calling PQisBusy()
[194685] pq_get_result_async: PQisBusy() = 1
[194685] conn_poll: returning 1
[194685] conn_poll: status = 2
[194685] conn_poll: status -> CONN_STATUS_*
[194685] conn_poll: async_status = ASYNC_READ
[194685] conn_poll: poll reading
[194685] pq_get_result_async: calling PQconsumeInput()
[194685] pq_get_result_async: PQconsumeInput() failed
[194685] conn_poll: returning 3

which, again, it's confusing because looking at pq_get_result_async()

https://github.com/psycopg/psycopg2/blob/5509e011080b7316daea0ca79ac1d8faff163e6d/psycopg/pqpath.c#L710-L712

it seems that PQconsumeInput() failed should have been printed right after the ErrorResponse trace.

The plot thickens...

[dvarrazzo]

I think I'm making sense of the problem in psycopg 3. Adding the trace of all the libpq calls on top of the communication, using:

This script

import os
os.environ["PSYCOPG_IMPL"] = "python"

import logging
from time import sleep

import psycopg
from psycopg import pq
from psycopg.pq._debug import PGconnDebug

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("psycopg.debug")
logger.setLevel(logging.INFO)

assert pq.__impl__ == "python"
pq.PGconn = PGconnDebug

with psycopg.connect("") as conn:

    conn.pgconn.trace(2)
    conn.pgconn.set_trace_flags(
        psycopg.pq.Trace.SUPPRESS_TIMESTAMPS | psycopg.pq.Trace.REGRESS_MODE)

    with conn.cursor() as cursor:

        cursor.execute("SET SESSION idle_in_transaction_session_timeout = 100")
        sleep(0.2)
        cursor.execute("SELECT * from pg_tables")

The relevant result is:

PGconn.send_query(b'SELECT * from pg_tables')
F	28	Query	 "SELECT * from pg_tables"
    <- None
PGconn.flush()
    <- 0
PGconn.is_busy()
    <- 1
PGconn.consume_input()
    <- None
PGconn.is_busy()
B	NN	ErrorResponse	 S "FATAL" V "FATAL" C "25P03" M "terminating connection due to idle-in-transaction timeout" F "SSSS" L "SSSS" R "SSSS" \x00
    <- 0
PGconn.notifies()
    <- None
PGconn.get_result()
    <- <psycopg.pq.pq_ctypes.PGresult [FATAL_ERROR] at 0x7c902e39e800>
PGconn.is_busy()
    <- 1
PGconn.consume_input()
PGconn.status -> 1
Traceback (most recent call last):
  File "/home/piro/dev/psycopg3/test_close_conn.py", line 33, in <module>
...
psycopg.OperationalError: consuming input failed: terminating connection due to idle-in-transaction timeout
server closed the connection unexpectedly
	This probably means the server terminated abnormally
	before or while processing the request.

So it seems that we indeed receive a result, but then we barf on a following PQconsumeInput(). This points to the _fetch_many() implementation:

psycopg/psycopg/psycopg/generators.py

Lines 178 to 196 in 5cb5635

	while True:
	res = yield from _fetch(pgconn)
	if not res:
	break
	results.append(res)
	status = res.status
	if status == COPY_IN or status == COPY_OUT or status == COPY_BOTH:
	# After entering copy mode the libpq will create a phony result
	# for every request so let's break the endless loop.
	break
	if status == PIPELINE_SYNC:
	# PIPELINE_SYNC is not followed by a NULL, but we return it alone
	# similarly to other result sets.
	assert len(results) == 1, results
	break
	return results

Here we fetch the error, but we dutifully follow the libpq instructions saying

PQgetResult must be called repeatedly until it returns a null pointer, indicating that the command is done.

But, as we try to continue the loop after receiving the error result, we end up calling PQconsumeInput with a closed connection in _fetch(), which throws an immediate exception and discards any error already received.

[dvarrazzo]

This patch:

$ git diff
diff --git a/psycopg/psycopg/generators.py b/psycopg/psycopg/generators.py
index 8eee7ccc..9fc2ced7 100644
--- a/psycopg/psycopg/generators.py
+++ b/psycopg/psycopg/generators.py
@@ -176,7 +176,19 @@ def _fetch_many(pgconn: PGconn) -> PQGen[list[PGresult]]:
     """
     results: list[PGresult] = []
     while True:
-        res = yield from _fetch(pgconn)
+        try:
+            res = yield from _fetch(pgconn)
+        except e.DatabaseError:
+            # What might have happened here is that a previuos error
+            # disconnected the connection, for example a idle in transaction
+            # timeout. Check if we had received an error before, and raise it
+            # as exception, because it should contain more details. See #988.
+            for res in results:
+                if res.status == pq.ExecStatus.FATAL_ERROR:
+                    raise e.error_from_result(res, encoding=pgconn._encoding)
+            else:
+                raise
+
         if not res:
             break

seems to solve the problem. Will test it more thoroughly and prepare a MR.

[b0uh]

I'm not sure to follow everything 😅, but a big thank you for how quick your are handling this!