Skip to content

Fix stale _top_xact after a failed Transaction.start() #1399

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
chenty2333 wants to merge 1 commit into
base: master
Choose a base branch
from
+336 −0
Open

Fix stale _top_xact after a failed Transaction.start() #1399

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 6 additions & 0 deletions asyncpg/transaction.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
330 changes: 330 additions & 0 deletions tests/test_transaction.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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(
Expand Down