Propagate errors across all results in a transaction (#973)

* Propagate errors across all results in a transaction

A new error `ResultFailedError` is introduced. It will be raised when using
a `Result` object after the result or another result in the same transaction
has failed.

User code would only ever run into this situation when catch exceptions and
deciding to ignore them. Now, an error will be raised instead of undefined
behavior. The undefined behavior before this fix could be (among other things)
protocol violations, incomplete summary data, and hard to interpret errors.

* Fix missing space in error message

* Add unit tests for transaction error propagation

Author: robsdedude

docs/source/api.rst

@@ -1902,6 +1902,8 @@ Client-side errors
 
   * :class:`neo4j.exceptions.ResultError`
 
+    * :class:`neo4j.exceptions.ResultFailedError`
+
     * :class:`neo4j.exceptions.ResultConsumedError`
 
     * :class:`neo4j.exceptions.ResultNotSingleError`
@@ -1946,6 +1948,9 @@ Client-side errors
     :show-inheritance:
     :members: result
 
+.. autoexception:: neo4j.exceptions.ResultFailedError()
+    :show-inheritance:
+
 .. autoexception:: neo4j.exceptions.ResultConsumedError()
     :show-inheritance:
 

src/neo4j/_async/work/result.py

@@ -38,6 +38,7 @@
 )
 from ...exceptions import (
     ResultConsumedError,
+    ResultFailedError,
     ResultNotSingleError,
 )
 from ...time import (
@@ -57,6 +58,10 @@
 _TResultKey = t.Union[int, str]
 
 
+_RESULT_FAILED_ERROR = (
+    "The result has failed. Either this result or another result in the same "
+    "transaction has encountered an error."
+)
 _RESULT_OUT_OF_SCOPE_ERROR = (
     "The result is out of scope. The associated transaction "
     "has been closed. Results can only be used while the "
@@ -76,8 +81,11 @@ class AsyncResult:
     """
 
     def __init__(self, connection, fetch_size, on_closed, on_error):
-        self._connection = ConnectionErrorHandler(connection, on_error)
+        self._connection = ConnectionErrorHandler(
+            connection, self._connection_error_handler
+        )
         self._hydration_scope = connection.new_hydration_scope()
+        self._on_error = on_error
         self._on_closed = on_closed
         self._metadata = None
         self._keys = None
@@ -101,6 +109,13 @@ def __init__(self, connection, fetch_size, on_closed, on_error):
         self._consumed = False
         # the result has been closed as a result of closing the transaction
         self._out_of_scope = False
+        # exception shared across all results of a transaction
+        self._exception = None
+
+    async def _connection_error_handler(self, exc):
+        self._exception = exc
+        self._attached = False
+        await AsyncUtil.callback(self._on_error, exc)
 
     @property
     def _qid(self):
@@ -257,6 +272,9 @@ async def __aiter__(self) -> t.AsyncIterator[Record]:
                 await self._connection.send_all()
 
         self._exhausted = True
+        if self._exception is not None:
+            raise ResultFailedError(self, _RESULT_FAILED_ERROR) \
+                from self._exception
         if self._out_of_scope:
             raise ResultConsumedError(self, _RESULT_OUT_OF_SCOPE_ERROR)
         if self._consumed:
@@ -346,6 +364,11 @@ async def _tx_end(self):
         await self._exhaust()
         self._out_of_scope = True
 
+    def _tx_failure(self, exc):
+        # Handle failure of the associated transaction.
+        self._attached = False
+        self._exception = exc
+
     async def consume(self) -> ResultSummary:
         """Consume the remainder of this result and return a :class:`neo4j.ResultSummary`.
 

src/neo4j/_async/work/transaction.py

@@ -92,6 +92,8 @@ async def _result_on_closed_handler(self):
 
     async def _error_handler(self, exc):
         self._last_error = exc
+        for result in self._results:
+            result._tx_failure(exc)
         if isinstance(exc, asyncio.CancelledError):
             self._cancel()
             return

src/neo4j/_sync/work/result.py

@@ -38,6 +38,7 @@
 )
 from ...exceptions import (
     ResultConsumedError,
+    ResultFailedError,
     ResultNotSingleError,
 )
 from ...time import (
@@ -57,6 +58,10 @@
 _TResultKey = t.Union[int, str]
 
 
+_RESULT_FAILED_ERROR = (
+    "The result has failed. Either this result or another result in the same "
+    "transaction has encountered an error."
+)
 _RESULT_OUT_OF_SCOPE_ERROR = (
     "The result is out of scope. The associated transaction "
     "has been closed. Results can only be used while the "
@@ -76,8 +81,11 @@ class Result:
     """
 
     def __init__(self, connection, fetch_size, on_closed, on_error):
-        self._connection = ConnectionErrorHandler(connection, on_error)
+        self._connection = ConnectionErrorHandler(
+            connection, self._connection_error_handler
+        )
         self._hydration_scope = connection.new_hydration_scope()
+        self._on_error = on_error
         self._on_closed = on_closed
         self._metadata = None
         self._keys = None
@@ -101,6 +109,13 @@ def __init__(self, connection, fetch_size, on_closed, on_error):
         self._consumed = False
         # the result has been closed as a result of closing the transaction
         self._out_of_scope = False
+        # exception shared across all results of a transaction
+        self._exception = None
+
+    def _connection_error_handler(self, exc):
+        self._exception = exc
+        self._attached = False
+        Util.callback(self._on_error, exc)
 
     @property
     def _qid(self):
@@ -257,6 +272,9 @@ def __iter__(self) -> t.Iterator[Record]:
                 self._connection.send_all()
 
         self._exhausted = True
+        if self._exception is not None:
+            raise ResultFailedError(self, _RESULT_FAILED_ERROR) \
+                from self._exception
         if self._out_of_scope:
             raise ResultConsumedError(self, _RESULT_OUT_OF_SCOPE_ERROR)
         if self._consumed:
@@ -346,6 +364,11 @@ def _tx_end(self):
         self._exhaust()
         self._out_of_scope = True
 
+    def _tx_failure(self, exc):
+        # Handle failure of the associated transaction.
+        self._attached = False
+        self._exception = exc
+
     def consume(self) -> ResultSummary:
         """Consume the remainder of this result and return a :class:`neo4j.ResultSummary`.
 

src/neo4j/_sync/work/transaction.py

@@ -92,6 +92,8 @@ def _result_on_closed_handler(self):
 
     def _error_handler(self, exc):
         self._last_error = exc
+        for result in self._results:
+            result._tx_failure(exc)
         if isinstance(exc, asyncio.CancelledError):
             self._cancel()
             return

src/neo4j/exceptions.py

@@ -40,6 +40,7 @@
   + TransactionError
     + TransactionNestingError
   + ResultError
+    + ResultFailedError
     + ResultConsumedError
     + ResultNotSingleError
   + BrokenRecordError
@@ -464,6 +465,17 @@ def __init__(self, result_, *args, **kwargs):
         self.result = result_
 
 
+# DriverError > ResultError > ResultFailedError
+class ResultFailedError(ResultError):
+    """Raised when trying to access records of a failed result.
+
+    A :class:`.Result` will be considered failed if
+     * itself encountered an error while fetching records
+     * another result within the same transaction encountered an error while
+       fetching records
+    """
+
+
 # DriverError > ResultError > ResultConsumedError
 class ResultConsumedError(ResultError):
     """Raised when trying to access records of a consumed result."""

testkitbackend/_async/requests.py

@@ -401,7 +401,7 @@ async def ExecuteQuery(backend, data):
 
 def resolution_func(backend, custom_resolver=False, custom_dns_resolver=False):
     # This solution (putting custom resolution together with DNS resolution
-    # into one function only works because the Python driver calls the custom
+    # into one function) only works because the Python driver calls the custom
     # resolver function for every connection, which is not true for all
     # drivers. Properly exposing a way to change the DNS lookup behavior is not
     # possible without changing the driver's code.

testkitbackend/_sync/requests.py

@@ -401,7 +401,7 @@ def ExecuteQuery(backend, data):
 
 def resolution_func(backend, custom_resolver=False, custom_dns_resolver=False):
     # This solution (putting custom resolution together with DNS resolution
-    # into one function only works because the Python driver calls the custom
+    # into one function) only works because the Python driver calls the custom
     # resolver function for every connection, which is not true for all
     # drivers. Properly exposing a way to change the DNS lookup behavior is not
     # possible without changing the driver's code.

testkitbackend/test_config.json

@@ -13,13 +13,7 @@
     "'neo4j.datatypes.test_temporal_types.TestDataTypes.test_should_echo_all_timezone_ids'":
       "test_subtest_skips.dt_conversion",
     "'neo4j.datatypes.test_temporal_types.TestDataTypes.test_date_time_cypher_created_tz_id'":
-      "test_subtest_skips.tz_id",
-    "'stub.tx_run.test_tx_run.TestTxRun.test_should_prevent_discard_after_tx_termination_on_run'":
-      "Fixme: transactions don't prevent further actions after failure.",
-    "'stub.tx_run.test_tx_run.TestTxRun.test_should_prevent_pull_after_tx_termination_on_pull'":
-      "Fixme: transactions don't prevent further actions after failure.",
-    "'stub.tx_run.test_tx_run.TestTxRun.test_should_prevent_pull_after_tx_termination_on_run'":
-      "Fixme: transactions don't prevent further actions after failure."
+      "test_subtest_skips.tz_id"
   },
   "features": {
     "Feature:API:BookmarkManager": true,

tests/unit/async_/fixtures/fake_connection.py

@@ -24,6 +24,7 @@
 from neo4j._async.io import AsyncBolt
 from neo4j._deadline import Deadline
 from neo4j.auth_management import AsyncAuthManager
+from neo4j.exceptions import Neo4jError
 
 
 __all__ = [
@@ -154,10 +155,12 @@ def set_script(self, callbacks):
                 [
                     ("run", {"on_success": ({},), "on_summary": None}),
                     ("pull", {
+                        "on_records": ([some_record],),
                         "on_success": None,
                         "on_summary": None,
-                        "on_records":
                     })
+                    # use any exception to throw it instead of calling handlers
+                    ("commit", RuntimeError("oh no!"))
                 ]
                 ```
                 Note that arguments can be `None`. In this case, ScriptedConnection
@@ -180,6 +183,9 @@ def func(*args, **kwargs):
                     self._script_pos += 1
 
                     async def callback():
+                        if isinstance(scripted_callbacks, BaseException):
+                            raise scripted_callbacks
+                        error = None
                         for cb_name, default_cb_args in (
                             ("on_ignored", ({},)),
                             ("on_failure", ({},)),
@@ -197,10 +203,14 @@ async def callback():
                             if cb_args is None:
                                 cb_args = default_cb_args
                             res = cb(*cb_args)
+                            if cb_name == "on_failure":
+                                error = Neo4jError.hydrate(**cb_args[0])
                             try:
                                 await res  # maybe the callback is async
                             except TypeError:
                                 pass  # or maybe it wasn't ;)
+                        if error is not None:
+                            raise error
 
                     self.callbacks.append(callback)
 

tests/unit/async_/work/test_transaction.py

@@ -17,7 +17,6 @@
 
 
 from unittest.mock import MagicMock
-from uuid import uuid4
 
 import pytest
 
@@ -26,6 +25,11 @@
     NotificationMinimumSeverity,
     Query,
 )
+from neo4j.exceptions import (
+    ClientError,
+    ResultFailedError,
+    ServiceUnavailable,
+)
 
 from ...._async_compat import mark_async_test
 
@@ -275,3 +279,51 @@ async def test_transaction_begin_pipelining(
         expected_calls.append(("send_all",))
         expected_calls.append(("fetch_all",))
     assert async_fake_connection.method_calls == expected_calls
+
+
+@pytest.mark.parametrize("error", ("server", "connection"))
+@mark_async_test
+async def test_server_error_propagates(async_scripted_connection, error):
+    connection = async_scripted_connection
+    script = [
+        # res 1
+        ("run", {"on_success": ({"fields": ["n"]},), "on_summary": None}),
+        ("pull", {"on_records": ([[1], [2]],),
+                  "on_success": ({"has_more": True},)}),
+        # res 2
+        ("run", {"on_success": ({"fields": ["n"]},), "on_summary": None}),
+        ("pull", {"on_records": ([[1], [2]],),
+                  "on_success": ({"has_more": True},)}),
+    ]
+    if error == "server":
+        script.append(
+            ("pull", {"on_failure": ({"code": "Neo.ClientError.Made.Up"},),
+                      "on_summary": None})
+        )
+        expected_error = ClientError
+    elif error == "connection":
+        script.append(("pull", ServiceUnavailable()))
+        expected_error = ServiceUnavailable
+    else:
+        raise ValueError(f"Unknown error type {error}")
+    connection.set_script(script)
+
+    tx = AsyncTransaction(
+        connection, 2, lambda *args, **kwargs: None,
+        lambda *args, **kwargs: None, lambda *args, **kwargs: None
+    )
+    res1 = await tx.run("UNWIND range(1, 1000) AS n RETURN n")
+    assert await res1.__anext__() == {"n": 1}
+
+    res2 = await tx.run("RETURN 'causes error later'")
+    assert await res2.fetch(2) == [{"n": 1}, {"n": 2}]
+    with pytest.raises(expected_error) as exc1:
+        await res2.__anext__()
+
+    # can finish the buffer
+    assert await res1.fetch(1) == [{"n": 2}]
+    # then fails because the connection was broken by res2
+    with pytest.raises(ResultFailedError) as exc2:
+        await res1.__anext__()
+
+    assert exc1.value is exc2.value.__cause__

tests/unit/sync/fixtures/fake_connection.py

@@ -24,6 +24,7 @@
 from neo4j._deadline import Deadline
 from neo4j._sync.io import Bolt
 from neo4j.auth_management import AuthManager
+from neo4j.exceptions import Neo4jError
 
 
 __all__ = [
@@ -154,10 +155,12 @@ def set_script(self, callbacks):
                 [
                     ("run", {"on_success": ({},), "on_summary": None}),
                     ("pull", {
+                        "on_records": ([some_record],),
                         "on_success": None,
                         "on_summary": None,
-                        "on_records":
                     })
+                    # use any exception to throw it instead of calling handlers
+                    ("commit", RuntimeError("oh no!"))
                 ]
                 ```
                 Note that arguments can be `None`. In this case, ScriptedConnection
@@ -180,6 +183,9 @@ def func(*args, **kwargs):
                     self._script_pos += 1
 
                     def callback():
+                        if isinstance(scripted_callbacks, BaseException):
+                            raise scripted_callbacks
+                        error = None
                         for cb_name, default_cb_args in (
                             ("on_ignored", ({},)),
                             ("on_failure", ({},)),
@@ -197,10 +203,14 @@ def callback():
                             if cb_args is None:
                                 cb_args = default_cb_args
                             res = cb(*cb_args)
+                            if cb_name == "on_failure":
+                                error = Neo4jError.hydrate(**cb_args[0])
                             try:
                                 res  # maybe the callback is async
                             except TypeError:
                                 pass  # or maybe it wasn't ;)
+                        if error is not None:
+                            raise error
 
                     self.callbacks.append(callback)