Don't close active connections if connection opener fails

Especially in concurrent environments, this could lead to all sorts of
unspecific errors.

Author: robsdedude

neo4j/io/__init__.py

@@ -695,7 +695,7 @@ def time_remaining():
                     try:
                         connection = self.opener(address, timeout)
                     except ServiceUnavailable:
-                        self.remove(address)
+                        self.deactivate(address)
                         raise
                     else:
                         connection.pool = self
@@ -779,31 +779,24 @@ def deactivate(self, address):
                         conn.close()
                     except OSError:
                         pass
-            if not connections:
-                self.remove(address)
+            if not self.connections[address]:
+                del self.connections[address]
 
     def on_write_failure(self, address):
         raise WriteServiceUnavailable("No write service available for pool {}".format(self))
 
-    def remove(self, address):
-        """ Remove an address from the connection pool, if present, closing
-        all connections to that address.
-        """
-        with self.lock:
-            for connection in self.connections.pop(address, ()):
-                try:
-                    connection.close()
-                except OSError:
-                    pass
-
     def close(self):
         """ Close all connections and empty the pool.
         This method is thread safe.
         """
         try:
             with self.lock:
                 for address in list(self.connections):
-                    self.remove(address)
+                    for connection in self.connections.pop(address, ()):
+                        try:
+                            connection.close()
+                        except OSError:
+                            pass
         except TypeError:
             pass
 

tests/unit/io/test_neo4j_pool.py

@@ -35,6 +35,10 @@
     RoutingConfig,
     WorkspaceConfig
 )
+from neo4j.exceptions import (
+    ServiceUnavailable,
+    SessionExpired
+)
 from neo4j.io import Neo4jPool
 
 
@@ -226,3 +230,14 @@ def test_release_does_not_resets_defunct_connections(opener):
     cx1.defunct.assert_called_once()
     cx1.is_reset_mock.asset_not_called()
     cx1.reset.asset_not_called()
+
+
+def test_failing_opener_leaves_connections_in_use_alone(opener):
+    pool = Neo4jPool(opener, PoolConfig(), WorkspaceConfig(), ROUTER_ADDRESS)
+    cx1 = pool.acquire(READ_ACCESS, 30, "test_db", None)
+
+    opener.side_effect = ServiceUnavailable("Server overloaded")
+    with pytest.raises((ServiceUnavailable, SessionExpired)):
+        pool.acquire(READ_ACCESS, 30, "test_db", None)
+
+    assert not cx1.closed()