Add timeout around request / response id matching while loop:

- Use the call timeout for the provider to set up a reasonable escape time for the while look that tries to match the response `id` with the request `id`. All providers should technically return matching ids in the response but we've run into issues with this in the past and this prevents an infinite loop, handling it somewhat gracefully.
- Make sure that the ``WebsocketProviderV2`` sets a default timeout.

Author: fselmo

tests/core/providers/test_wsv2_provider.py

@@ -6,6 +6,9 @@
     to_bytes,
 )
 
+from web3.exceptions import (
+    TimeExhausted,
+)
 from web3.providers.websocket import (
     WebsocketProviderV2,
 )
@@ -14,24 +17,28 @@
 )
 
 
+def _mock_ws(provider):
+    # move to top of file when python 3.7 is no longer supported in web3.py
+    from unittest.mock import (
+        AsyncMock,
+    )
+
+    provider._ws = AsyncMock()
+
+
 @pytest.mark.asyncio
 @pytest.mark.skipif(
     # TODO: remove when python 3.7 is no longer supported in web3.py
     #  python 3.7 is already sunset so this feels like a reasonable tradeoff
     sys.version_info < (3, 8),
-    reason="Mock args behave differently in python 3.7 but test should still pass.",
+    reason="Uses AsyncMock, not supported by python 3.7",
 )
 async def test_async_make_request_caches_all_undesired_responses_and_returns_desired():
-    # move to top of file when python 3.7 is no longer supported in web3.py
-    from unittest.mock import (
-        AsyncMock,
-    )
-
     provider = WebsocketProviderV2("ws://mocked")
 
     method_under_test = provider.make_request
 
-    provider._ws = AsyncMock()
+    _mock_ws(provider)
     undesired_responses_count = 10
     ws_recv_responses = [
         to_bytes(
@@ -90,3 +97,25 @@ async def test_async_make_request_returns_cached_response_with_no_recv_if_cached
 
     assert len(provider._request_processor._raw_response_cache) == 0
     assert not provider._ws.recv.called  # type: ignore
+
+
+@pytest.mark.asyncio
+@pytest.mark.skipif(
+    # TODO: remove when python 3.7 is no longer supported in web3.py
+    #  python 3.7 is already sunset so this feels like a reasonable tradeoff
+    sys.version_info < (3, 8),
+    reason="Uses AsyncMock, not supported by python 3.7",
+)
+async def test_async_make_request_times_out_of_while_loop_looking_for_response():
+    provider = WebsocketProviderV2("ws://mocked", call_timeout=0.1)
+
+    method_under_test = provider.make_request
+
+    _mock_ws(provider)
+    provider._ws.recv.side_effect = lambda *args, **kwargs: b'{"jsonrpc": "2.0"}'
+
+    with pytest.raises(
+        TimeExhausted,
+        match="Timed out waiting for response to request id `0` after 0.1 seconds.",
+    ):
+        await method_under_test(RPCEndpoint("some_method"), ["desired_params"])

web3/providers/persistent.py

@@ -34,7 +34,7 @@ def __init__(
         self,
         endpoint_uri: str,
         request_cache_size: int = 100,
-        call_timeout: int = DEFAULT_PERSISTENT_CONNECTION_TIMEOUT,
+        call_timeout: float = DEFAULT_PERSISTENT_CONNECTION_TIMEOUT,
     ) -> None:
         super().__init__()
         self.endpoint_uri = endpoint_uri

web3/providers/websocket/websocket_v2.py

@@ -27,9 +27,11 @@
 )
 from web3.exceptions import (
     ProviderConnectionError,
+    TimeExhausted,
     Web3ValidationError,
 )
 from web3.providers.persistent import (
+    DEFAULT_PERSISTENT_CONNECTION_TIMEOUT,
     PersistentConnectionProvider,
 )
 from web3.types import (
@@ -64,7 +66,7 @@ def __init__(
         self,
         endpoint_uri: Optional[Union[URI, str]] = None,
         websocket_kwargs: Optional[Dict[str, Any]] = None,
-        call_timeout: Optional[int] = None,
+        call_timeout: Optional[float] = DEFAULT_PERSISTENT_CONNECTION_TIMEOUT,
     ) -> None:
         self.endpoint_uri = URI(endpoint_uri)
         if self.endpoint_uri is None:
@@ -167,21 +169,40 @@ async def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
         return response
 
     async def _get_response_for_request_id(self, request_id: RPCId) -> RPCResponse:
-        response_id = None
-        response = None
-        while response_id != request_id:
-            response = await self._ws_recv()
-            response_id = response.get("id")
+        async def _match_response_id_to_request_id() -> RPCResponse:
+            response_id = None
+            response = None
+            while response_id != request_id:
+                response = await self._ws_recv()
+                response_id = response.get("id")
+
+                if response_id == request_id:
+                    break
+                else:
+                    # cache all responses that are not the desired response
+                    await self._request_processor.cache_raw_response(
+                        response,
+                    )
+                    await asyncio.sleep(0.1)
+
+            return response
 
-            if response_id == request_id:
-                break
-            else:
-                # cache all responses that are not the desired response
-                await self._request_processor.cache_raw_response(
-                    response,
-                )
-
-        return response
+        try:
+            # Enters a while loop, looking for a response id match to the request id.
+            # If the provider does not give responses with matching ids, this will
+            # hang forever. The JSON-RPC spec requires that providers respond with
+            # the same id that was sent in the request, but we need to handle these
+            # "bad" cases somewhat gracefully.
+            return await asyncio.wait_for(
+                _match_response_id_to_request_id(), self.call_timeout
+            )
+        except asyncio.TimeoutError:
+            raise TimeExhausted(
+                f"Timed out waiting for response to request id `{request_id}` after "
+                f"{self.call_timeout} seconds. This is likely due to the provider not "
+                "issuing a response with the same id that was sent in the request, "
+                "which is required by the JSON-RPC spec."
+            )
 
     async def _ws_recv(self) -> RPCResponse:
         return json.loads(