Repository: ethereum/web3.py
Pull request #3623 (2025-02-20 16:32:36)

handle when a client returns the response id more than once (JSON-RPC spec error)
### What was wrong?

- Handle an edge case when a client sends an error response that isn't tied to any outstanding requests. A user reported a client returning the same response `id` more than once, the second with an error. This does not adhere to the JSON-RPC spec but was reported by a user on Discord as leaving the library hanging indefinitely, so we should try to fail hard under these edge cases.

### How was it fixed?

- After every request, if the request / response cache still has anything in it, check if there are any errors and if they are not tied to outstanding requests, raise them.
- Add a simple test to check that this logic is in place.
- Move the listener task exception check above looking for a response with `id` so we can fail hard and fast if an exception has already been raised in the listener.
- Some refactoring of formatters and validation methods out of `manager.py` and into `_utils.formatters.py` and `_utils/validation.py`, respectively.

### Todo:

- [x] Clean up commit history
- [x] Add entry to the [release notes](https://github.com/ethereum/web3.py/blob/main/newsfragments/README.md)

#### Cute Animal Picture

```
      /^-----^\
      V  o o  V
       |  Y  |
        \ Q /
       / - \
       |    \
       |     \     )
       || (___\==== 
```

fselmo: Actually, this is a terrible idea as `id`s don't always come in order. I'm going to re-think this but since it's an issue on the client side I'm not 100% sure we need a great answer here. Maybe looking for stray `error` messages might be the best way to go about this. Disregard the review for now.
fselmo: @pacrob @kclowes ok, this is much better. We can look for stray error responses in the cache that have no outstanding requests and raise if any are present. This still allows a single request with `id` to retrieve its response by `id` and raise more appropriately in the `manager.py`. But if we find any stragglers in the cache we still account for them.

Again, not really something that's expected, but better than leaving the library hanging like in the case that was reported.

Patch:
diff --git a/newsfragments/3623.misc.rst b/newsfragments/3623.misc.rst
new file mode 100644
index 00000000..dbb5aa53
--- /dev/null
+++ b/newsfragments/3623.misc.rst
@@ -0,0 +1,1 @@
+Raise stray errors a client may send that is not tied to any outstanding requests.
diff --git a/web3/_utils/formatters.py b/web3/_utils/formatters.py
index 7d06f690..c01fcb8d 100644
--- a/web3/_utils/formatters.py
+++ b/web3/_utils/formatters.py
@@ -6,6 +6,7 @@ from typing import (
     Callable,
     Dict,
     Iterable,
+    Optional,
     Tuple,
     TypeVar,
 )
@@ -26,11 +27,15 @@ from eth_utils.toolz import (
     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
diff --git a/web3/_utils/validation.py b/web3/_utils/validation.py
index 4241cffa..8cd2692e 100644
--- a/web3/_utils/validation.py
+++ b/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 @@ from web3._utils.abi import (
     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)
diff --git a/web3/manager.py b/web3/manager.py
index 69e9beb4..0ee3e606 100644
--- a/web3/manager.py
+++ b/web3/manager.py
@@ -8,7 +8,6 @@ from typing import (
     Coroutine,
     Dict,
     List,
-    NoReturn,
     Optional,
     Sequence,
     Tuple,
@@ -32,17 +31,19 @@ from web3._utils.caching import (
 from web3._utils.compat import (
     Self,
 )
+from web3._utils.formatters import (
+    apply_null_result_formatters,
+)
+from web3._utils.validation import (
+    raise_error_for_batch_response,
+    validate_rpc_response_and_raise_if_error,
+)
 from web3.datastructures import (
     NamedElementOnion,
 )
 from web3.exceptions import (
-    BadResponseFormat,
-    MethodUnavailable,
     ProviderConnectionError,
-    RequestTimedOut,
     TaskNotRunning,
-    TransactionNotFound,
-    Web3RPCError,
     Web3TypeError,
 )
 from web3.method import (
@@ -95,200 +96,6 @@ if TYPE_CHECKING:
 
 
 NULL_RESPONSES = [None, HexBytes("0x"), "0x"]
-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 _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 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
-
-
-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 _validate_response(
-    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)
-        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)
-
-
-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_response(
-        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."
-    )
 
 
 class RequestManager:
@@ -388,7 +195,7 @@ class RequestManager:
             and response["params"].get("result") is not None
         )
 
-        _validate_response(
+        validate_rpc_response_and_raise_if_error(
             response,
             error_formatters,
             is_subscription_response=is_subscription_response,
@@ -479,7 +286,7 @@ class RequestManager:
             return list(formatted_responses)
         else:
             # expect a single response with an error
-            _raise_error_for_batch_response(response, self.logger)
+            raise_error_for_batch_response(response, self.logger)
 
     async def _async_make_batch_request(
         self,
@@ -522,7 +329,7 @@ class RequestManager:
             return list(formatted_responses)
         else:
             # expect a single response with an error
-            _raise_error_for_batch_response(response, self.logger)
+            raise_error_for_batch_response(response, self.logger)
 
     def _format_batched_response(
         self,
@@ -530,7 +337,7 @@ class RequestManager:
         response: RPCResponse,
     ) -> RPCResponse:
         result_formatters, error_formatters, null_result_formatters = requests_info[1]
-        _validate_response(
+        validate_rpc_response_and_raise_if_error(
             response,
             error_formatters,
             is_subscription_response=False,
diff --git a/web3/providers/persistent/persistent.py b/web3/providers/persistent/persistent.py
index ff43830f..bcf55de1 100644
--- a/web3/providers/persistent/persistent.py
+++ b/web3/providers/persistent/persistent.py
@@ -33,6 +33,9 @@ from web3._utils.caching.caching_utils import (
     async_handle_recv_caching,
     async_handle_send_caching,
 )
+from web3._utils.validation import (
+    validate_rpc_response_and_raise_if_error,
+)
 from web3.exceptions import (
     PersistentConnectionClosedOK,
     ProviderConnectionError,
@@ -302,6 +305,27 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
             TaskNotRunning(message_listener_task)
         )
 
+    def _raise_stray_errors_from_cache(self) -> None:
+        """
+        Check the request response cache for any errors not tied to current requests
+        and raise them if found.
+        """
+        if not self._is_batching:
+            for (
+                response
+            ) in self._request_processor._request_response_cache._data.values():
+                request = (
+                    self._request_processor._request_information_cache.get_cache_entry(
+                        generate_cache_key(response["id"])
+                    )
+                )
+                if "error" in response and request is None:
+                    # if we find an error response in the cache without a corresponding
+                    # request, raise the error
+                    validate_rpc_response_and_raise_if_error(
+                        response, None, logger=self.logger
+                    )
+
     async def _message_listener(self) -> None:
         self.logger.info(
             f"{self.__class__.__qualname__} listener background task started. Storing "
@@ -327,6 +351,7 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
                 await self._request_processor.cache_raw_response(
                     response, subscription=subscription
                 )
+                self._raise_stray_errors_from_cache()
             except PersistentConnectionClosedOK as e:
                 self.logger.info(
                     "Message listener background task has ended gracefully: "
@@ -376,6 +401,10 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
             request_cache_key = generate_cache_key(request_id)
 
             while True:
+                # check if an exception was recorded in the listener task and raise
+                # it in the main loop if so
+                self._handle_listener_task_exceptions()
+
                 if request_cache_key in self._request_processor._request_response_cache:
                     self.logger.debug(
                         f"Popping response for id {request_id} from cache."
@@ -385,9 +414,6 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
                     )
                     return popped_response
                 else:
-                    # check if an exception was recorded in the listener task and raise
-                    # it in the main loop if so
-                    self._handle_listener_task_exceptions()
                     await asyncio.sleep(0)
 
         try:


Test patch:
diff --git a/tests/core/providers/test_async_ipc_provider.py b/tests/core/providers/test_async_ipc_provider.py
index 452dcb6b..2d7ad90b 100644
--- a/tests/core/providers/test_async_ipc_provider.py
+++ b/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", [])
diff --git a/tests/core/providers/test_ipc_provider.py b/tests/core/providers/test_ipc_provider.py
index 6687780f..77b62370 100644
--- a/tests/core/providers/test_ipc_provider.py
+++ b/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", [])
 
diff --git a/tests/core/providers/test_websocket_provider.py b/tests/core/providers/test_websocket_provider.py
index bf18d10d..b8478bf9 100644
--- a/tests/core/providers/test_websocket_provider.py
+++ b/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