Handle single rpc error responses for batch requests:

- With some clients, when executing a batch request, the client will
  return a single error response for the entire batch. Previously,
  we were assuming that batch requests always received a list of
  responses. This commit updates the batch request logic to handle
  this case.

- Add tests for batch requests with a single error response.

Author: fselmo

newsfragments/3585.bugfix.rst

@@ -0,0 +1 @@
+Handle the case when a single RPC error response is returned for a batch request, instead of always expecting a list of responses.

tests/core/providers/test_async_http_provider.py

@@ -1,4 +1,8 @@
 import pytest
+from unittest.mock import (
+    AsyncMock,
+    patch,
+)
 
 from aiohttp import (
     ClientSession,
@@ -13,6 +17,7 @@
 )
 from web3.exceptions import (
     ProviderConnectionError,
+    Web3RPCError,
 )
 from web3.geth import (
     AsyncGeth,
@@ -107,3 +112,21 @@ def test_get_request_headers(provider):
         headers["User-Agent"] == f"web3.py/{web3py_version}/"
         f"{AsyncHTTPProvider.__module__}.{AsyncHTTPProvider.__qualname__}"
     )
+
+
+@patch(
+    "web3._utils.http_session_manager.HTTPSessionManager.async_make_post_request",
+    new_callable=AsyncMock,
+)
+@pytest.mark.asyncio
+async def test_async_http_empty_batch_response(mock_async_post):
+    mock_async_post.return_value = (
+        b'{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"empty batch"}}'
+    )
+    async_w3 = AsyncWeb3(AsyncHTTPProvider())
+    async with async_w3.batch_requests() as batch:
+        with pytest.raises(Web3RPCError, match="empty batch"):
+            await batch.async_execute()
+
+    # assert that event though there was an error, we have reset the batching state
+    assert not async_w3.provider._is_batching

tests/core/providers/test_async_ipc_provider.py

@@ -25,6 +25,7 @@
 )
 from web3.exceptions import (
     ReadBufferLimitReached,
+    Web3RPCError,
 )
 from web3.providers import (
     AsyncIPCProvider,
@@ -366,3 +367,24 @@ async def test_async_ipc_provider_write_messages_end_with_new_line_delimiter(
 
         request_data = b'{"id": 0, "jsonrpc": "2.0", "method": "method", "params": []}'
         w3.provider._writer.write.assert_called_with(request_data + b"\n")
+
+
+@pytest.mark.asyncio
+async def test_persistent_connection_provider_empty_batch_response(
+    simple_ipc_server,
+    jsonrpc_ipc_pipe_path,
+):
+    async with AsyncWeb3(
+        AsyncIPCProvider(pathlib.Path(jsonrpc_ipc_pipe_path))
+    ) as async_w3:
+        async_w3.provider._reader.readline = AsyncMock()
+        async_w3.provider._reader.readline.return_value = (
+            b'{"jsonrpc": "2.0","id":null,"error": {"code": -32600, "message": '
+            b'"empty batch"}}\n'
+        )
+        async with async_w3.batch_requests() as batch:
+            with pytest.raises(Web3RPCError, match="empty batch"):
+                await batch.async_execute()
+
+        # assert that event though there was an error, we have reset the batching state
+        assert not async_w3.provider._is_batching

tests/core/providers/test_http_provider.py

@@ -1,4 +1,8 @@
 import pytest
+from unittest.mock import (
+    Mock,
+    patch,
+)
 
 from requests import (
     Session,
@@ -16,6 +20,7 @@
 )
 from web3.exceptions import (
     ProviderConnectionError,
+    Web3RPCError,
 )
 from web3.geth import (
     Geth,
@@ -110,3 +115,20 @@ def test_get_request_headers(provider):
         headers["User-Agent"] == f"web3.py/{web3py_version}/"
         f"{HTTPProvider.__module__}.{HTTPProvider.__qualname__}"
     )
+
+
+@patch(
+    "web3._utils.http_session_manager.HTTPSessionManager.make_post_request",
+    new_callable=Mock,
+)
+def test_http_empty_batch_response(mock_post):
+    mock_post.return_value = (
+        b'{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"empty batch"}}'
+    )
+    w3 = Web3(HTTPProvider())
+    with w3.batch_requests() as batch:
+        with pytest.raises(Web3RPCError, match="empty batch"):
+            batch.execute()
+
+    # assert that event though there was an error, we have reset the batching state
+    assert not w3.provider._is_batching

tests/core/providers/test_websocket_provider.py

@@ -27,6 +27,7 @@
 )
 from web3.exceptions import (
     TimeExhausted,
+    Web3RPCError,
 )
 from web3.providers.persistent import (
     WebSocketProvider,
@@ -432,3 +433,24 @@ async def test_listener_task_breaks_out_of_handle_subscriptions_when_cancelled()
     # up the exception.
     with pytest.raises(asyncio.CancelledError):
         await async_w3.subscription_manager.handle_subscriptions(run_forever=True)
+
+
+@pytest.mark.asyncio
+async def test_persistent_connection_provider_empty_batch_response():
+    with patch(
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
+    ):
+        async with AsyncWeb3(WebSocketProvider("ws://mocked")) as async_w3:
+            async_w3.provider._ws.recv = AsyncMock()
+            async_w3.provider._ws.recv.return_value = (
+                b'{"jsonrpc": "2.0","id":null,"error": {"code": -32600, "message": '
+                b'"empty batch"}}\n'
+            )
+            async with async_w3.batch_requests() as batch:
+                with pytest.raises(Web3RPCError, match="empty batch"):
+                    await batch.async_execute()
+
+            # assert that event though there was an error, we have reset the batching
+            # state
+            assert not async_w3.provider._is_batching

web3/manager.py

@@ -436,15 +436,27 @@ def _make_batch_request(
         request_func = provider.batch_request_func(
             cast("Web3", self.w3), cast("MiddlewareOnion", self.middleware_onion)
         )
-        responses = request_func(
+        response = request_func(
             [
                 (method, params)
                 for (method, params), _response_formatters in requests_info
             ]
         )
+
+        if not isinstance(response, list):
+            # Expect a JSON-RPC error response and call _validate_response to raise
+            # the appropriate exception
+            _validate_response(
+                cast(RPCResponse, response),
+                None,
+                is_subscription_response=False,
+                logger=self.logger,
+                params=[],
+            )
+
         formatted_responses = [
-            self._format_batched_response(info, resp)
-            for info, resp in zip(requests_info, responses)
+            self._format_batched_response(info, cast(RPCResponse, resp))
+            for info, resp in zip(requests_info, response)
         ]
         return list(formatted_responses)
 
@@ -465,23 +477,35 @@ async def _async_make_batch_request(
         # since we add items to the batch without awaiting, we unpack the coroutines
         # and await them all here
         unpacked_requests_info = await asyncio.gather(*requests_info)
-        responses = await request_func(
+        response = await request_func(
             [
                 (method, params)
                 for (method, params), _response_formatters in unpacked_requests_info
             ]
         )
 
+        if not isinstance(response, list):
+            # Expect a JSON-RPC error response and call _validate_response to raise
+            # the appropriate exception
+            _validate_response(
+                cast(RPCResponse, response),
+                None,
+                is_subscription_response=False,
+                logger=self.logger,
+                params=[],
+            )
+
+        response = cast(List[RPCResponse], response)
         if isinstance(self.provider, PersistentConnectionProvider):
             # call _process_response for each response in the batch
             return [
                 cast(RPCResponse, await self._process_response(resp))
-                for resp in responses
+                for resp in response
             ]
 
         formatted_responses = [
             self._format_batched_response(info, resp)
-            for info, resp in zip(unpacked_requests_info, responses)
+            for info, resp in zip(unpacked_requests_info, response)
         ]
         return list(formatted_responses)
 
@@ -491,6 +515,13 @@ def _format_batched_response(
         response: RPCResponse,
     ) -> RPCResponse:
         result_formatters, error_formatters, null_result_formatters = requests_info[1]
+        _validate_response(
+            response,
+            error_formatters,
+            is_subscription_response=False,
+            logger=self.logger,
+            params=requests_info[0][1],
+        )
         return apply_result_formatters(
             result_formatters,
             self.formatted_response(

web3/middleware/base.py

@@ -62,15 +62,19 @@ def wrap_make_batch_request(
     ) -> "MakeBatchRequestFn":
         def middleware(
             requests_info: List[Tuple["RPCEndpoint", Any]]
-        ) -> List["RPCResponse"]:
+        ) -> Union[List["RPCResponse"], "RPCResponse"]:
             req_processed = [
                 self.request_processor(method, params)
                 for (method, params) in requests_info
             ]
-            responses = make_batch_request(req_processed)
+            response = make_batch_request(req_processed)
+            if not isinstance(response, list):
+                # RPC errors return only one response with the error object
+                return response
+
             methods, _params = zip(*req_processed)
             formatted_responses = [
-                self.response_processor(m, r) for m, r in zip(methods, responses)
+                self.response_processor(m, r) for m, r in zip(methods, response)
             ]
             return formatted_responses
 
@@ -103,16 +107,20 @@ async def async_wrap_make_batch_request(
     ) -> "AsyncMakeBatchRequestFn":
         async def middleware(
             requests_info: List[Tuple["RPCEndpoint", Any]]
-        ) -> List["RPCResponse"]:
+        ) -> Union[List["RPCResponse"], "RPCResponse"]:
             req_processed = [
                 await self.async_request_processor(method, params)
                 for (method, params) in requests_info
             ]
-            responses = await make_batch_request(req_processed)
+            response = await make_batch_request(req_processed)
+            if not isinstance(response, list):
+                # RPC errors return only one response with the error object
+                return response
+
             methods, _params = zip(*req_processed)
             formatted_responses = [
                 await self.async_response_processor(m, r)
-                for m, r in zip(methods, responses)
+                for m, r in zip(methods, response)
             ]
             return formatted_responses
 

web3/providers/async_base.py

@@ -77,7 +77,8 @@ class AsyncBaseProvider:
 
     _is_batching: bool = False
     _batch_request_func_cache: Tuple[
-        Tuple[Middleware, ...], Callable[..., Coroutine[Any, Any, List[RPCResponse]]]
+        Tuple[Middleware, ...],
+        Callable[..., Coroutine[Any, Any, Union[List[RPCResponse], RPCResponse]]],
     ] = (None, None)
 
     is_async = True
@@ -119,7 +120,7 @@ async def request_func(
 
     async def batch_request_func(
         self, async_w3: "AsyncWeb3", middleware_onion: MiddlewareOnion
-    ) -> Callable[..., Coroutine[Any, Any, List[RPCResponse]]]:
+    ) -> Callable[..., Coroutine[Any, Any, Union[List[RPCResponse], RPCResponse]]]:
         middleware: Tuple[Middleware, ...] = middleware_onion.as_tuple_of_middleware()
 
         cache_key = self._batch_request_func_cache[0]
@@ -141,8 +142,8 @@ async def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
 
     async def make_batch_request(
         self, requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
-        raise NotImplementedError("Only AsyncHTTPProvider supports this method")
+    ) -> Union[List[RPCResponse], RPCResponse]:
+        raise NotImplementedError("Providers must implement this method")
 
     async def is_connected(self, show_traceback: bool = False) -> bool:
         raise NotImplementedError("Providers must implement this method")

web3/providers/base.py

@@ -118,7 +118,7 @@ class JSONBaseProvider(BaseProvider):
 
     _is_batching: bool = False
     _batch_request_func_cache: Tuple[
-        Tuple[Middleware, ...], Callable[..., List[RPCResponse]]
+        Tuple[Middleware, ...], Callable[..., Union[List[RPCResponse], RPCResponse]]
     ] = (None, None)
 
     def __init__(self, **kwargs: Any) -> None:
@@ -168,7 +168,7 @@ def is_connected(self, show_traceback: bool = False) -> bool:
 
     def batch_request_func(
         self, w3: "Web3", middleware_onion: MiddlewareOnion
-    ) -> Callable[..., List[RPCResponse]]:
+    ) -> Callable[..., Union[List[RPCResponse], RPCResponse]]:
         middleware: Tuple[Middleware, ...] = middleware_onion.as_tuple_of_middleware()
 
         cache_key = self._batch_request_func_cache[0]
@@ -199,5 +199,5 @@ def encode_batch_rpc_request(
 
     def make_batch_request(
         self, requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
+    ) -> Union[List[RPCResponse], RPCResponse]:
         raise NotImplementedError("Providers must implement this method")

web3/providers/persistent/request_processor.py

@@ -6,6 +6,7 @@
     Callable,
     Dict,
     Generic,
+    List,
     Optional,
     Tuple,
     TypeVar,
@@ -270,6 +271,15 @@ def append_middleware_response_processor(
 
     # raw response cache
 
+    def _is_batch_response(
+        self, raw_response: Union[List[RPCResponse], RPCResponse]
+    ) -> bool:
+        return isinstance(raw_response, list) or (
+            isinstance(raw_response, dict)
+            and raw_response.get("id") is None
+            and self._provider._is_batching
+        )
+
     async def cache_raw_response(
         self, raw_response: Any, subscription: bool = False
     ) -> None:
@@ -296,7 +306,7 @@ async def cache_raw_response(
                 # otherwise, put it in the subscription response queue so a response
                 # can be yielded by the message stream
                 await self._subscription_response_queue.put(raw_response)
-        elif isinstance(raw_response, list):
+        elif self._is_batch_response(raw_response):
             # Since only one batch should be in the cache at all times, we use a
             # constant cache key for the batch response.
             cache_key = generate_cache_key(BATCH_REQUEST_ID)