Fix socket timeouts

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,