connections not returning to connection pool on CancelledError

[samuelcolvin]

• asyncpg version: 0.9.0
• PostgreSQL version: 9.6
• Python version: 3.6
• Platform: Heroku, ubuntu locally
• Do you use pgbouncer?: no
• Did you install asyncpg with pip?: yes
• If you built asyncpg locally, which version of Cython did you use?: -
• Can the issue be reproduced under both asyncio and
  uvloop?: yes

In a websocket handler with aiohttp (2.0.4) I have (simplified for brevity):

    ws = WebSocketResponse()
    await ws.prepare(request)
    async with request.app['pg'].acquire(timeout=5) as conn:
        await conn.add_listener('events', send_event_)
        ...  # do stuff, including executing queries
        async for msg in ws:
            ...  # do stuff with msg, including executing queries
    return ws

This is causing problems; if a browser is closed without calling websocket.close() in js, the handler is terminated with a CancelledError, __aexit__ also gets cancelled and the connection is never returned to the pool.

If I use

    ws = WebSocketResponse()
    await ws.prepare(request)
    conn = await request.app['pg']._acquire(timeout=5)
    try:
        await conn.add_listener('events', send_event_)
        ...  # do stuff, including executing queries
        async for msg in ws:
            ...  # do stuff with msg, including executing queries
    finally:
        async def release(app, conn):
            await app['pg'].release(conn)
        request.app.loop.create_task(release(request.app, conn))

The connection is correctly released and the pool doesn't run out of connections.

(actual code is here if that helps.)

I'm not sure whether asyncpg can resolve this or whether it's a problem with aiohttp?

Would it be possible for the pool manager to somehow recover those connections which weren't released?

Or could there be an alternative acquire used with with await pool.acquire() as conn: so __exit__ was not a coroutine?

[elprans]

This is not specific to asyncpg. aiohttp cancels the entire task, including the PoolAcquiteContext.__aexit__ call which is supposed to return the connection to the pool.

@1st1 : ideas on the best way of handling this? One way is to create_task in __aexit__. Overall, I think that task.cancel() is somewhat broken with respect to finalization, as it essentially breaks all async code in finally and __aexit__.

[samuelcolvin]

I understand this isn't specific to asyncpg.

It's possible to make task cancellation unlikely but not impossible. (eg. this app could call websocket.close() in onbeforeunload, but you couldn't be certain it would always be called.) So you could be confident that zombie connections would occur rarely but not never.

Therefore would it be possible to have a cleanup method on the pool which released any connections which are not referenced? cleanup could be called periodically or if acquiring a connection timed out. I know this seems ugly but otherwise I don't see you can be certain that connections won't be lost from the pool over a long period.

[elprans]

Well, the app shouldn't be using a pool if it plans to hold onto a connection for a long time. In your case there's no reason why a db connection should be open for the entirety of a websocket connection. You're essentially limiting the number of concurrent users with the size of the pool. acquire on as-needed basis, i.e. inside the while True loop.

[samuelcolvin]

Ye, I realise that but the long-lived connection was required for the listener. I've now changed it to use redis pub-sub so the pg connections are only acquired when required.

[elprans]

You only really need one listener in a background task and a list of currently open WS connections.

[samuelcolvin]

I realised that exact thing just after hitting comment. 😄

[elprans]

Oh, and acquire() can be used directly, not just as a context manager. Don't use _acquire().

[1st1]

@1st1 : ideas on the best way of handling this? One way is to create_task in aexit. Overall, I think that task.cancel() is somewhat broken with respect to finalization, as it essentially breaks all async code in finally and aexit.

We can wrap __aexit__ in asyncio.shield.

[elprans]

We can wrap __aexit__ in asyncio.shield

This works! I implemented the fix in #98.