Handle additional InterfaceError types when inside a transaction

[mbrancato]

• asyncpg version: 0.26.0
• PostgreSQL version: PostgreSQL 14.5 (Debian 14.5-1.pgdg110+1) on x86_64-pc-linux-gnu, compiled by gcc (Debian 10.2.1-6) 10.2.1 20210110, 64-bit - Also fails on older versions such as PG 13
• Do you use a PostgreSQL SaaS? If so, which? Can you reproduce
  the issue with a local PostgreSQL install?: Google CloudSQL, but I'm reproducing this locally in a test.
• Python version: 3.10
• Platform: MacOS
• Do you use pgbouncer?: No
• Did you install asyncpg with pip?: yes
• If you built asyncpg locally, which version of Cython did you use?: n/a
• Can the issue be reproduced under both asyncio and
  uvloop?: yes

When an InterfaceError is raised during a transaction, asyncpg does not explicitly handle all variants of this exception, and raises it.

asyncpg/asyncpg/transaction.py

Lines 65 to 79 in 5f908e6

	try:
	self._check_conn_validity('__aexit__')
	except apg_errors.InterfaceError:
	if extype is GeneratorExit:
	# When a PoolAcquireContext is being exited, and there
	# is an open transaction in an async generator that has
	# not been iterated fully, there is a possibility that
	# Pool.release() would race with this __aexit__(), since
	# both would be in concurrent tasks. In such case we
	# yield to Pool.release() to do the ROLLBACK for us.
	# See https://github.com/MagicStack/asyncpg/issues/232
	# for an example.
	return
	else:
	raise

For example, it does not handle the type asyncpg.exceptions.ConnectionDoesNotExistError - and as such, this leads to leftover connections while trying to close the pool. Simple reproduction that forces the error by setting idle_in_transaction_session_timeout to be very small - basically telling the server to close a connection before it can complete a query:

pool = await asyncpg.create_pool(
    dsn=dsn,
    min_size=1,
    max_size=4,
    server_settings={
        "idle_in_transaction_session_timeout": "1",
    },
)


async def query_gen():
    async with pool.acquire(timeout=5) as con:
        async with con.transaction(readonly=True):
            await con.fetch(
                "SELECT * FROM my_db.my_schema.my_table;", timeout=15
            )


tasks = []
for i in range(40):
    tasks.append(query_gen())
results = await asyncio.gather(*tasks, return_exceptions=True)

print(results, flush=True)

await pool.close()

The results List is all exceptions:

[InterfaceError('cannot call Transaction.__aexit__(): the underlying connection is closed'), InterfaceError('cannot call Transaction.__aexit__(): the underlying connection is closed'), InterfaceError('cannot call Transaction.__aexit__(): the underlying connection is closed'), InterfaceError('cannot call Transaction.__aexit__(): the underlying connection is closed'), ...]

The exception type is asyncpg.exceptions.ConnectionDoesNotExistError. And the pool.close() will print a warning:

asyncpg.pool:Pool.close() is taking over 60 seconds to complete. Check if you have any unreleased connections left. Use asyncio.wait_for() to set a timeout for Pool.close().

When this happens, I do believe the connection pool should set the _in_use attribute for the connection holder to False and / or delete the pool connection holder.

This does not happen when not using the async with con.transaction() context manager.

Note: Under normal conditions / default settings, I see this rarely, however, I do see the pool.close() warning on occasion.