Fix socket timeouts (#1227)

## Fix connection hint timeout
The driver may receive a configuration hint from the server during connection
setup. The hint is a timeout: if the driver is waiting for a response from the
server but has not heard back within the timeout window, the driver shall assume
the connection is dead. The server can send `NOOP` ping messages to reset the
timer ticking on the client-side.

The driver applied this hint to both reads and writes. This PR changes it so
that the hint, as intended, only applies to read operations on connections.

## New write timeout configuraiton option
To avoid the driver hanging indefinitely on write operations, a new driver-level
configuration option `connection_write_timeout`, which defaults to `30 seconds`
is introduced. As the name suggests, this timeout is applied to all write
operations on connections.

Author: robsdedude

CHANGELOG.md

@@ -190,6 +190,9 @@ See also https://github.com/neo4j/neo4j-python-driver/wiki for a full changelog.
 - On failed liveness check (s. `liveness_check_timeout` configuration option), the driver will no longer remove the
   remote from the cached routing tables, but only close the connection under test.  
   This aligns the driver with the other official Neo4j drivers.
+- The driver incorrectly applied a timeout hint received from the server to both read and write I/O operations.
+  It is now only applied to read I/O operations.  
+  In turn, a new configuration option `connection_write_timeout` with a default value of `30 seconds` is introduced.
 
 
 ## Version 5.28

docs/source/api.rst

@@ -399,6 +399,7 @@ Additional configuration can be provided via the :class:`neo4j.Driver` construct
 
 + :ref:`connection-acquisition-timeout-ref`
 + :ref:`connection-timeout-ref`
++ :ref:`connection-write-timeout-ref`
 + :ref:`encrypted-ref`
 + :ref:`keep-alive-ref`
 + :ref:`max-connection-lifetime-ref`
@@ -430,7 +431,7 @@ it should be chosen larger than :ref:`connection-timeout-ref`.
 :Type: ``float``
 :Default: ``60.0``
 
-.. versionadded:: 6.0
+.. versionchanged:: 6.0
     The setting now entails *anything* required to acquire a connection.
     This includes potential fetching of routing tables which in itself requires acquiring a connection.
     Previously, the timeout would be restarted for such auxiliary connection acquisitions.
@@ -450,6 +451,16 @@ connection can be used to perform database related work.
 :Default: ``30.0``
 
 
+.. _connection-write-timeout-ref:
+
+``connection_write_timeout``
+----------------------------
+The maximum amount of time in seconds to wait for TCP write operations to complete.
+
+:Type: ``float``
+:Default: ``30.0``
+
+
 .. _encrypted-ref:
 
 ``encrypted``

src/neo4j/_async/config.py

@@ -54,6 +54,10 @@ class AsyncPoolConfig(Config):
     # The maximum amount of time to wait for a TCP connection to be
     # established.
 
+    #: Connection Write Timeout
+    connection_write_timeout = 30.0  # seconds
+    # The maximum amount of time to wait for I/O write operations to complete.
+
     #: Custom Resolver
     resolver = None
     # Custom resolver function, returning list of resolved addresses.

src/neo4j/_async/driver.py

@@ -127,6 +127,7 @@ def driver(
             liveness_check_timeout: float | None = ...,
             max_connection_pool_size: int = ...,
             connection_timeout: float = ...,
+            connection_write_timeout: float = ...,
             resolver: (
                 t.Callable[[Address], t.Iterable[Address]]
                 | t.Callable[[Address], t.Awaitable[t.Iterable[Address]]]

src/neo4j/_async/io/_bolt.py

@@ -418,11 +418,13 @@ async def open(
         )
 
         try:
-            connection.socket.set_deadline(deadline)
+            connection.socket.set_read_deadline(deadline)
+            connection.socket.set_write_deadline(deadline)
             try:
                 await connection.hello()
             finally:
-                connection.socket.set_deadline(None)
+                connection.socket.set_read_deadline(None)
+                connection.socket.set_write_deadline(None)
         except (
             Exception,
             # Python 3.8+: CancelledError is a subclass of BaseException

src/neo4j/_async/io/_bolt4.py

@@ -626,7 +626,7 @@ def on_success(metadata):
                     "connection.recv_timeout_seconds"
                 ]
                 if isinstance(recv_timeout, int) and recv_timeout > 0:
-                    self.socket.settimeout(recv_timeout)
+                    self.socket.set_read_timeout(recv_timeout)
                 else:
                     log.info(
                         "[#%04X]  _: <CONNECTION> Server supplied an "

src/neo4j/_async/io/_bolt5.py

@@ -149,7 +149,7 @@ def on_success(metadata):
                     "connection.recv_timeout_seconds"
                 ]
                 if isinstance(recv_timeout, int) and recv_timeout > 0:
-                    self.socket.settimeout(recv_timeout)
+                    self.socket.set_read_timeout(recv_timeout)
                 else:
                     log.info(
                         "[#%04X]  _: <CONNECTION> Server supplied an "
@@ -622,7 +622,7 @@ def on_success(metadata):
                     "connection.recv_timeout_seconds"
                 ]
                 if isinstance(recv_timeout, int) and recv_timeout > 0:
-                    self.socket.settimeout(recv_timeout)
+                    self.socket.set_read_timeout(recv_timeout)
                 else:
                     log.info(
                         "[#%04X]  _: <CONNECTION> Server supplied an "
@@ -708,7 +708,7 @@ def on_success(metadata):
                     "connection.recv_timeout_seconds"
                 ]
                 if isinstance(recv_timeout, int) and recv_timeout > 0:
-                    self.socket.settimeout(recv_timeout)
+                    self.socket.set_read_timeout(recv_timeout)
                 else:
                     log.info(
                         "[#%04X]  _: <CONNECTION> Server supplied an "

src/neo4j/_async/io/_bolt6.py

@@ -158,7 +158,7 @@ def on_success(metadata):
                     "connection.recv_timeout_seconds"
                 ]
                 if isinstance(recv_timeout, int) and recv_timeout > 0:
-                    self.socket.settimeout(recv_timeout)
+                    self.socket.set_read_timeout(recv_timeout)
                 else:
                     log.info(
                         "[#%04X]  _: <CONNECTION> Server supplied an "

src/neo4j/_async/io/_bolt_socket.py

@@ -32,7 +32,10 @@
     BoltError,
     BoltProtocolError,
 )
-from ..._io import BoltProtocolVersion
+from ..._io import (
+    BoltProtocolVersion,
+    min_timeout,
+)
 from ...exceptions import (
     DriverError,
     ServiceUnavailable,
@@ -157,8 +160,8 @@ def _encode_varint(n: int) -> bytearray:
         return res
 
     async def _handshake_read(self, ctx: HandshakeCtx, n: int) -> bytes:
-        original_timeout = self.gettimeout()
-        self.settimeout(ctx.deadline.to_timeout())
+        original_timeout = self.get_read_timeout()
+        self.set_read_timeout(ctx.deadline.to_timeout())
         try:
             response = await self.recv(n)
             ctx.full_response.extend(response)
@@ -168,7 +171,7 @@ async def _handshake_read(self, ctx: HandshakeCtx, n: int) -> bytes:
                 f"{ctx.resolved_address!r} (deadline {ctx.deadline})"
             ) from exc
         finally:
-            self.settimeout(original_timeout)
+            self.set_read_timeout(original_timeout)
         data_size = len(response)
         if data_size == 0:
             # If no data is returned after a successful select
@@ -192,9 +195,11 @@ async def _handshake_read(self, ctx: HandshakeCtx, n: int) -> bytes:
 
         return response
 
-    async def _handshake_send(self, ctx, data):
-        original_timeout = self.gettimeout()
-        self.settimeout(ctx.deadline.to_timeout())
+    async def _handshake_send(self, ctx, data, write_timeout=None):
+        original_timeout = self.get_write_timeout()
+        self.set_write_timeout(
+            min_timeout(ctx.deadline.to_timeout(), write_timeout)
+        )
         try:
             await self.sendall(data)
         except OSError as exc:
@@ -203,7 +208,7 @@ async def _handshake_send(self, ctx, data):
                 f"{ctx.resolved_address!r} (deadline {ctx.deadline})"
             ) from exc
         finally:
-            self.settimeout(original_timeout)
+            self.set_write_timeout(original_timeout)
 
     async def _handshake(
         self,