Repository: neo4j/neo4j-python-driver

Issue:
Async driver keeping connections from pool open?
I encountered a new issue with the Async driver version 5.0.0a2.

The error states: 

ERROR in app: Exception on request POST /recommend_contact
Traceback (most recent call last):
  File "/usr/local/lib/python3.7/site-packages/quart/app.py", line 1489, in handle_request
    return await self.full_dispatch_request(request_context)
  File "/usr/local/lib/python3.7/site-packages/quart/app.py", line 1514, in full_dispatch_request
    result = await self.handle_user_exception(error)
  File "/usr/local/lib/python3.7/site-packages/quart/app.py", line 964, in handle_user_exception
    raise error
  File "/usr/local/lib/python3.7/site-packages/quart/app.py", line 1512, in full_dispatch_request
    result = await self.dispatch_request(request_context)
  File "/usr/local/lib/python3.7/site-packages/quart/app.py", line 1557, in dispatch_request
    return await self.ensure_async(handler)(**request_.view_args)
  File "/usr/local/lib/python3.7/site-packages/quart/views.py", line 57, in view
    return await current_app.ensure_async(self.dispatch_request)(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/quart_openapi/resource.py", line 71, in dispatch_request
    return await handler(*args, **kwargs)
  File "/api/app.py", line 119, in post
    recommendations = await asyncio.gather(*tasks)
  File "/api/app.py", line 108, in recommendation_task
    el['offsite'])
  File "/usr/local/lib/python3.7/site-packages/recommendation_engine-0.0.1-py3.7.egg/recommendation_engine/recommendation/recommender.py", line 32, in get_popular_contact
    ret = await self.graph_handler.run_query_async(graph_query, contact_id=contact_id, p=p, offsite=offsite)
  File "/usr/local/lib/python3.7/site-packages/recommendation_engine-0.0.1-py3.7.egg/recommendation_engine/graph_handler.py", line 636, in run_query_async
    return await session.read_transaction(self.run_transaction, query, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/neo4j/_meta.py", line 73, in inner
    return await f(*args, **kwargs)
  File "/usr/local/lib/python3.7/site-packages/neo4j/_async/work/session.py", line 656, in read_transaction
    READ_ACCESS, transaction_function, *args, **kwargs
  File "/usr/local/lib/python3.7/site-packages/neo4j/_async/work/session.py", line 478, in _run_transaction
    timeout=timeout
  File "/usr/local/lib/python3.7/site-packages/neo4j/_async/work/session.py", line 392, in _open_transaction
    await self._connect(access_mode=access_mode)
  File "/usr/local/lib/python3.7/site-packages/neo4j/_async/work/session.py", line 122, in _connect
    await super()._connect(access_mode, **access_kwargs)
  File "/usr/local/lib/python3.7/site-packages/neo4j/_async/work/workspace.py", line 194, in _connect
    self._connection = await self._pool.acquire(**acquire_kwargs_)
  File "/usr/local/lib/python3.7/site-packages/neo4j/_async/io/_pool.py", line 403, in acquire
    self.address, deadline, liveness_check_timeout
  File "/usr/local/lib/python3.7/site-packages/neo4j/_async/io/_pool.py", line 216, in _acquire
    "{!r}s (timeout)".format(deadline.original_timeout)
neo4j.exceptions.ClientError: {code: None} {message: None}

I suspected it is the issue with connections being kept alive or being alive for a prolonged period of time, as when I checked "/usr/local/lib/python3.7/site-packages/neo4j/_async/io/_pool.py" I noticed the error is thrown when the pool is full and there are no avaliable connections. Also the code and message are shown as None, so I am not sure if that is a bug also.

We are running Neo4j Version 4.4.3 Community edition as a single instance.

Discussion:

Patch:
diff --git a/neo4j/_async/io/_pool.py b/neo4j/_async/io/_pool.py
index d0a9acd4..f61ebd46 100644
--- a/neo4j/_async/io/_pool.py
+++ b/neo4j/_async/io/_pool.py
@@ -23,7 +23,6 @@ from collections import (
     defaultdict,
     deque,
 )
-from contextlib import asynccontextmanager
 from logging import getLogger
 from random import choice
 
@@ -161,11 +160,10 @@ class AsyncIOPool(abc.ABC):
             connections = self.connections[address]
             pool_size = (len(connections)
                          + self.connections_reservations[address])
-            can_create_new_connection = (infinite_pool_size
-                                         or pool_size < max_pool_size)
-            self.connections_reservations[address] += 1
-        if can_create_new_connection:
-            return connection_creator
+            if infinite_pool_size or pool_size < max_pool_size:
+                # there's room for a new connection
+                self.connections_reservations[address] += 1
+                return connection_creator
         return None
 
     async def _acquire(self, address, deadline, liveness_check_timeout):
diff --git a/neo4j/_sync/io/_pool.py b/neo4j/_sync/io/_pool.py
index d031ad8c..0a379009 100644
--- a/neo4j/_sync/io/_pool.py
+++ b/neo4j/_sync/io/_pool.py
@@ -23,7 +23,6 @@ from collections import (
     defaultdict,
     deque,
 )
-from contextlib import contextmanager
 from logging import getLogger
 from random import choice
 
@@ -161,11 +160,10 @@ class IOPool(abc.ABC):
             connections = self.connections[address]
             pool_size = (len(connections)
                          + self.connections_reservations[address])
-            can_create_new_connection = (infinite_pool_size
-                                         or pool_size < max_pool_size)
-            self.connections_reservations[address] += 1
-        if can_create_new_connection:
-            return connection_creator
+            if infinite_pool_size or pool_size < max_pool_size:
+                # there's room for a new connection
+                self.connections_reservations[address] += 1
+                return connection_creator
         return None
 
     def _acquire(self, address, deadline, liveness_check_timeout):


Test patch:
diff --git a/tests/unit/async_/io/test_neo4j_pool.py b/tests/unit/async_/io/test_neo4j_pool.py
index b9bbc4e4..e4dec78b 100644
--- a/tests/unit/async_/io/test_neo4j_pool.py
+++ b/tests/unit/async_/io/test_neo4j_pool.py
@@ -25,6 +25,7 @@ from neo4j import (
     WRITE_ACCESS,
 )
 from neo4j._async.io import AsyncNeo4jPool
+from neo4j._async_compat.util import AsyncUtil
 from neo4j._conf import (
     PoolConfig,
     RoutingConfig,
@@ -437,3 +438,33 @@ async def test_failing_opener_leaves_connections_in_use_alone(opener):
     with pytest.raises((ServiceUnavailable, SessionExpired)):
         await pool.acquire(READ_ACCESS, 30, "test_db", None, None)
     assert not cx1.closed()
+
+
+@mark_async_test
+async def test__acquire_new_later_with_room(opener):
+    config = PoolConfig()
+    config.max_connection_pool_size = 1
+    pool = AsyncNeo4jPool(
+        opener, config, WorkspaceConfig(), ROUTER_ADDRESS
+    )
+    assert pool.connections_reservations[READER_ADDRESS] == 0
+    creator = pool._acquire_new_later(READER_ADDRESS, Deadline(1))
+    assert pool.connections_reservations[READER_ADDRESS] == 1
+    assert callable(creator)
+    if AsyncUtil.is_async_code:
+        assert inspect.iscoroutinefunction(creator)
+
+
+@mark_async_test
+async def test__acquire_new_later_without_room(opener):
+    config = PoolConfig()
+    config.max_connection_pool_size = 1
+    pool = AsyncNeo4jPool(
+        opener, config, WorkspaceConfig(), ROUTER_ADDRESS
+    )
+    _ = await pool.acquire(READ_ACCESS, 30, "test_db", None, None)
+    # pool is full now
+    assert pool.connections_reservations[READER_ADDRESS] == 0
+    creator = pool._acquire_new_later(READER_ADDRESS, Deadline(1))
+    assert pool.connections_reservations[READER_ADDRESS] == 0
+    assert creator is None
diff --git a/tests/unit/sync/io/test_neo4j_pool.py b/tests/unit/sync/io/test_neo4j_pool.py
index bca9d444..17507c71 100644
--- a/tests/unit/sync/io/test_neo4j_pool.py
+++ b/tests/unit/sync/io/test_neo4j_pool.py
@@ -24,6 +24,7 @@ from neo4j import (
     READ_ACCESS,
     WRITE_ACCESS,
 )
+from neo4j._async_compat.util import Util
 from neo4j._conf import (
     PoolConfig,
     RoutingConfig,
@@ -437,3 +438,33 @@ def test_failing_opener_leaves_connections_in_use_alone(opener):
     with pytest.raises((ServiceUnavailable, SessionExpired)):
         pool.acquire(READ_ACCESS, 30, "test_db", None, None)
     assert not cx1.closed()
+
+
+@mark_sync_test
+def test__acquire_new_later_with_room(opener):
+    config = PoolConfig()
+    config.max_connection_pool_size = 1
+    pool = Neo4jPool(
+        opener, config, WorkspaceConfig(), ROUTER_ADDRESS
+    )
+    assert pool.connections_reservations[READER_ADDRESS] == 0
+    creator = pool._acquire_new_later(READER_ADDRESS, Deadline(1))
+    assert pool.connections_reservations[READER_ADDRESS] == 1
+    assert callable(creator)
+    if Util.is_async_code:
+        assert inspect.iscoroutinefunction(creator)
+
+
+@mark_sync_test
+def test__acquire_new_later_without_room(opener):
+    config = PoolConfig()
+    config.max_connection_pool_size = 1
+    pool = Neo4jPool(
+        opener, config, WorkspaceConfig(), ROUTER_ADDRESS
+    )
+    _ = pool.acquire(READ_ACCESS, 30, "test_db", None, None)
+    # pool is full now
+    assert pool.connections_reservations[READER_ADDRESS] == 0
+    creator = pool._acquire_new_later(READER_ADDRESS, Deadline(1))
+    assert pool.connections_reservations[READER_ADDRESS] == 0
+    assert creator is None