[4.4] Fix handling of sub-ms transaction timeouts (#1034)

Transaction timeouts are specified in seconds as float.
However, the server expects it in milliseconds as int. This would lead to

 1) rounding issues: previously, the driver would multiply by 1000 and
    then truncate to int. E.g., 256.4 seconds would be turned into 256399 ms
    because of float imprecision.
    Therefore, the built-in `round` is now used instead.
 2) values below 1 ms (e.g., 0.0001) would be rounded down to 0. However, 0 is
    a special value that instructs the server to not apply any timeout. This
    is likely to surprise the user which specified a non-zero timeout. In this
    special case, the driver now rounds up to 1 ms.

Backport of: #940

Backport adjustment:
For better backwards compatibility, we're sending negative timeouts to the
DBMS as the 4.4 driver did before (even though those values might not have the
intuitive effect, we don't want to change the behavior of the driver too much in
a patch release).

Author: robsdedude

neo4j/io/_bolt3.py

@@ -47,6 +47,7 @@
     CommitResponse,
     InitResponse,
     Response,
+    tx_timeout_as_ms,
 )
 
 
@@ -225,11 +226,8 @@ def run(self, query, parameters=None, mode=None, bookmarks=None,
                 extra["tx_metadata"] = dict(metadata)
             except TypeError:
                 raise TypeError("Metadata must be coercible to a dict")
-        if timeout:
-            try:
-                extra["tx_timeout"] = int(1000 * timeout)
-            except TypeError:
-                raise TypeError("Timeout must be specified as a number of seconds")
+        if timeout or (isinstance(timeout, (float, int)) and timeout == 0):
+            extra["tx_timeout"] = tx_timeout_as_ms(timeout)
         fields = (query, parameters, extra)
         log.debug("[#%04X]  C: RUN %s", self.local_port, " ".join(map(repr, fields)))
         if query.upper() == u"COMMIT":
@@ -277,12 +275,8 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
                 extra["tx_metadata"] = dict(metadata)
             except TypeError:
                 raise TypeError("Metadata must be coercible to a dict")
-        if timeout:
-            try:
-                extra["tx_timeout"] = int(1000 * timeout)
-            except TypeError:
-                raise TypeError("Timeout must be specified as a number of seconds")
-        log.debug("[#%04X]  C: BEGIN %r", self.local_port, extra)
+        if timeout or (isinstance(timeout, (float, int)) and timeout == 0):
+            extra["tx_timeout"] = tx_timeout_as_ms(timeout)
         self._append(b"\x11", (extra,), Response(self, "begin", **handlers))
 
     def commit(self, **handlers):

neo4j/io/_bolt4.py

@@ -34,7 +34,6 @@
 from neo4j.exceptions import (
     ConfigurationError,
     DatabaseUnavailable,
-    DriverError,
     ForbiddenOnReadOnlyDatabase,
     Neo4jError,
     NotALeader,
@@ -48,6 +47,7 @@
     CommitResponse,
     InitResponse,
     Response,
+    tx_timeout_as_ms,
 )
 from neo4j.io._bolt3 import (
     ServerStateManager,
@@ -178,11 +178,8 @@ def run(self, query, parameters=None, mode=None, bookmarks=None,
                 extra["tx_metadata"] = dict(metadata)
             except TypeError:
                 raise TypeError("Metadata must be coercible to a dict")
-        if timeout:
-            try:
-                extra["tx_timeout"] = int(1000 * timeout)
-            except TypeError:
-                raise TypeError("Timeout must be specified as a number of seconds")
+        if timeout or (isinstance(timeout, (float, int)) and timeout == 0):
+            extra["tx_timeout"] = tx_timeout_as_ms(timeout)
         fields = (query, parameters, extra)
         log.debug("[#%04X]  C: RUN %s", self.local_port, " ".join(map(repr, fields)))
         if query.upper() == u"COMMIT":
@@ -229,11 +226,8 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
                 extra["tx_metadata"] = dict(metadata)
             except TypeError:
                 raise TypeError("Metadata must be coercible to a dict")
-        if timeout:
-            try:
-                extra["tx_timeout"] = int(1000 * timeout)
-            except TypeError:
-                raise TypeError("Timeout must be specified as a number of seconds")
+        if timeout or (isinstance(timeout, (float, int)) and timeout == 0):
+            extra["tx_timeout"] = tx_timeout_as_ms(timeout)
         log.debug("[#%04X]  C: BEGIN %r", self.local_port, extra)
         self._append(b"\x11", (extra,), Response(self, "begin", **handlers))
 
@@ -490,12 +484,8 @@ def run(self, query, parameters=None, mode=None, bookmarks=None,
                 extra["tx_metadata"] = dict(metadata)
             except TypeError:
                 raise TypeError("Metadata must be coercible to a dict")
-        if timeout:
-            try:
-                extra["tx_timeout"] = int(1000 * timeout)
-            except TypeError:
-                raise TypeError("Timeout must be specified as a number of "
-                                "seconds")
+        if timeout or (isinstance(timeout, (float, int)) and timeout == 0):
+            extra["tx_timeout"] = tx_timeout_as_ms(timeout)
         fields = (query, parameters, extra)
         log.debug("[#%04X]  C: RUN %s", self.local_port,
                   " ".join(map(repr, fields)))
@@ -525,11 +515,7 @@ def begin(self, mode=None, bookmarks=None, metadata=None, timeout=None,
                 extra["tx_metadata"] = dict(metadata)
             except TypeError:
                 raise TypeError("Metadata must be coercible to a dict")
-        if timeout:
-            try:
-                extra["tx_timeout"] = int(1000 * timeout)
-            except TypeError:
-                raise TypeError("Timeout must be specified as a number of "
-                                "seconds")
+        if timeout or (isinstance(timeout, (float, int)) and timeout == 0):
+            extra["tx_timeout"] = tx_timeout_as_ms(timeout)
         log.debug("[#%04X]  C: BEGIN %r", self.local_port, extra)
         self._append(b"\x11", (extra,), Response(self, "begin", **handlers))

neo4j/io/_common.py

@@ -270,3 +270,31 @@ def on_failure(self, metadata):
 class CommitResponse(Response):
 
     pass
+
+
+def tx_timeout_as_ms(timeout: float) -> int:
+    """
+    Round transaction timeout to milliseconds.
+
+    Values in (0, 1], else values are rounded using the built-in round()
+    function (round n.5 values to nearest even).
+
+    :param timeout: timeout in seconds
+
+    :returns: timeout in milliseconds (rounded)
+
+    :raise ValueError: if timeout is negative
+    """
+    try:
+        timeout = float(timeout)
+    except (TypeError, ValueError) as e:
+        err_type = type(e)
+        msg = "Timeout must be specified as a number of seconds"
+        raise err_type(msg) from e
+    ms = int(round(1000 * timeout))
+    if ms == 0 and timeout > 0:
+        # Special case for 0 < timeout < 0.5 ms.
+        # This would be rounded to 0 ms, but the server interprets this as
+        # infinite timeout. So we round to the smallest possible timeout: 1 ms.
+        ms = 1
+    return ms

neo4j/work/simple.py

@@ -281,10 +281,19 @@ def begin_transaction(self, metadata=None, timeout=None):
 
         :param timeout:
             the transaction timeout in seconds.
-            Transactions that execute longer than the configured timeout will be terminated by the database.
-            This functionality allows to limit query/transaction execution time.
-            Specified timeout overrides the default timeout configured in the database using ``dbms.transaction.timeout`` setting.
-            Value should not represent a duration of zero or negative duration.
+            Transactions that execute longer than the configured timeout will
+            be terminated by the database.
+            This functionality allows user code to limit query/transaction
+            execution time.
+            The specified timeout overrides the default timeout configured in
+            the database using the ``db.transaction.timeout`` setting
+            (``dbms.transaction.timeout`` before Neo4j 5.0).
+            Values higher than ``db.transaction.timeout`` will be ignored and
+            will fall back to the default for server versions between 4.2 and
+            5.2 (inclusive).
+            The value should not represent a negative duration.
+            A ``0`` duration will make the transaction execute indefinitely.
+            :data:`None` will use the default timeout configured on the server.
         :type timeout: int
 
         :returns: A new transaction instance.
@@ -441,7 +450,21 @@ class Query:
     :type text: str
     :param metadata: metadata attached to the query.
     :type metadata: dict
-    :param timeout: seconds.
+    :param timeout:
+        the transaction timeout in seconds.
+        Transactions that execute longer than the configured timeout will
+        be terminated by the database.
+        This functionality allows user code to limit query/transaction
+        execution time.
+        The specified timeout overrides the default timeout configured in
+        the database using the ``db.transaction.timeout`` setting
+        (``dbms.transaction.timeout`` before Neo4j 5.0).
+        Values higher than ``db.transaction.timeout`` will be ignored and
+        will fall back to the default for server versions between 4.2 and
+        5.2 (inclusive).
+        The value should not represent a negative duration.
+        A ``0`` duration will make the transaction execute indefinitely.
+        :data:`None` will use the default timeout configured on the server.
     :type timeout: float or :const:`None`
     """
     def __init__(self, text, metadata=None, timeout=None):
@@ -476,12 +499,19 @@ def count_people_tx(tx):
 
     :param timeout:
         the transaction timeout in seconds.
-        Transactions that execute longer than the configured timeout will be terminated by the database.
-        This functionality allows to limit query/transaction execution time.
-        Specified timeout overrides the default timeout configured in the database using ``dbms.transaction.timeout`` setting.
-        Values higher than ``dbms.transaction.timeout`` will be ignored and
-        will fall back to default (unless using Neo4j < 4.2).
-        Value should not represent a duration of zero or negative duration.
+        Transactions that execute longer than the configured timeout will
+        be terminated by the database.
+        This functionality allows user code to limit query/transaction
+        execution time.
+        The specified timeout overrides the default timeout configured in
+        the database using the ``db.transaction.timeout`` setting
+        (``dbms.transaction.timeout`` before Neo4j 5.0).
+        Values higher than ``db.transaction.timeout`` will be ignored and
+        will fall back to the default for server versions between 4.2 and
+        5.2 (inclusive).
+        The value should not represent a negative duration.
+        A ``0`` duration will make the transaction execute indefinitely.
+        :data:`None` will use the default timeout configured on the server.
     :type timeout: float or :const:`None`
     """
 

tests/unit/io/test_class_bolt3.py

@@ -110,3 +110,56 @@ def test_hint_recv_timeout_seconds_gets_ignored(fake_socket_pair, recv_timeout):
                        PoolConfig.max_connection_lifetime)
     connection.hello()
     sockets.client.settimeout.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("func", "args", "extra_idx"),
+    (
+        ("run", ("RETURN 1",), 2),
+        ("begin", (), 0),
+    )
+)
+@pytest.mark.parametrize(
+    ("timeout", "res"),
+    (
+        (None, None),
+        (0, 0),
+        (0.1, 100),
+        (0.001, 1),
+        (1e-15, 1),
+        (0.0005, 1),
+        (0.0001, 1),
+        (1.0015, 1002),
+        (1.000499, 1000),
+        (1.0025, 1002),
+        (3.0005, 3000),
+        (3.456, 3456),
+        (1, 1000),
+        (
+            "foo",
+            ValueError("Timeout must be specified as a number of seconds")
+        ),
+        (
+            [1, 2],
+            TypeError("Timeout must be specified as a number of seconds")
+        )
+    )
+)
+def test_tx_timeout(fake_socket_pair, func, args, extra_idx, timeout, res):
+    address = ("127.0.0.1", 7687)
+    sockets = fake_socket_pair(address)
+    sockets.server.send_message(0x70, {})
+    connection = Bolt3(address, sockets.client, 0)
+    func = getattr(connection, func)
+    if isinstance(res, Exception):
+        with pytest.raises(type(res), match=str(res)):
+            func(*args, timeout=timeout)
+    else:
+        func(*args, timeout=timeout)
+        connection.send_all()
+        tag, fields = sockets.server.pop_message()
+        extra = fields[extra_idx]
+        if timeout is None:
+            assert "tx_timeout" not in extra
+        else:
+            assert extra["tx_timeout"] == res

tests/unit/io/test_class_bolt4x0.py

@@ -197,3 +197,56 @@ def test_hint_recv_timeout_seconds_gets_ignored(fake_socket_pair, recv_timeout):
                          PoolConfig.max_connection_lifetime)
     connection.hello()
     sockets.client.settimeout.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("func", "args", "extra_idx"),
+    (
+        ("run", ("RETURN 1",), 2),
+        ("begin", (), 0),
+    )
+)
+@pytest.mark.parametrize(
+    ("timeout", "res"),
+    (
+        (None, None),
+        (0, 0),
+        (0.1, 100),
+        (0.001, 1),
+        (1e-15, 1),
+        (0.0005, 1),
+        (0.0001, 1),
+        (1.0015, 1002),
+        (1.000499, 1000),
+        (1.0025, 1002),
+        (3.0005, 3000),
+        (3.456, 3456),
+        (1, 1000),
+        (
+            "foo",
+            ValueError("Timeout must be specified as a number of seconds")
+        ),
+        (
+            [1, 2],
+            TypeError("Timeout must be specified as a number of seconds")
+        )
+    )
+)
+def test_tx_timeout(fake_socket_pair, func, args, extra_idx, timeout, res):
+    address = ("127.0.0.1", 7687)
+    sockets = fake_socket_pair(address)
+    sockets.server.send_message(0x70, {})
+    connection = Bolt4x0(address, sockets.client, 0)
+    func = getattr(connection, func)
+    if isinstance(res, Exception):
+        with pytest.raises(type(res), match=str(res)):
+            func(*args, timeout=timeout)
+    else:
+        func(*args, timeout=timeout)
+        connection.send_all()
+        tag, fields = sockets.server.pop_message()
+        extra = fields[extra_idx]
+        if timeout is None:
+            assert "tx_timeout" not in extra
+        else:
+            assert extra["tx_timeout"] == res

tests/unit/io/test_class_bolt4x1.py

@@ -210,3 +210,56 @@ def test_hint_recv_timeout_seconds_gets_ignored(fake_socket_pair, recv_timeout):
                          PoolConfig.max_connection_lifetime)
     connection.hello()
     sockets.client.settimeout.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    ("func", "args", "extra_idx"),
+    (
+        ("run", ("RETURN 1",), 2),
+        ("begin", (), 0),
+    )
+)
+@pytest.mark.parametrize(
+    ("timeout", "res"),
+    (
+        (None, None),
+        (0, 0),
+        (0.1, 100),
+        (0.001, 1),
+        (1e-15, 1),
+        (0.0005, 1),
+        (0.0001, 1),
+        (1.0015, 1002),
+        (1.000499, 1000),
+        (1.0025, 1002),
+        (3.0005, 3000),
+        (3.456, 3456),
+        (1, 1000),
+        (
+            "foo",
+            ValueError("Timeout must be specified as a number of seconds")
+        ),
+        (
+            [1, 2],
+            TypeError("Timeout must be specified as a number of seconds")
+        )
+    )
+)
+def test_tx_timeout(fake_socket_pair, func, args, extra_idx, timeout, res):
+    address = ("127.0.0.1", 7687)
+    sockets = fake_socket_pair(address)
+    sockets.server.send_message(0x70, {})
+    connection = Bolt4x1(address, sockets.client, 0)
+    func = getattr(connection, func)
+    if isinstance(res, Exception):
+        with pytest.raises(type(res), match=str(res)):
+            func(*args, timeout=timeout)
+    else:
+        func(*args, timeout=timeout)
+        connection.send_all()
+        tag, fields = sockets.server.pop_message()
+        extra = fields[extra_idx]
+        if timeout is None:
+            assert "tx_timeout" not in extra
+        else:
+            assert extra["tx_timeout"] == res