How to do more than one cleanup task in cancelled state?

[zzzeek]

• asyncpg version: 0.23.0
• PostgreSQL version: 12.6
• Do you use a PostgreSQL SaaS? If so, which? Can you reproduce
  the issue with a local PostgreSQL install?: no/ yes
• Python version:3.9.5
• Platform: Fedora Linux
• Do you use pgbouncer?: no
• Did you install asyncpg with pip?: yes
• If you built asyncpg locally, which version of Cython did you use?: n/a
• Can the issue be reproduced under both asyncio and
  uvloop?: have only tested w/ asyncio directly

We have a user illustrating the case that a task being cancelled will allow us to reach a finally: block where we can at most run only one awaitable cleanup task on asyncpg in order to close out the connection. if we have more than one thing to await, such as emitting a ROLLBACK or anything else, we don't get the chance to close() the connection. It then seems to go into some place where we no longer have any reference to this connection yet asyncpg still leaves it opened; our own GC handlers that are supposed to take care of this are never called.

One way to illustrate it is the use case in such a way that indicates how I'm looking for "how to solve this problem?", of having two separate asycnpg connections that suppose we are doing some kind of work on separately in the same awaitable. if a cancel() is called, I can reach the finally: block, and I can then close at most one of the connections, but not both. in the real case, we are using only one connection but we are trying to emit a ROLLBACK and also do other awaitable things before we get to the .close().

What I dont understand is why gc isn't collecting these connections or why they aren't getting closed.

import asyncio

from asyncio import current_task
import asyncpg


async def get_and_cancel():
    c1 = await asyncpg.connect(
        user="scott", password="tiger", host="localhost", database="test"
    )
    c2 = await asyncpg.connect(
        user="scott", password="tiger", host="localhost", database="test"
    )
    try:
        r1 = await c1.fetch("SELECT 1")
        r2 = await c2.fetch("SELECT 1")
        current_task().cancel()
    finally:

        # we get here...

        # this seems to affect the asyncpg connection, the await is
        # honored....
        await c1.close()

        # but we never get here. connection leaks.  canonical way to
        # solve this issue?  
        await c2.close()


async def main():
    while True:
        try:
            await get_and_cancel()
        except asyncio.exceptions.CancelledError:
            pass


asyncio.run(main())

the stack trace is that we've run out of connections:

Traceback (most recent call last):
  File "/home/classic/dev/sqlalchemy/test4.py", line 39, in <module>
    asyncio.run(main())
  File "/usr/lib64/python3.9/asyncio/runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "/usr/lib64/python3.9/asyncio/base_events.py", line 642, in run_until_complete
    return future.result()
  File "/home/classic/dev/sqlalchemy/test4.py", line 34, in main
    await get_and_cancel()
  File "/home/classic/dev/sqlalchemy/test4.py", line 11, in get_and_cancel
    c2 = await asyncpg.connect(
  File "/home/classic/.venv3/lib64/python3.9/site-packages/asyncpg/connection.py", line 1981, in connect
    return await connect_utils._connect(
  File "/home/classic/.venv3/lib64/python3.9/site-packages/asyncpg/connect_utils.py", line 732, in _connect
    con = await _connect_addr(
  File "/home/classic/.venv3/lib64/python3.9/site-packages/asyncpg/connect_utils.py", line 632, in _connect_addr
    return await __connect_addr(params, timeout, True, *args)
  File "/home/classic/.venv3/lib64/python3.9/site-packages/asyncpg/connect_utils.py", line 682, in __connect_addr
    await compat.wait_for(connected, timeout=timeout)
  File "/home/classic/.venv3/lib64/python3.9/site-packages/asyncpg/compat.py", line 103, in wait_for
    return await asyncio.wait_for(fut, timeout)
  File "/usr/lib64/python3.9/asyncio/tasks.py", line 481, in wait_for
    return fut.result()
asyncpg.exceptions.TooManyConnectionsError: remaining connection slots are reserved for non-replication superuser connections

[zzzeek]

good news magicstack, I found the issue on our end where we were catching an Exception rather than a BaseException, which is as you know where CancelledError derives. /me drives off in his clown car

[zzzeek]

although FTR I am still not following yet where "c2" above goes and why it is seemingly not garbage collected.

[fantix]

Hi Mike! That is a very interesting issue. I think the reason is that, there is no await between the cancellation and the end of the try: block, therefore the actual cancellation happens during the first await in this task after that cancellation, which is the first close() in the finally block, and the first close() got cancelled.

---

In short, current_task().cancel() doesn't raise a CancelledError - the first await after it does.

[zzzeek]

@fantix if you run the above program, it runs out of connections. There are like hundreds that are all opened. Where are they ? c2 only points to one of them at a time. does asyncpg have some kind of strong-referencing of connections that aren't closed going on or otherwise some logic that doesn't end the socket when the connection is de-referenced?

[fantix]

Ah I see what you mean. Looks like without the cancellation, the connection will be recycled properly. 🤔 Looking into it now

---

Hmm no, as soon as the the connection is used, it's not recycled without an explicit close() 🤔

[fantix]

Aha, looks like there is an asyncio timer handle (300s) to clear some statement cache (_on_entry_expired) that has a callback handle _on_remove which is a bound method of the connection object - let me try fixing that.