Ensure pool connection is released in acquire when cancelled

Closes MagicStack#547

When wait_for is cancelled, there is a chance that the waited
task has already been completed, leaving the connection
looking like it is in use. This fix ensures that the connection
is returned to the pool in this situation.

For context, see:
https://bugs.python.org/issue37658
MagicStack#467

Author: aaliddell

asyncpg/pool.py

@@ -605,15 +605,29 @@ async def _acquire_impl():
                 ch._timeout = timeout
                 return proxy
 
+        def _release_leaked_connection(fut):
+            if not fut.cancelled():
+                asyncio.ensure_future(self.release(fut.result()))
+
         if self._closing:
             raise exceptions.InterfaceError('pool is closing')
         self._check_init()
 
+        acquire_fut = asyncio.ensure_future(_acquire_impl())
         if timeout is None:
-            return await _acquire_impl()
+            return await acquire_fut
         else:
-            return await asyncio.wait_for(
-                _acquire_impl(), timeout=timeout)
+            try:
+                return await asyncio.wait_for(
+                    acquire_fut, timeout=timeout)
+            except asyncio.CancelledError:
+                # Ensure connection is marked as not in use.
+                # The cancellation may have raced the acquire, leading
+                # to the acquire completing but the wait_for to be
+                # cancelled.
+                # See: https://bugs.python.org/issue37658
+                acquire_fut.add_done_callback(_release_leaked_connection)
+                raise
 
     async def release(self, connection, *, timeout=None):
         """Release a database connection back to the pool.