Merge branch '5.0' into async

Author: robsdedude

neo4j/_async/io/_pool.py

@@ -21,6 +21,7 @@
     defaultdict,
     deque,
 )
+import logging
 from logging import getLogger
 from random import choice
 from time import perf_counter
@@ -99,11 +100,19 @@ def time_remaining():
                 # try to find a free connection in pool
                 for connection in list(self.connections.get(address, [])):
                     if (connection.closed() or connection.defunct()
-                            or connection.stale()):
+                            or (connection.stale() and not connection.in_use)):
                         # `close` is a noop on already closed connections.
-                        # This is to make sure that the connection is gracefully
-                        # closed, e.g. if it's just marked as `stale` but still
-                        # alive.
+                        # This is to make sure that the connection is
+                        # gracefully closed, e.g. if it's just marked as
+                        # `stale` but still alive.
+                        if log.isEnabledFor(logging.DEBUG):
+                            log.debug(
+                                "[#%04X]  C: <POOL> removing old connection "
+                                "(closed=%s, defunct=%s, stale=%s, in_use=%s)",
+                                connection.local_port,
+                                connection.closed(), connection.defunct(),
+                                connection.stale(), connection.in_use
+                            )
                         await connection.close()
                         try:
                             self.connections.get(address, []).remove(connection)

tests/unit/async_/io/test_neo4j_pool.py

@@ -185,6 +185,37 @@ def break_connection():
     assert cx2 in pool.connections[cx2.addr]
 
 
+@mark_async_test
+async def test_does_not_close_stale_connections_in_use(opener):
+    pool = AsyncNeo4jPool(
+        opener, PoolConfig(), WorkspaceConfig(), ROUTER_ADDRESS
+    )
+    cx1 = await pool.acquire(READ_ACCESS, 30, "test_db", None)
+    assert cx1 in pool.connections[cx1.addr]
+    # simulate connection going stale (e.g. exceeding) while being in use
+    cx1.stale.return_value = True
+    cx2 = await pool.acquire(READ_ACCESS, 30, "test_db", None)
+    await pool.release(cx2)
+    cx1.close.assert_not_called()
+    assert cx2 is not cx1
+    assert cx2.addr == cx1.addr
+    assert cx1 in pool.connections[cx1.addr]
+    assert cx2 in pool.connections[cx2.addr]
+
+    await pool.release(cx1)
+    # now that cx1 is back in the pool and still stale,
+    # it should be closed when trying to acquire the next connection
+    cx1.close.assert_not_called()
+
+    cx3 = await pool.acquire(READ_ACCESS, 30, "test_db", None)
+    await pool.release(cx3)
+    cx1.close.assert_called_once()
+    assert cx2 is cx3
+    assert cx3.addr == cx1.addr
+    assert cx1 not in pool.connections[cx1.addr]
+    assert cx3 in pool.connections[cx2.addr]
+
+
 @mark_async_test
 async def test_release_resets_connections(opener):
     pool = AsyncNeo4jPool(