Connection not returning to pool after ConnectionDoesNotExistError exception

[feyukio]

• asyncpg version: 0.18.1
• PostgreSQL version: PostgreSQL 9.6.6 on x86_64-pc-linux-musl, compiled by gcc (Alpine 6.2.1) 6.2.1 20160822, 64-bit and PostgreSQL 9.5
• Do you use a PostgreSQL SaaS? If so, which? Can you reproduce
  the issue with a local PostgreSQL install?: Yes, I can reproduce. AWS and local
• Python version: 3.6.6
• Platform: 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?: no
• Can the issue be reproduced under both asyncio and
  uvloop?: didn't try using uvloop

When a connection from a pool is killed from database (asyncpg.exceptions.ConnectionDoesNotExistError), the pool cannot recovery this connection and the pool size will be compromised, reaching zero connections available even though no queries are executing.

This can happen other ways besides the ConnectionDoesNotExistError exception.

The following code can be used to reproduce this issue.

import asyncio
import asyncpg

_pool = None


async def connect():
    global _pool
    _pool = await asyncpg.create_pool(
        host="localhost",
        user="postgres",
        password="postgres",
        database="postgres",
        port=5432,
        min_size=0,
        max_size=3
    )


async def fetch(query, *args):
    # Acquire a connection
    async with _pool.acquire() as connection:
        try:
            result = await connection.fetch(query, *args)
            print(".")
            return [dict(row) for row in result]
        except asyncpg.exceptions.ConnectionDoesNotExistError:
            print("asyncpg.exceptions.ConnectionDoesNotExistError")


def print_counts():
    print("Queue size =", _pool._queue.qsize())


query = "select pg_sleep(2);"
terminate_query = "select pg_terminate_backend(pid) from pg_stat_activity;"


async def run():
    await connect()
    for i in range(5):
        futures = [fetch(query) for _ in range(3)]
        await asyncio.wait(futures)

        print_counts()
        await fetch(terminate_query)
        print_counts()


def main():
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(run())
    finally:
        loop.close()


if __name__ == "__main__":
    main()

[feyukio]

Update: To solve this issue, I call close() inside except, pool recovery its connection, but connection.is_closed() return True before close() call:

 except asyncpg.exceptions.ConnectionDoesNotExistError:
     print("asyncpg.exceptions.ConnectionDoesNotExistError")
     print("is_closed", connection.is_closed())
     await connection.close()

[feyukio]

Thanks =D