Don't remove broken connections multiple times from the pool

Fixes `ValueError: deque.remove(x): x not in deque` when multiple connections
to the same server brake more or less simultaneously.

Author: robsdedude

neo4j/_async/io/_pool.py

@@ -241,13 +241,16 @@ async def deactivate(self, address):
                 connections = self.connections[address]
             except KeyError:  # already removed from the connection pool
                 return
-            for conn in list(connections):
-                if not conn.in_use:
-                    connections.remove(conn)
-                    try:
-                        await conn.close()
-                    except OSError:
-                        pass
+            closable_connections = [
+                conn for conn in connections if not conn.in_use
+            ]
+            # First remove all connections in question, then try to close them.
+            # If closing of a connection fails, we will end up in this method
+            # again.
+            for conn in closable_connections:
+                connections.remove(conn)
+            for conn in closable_connections:
+                await conn.close()
             if not self.connections[address]:
                 del self.connections[address]
 

neo4j/_sync/io/_pool.py

@@ -241,13 +241,16 @@ def deactivate(self, address):
                 connections = self.connections[address]
             except KeyError:  # already removed from the connection pool
                 return
-            for conn in list(connections):
-                if not conn.in_use:
-                    connections.remove(conn)
-                    try:
-                        conn.close()
-                    except OSError:
-                        pass
+            closable_connections = [
+                conn for conn in connections if not conn.in_use
+            ]
+            # First remove all connections in question, then try to close them.
+            # If closing of a connection fails, we will end up in this method
+            # again.
+            for conn in closable_connections:
+                connections.remove(conn)
+            for conn in closable_connections:
+                conn.close()
             if not self.connections[address]:
                 del self.connections[address]
 

tests/unit/async_/io/test_neo4j_pool.py

@@ -391,6 +391,39 @@ def liveness_side_effect(*args, **kwargs):
 
 
 
+@mark_async_test
+async def test_multiple_broken_connections_on_close(opener):
+    def mock_connection_breaks_on_close(cx):
+        async def close_side_effect():
+            cx.closed.return_value = True
+            cx.defunct.return_value = True
+            await pool.deactivate(READER_ADDRESS)
+
+        cx.attach_mock(AsyncMock(side_effect=close_side_effect), "close")
+
+    # create pool with 2 idle connections
+    pool = AsyncNeo4jPool(
+        opener, PoolConfig(), WorkspaceConfig(), ROUTER_ADDRESS
+    )
+    cx1 = await pool.acquire(READ_ACCESS, 30, "test_db", None)
+    cx2 = await pool.acquire(READ_ACCESS, 30, "test_db", None)
+    await pool.release(cx1)
+    await pool.release(cx2)
+
+    # both will loose connection
+    mock_connection_breaks_on_close(cx1)
+    mock_connection_breaks_on_close(cx2)
+
+    # force pool to close cx1, which will make it realize that the server is
+    # unreachable
+    cx1.stale.return_value = True
+
+    cx3 = await pool.acquire(READ_ACCESS, 30, "test_db", None)
+
+    assert cx3 is not cx1
+    assert cx3 is not cx2
+
+
 @mark_async_test
 async def test_failing_opener_leaves_connections_in_use_alone(opener):
     pool = AsyncNeo4jPool(

tests/unit/sync/io/test_neo4j_pool.py

@@ -391,6 +391,39 @@ def liveness_side_effect(*args, **kwargs):
 
 
 
+@mark_sync_test
+def test_multiple_broken_connections_on_close(opener):
+    def mock_connection_breaks_on_close(cx):
+        def close_side_effect():
+            cx.closed.return_value = True
+            cx.defunct.return_value = True
+            pool.deactivate(READER_ADDRESS)
+
+        cx.attach_mock(Mock(side_effect=close_side_effect), "close")
+
+    # create pool with 2 idle connections
+    pool = Neo4jPool(
+        opener, PoolConfig(), WorkspaceConfig(), ROUTER_ADDRESS
+    )
+    cx1 = pool.acquire(READ_ACCESS, 30, "test_db", None)
+    cx2 = pool.acquire(READ_ACCESS, 30, "test_db", None)
+    pool.release(cx1)
+    pool.release(cx2)
+
+    # both will loose connection
+    mock_connection_breaks_on_close(cx1)
+    mock_connection_breaks_on_close(cx2)
+
+    # force pool to close cx1, which will make it realize that the server is
+    # unreachable
+    cx1.stale.return_value = True
+
+    cx3 = pool.acquire(READ_ACCESS, 30, "test_db", None)
+
+    assert cx3 is not cx1
+    assert cx3 is not cx2
+
+
 @mark_sync_test
 def test_failing_opener_leaves_connections_in_use_alone(opener):
     pool = Neo4jPool(