Handle inactive connection closes while stored in the pool

asyncpg/pool.py

@@ -143,7 +143,8 @@ async def connect(self):
         self._con = con
 
     async def acquire(self) -> PoolConnectionProxy:
-        if self._con is None:
+        if self._con is None or self._con.is_closed():
+            self._con = None
             await self.connect()
 
         self._maybe_cancel_inactive_callback()

tests/test_pool.py

@@ -612,6 +612,27 @@ async def worker(pool):
 
         self.assertGreaterEqual(N, 50)
 
+    async def test_pool_handles_inactive_connection_errors(self):
+        pool = await self.create_pool(database='postgres',
+                                      min_size=1, max_size=1)
+
+        con = await pool.acquire(timeout=POOL_NOMINAL_TIMEOUT)
+
+        true_con = con._con
+
+        await pool.release(con)
+
+        # we simulate network error by terminating the connection
+        true_con.terminate()
+
+        # now pool should reopen terminated connection
+        con = await pool.acquire(timeout=POOL_NOMINAL_TIMEOUT)
+
+        self.assertEqual(await con.fetchval('SELECT 1'), 1)
+
+        await con.close()
+        await pool.close()
+
 
 @unittest.skipIf(os.environ.get('PGHOST'), 'using remote cluster for testing')
 class TestHotStandby(tb.ConnectedTestCase):