From ff1bb8d6a5d1aced66e92117a0457898b5743049 Mon Sep 17 00:00:00 2001 From: Tianyi Chen Date: Thu, 22 Oct 2026 08:27:02 +0900 Subject: [PATCH] Fix stale _top_xact after a failed Transaction.start() Transaction.start() records itself as Connection._top_xact before sending BEGIN. If the command fails (for example, the connection is busy, or the task is cancelled after the server has started the transaction), the transaction was left registered as the top-level transaction even though its state became FAILED. Subsequent transactions were then treated as nested and sent SAVEPOINT; when BEGIN had reached the server, the savepoint was released but the abandoned transaction was never committed, silently losing writes. Clear Connection._top_xact when the transaction that failed to start is still the current one, following the guard suggested in #1386. The identity check keeps the outer transaction in place when a nested SAVEPOINT fails. Add regression tests for pre-send failures, cancelled BEGIN, nested savepoint failures, and pool release/reset. --- asyncpg/transaction.py | 6 + tests/test_transaction.py | 330 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 336 insertions(+) diff --git a/asyncpg/transaction.py b/asyncpg/transaction.py index 562811e6..290d04d8 100644 --- a/asyncpg/transaction.py +++ b/asyncpg/transaction.py @@ -146,6 +146,12 @@ async def start(self): await self._connection.execute(query) except BaseException: self._state = TransactionState.FAILED + # A top-level transaction that failed to start must not remain + # current, or the next transaction would be treated as a nested + # savepoint. Leave the outer transaction in place when a + # nested block fails. + if con._top_xact is self: + con._top_xact = None raise else: self._state = TransactionState.STARTED diff --git a/tests/test_transaction.py b/tests/test_transaction.py index f84cf7c0..3ba1156c 100644 --- a/tests/test_transaction.py +++ b/tests/test_transaction.py @@ -5,6 +5,8 @@ # the Apache 2.0 License: http://www.apache.org/licenses/LICENSE-2.0 +import asyncio + import asyncpg from asyncpg import _testbase as tb @@ -180,6 +182,334 @@ async def test_transaction_within_manual_transaction(self): self.assertIsNone(self.con._top_xact) self.assertFalse(self.con.is_in_transaction()) + async def test_start_failure_does_not_poison_connection(self): + other = await self.connect() + slow = None + try: + await self.con.execute(''' + DROP TABLE IF EXISTS test_start_failure; + CREATE TABLE test_start_failure (a int); + ''') + + slow = self.loop.create_task( + self.con.fetchval('SELECT pg_sleep(1)')) + await asyncio.sleep(0.1) + + with self.assertRaisesRegex( + asyncpg.InterfaceError, + 'cannot perform operation: another operation is in ' + 'progress'): + await self.con.transaction().start() + + # BEGIN never reached the server, so the failed transaction + # must not stay registered as the top-level transaction. + self.assertIsNone(self.con._top_xact) + + await slow + self.assertFalse(self.con.is_in_transaction()) + + async with self.con.transaction(): + await self.con.execute( + 'INSERT INTO test_start_failure VALUES (1)') + + self.assertIsNone(self.con._top_xact) + self.assertFalse(self.con.is_in_transaction()) + self.assertEqual( + await other.fetch('SELECT a FROM test_start_failure'), + [(1,)]) + finally: + if slow is not None and not slow.done(): + slow.cancel() + try: + await slow + except asyncio.CancelledError: + pass + if (self.con.is_in_transaction() or + self.con._top_xact is not None): + self.con.terminate() + try: + await other.execute( + 'DROP TABLE IF EXISTS test_start_failure') + finally: + await other.close() + + async def test_cancelled_start_does_not_poison_connection(self): + other = await self.connect() + try: + await self.con.execute(''' + DROP TABLE IF EXISTS test_cancelled_start; + CREATE TABLE test_cancelled_start (a int); + ''') + + tr = self.con.transaction() + + async def start_transaction(): + async with tr: + await self.con.execute( + 'INSERT INTO test_cancelled_start VALUES (1)') + + task = self.loop.create_task(start_transaction()) + self.loop.call_soon(task.cancel) + + with self.assertRaises(asyncio.CancelledError): + await task + + tr = self.con.transaction() + try: + await tr.start() + except asyncpg.InterfaceError as ex: + # The server did execute BEGIN and is still inside the + # abandoned transaction, so reuse is explicitly refused. + self.assertEqual( + str(ex), + 'cannot use Connection.transaction() in a manually ' + 'started transaction') + self.assertTrue(self.con.is_in_transaction()) + + # Resetting rolls back the abandoned transaction and is + # expected to report it. + with self.assertLoopErrorHandlerCalled( + 'Resetting connection with an active transaction'): + await self.con.reset() + + self.assertFalse(self.con.is_in_transaction()) + self.assertIsNone(self.con._top_xact) + + tr = self.con.transaction() + await tr.start() + + await self.con.execute( + 'INSERT INTO test_cancelled_start VALUES (1)') + await tr.commit() + + # Either the transaction above joined the abandoned server + # transaction and committed it, or it was refused and started + # after a reset. In both cases the write must be visible. + self.assertEqual( + await other.fetch('SELECT a FROM test_cancelled_start'), + [(1,)]) + self.assertIsNone(self.con._top_xact) + self.assertFalse(self.con.is_in_transaction()) + finally: + if (self.con.is_in_transaction() or + self.con._top_xact is not None): + self.con.terminate() + try: + await other.execute( + 'DROP TABLE IF EXISTS test_cancelled_start') + finally: + await other.close() + + async def test_cancelled_start_refuses_reuse_until_reset(self): + other = await self.connect() + try: + await self.con.execute(''' + DROP TABLE IF EXISTS test_cancelled_refused; + CREATE TABLE test_cancelled_refused (a int); + ''') + + tr = self.con.transaction() + + async def start_transaction(): + async with tr: + pass + + task = self.loop.create_task(start_transaction()) + self.loop.call_soon(task.cancel) + + with self.assertRaises(asyncio.CancelledError): + await task + + # Wait for the cancellation to be acknowledged. The server + # has executed BEGIN, so the connection is inside a + # transaction that has no Transaction object anymore. + await self.con.execute('SELECT 1') + + self.assertIsNone(self.con._top_xact) + self.assertTrue(self.con.is_in_transaction()) + + with self.assertRaisesRegex( + asyncpg.InterfaceError, + r'cannot use Connection.transaction\(\) in a manually ' + 'started transaction'): + await self.con.transaction().start() + + self.assertIsNone(self.con._top_xact) + self.assertTrue(self.con.is_in_transaction()) + + with self.assertLoopErrorHandlerCalled( + 'Resetting connection with an active transaction'): + await self.con.reset() + + self.assertFalse(self.con.is_in_transaction()) + self.assertIsNone(self.con._top_xact) + + async with self.con.transaction(): + await self.con.execute( + 'INSERT INTO test_cancelled_refused VALUES (1)') + + self.assertEqual( + await other.fetch('SELECT a FROM test_cancelled_refused'), + [(1,)]) + finally: + if (self.con.is_in_transaction() or + self.con._top_xact is not None): + self.con.terminate() + try: + await other.execute( + 'DROP TABLE IF EXISTS test_cancelled_refused') + finally: + await other.close() + + async def test_nested_start_failure_keeps_outer_transaction(self): + other = await self.connect() + try: + await self.con.execute(''' + DROP TABLE IF EXISTS test_nested_start; + CREATE TABLE test_nested_start (a int); + ''') + + outer = self.con.transaction(isolation='serializable') + await outer.start() + await self.con.execute( + 'INSERT INTO test_nested_start VALUES (1)') + + with self.assertRaisesRegex( + asyncpg.InterfaceError, + 'nested transaction has a different isolation level'): + await self.con.transaction( + isolation='read_committed').start() + + self.assertIs(self.con._top_xact, outer) + self.assertTrue(self.con.is_in_transaction()) + + await self.con.execute( + 'INSERT INTO test_nested_start VALUES (2)') + await outer.commit() + + self.assertIsNone(self.con._top_xact) + self.assertFalse(self.con.is_in_transaction()) + + async with self.con.transaction(): + await self.con.execute( + 'INSERT INTO test_nested_start VALUES (3)') + + self.assertEqual( + await other.fetch( + 'SELECT a FROM test_nested_start ORDER BY a'), + [(1,), (2,), (3,)]) + finally: + if (self.con.is_in_transaction() or + self.con._top_xact is not None): + self.con.terminate() + try: + await other.execute( + 'DROP TABLE IF EXISTS test_nested_start') + finally: + await other.close() + + async def test_nested_savepoint_failure_keeps_outer_transaction(self): + other = await self.connect() + try: + await self.con.execute(''' + DROP TABLE IF EXISTS test_nested_savepoint; + CREATE TABLE test_nested_savepoint (a int); + ''') + + outer = self.con.transaction() + await outer.start() + await self.con.execute( + 'INSERT INTO test_nested_savepoint VALUES (1)') + + with self.assertRaises(asyncpg.DivisionByZeroError): + await self.con.execute('SELECT 1 / 0') + + with self.assertRaises(asyncpg.InFailedSQLTransactionError): + async with self.con.transaction(): + await self.con.execute( + 'INSERT INTO test_nested_savepoint VALUES (2)') + + self.assertIs(self.con._top_xact, outer) + self.assertTrue(self.con.is_in_transaction()) + + await outer.rollback() + + self.assertIsNone(self.con._top_xact) + self.assertFalse(self.con.is_in_transaction()) + + async with self.con.transaction(): + await self.con.execute( + 'INSERT INTO test_nested_savepoint VALUES (3)') + + self.assertEqual( + await other.fetch( + 'SELECT a FROM test_nested_savepoint ORDER BY a'), + [(3,)]) + finally: + if (self.con.is_in_transaction() or + self.con._top_xact is not None): + self.con.terminate() + try: + await other.execute( + 'DROP TABLE IF EXISTS test_nested_savepoint') + finally: + await other.close() + + async def test_cancelled_start_pool_release(self): + other = await self.connect() + try: + await self.con.execute(''' + DROP TABLE IF EXISTS test_pool_cancelled_start; + CREATE TABLE test_pool_cancelled_start (a int); + ''') + + async with self.create_pool( + min_size=1, max_size=1) as pool: + with self.assertLoopErrorHandlerCalled( + 'Resetting connection with an active transaction'): + async with pool.acquire() as con: + tr = con.transaction() + + async def start_transaction(): + async with tr: + await con.execute( + 'INSERT INTO ' + 'test_pool_cancelled_start VALUES (1)') + + task = self.loop.create_task(start_transaction()) + self.loop.call_soon(task.cancel) + + with self.assertRaises(asyncio.CancelledError): + await task + + # Wait for the cancellation to be acknowledged. + # The server has executed BEGIN by now and the + # connection is inside a transaction it does not + # own anymore. + await con.execute('SELECT 1') + + self.assertIsNone(con._con._top_xact) + self.assertTrue(con.is_in_transaction()) + + async with pool.acquire() as con: + self.assertIsNone(con._con._top_xact) + self.assertFalse(con.is_in_transaction()) + async with con.transaction(): + await con.execute( + 'INSERT INTO test_pool_cancelled_start ' + 'VALUES (2)') + + self.assertEqual( + await other.fetch( + 'SELECT a FROM test_pool_cancelled_start'), + [(2,)]) + finally: + try: + await other.execute( + 'DROP TABLE IF EXISTS test_pool_cancelled_start') + finally: + await other.close() + async def test_isolation_level(self): await self.con.reset() default_isolation = await self.con.fetchval(