Transaction can't recover a session whose connection was lost outside the block #74

Transaction cannot recover a session whose connection was lost outside the block. A retried function that runs a query before or after async with transaction: fails for good if the connection drops during that query: the retry re-enters Transaction, and the first statement raises PendingRollbackError: Can't reconnect until invalid transaction is rolled back.

This is why the README requires the retried coroutine to own its whole transaction (#70). Downstream projects follow that rule by hand. modern-python/chat-app#54 had to move reads into the block in three use cases after its connection-loss tests caught them. With pool_pre_ping off, as in chat-app and both templates now, the first statement after an idle connection dies is exposed to this.

Reproduction

SQLAlchemy 2.1.4, asyncpg 0.32.0, Postgres 18 (db-retry's test container). chat-app hit the same failure on SQLAlchemy 2.0.54 in modern-python/chat-app#54. One idle pooled connection whose backend was terminated, then:

@postgres_retry(retries=3)
async def use_case() -> int | None:
    nonlocal attempts
    attempts += 1
    if attempts == 1:
        await session.scalar(sqlalchemy.text("SELECT 1"))  # outside the block, hits the dead connection
    async with transaction:
        return await session.scalar(sqlalchemy.text("SELECT 1"))

Result: PendingRollbackError after 2 attempts.

Why __aenter__ doesn't notice

After the disconnect, __aenter__ sees session.in_transaction() is True, skips begin() and joins the dead transaction. The session-level flags give no hint:

Session.is_active is False only in the "partial rollback" state after a failed flush, so it can't be used here. The Core Connection.invalidated flag is public, and session.connection() returns the existing connection without sending SQL when the session is already in a transaction.

Proposal

In Transaction.__aenter__, roll back first when the session's transaction sits on an invalidated connection:

if self.session.in_transaction() and (await self.session.connection()).invalidated:
    await self.session.rollback()

This only throws away a transaction Postgres has already discarded, so no caller work is lost. A live outer transaction is still joined as today. I prototyped it as a subclass: the reproduction above succeeds on attempt 2, and an outer transaction with an uncommitted insert still sees its row inside the block.

Open points:

• The isolation-level branch in __aenter__ calls session.connection(execution_options=...) first. The check should run before it, so the new connection gets the isolation level.
• Docs: the README precondition can relax from "the retried coroutine must own its whole transaction" to "every retried attempt must enter Transaction". A connection lost on a query outside the block would no longer break the retry. ADR 0002 or a new ADR should record it.
• Tests: the reproduction above and the live-outer-transaction case, in tests/test_transaction.py.