handle when a client might return an error unrelated to a request:

Handle the edge case when a client returns an error that is not
related to a request. This was reported from a user having received
a response for a request `id` followed by an error with the same
response `id` but we were no longer looking for that `id` since the
response was already processed. This is on the client, but we
should handle it gracefully within the library without hanging
indefinitely.

- Refactor rpc response validation to ``_utils/validation.py`` and use
  this in `PersistentConnectionProvider` to handle any errors that are
  found in the cache. Refactor applying error and null formatters to
  ``_utils/formatters.py``

- Add a test case that covers a similar scenario to what the user
  reported.

- Add a newsfragment for this change.

Author: fselmo

newsfragments/3623.misc.rst

@@ -0,0 +1 @@
+Raise stray errors a client may send that is not tied to any outstanding requests.

tests/core/providers/test_async_ipc_provider.py

@@ -360,7 +360,7 @@ async def test_async_ipc_provider_write_messages_end_with_new_line_delimiter(
     async with AsyncWeb3(AsyncIPCProvider(pathlib.Path(jsonrpc_ipc_pipe_path))) as w3:
         w3.provider._writer.write = Mock()
         w3.provider._reader.readline = AsyncMock(
-            return_value=b'{"id": 0, "result": {}}\n'
+            return_value=b'{"id": 0, "jsonrpc": "2.0", "result": {}}\n'
         )
 
         await w3.provider.make_request("method", [])

tests/core/providers/test_ipc_provider.py

@@ -191,7 +191,9 @@ def test_web3_auto_gethdev(request_mocker):
 def test_ipc_provider_write_messages_end_with_new_line_delimiter(jsonrpc_ipc_pipe_path):
     provider = IPCProvider(pathlib.Path(jsonrpc_ipc_pipe_path), timeout=3)
     provider._socket.sock = Mock()
-    provider._socket.sock.recv.return_value = b'{"id":1, "result": {}}\n'
+    provider._socket.sock.recv.return_value = (
+        b'{"id":0, "jsonrpc": "2.0", "result": {}}\n'
+    )
 
     provider.make_request("method", [])
 

tests/core/providers/test_websocket_provider.py

@@ -493,3 +493,20 @@ async def test_websocket_provider_use_text_frames(use_text_frames, expected_send
 
     await provider.make_request(RPCEndpoint("eth_getBlockByNumber"), ["latest", False])
     provider._ws.send.assert_called_once_with(expected_send_arg)
+
+
+@pytest.mark.asyncio
+async def test_websocket_provider_raises_errors_from_cache_not_tied_to_a_request():
+    with patch(
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: WebSocketMessageStreamMock(
+            messages=[
+                b'{"id": 0, "jsonrpc": "2.0", "result": "0x0"}\n',
+                b'{"id": null, "jsonrpc": "2.0", "error": {"code": 21, "message": "Request shutdown"}}\n',  # noqa: E501
+            ]
+        ),
+    ):
+        async_w3 = await AsyncWeb3(WebSocketProvider("ws://mocked"))
+        with pytest.raises(Web3RPCError, match="Request shutdown"):
+            await asyncio.sleep(0.1)
+            await async_w3.eth.block_number

web3/_utils/formatters.py

@@ -6,6 +6,7 @@
     Callable,
     Dict,
     Iterable,
+    Optional,
     Tuple,
     TypeVar,
 )
@@ -26,11 +27,15 @@
     compose,
     curry,
     dissoc,
+    pipe,
 )
 
 from web3._utils.decorators import (
     reject_recursive_repeats,
 )
+from web3.types import (
+    RPCResponse,
+)
 
 TReturn = TypeVar("TReturn")
 TValue = TypeVar("TValue")
@@ -131,3 +136,26 @@ def remove_key_if(
         return dissoc(input_dict, key)
     else:
         return input_dict
+
+
+def apply_error_formatters(
+    error_formatters: Callable[..., Any],
+    response: RPCResponse,
+) -> RPCResponse:
+    if error_formatters:
+        formatted_resp = pipe(response, error_formatters)
+        return formatted_resp
+    else:
+        return response
+
+
+def apply_null_result_formatters(
+    null_result_formatters: Callable[..., Any],
+    response: RPCResponse,
+    params: Optional[Any] = None,
+) -> RPCResponse:
+    if null_result_formatters:
+        formatted_resp = pipe(params, null_result_formatters)
+        return formatted_resp
+    else:
+        return response

web3/_utils/validation.py

@@ -1,7 +1,11 @@
 import itertools
+import logging
 from typing import (
     Any,
+    Callable,
     Dict,
+    NoReturn,
+    Optional,
 )
 
 from eth_typing import (
@@ -53,11 +57,22 @@
     length_of_array_type,
     sub_type_of_array_type,
 )
+from web3._utils.formatters import (
+    apply_error_formatters,
+)
 from web3.exceptions import (
+    BadResponseFormat,
     InvalidAddress,
+    MethodUnavailable,
+    RequestTimedOut,
+    TransactionNotFound,
+    Web3RPCError,
     Web3TypeError,
     Web3ValueError,
 )
+from web3.types import (
+    RPCResponse,
+)
 
 
 def _prepare_selector_collision_msg(duplicates: Dict[HexStr, ABIFunction]) -> str:
@@ -211,3 +226,179 @@ def assert_one_val(*args: Any, **kwargs: Any) -> None:
             "Exactly one of the passed values can be specified. "
             f"Instead, values were: {args!r}, {kwargs!r}"
         )
+
+
+# -- RPC Response Validation -- #
+
+KNOWN_REQUEST_TIMEOUT_MESSAGING = {
+    # Note: It's important to be very explicit here and not too broad. We don't want
+    # to accidentally catch a message that is not for a request timeout. In the worst
+    # case, we raise something more generic like `Web3RPCError`. JSON-RPC unfortunately
+    # has not standardized error codes for request timeouts.
+    "request timed out",  # go-ethereum
+}
+METHOD_NOT_FOUND = -32601
+
+
+def _validate_subscription_fields(response: RPCResponse) -> None:
+    params = response["params"]
+    subscription = params["subscription"]
+    if not isinstance(subscription, str) and not len(subscription) == 34:
+        _raise_bad_response_format(
+            response, "eth_subscription 'params' must include a 'subscription' field."
+        )
+
+
+def _raise_bad_response_format(response: RPCResponse, error: str = "") -> None:
+    message = "The response was in an unexpected format and unable to be parsed."
+    raw_response = f"The raw response is: {response}"
+
+    if error is not None and error != "":
+        error = error[:-1] if error.endswith(".") else error
+        message = f"{message} {error}. {raw_response}"
+    else:
+        message = f"{message} {raw_response}"
+
+    raise BadResponseFormat(message)
+
+
+def raise_error_for_batch_response(
+    response: RPCResponse,
+    logger: Optional[logging.Logger] = None,
+) -> NoReturn:
+    error = response.get("error")
+    if error is None:
+        _raise_bad_response_format(
+            response,
+            "Batch response must be formatted as a list of responses or "
+            "as a single JSON-RPC error response.",
+        )
+    validate_rpc_response_and_raise_if_error(
+        response,
+        None,
+        is_subscription_response=False,
+        logger=logger,
+        params=[],
+    )
+    # This should not be reached, but if it is, raise a generic `BadResponseFormat`
+    raise BadResponseFormat(
+        "Batch response was in an unexpected format and unable to be parsed."
+    )
+
+
+def validate_rpc_response_and_raise_if_error(
+    response: RPCResponse,
+    error_formatters: Optional[Callable[..., Any]],
+    is_subscription_response: bool = False,
+    logger: Optional[logging.Logger] = None,
+    params: Optional[Any] = None,
+) -> None:
+    if "jsonrpc" not in response or response["jsonrpc"] != "2.0":
+        _raise_bad_response_format(
+            response, 'The "jsonrpc" field must be present with a value of "2.0".'
+        )
+
+    response_id = response.get("id")
+    if "id" in response:
+        int_error_msg = (
+            '"id" must be an integer or a string representation of an integer.'
+        )
+        if response_id is None and "error" in response:
+            # errors can sometimes have null `id`, according to the JSON-RPC spec
+            pass
+        elif not isinstance(response_id, (str, int)):
+            _raise_bad_response_format(response, int_error_msg)
+        elif isinstance(response_id, str):
+            try:
+                int(response_id)
+            except ValueError:
+                _raise_bad_response_format(response, int_error_msg)
+    elif is_subscription_response:
+        # if `id` is not present, this must be a subscription response
+        _validate_subscription_fields(response)
+    else:
+        _raise_bad_response_format(
+            response,
+            'Response must include an "id" field or be formatted as an '
+            "`eth_subscription` response.",
+        )
+
+    if all(key in response for key in {"error", "result"}):
+        _raise_bad_response_format(
+            response, 'Response cannot include both "error" and "result".'
+        )
+    elif (
+        not any(key in response for key in {"error", "result"})
+        and not is_subscription_response
+    ):
+        _raise_bad_response_format(
+            response, 'Response must include either "error" or "result".'
+        )
+    elif "error" in response:
+        web3_rpc_error: Optional[Web3RPCError] = None
+        error = response["error"]
+
+        # raise the error when the value is a string
+        if error is None or not isinstance(error, dict):
+            _raise_bad_response_format(
+                response,
+                'response["error"] must be a valid object as defined by the '
+                "JSON-RPC 2.0 specification.",
+            )
+
+        # errors must include a message
+        error_message = error.get("message")
+        if not isinstance(error_message, str):
+            _raise_bad_response_format(
+                response, 'error["message"] is required and must be a string value.'
+            )
+        elif error_message == "transaction not found":
+            transaction_hash = params[0]
+            web3_rpc_error = TransactionNotFound(
+                repr(error),
+                rpc_response=response,
+                user_message=(f"Transaction with hash {transaction_hash!r} not found."),
+            )
+
+        # errors must include an integer code
+        code = error.get("code")
+        if not isinstance(code, int):
+            _raise_bad_response_format(
+                response, 'error["code"] is required and must be an integer value.'
+            )
+        elif code == METHOD_NOT_FOUND:
+            web3_rpc_error = MethodUnavailable(
+                repr(error),
+                rpc_response=response,
+                user_message=(
+                    "This method is not available. Check your node provider or your "
+                    "client's API docs to see what methods are supported and / or "
+                    "currently enabled."
+                ),
+            )
+        elif any(
+            # parse specific timeout messages
+            timeout_str in error_message.lower()
+            for timeout_str in KNOWN_REQUEST_TIMEOUT_MESSAGING
+        ):
+            web3_rpc_error = RequestTimedOut(
+                repr(error),
+                rpc_response=response,
+                user_message=(
+                    "The request timed out. Check the connection to your node and "
+                    "try again."
+                ),
+            )
+
+        if web3_rpc_error is None:
+            # if no condition was met above, raise a more generic `Web3RPCError`
+            web3_rpc_error = Web3RPCError(repr(error), rpc_response=response)
+
+        response = apply_error_formatters(error_formatters, response)
+        if logger is not None:
+            logger.debug(f"RPC error response: {response}")
+
+        raise web3_rpc_error
+
+    elif "result" not in response and not is_subscription_response:
+        _raise_bad_response_format(response)