Fix connection pool deadlock

asyncpg/pool.py

@@ -199,8 +199,13 @@ async def release(self, timeout: Optional[float]) -> None:
                 'a free connection holder')
 
         if self._con.is_closed():
-            # When closing, pool connections perform the necessary
-            # cleanup, so we don't have to do anything else here.
+            # The connection was closed/aborted, possibly without going
+            # through Connection.close() or Connection.terminate() (e.g.
+            # when the protocol calls abort() on a fatal server error).
+            # In that case, _release_on_close() may never have been called,
+            # so we call _release() here to ensure the holder is returned
+            # to the pool queue.
+            self._release()
             return
 
         self._timeout = None

tests/test_pool.py

@@ -1004,6 +1004,19 @@ async def worker():
         conn = await pool.acquire(timeout=0.1)
         await pool.release(conn)
 
+    async def test_pool_release_after_protocol_abort(self):
+        pool = await self.create_pool(min_size=1, max_size=1)
+
+        conn = await pool.acquire()
+        conn._con._protocol.abort()
+
+        await pool.release(conn)
+
+        # Check the connection has been released back to the pool
+        conn2 = await pool.acquire(timeout=1.0)
+        await pool.release(conn2)
+        await pool.close()
+
 
 @unittest.skipIf(os.environ.get('PGHOST'), 'unmanaged cluster')
 class TestPoolReconnectWithTargetSessionAttrs(tb.ClusterTestCase):