Reset Connection._top_xact when BEGIN fails in Transaction.start()

[devtechedge]

Transaction.start() sets con._top_xact = self before it sends BEGIN.

If BEGIN raises, the transaction is marked FAILED, but _top_xact stays pointed at it.

Only __commit() and __rollback() ever clear it, and both refuse to run on a failed transaction, so nothing releases it after that.

Every later conn.transaction() on the connection is then treated as nested and sent as SAVEPOINT, which the server rejects with NoActiveSQLTransactionError until the connection is reset or closed.

This change clears _top_xact in the except branch of start() when it still points at the failed transaction, the same way __commit() and __rollback() do.

The is self check leaves the outer transaction alone when a nested SAVEPOINT fails.

I added test_transaction_failed_begin to tests/test_transaction.py. It patches Connection.execute so BEGIN raises PostgresConnectionError, then checks that the same error comes back, that execute was awaited once with BEGIN;, that _top_xact is None, and that the next conn.transaction() opens a real top-level transaction.

The test fails if that clear is removed, and passes with the fix. I re-ran tests.test_transaction on Python 3.12 against Postgres 16, and flake8 on the touched files was clean.

Fixes #1386

[elprans]

Can we do something like this to avoid relying on flaky loop timing?

Suggested change

	# Make BEGIN fail by running it while another
	# operation is in progress on the connection.
	busy = asyncio.ensure_future(self.con.execute('SELECT pg_sleep(0.5)'))
	await asyncio.sleep(0.1)
	with self.assertRaisesRegex(asyncpg.InterfaceError,
	'another operation is in progress'):
	await self.con.transaction().start()
	await busy
	tr = self.con.transaction()
	error = asyncpg.PostgresConnectionError(
	'Timed-out waiting to acquire database connection.'
	)
	with mock.patch.object(
	type(self.con), 'execute', side_effect=error
	) as execute:
	with self.assertRaises(asyncpg.PostgresConnectionError) as caught:
	await tr.start()
	self.assertIs(caught.exception, error)
	execute.assert_awaited_once_with('BEGIN;')

[devtechedge]

Thanks for catching that, Elvis.

The sleep was a bad way to force BEGIN to fail, and a slow loop could miss the window.

I switched the test to patch execute with a PostgresConnectionError, the way you suggested.

It still checks that _top_xact is cleared and that the next transaction() is a real top-level one, because those are the checks that fail without the fix.

The mock only wraps start(), so that follow-up runs on the live connection.

Happy to trim it if you'd rather the test stop where your suggestion did.