[5.28] Fix race condition in bolt connection pool (neo4j#1348)

The pool tries to acquire an idle connection first. If no idle connection is
available, it puts itself into a queue waiting for another thread/task to return
a connection before trying again.

However, there is no protection against a connection being returned between the
attempt to acquire an idle connection and the enqueuing. Therefore, it could
happen that a thread waits indefinitely like this:

Assume max pool size of 1

 * T1 creates & acquires con1
 * T2 checks for idle connections => None
 * T1 releases con1
 * T2 acquires lock to be able to wait for the release event
 * T1 terminates
 * T2 times out waiting for Godot

Author: robsdedude

src/neo4j/_async/io/_pool.py

@@ -394,6 +394,13 @@ async def health_check(connection_, deadline_):
                 return connection
             # all connections in pool are in-use
             with self.lock:
+                if any(not con.in_use for con in self.connections[address]):
+                    # between trying to acquire an idle connection and
+                    # acquiring the lock to wait for space in the pool, a
+                    # connection was released back into the pool.
+                    # => Try to acquire an idle connection again instead of
+                    #    waiting.
+                    continue
                 connection_creator = self._acquire_new_later(
                     address,
                     auth,

src/neo4j/_sync/io/_pool.py

@@ -20,6 +20,7 @@
 import asyncio
 import threading
 import time
+import traceback
 from asyncio import Event as AsyncEvent
 from threading import (
     Event,
@@ -68,18 +69,25 @@ def test_multithread(self, pre_populated, fake_connection_generator):
         connections_lock = Lock()
         connections = []
         pre_populated_connections = []
+        thread_exceptions = []
 
         def acquire_release_conn(
             pool_, address_, acquired_counter_, release_event_
         ):
-            nonlocal connections, connections_lock
-            conn_ = pool_._acquire(address_, None, Deadline(3), None)
-            with connections_lock:
-                if connections is not None:
-                    connections.append(conn_)
-            acquired_counter_.increment()
-            release_event_.wait()
-            pool_.release(conn_)
+            nonlocal connections, connections_lock, thread_exceptions
+            try:
+                conn_ = pool_._acquire(address_, None, Deadline(3), None)
+                try:
+                    with connections_lock:
+                        if connections is not None:
+                            connections.append(conn_)
+                    acquired_counter_.increment()
+                    release_event_.wait()
+                finally:
+                    pool_.release(conn_)
+            except BaseException as e:
+                traceback.print_exception(e)
+                thread_exceptions.append(e)
 
         with FakeBoltPool(
             fake_connection_generator, (), max_connection_pool_size=5
@@ -126,6 +134,10 @@ def acquire_release_conn(
                 t.join(timeout=5)
                 if t.is_alive():
                     raise TimeoutError(f"Joining thread timed out: {t!r}")
+            if thread_exceptions:
+                raise RuntimeError(
+                    "thread raised an exception"
+                ) from thread_exceptions[0]
             # The pool size is still 5, but all are free
             self.assert_pool_size(address, 0, 5, pool)