InterfaceError On Exception Inside PoolAcquireContext But Outside Transaction

[pauldw]

• asyncpg version: 0.13.0
• PostgreSQL version: 9.6.3
• Do you use a PostgreSQL SaaS? If so, which? Can you reproduce
  the issue with a local PostgreSQL install?: Using local install.
• Python version: 3.6.2
• Platform: Client on macOS 10.12.6, PostgreSQL inside linux x86_64 docker container.
• Do you use pgbouncer?: No.
• Did you install asyncpg with pip?: Yes.
• If you built asyncpg locally, which version of Cython did you use?: Not built locally.
• Can the issue be reproduced under both asyncio and
  uvloop?: Unsure.

Unexpected Behaviour Example

If an exception occurs outside of a transaction block but within a pool acquire block, an InterfaceError is raised.

import asyncpg
import asyncio

loop = asyncio.get_event_loop()

async def iterate(con):
    async with con.transaction():
        async for record in con.cursor("SELECT 1"):
            yield record

async def run():
    pool = await asyncpg.create_pool(user='postgres')
    async with pool.acquire() as con:
        async for _ in iterate(con):
            raise Exception()

loop.run_until_complete(run())

This will cause asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress which I did not expect.

Expected Behaviour Example

If the transaction block is moved to encompass the exception, this behaves as expected.

import asyncpg
import asyncio

loop = asyncio.get_event_loop()

async def iterate(con):
    async for record in con.cursor("SELECT 1"):
        yield record

async def run():
    pool = await asyncpg.create_pool(user='postgres')
    async with pool.acquire() as con:
        async with con.transaction():
            async for _ in iterate(con):
                raise Exception()

loop.run_until_complete(run())

This will cause Exception as I expected.

[elprans]

Interesting case. Here's what I think happens when Exception is raised in the first case:

               run()                                      iterate()
-----------------------------------------------------------------------------------
raise Exception()
Loop.create_task(agen.aclose())
PoolAcquireContext.__aexit__()
Loop.create_task(Pool._release_impl())
                                             # agen.aclose() task
                                             Transaction.__aexit__()
                                             Transaction.rollback()
                                             await Connection.execute('ROLLBACK;')

# Pool._release_impl() task
_release_impl() -> Connection.reset()
await Connection.execute('RESET...')
# Two simultaneously running queries!
raise InterfaceError()

Basically, Pool.release() and Transaction.rollback() are executed concurrently, and they really shouldn't.

The second case, in contrast, has no agen.close() task, so things look better, but there are still issues there.

The proper fix here would be to serialize transaction exit with pool release somehow.