Repository: ethereum/web3.py

Issue:
Excessive waits when listening to events through a websocket
* Version: 6.12.0
* Python: 3.12
* OS: osx/linux/win
* `pip freeze` output

```
aiohttp==3.9.1
aiosignal==1.3.1
annotated-types==0.6.0
anyio==3.7.1
asyncpg==0.29.0
attrs==23.1.0
beautifulsoup4==4.12.2
bitarray==2.8.5
cachetools==5.3.2
certifi==2023.11.17
charset-normalizer==3.3.2
click==8.1.7
cytoolz==0.12.2
eth-abi==4.2.1
eth-account==0.10.0
eth-hash==0.5.2
eth-keyfile==0.7.0
eth-keys==0.4.0
eth-rlp==1.0.0
eth-typing==3.5.2
eth-utils==2.3.1
fastapi==0.104.1
frozenlist==1.4.0
h11==0.14.0
hexbytes==0.3.1
idna==3.6
jsonschema==4.20.0
jsonschema-specifications==2023.11.2
lru-dict==1.2.0
multidict==6.0.4
parsimonious==0.9.0
pipdeptree==2.13.1
proto-plus==1.23.0
protobuf==4.25.1
pyasn1==0.5.1
pyasn1-modules==0.3.0
pycryptodome==3.19.0
pydantic==2.5.2
pydantic-settings==2.0.3
pydantic_core==2.14.5
python-dotenv==1.0.0
pyunormalize==15.1.0
PyYAML==6.0.1
referencing==0.32.0
regex==2023.10.3
requests==2.31.0
rlp==4.0.0
rpds-py==0.13.2
rsa==4.9
setuptools==69.0.2
six==1.16.0
sniffio==1.3.0
soupsieve==2.5
starlette==0.27.0
toolz==0.12.0
typing_extensions==4.9.0
urllib3==2.1.0
uvicorn==0.23.2
web3==6.12.0
websockets==12.0
yarl==1.9.4
```


### What was wrong?

When user does `listen_to_websocket()`, web3.py ends up doing:
```
response = await self._provider._ws_recv(timeout=0.5)
```

This means that web3.py will wait 0.5 secs for a possible event since websockets' `recv()` will block.
We noticed the issue since we are running a web server with several ws connections to different RPCs. Each ws connection have different subscriptions.

In our tests, we were seeing events arriving several seconds after the websockets library was getting the frame.

### How can it be fixed?

As mentioned in
https://github.com/python-websockets/websockets/issues/162

instead of timeout on websockets' `recv()`, web3.py can check the `websocket.messages` queue directly for event listening.



Discussion:
fselmo: I just rebased due to significant changes on main since this was opened. No code should have changed. 👀 

Patch:
diff --git a/docs/internals.rst b/docs/internals.rst
index 5f311a00..8672ada3 100644
--- a/docs/internals.rst
+++ b/docs/internals.rst
@@ -203,6 +203,19 @@ not adhere to the `JSON-RPC 2.0 specification <https://www.jsonrpc.org/specifica
 in this way will not work with ``PersistentConnectionProvider`` instances. The specifics
 of how the request processor handles this are outlined below.
 
+Listening for Responses
+~~~~~~~~~~~~~~~~~~~~~~~
+
+Implementations of the ``PersistentConnectionProvider`` class have a message listener
+background task that is called when the websocket connection is established. This task
+is responsible for listening for any and all messages coming in over the websocket
+connection and storing them in the ``RequestProcessor`` instance internal to the
+``PersistentConnectionProvider`` instance. The ``RequestProcessor`` instance is
+responsible for storing the messages in the correct cache, either the one-to-one cache
+or the one-to-many (subscriptions) queue, depending on whether the message has a
+JSON-RPC *id* value or not.
+
+
 One-To-One Requests
 ~~~~~~~~~~~~~~~~~~~
 
@@ -221,12 +234,17 @@ back. An example is using the ``eth`` module API to request the latest block num
     >>> asyncio.run(wsV2_one_to_one_example())
 
 With websockets we have to call ``send()`` and asynchronously receive responses via
-``recv()``. In order to make the one-to-one request-to-response call work, we
-have to save the request information somewhere so that when the response is received
-we can match it to the original request that was made i.e. the request with a matching
-*id* to the response that was received). The stored request information is then used to
-process the response when it is received, piping it through the response formatters and
-middlewares internal to the *web3.py* library.
+another means, generally by calling ``recv()`` or by iterating on the websocket
+connection for messages. As outlined above, the ``PersistentConnectionProvider`` class
+has a message listener background task that handles the receiving of messages.
+
+Due to this asynchronous nature of sending and receiving, in order to make one-to-one
+request-to-response calls work, we have to save the request information somewhere so
+that, when the response is received, we can match it to the original request that was
+made (i.e. the request with a matching *id* to the response that was received). The
+stored request information is then used to process the response when it is received,
+piping it through the response formatters and middlewares internal to the *web3.py*
+library.
 
 In order to store the request information, the ``RequestProcessor`` class has an
 internal ``RequestInformation`` cache. The ``RequestInformation`` class saves important
@@ -264,20 +282,24 @@ information about a request.
 One-to-one responses, those that include a JSON-RPC *id* in the response object, are
 stored in an internal ``SimpleCache`` class, isolated from any one-to-many responses.
 When the ``PersistentConnectionProvider`` is looking for a response internally, it will
-cycle within a ``while`` loop, alternating between checking this cache (in case
-somewhere else in the code our desired response was cached by another call) and calling
-``recv()`` on the websocket connection to see if it is yet to come.
+expect the message listener task to store the response in this cache. Since the request
+*id* is used in the cache key generation, it will then look for a cache key that matches
+the response *id* with that of the request *id*. If the cache key is found, the response
+is processed and returned to the user. If the cache key is not found, the operation will
+time out and raise a ``TimeExhausted`` exception. This timeout can be configured by the
+user when instantiating the ``PersistentConnectionProvider`` instance via the
+``response_timeout`` keyword argument.
 
 One-To-Many Requests
 ~~~~~~~~~~~~~~~~~~~~
 
 One-to-many requests can be summarized by any request that expects many responses as a
-result of the initial request. An example is the ``eth_subscribe`` request. The initial
-``eth_subscribe`` request expects only one response, the subscription *id* value, but
-it also expects to receive many ``eth_subscription`` messages if and when the request is
-successful. For this reason, the original request is considered a one-to-one request
-so that a subscription *id* can be returned to the user on the same line, but the
-``listen_to_websocket()`` method on the
+result of the initial request. The only current example is the ``eth_subscribe``
+request. The initial ``eth_subscribe`` request expects only one response, the
+subscription *id* value, but it also expects to receive many ``eth_subscription``
+messages if and when the request is successful. For this reason, the original request
+is considered a one-to-one request so that a subscription *id* can be returned to the
+user on the same line, but the ``process_subscriptions()`` method on the
 :class:`~web3.providers.websocket.WebsocketConnection` class, the public API for
 interacting with the active websocket connection, is set up to receive
 ``eth_subscription`` responses over an asynchronous interator pattern.
@@ -293,8 +315,8 @@ interacting with the active websocket connection, is set up to receive
     ...         subscription_id = await w3.eth.subscribe("newHeads")
     ...
     ...         # Listen to the websocket for the many responses utilizing the ``w3.ws``
-    ...         # ``WebsocketConnection`` public API method ``listen_to_websocket()``
-    ...         async for response in w3.ws.listen_to_websocket():
+    ...         # ``WebsocketConnection`` public API method ``process_subscriptions()``
+    ...         async for response in w3.ws.process_subscriptions():
     ...             # Receive only one-to-many responses here so that we don't
     ...             # accidentally return the response for a one-to-one request in this
     ...             # block
@@ -310,13 +332,22 @@ interacting with the active websocket connection, is set up to receive
     >>> asyncio.run(ws_v2_subscription_example())
 
 One-to-many responses, those that do not include a JSON-RPC *id* in the response object,
-are stored in an internal ``collections.deque`` instance, isolated from any one-to-one
-responses. When the ``PersistentConnectionProvider`` is looking for a one-to-many
-response internally, it will cycle within a ``while`` loop, alternating between
-checking this deque (in case somewhere else in the code, subscriptions responses were
-put in the deque by another call looking for another response type) and calling
-``recv()`` on the websocket connection to see if the response is yet to be received.
-With each iteration on this iterator, if the ``deque`` is non-empty, it will yield
-messages from the ``deque`` as FIFO order until the deque is empty before receiving any
-new messages from the websocket connection, guaranteeing the messages are yielded in the
-order they were received.
+are stored in an internal ``asyncio.Queue`` instance, isolated from any one-to-one
+responses. When the ``PersistentConnectionProvider`` is looking for one-to-many
+responses internally, it will expect the message listener task to store these messages
+in this queue. Since the order of the messages is important, the queue is a FIFO queue.
+The ``process_subscriptions()`` method on the ``WebsocketConnection`` class is set up
+to pop messages from this queue as FIFO over an asynchronous iterator pattern.
+
+If the stream of messages from the websocket is not being interrupted by any other
+tasks, the queue will generally be in sync with the messages coming in over the
+websocket. That is, the message listener will put a message in the queue and the
+``process_subscriptions()`` method will pop that message from the queue and yield
+control of the loop back to the listener. This will continue until the websocket
+connection is closed or the user unsubscribes from the subscription. If the stream of
+messages lags a bit, or the provider is not consuming messages but has subscribed to
+a subscription, this internal queue may fill up with messages until it reaches its max
+size and then trigger a waiting ``asyncio.Event`` until the provider begins consuming
+messages from the queue again. For this reason, it's important to begin consuming
+messages from the queue, via the ``process_subscriptions()`` method, as soon as a
+subscription is made.
diff --git a/docs/providers.rst b/docs/providers.rst
index 6f983ab7..d0e36572 100644
--- a/docs/providers.rst
+++ b/docs/providers.rst
@@ -273,7 +273,7 @@ asynchronous context manager, can be found in the `websockets connection`_ docs.
         ...         # subscribe to new block headers
         ...         subscription_id = await w3.eth.subscribe("newHeads")
         ...
-        ...         async for response in w3.ws.listen_to_websocket():
+        ...         async for response in w3.ws.process_subscriptions():
         ...             print(f"{response}\n")
         ...             # handle responses here
         ...
@@ -319,6 +319,7 @@ and reconnect automatically if the connection is lost. A similar example, using
     ...         except websockets.ConnectionClosed:
     ...             continue
 
+    # run the example
     >>> asyncio.run(ws_v2_subscription_iterator_example())
 
 
@@ -338,6 +339,9 @@ provider in separate lines. Both of these examples are shown below.
     ...     # manual cleanup
     ...     await w3.provider.disconnect()
 
+    # run the example
+    >>> asyncio.run(ws_v2_alternate_init_example_1)
+
     >>> async def ws_v2_alternate_init_example_2():
     ...     # instantiation and connection via the provider as separate lines
     ...     w3 = AsyncWeb3.persistent_websocket(WebsocketProviderV2(f"ws://127.0.0.1:8546"))
@@ -348,8 +352,7 @@ provider in separate lines. Both of these examples are shown below.
     ...     # manual cleanup
     ...     await w3.provider.disconnect()
 
-    >>> # run the examples:
-    >>> asyncio.run(ws_v2_alternate_init_example_1)
+    # run the example
     >>> asyncio.run(ws_v2_alternate_init_example_2)
 
 The ``WebsocketProviderV2`` class uses the
@@ -393,11 +396,11 @@ Interacting with the Websocket Connection
         the subscription ``id`` to a dict of metadata about the subscription
         request.
 
-    .. py:method:: listen_to_websocket()
+    .. py:method:: process_subscriptions()
 
-        This method is available for listening to websocket responses indefinitely.
+        This method is available for listening to websocket subscriptions indefinitely.
         It is an asynchronous iterator that yields strictly one-to-many
-        (e.g. ``eth_subscription`` responses) request-to-response responses from the
+        (e.g. ``eth_subscription`` responses) request-to-response messages from the
         websocket connection. To receive responses for one-to-one request-to-response
         calls, use the standard API for making requests via the appropriate module
         (e.g. ``block_num = await w3.eth.block_number``)
@@ -410,9 +413,12 @@ Interacting with the Websocket Connection
 
         The ``recv()`` method can be used to receive the next message from the
         websocket. The response from this method is formatted by web3.py formatters
-        and run through the middlewares before being returned. This is useful for
-        receiving singled responses for one-to-many requests such receiving the
-        next ``eth_subscribe`` subscription response.
+        and run through the middlewares before being returned. This is not the
+        recommended way to receive a message as the ``process_subscriptions()`` method
+        is available for listening to websocket subscriptions and the standard API for
+        making requests via the appropriate module
+        (e.g. ``block_num = await w3.eth.block_number``) is available for receiving
+        responses for one-to-one request-to-response calls.
 
     .. py:method:: send(method: RPCEndpoint, params: Sequence[Any])
 
diff --git a/newsfragments/3179.breaking.rst b/newsfragments/3179.breaking.rst
new file mode 100644
index 00000000..979e209e
--- /dev/null
+++ b/newsfragments/3179.breaking.rst
@@ -0,0 +1,1 @@
+Use a message listener background task for ``WebsocketProviderV2`` rather than relying on ``ws.recv()`` blocking. Some breaking changes to API, notably ``listen_to_websocket`` -> ``process_subscriptions``.
diff --git a/web3/manager.py b/web3/manager.py
index 49b99955..de11789f 100644
--- a/web3/manager.py
+++ b/web3/manager.py
@@ -35,6 +35,7 @@ from web3.datastructures import (
 from web3.exceptions import (
     BadResponseFormat,
     MethodUnavailable,
+    ProviderConnectionError,
 )
 from web3.middleware import (
     attrdict_middleware,
@@ -311,7 +312,8 @@ class RequestManager:
             response, params, error_formatters, null_result_formatters
         )
 
-    # persistent connection
+    # -- persistent connection -- #
+
     async def ws_send(self, method: RPCEndpoint, params: Any) -> RPCResponse:
         provider = cast(PersistentConnectionProvider, self._provider)
         request_func = await provider.request_func(
@@ -324,47 +326,35 @@ class RequestManager:
         response = await request_func(method, params)
         return await self._process_ws_response(response)
 
-    async def ws_recv(self) -> Any:
-        return await self._ws_recv_stream().__anext__()
+    def _persistent_message_stream(self) -> "_AsyncPersistentMessageStream":
+        return _AsyncPersistentMessageStream(self)
 
-    def _persistent_recv_stream(self) -> "_AsyncPersistentRecvStream":
-        return _AsyncPersistentRecvStream(self)
+    async def _get_next_ws_message(self) -> Any:
+        return await self._ws_message_stream().__anext__()
 
-    async def _ws_recv_stream(self) -> AsyncGenerator[RPCResponse, None]:
+    async def _ws_message_stream(self) -> AsyncGenerator[RPCResponse, None]:
         if not isinstance(self._provider, PersistentConnectionProvider):
             raise TypeError(
                 "Only websocket providers that maintain an open, persistent connection "
                 "can listen to websocket recv streams."
             )
 
+        if self._provider._message_listener_task is None:
+            raise ProviderConnectionError("No listener found for websocket connection.")
+
         while True:
-            # sleep(0) here seems to be the most efficient way to yield control back to
-            # the event loop while waiting for the response to be cached or received on
-            # the websocket.
+            # sleep(0) here seems to be the most efficient way to yield control
+            # back to the event loop while waiting for the response in the queue.
             await asyncio.sleep(0)
 
-            # look in the cache for a response
             response = self._request_processor.pop_raw_response(subscription=True)
-            if response is not None:
-                break
-            else:
-                # if no response in the cache, check the websocket connection
-                if not self._provider._ws_lock.locked():
-                    async with self._provider._ws_lock:
-                        try:
-                            # keep timeout low but reasonable to check both the cache
-                            # and the websocket connection for new responses
-                            response = await self._provider._ws_recv(timeout=0.5)
-                        except asyncio.TimeoutError:
-                            # if no response received, continue to next iteration
-                            continue
-
-                    if response.get("method") == "eth_subscription":
-                        break
-                    else:
-                        self._provider._request_processor.cache_raw_response(response)
-
-        yield await self._process_ws_response(response)
+            if (
+                response is not None
+                and response.get("params", {}).get("subscription")
+                in self._request_processor.active_subscriptions
+            ):
+                # if response is an active subscription response, process it
+                yield await self._process_ws_response(response)
 
     async def _process_ws_response(self, response: RPCResponse) -> RPCResponse:
         provider = cast(PersistentConnectionProvider, self._provider)
@@ -412,15 +402,18 @@ class RequestManager:
             return apply_result_formatters(result_formatters, partly_formatted_response)
 
 
-class _AsyncPersistentRecvStream:
+class _AsyncPersistentMessageStream:
     """
-    Async generator for receiving responses from a persistent connection. This
-    abstraction is necessary to define the `__aiter__()` method required for
-    use with "async for" loops.
+    Async generator for pulling subscription responses from the request processor
+    subscription queue. This abstraction is necessary to define the `__aiter__()`
+    method required for use with "async for" loops.
     """
 
     def __init__(self, manager: RequestManager, *args: Any, **kwargs: Any) -> None:
         self.manager = manager
+        self.provider: PersistentConnectionProvider = cast(
+            PersistentConnectionProvider, manager._provider
+        )
         super().__init__(*args, **kwargs)
 
     def __aiter__(self) -> Self:
@@ -428,6 +421,6 @@ class _AsyncPersistentRecvStream:
 
     async def __anext__(self) -> RPCResponse:
         try:
-            return await self.manager.ws_recv()
+            return await self.manager._get_next_ws_message()
         except ConnectionClosedOK:
             raise StopAsyncIteration
diff --git a/web3/providers/persistent.py b/web3/providers/persistent.py
index bc23ae3b..fcd92b05 100644
--- a/web3/providers/persistent.py
+++ b/web3/providers/persistent.py
@@ -17,9 +17,6 @@ from web3.providers.async_base import (
 from web3.providers.websocket.request_processor import (
     RequestProcessor,
 )
-from web3.types import (
-    RPCResponse,
-)
 
 DEFAULT_PERSISTENT_CONNECTION_TIMEOUT = 50
 
@@ -29,20 +26,21 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
     has_persistent_connection = True
 
     _ws: Optional[WebSocketClientProtocol] = None
-    _ws_lock: asyncio.Lock = asyncio.Lock()
     _request_processor: RequestProcessor
+    _message_listener_task: Optional["asyncio.Task[None]"] = None
+    _listen_event: asyncio.Event = asyncio.Event()
 
     def __init__(
         self,
         endpoint_uri: str,
         request_timeout: float = DEFAULT_PERSISTENT_CONNECTION_TIMEOUT,
-        subscription_response_deque_size: int = 500,
+        subscription_response_queue_size: int = 500,
     ) -> None:
         super().__init__()
         self.endpoint_uri = endpoint_uri
         self._request_processor = RequestProcessor(
             self,
-            subscription_response_deque_size=subscription_response_deque_size,
+            subscription_response_queue_size=subscription_response_queue_size,
         )
         self.request_timeout = request_timeout
 
@@ -52,5 +50,5 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
     async def disconnect(self) -> None:
         raise NotImplementedError("Must be implemented by subclasses")
 
-    async def _ws_recv(self, timeout: float = None) -> RPCResponse:
+    async def _ws_message_listener(self) -> None:
         raise NotImplementedError("Must be implemented by subclasses")
diff --git a/web3/providers/websocket/request_processor.py b/web3/providers/websocket/request_processor.py
index 1b938ab5..d1641c8c 100644
--- a/web3/providers/websocket/request_processor.py
+++ b/web3/providers/websocket/request_processor.py
@@ -1,6 +1,4 @@
-from collections import (
-    deque,
-)
+import asyncio
 from copy import (
     copy,
 )
@@ -8,7 +6,6 @@ from typing import (
     TYPE_CHECKING,
     Any,
     Callable,
-    Deque,
     Dict,
     Optional,
     Tuple,
@@ -33,21 +30,19 @@ if TYPE_CHECKING:
 
 
 class RequestProcessor:
-    _request_information_cache: SimpleCache
-    _request_response_cache: SimpleCache
-    _subscription_response_deque: Deque[RPCResponse]
+    _subscription_queue_synced_with_ws_stream: bool = False
 
     def __init__(
         self,
         provider: "PersistentConnectionProvider",
-        subscription_response_deque_size: int = 500,
+        subscription_response_queue_size: int = 500,
     ) -> None:
         self._provider = provider
 
-        self._request_information_cache = SimpleCache(500)
-        self._request_response_cache = SimpleCache(500)
-        self._subscription_response_deque = deque(
-            maxlen=subscription_response_deque_size
+        self._request_information_cache: SimpleCache = SimpleCache(500)
+        self._request_response_cache: SimpleCache = SimpleCache(500)
+        self._subscription_response_queue: asyncio.Queue[RPCResponse] = asyncio.Queue(
+            maxsize=subscription_response_queue_size
         )
 
     @property
@@ -184,16 +179,6 @@ class RequestProcessor:
                 subscribe_cache_key = generate_cache_key(subscription_id)
                 self.pop_cached_request_information(subscribe_cache_key)
 
-                # rebuild the deque without the unsubscribed subscription responses
-                self._subscription_response_deque = deque(
-                    filter(
-                        lambda sub_response: sub_response["params"]["subscription"]
-                        != subscription_id,
-                        self._subscription_response_deque,
-                    ),
-                    maxlen=self._subscription_response_deque.maxlen,
-                )
-
         return request_info
 
     def append_middleware_response_processor(
@@ -225,12 +210,21 @@ class RequestProcessor:
 
     # raw response cache
 
-    def cache_raw_response(self, raw_response: Any, subscription: bool = False) -> None:
+    async def cache_raw_response(
+        self, raw_response: Any, subscription: bool = False
+    ) -> None:
         if subscription:
+            if self._subscription_response_queue.full():
+                self._provider.logger.info(
+                    "Subscription queue is full. Waiting for provider to consume "
+                    "messages before caching."
+                )
+                await self._provider._listen_event.wait()
+
             self._provider.logger.debug(
                 f"Caching subscription response:\n    response={raw_response}"
             )
-            self._subscription_response_deque.append(raw_response)
+            await self._subscription_response_queue.put(raw_response)
         else:
             response_id = raw_response.get("id")
             cache_key = generate_cache_key(response_id)
@@ -244,17 +238,31 @@ class RequestProcessor:
         self, cache_key: str = None, subscription: bool = False
     ) -> Any:
         if subscription:
-            deque_length = len(self._subscription_response_deque)
-            if deque_length == 0:
+            qsize = self._subscription_response_queue.qsize()
+            if qsize == 0:
                 return None
 
-            raw_response = self._subscription_response_deque.popleft()
-            self._provider.logger.debug(
-                f"Subscription response deque is not empty. Processing {deque_length} "
-                "subscription(s) as FIFO before receiving new response."
-            )
+            if not self._provider._listen_event.is_set():
+                self._provider._listen_event.set()
+
+            raw_response = self._subscription_response_queue.get_nowait()
+            if qsize == 1:
+                if not self._subscription_queue_synced_with_ws_stream:
+                    self._subscription_queue_synced_with_ws_stream = True
+                    self._provider.logger.info(
+                        "Subscription response queue synced with websocket message "
+                        "stream."
+                    )
+            else:
+                if self._subscription_queue_synced_with_ws_stream:
+                    self._subscription_queue_synced_with_ws_stream = False
+                self._provider.logger.info(
+                    f"Subscription response queue has {qsize} subscriptions. "
+                    "Processing as FIFO."
+                )
+
             self._provider.logger.debug(
-                "Cached subscription response popped from deque to be processed:\n"
+                "Subscription response popped from queue to be processed:\n"
                 f"    raw_response={raw_response}"
             )
         else:
@@ -282,4 +290,6 @@ class RequestProcessor:
 
         self._request_information_cache.clear()
         self._request_response_cache.clear()
-        self._subscription_response_deque.clear()
+        self._subscription_response_queue = asyncio.Queue(
+            maxsize=self._subscription_response_queue.maxsize
+        )
diff --git a/web3/providers/websocket/websocket_connection.py b/web3/providers/websocket/websocket_connection.py
index 22555a25..c2fd50e6 100644
--- a/web3/providers/websocket/websocket_connection.py
+++ b/web3/providers/websocket/websocket_connection.py
@@ -14,7 +14,7 @@ if TYPE_CHECKING:
         _PersistentConnectionWeb3,
     )
     from web3.manager import (  # noqa: F401
-        _AsyncPersistentRecvStream,
+        _AsyncPersistentMessageStream,
     )
 
 
@@ -27,6 +27,7 @@ class WebsocketConnection:
     def __init__(self, w3: "_PersistentConnectionWeb3"):
         self._manager = w3.manager
 
+    # -- public methods -- #
     @property
     def subscriptions(self) -> Dict[str, Any]:
         return self._manager._request_processor.active_subscriptions
@@ -35,7 +36,7 @@ class WebsocketConnection:
         return await self._manager.ws_send(method, params)
 
     async def recv(self) -> Any:
-        return await self._manager.ws_recv()
+        return await self._manager._get_next_ws_message()
 
-    def listen_to_websocket(self) -> "_AsyncPersistentRecvStream":
-        return self._manager._persistent_recv_stream()
+    def process_subscriptions(self) -> "_AsyncPersistentMessageStream":
+        return self._manager._persistent_message_stream()
diff --git a/web3/providers/websocket/websocket_v2.py b/web3/providers/websocket/websocket_v2.py
index 3189c699..40d0fec8 100644
--- a/web3/providers/websocket/websocket_v2.py
+++ b/web3/providers/websocket/websocket_v2.py
@@ -32,7 +32,6 @@ from web3.exceptions import (
     Web3ValidationError,
 )
 from web3.providers.persistent import (
-    DEFAULT_PERSISTENT_CONNECTION_TIMEOUT,
     PersistentConnectionProvider,
 )
 from web3.types import (
@@ -67,7 +66,9 @@ class WebsocketProviderV2(PersistentConnectionProvider):
         self,
         endpoint_uri: Optional[Union[URI, str]] = None,
         websocket_kwargs: Optional[Dict[str, Any]] = None,
-        request_timeout: Optional[float] = DEFAULT_PERSISTENT_CONNECTION_TIMEOUT,
+        raise_listener_task_exceptions: bool = False,
+        # `PersistentConnectionProvider` kwargs can be passed through
+        **kwargs: Any,
     ) -> None:
         self.endpoint_uri = URI(endpoint_uri)
         if self.endpoint_uri is None:
@@ -78,7 +79,7 @@ class WebsocketProviderV2(PersistentConnectionProvider):
             for prefix in VALID_WEBSOCKET_URI_PREFIXES
         ):
             raise Web3ValidationError(
-                f"Websocket endpoint uri must begin with 'ws://' or 'wss://': "
+                "Websocket endpoint uri must begin with 'ws://' or 'wss://': "
                 f"{self.endpoint_uri}"
             )
 
@@ -93,8 +94,9 @@ class WebsocketProviderV2(PersistentConnectionProvider):
                 )
 
         self.websocket_kwargs = merge(DEFAULT_WEBSOCKET_KWARGS, websocket_kwargs or {})
+        self.raise_listener_task_exceptions = raise_listener_task_exceptions
 
-        super().__init__(endpoint_uri, request_timeout=request_timeout)
+        super().__init__(endpoint_uri, **kwargs)
 
     def __str__(self) -> str:
         return f"Websocket connection: {self.endpoint_uri}"
@@ -123,6 +125,9 @@ class WebsocketProviderV2(PersistentConnectionProvider):
             try:
                 _connection_attempts += 1
                 self._ws = await connect(self.endpoint_uri, **self.websocket_kwargs)
+                self._message_listener_task = asyncio.create_task(
+                    self._ws_message_listener()
+                )
                 break
             except WebSocketException as e:
                 if _connection_attempts == self._max_connection_retries:
@@ -146,6 +151,11 @@ class WebsocketProviderV2(PersistentConnectionProvider):
                 f'Successfully disconnected from endpoint: "{self.endpoint_uri}'
             )
 
+        try:
+            self._message_listener_task.cancel()
+            await self._message_listener_task
+        except (asyncio.CancelledError, StopAsyncIteration):
+            pass
         self._request_processor.clear_caches()
 
     @async_handle_request_caching
@@ -172,53 +182,18 @@ class WebsocketProviderV2(PersistentConnectionProvider):
 
             while True:
                 # sleep(0) here seems to be the most efficient way to yield control
-                # back to the event loop while waiting for the response to be cached
-                # or received on the websocket.
+                # back to the event loop while waiting for the response to be in the
+                # queue.
                 await asyncio.sleep(0)
 
                 if request_cache_key in self._request_processor._request_response_cache:
-                    # if response is already cached, pop it from cache
                     self.logger.debug(
-                        f"Response for id {request_id} is already cached, pop it "
-                        "from the cache."
+                        f"Popping response for id {request_id} from cache."
                     )
-                    return self._request_processor.pop_raw_response(
+                    popped_response = self._request_processor.pop_raw_response(
                         cache_key=request_cache_key,
                     )
-
-                else:
-                    if not self._ws_lock.locked():
-                        async with self._ws_lock:
-                            self.logger.debug(
-                                f"Response for id {request_id} is not cached, calling "
-                                "`recv()` on websocket."
-                            )
-                            try:
-                                # keep timeout low but reasonable to check both the
-                                # cache and the websocket connection for new responses
-                                response = await self._ws_recv(timeout=0.5)
-                            except asyncio.TimeoutError:
-                                # keep the request timeout around the whole of this
-                                # while loop in case the response sneaks into the cache
-                                # from another call.
-                                continue
-
-                        response_id = response.get("id")
-
-                        if response_id == request_id:
-                            self.logger.debug(
-                                f"Received and returning response for id {request_id}."
-                            )
-                            return response
-                        else:
-                            # cache all responses that are not the desired response
-                            self.logger.debug("Undesired response received, caching.")
-                            is_subscription = (
-                                response.get("method") == "eth_subscription"
-                            )
-                            self._request_processor.cache_raw_response(
-                                response, subscription=is_subscription
-                            )
+                    return popped_response
 
         try:
             # Add the request timeout around the while loop that checks the request
@@ -236,5 +211,31 @@ class WebsocketProviderV2(PersistentConnectionProvider):
                 "allowed to continue."
             )
 
-    async def _ws_recv(self, timeout: float = None) -> RPCResponse:
-        return json.loads(await asyncio.wait_for(self._ws.recv(), timeout=timeout))
+    async def _ws_message_listener(self) -> None:
+        self.logger.info(
+            "Websocket listener background task started. Storing all messages in "
+            "appropriate request processor queues / caches to be processed."
+        )
+        while True:
+            # the use of sleep(0) seems to be the most efficient way to yield control
+            # back to the event loop to share the loop with other tasks.
+            await asyncio.sleep(0)
+
+            try:
+                async for raw_message in self._ws:
+                    await asyncio.sleep(0)
+
+                    response = json.loads(raw_message)
+                    subscription = response.get("method") == "eth_subscription"
+                    await self._request_processor.cache_raw_response(
+                        response, subscription=subscription
+                    )
+            except Exception as e:
+                if self.raise_listener_task_exceptions:
+                    # If ``True``, raise; else, error log & keep task alive
+                    raise e
+
+                self.logger.error(
+                    "Exception caught in listener, error logging and keeping listener "
+                    f"background task alive.\n    error={e}"
+                )


Test patch:
diff --git a/tests/core/providers/test_wsv2_provider.py b/tests/core/providers/test_wsv2_provider.py
index 13750a9d..85172129 100644
--- a/tests/core/providers/test_wsv2_provider.py
+++ b/tests/core/providers/test_wsv2_provider.py
@@ -3,12 +3,23 @@ import json
 import pytest
 from unittest.mock import (
     AsyncMock,
+    Mock,
+    patch,
 )
 
 from eth_utils import (
     to_bytes,
 )
 
+from web3 import (
+    AsyncWeb3,
+)
+from web3._utils.caching import (
+    RequestInformation,
+)
+from web3._utils.module_testing.module_testing_utils import (
+    WebsocketMessageStreamMock,
+)
 from web3.exceptions import (
     TimeExhausted,
 )
@@ -24,15 +35,24 @@ def _mock_ws(provider):
     provider._ws = AsyncMock()
 
 
+async def _coro():
+    return None
+
+
 @pytest.mark.asyncio
-async def test_async_make_request_caches_all_undesired_responses_and_returns_desired():
+async def test_async_make_request_returns_desired_response():
     provider = WebsocketProviderV2("ws://mocked")
 
-    method_under_test = provider.make_request
+    with patch(
+        "web3.providers.websocket.websocket_v2.connect", new=lambda *_1, **_2: _coro()
+    ):
+        await provider.connect()
 
     _mock_ws(provider)
+    method_under_test = provider.make_request
+
     undesired_responses_count = 10
-    ws_recv_responses = [
+    ws_messages = [
         to_bytes(
             text=json.dumps(
                 {
@@ -42,44 +62,30 @@ async def test_async_make_request_caches_all_undesired_responses_and_returns_des
                 }
             )
         )
-        for i in range(0, undesired_responses_count)
+        for i in range(undesired_responses_count)
     ]
     # The first request we make should have an id of `0`, expect the response to match
     # that id. Append it as the last response in the list.
-    ws_recv_responses.append(b'{"jsonrpc": "2.0", "id":0, "result": "0x1337"}')
-    provider._ws.recv.side_effect = ws_recv_responses
+    ws_messages.append(b'{"jsonrpc": "2.0", "id": 0, "result": "0x1337"}')
+    provider._ws = WebsocketMessageStreamMock(messages=ws_messages)
 
     response = await method_under_test(RPCEndpoint("some_method"), ["desired_params"])
-    assert response == json.loads(ws_recv_responses.pop())  # pop the expected response
+    assert response == json.loads(ws_messages.pop())
 
-    assert (
-        len(provider._request_processor._subscription_response_deque)
-        == len(ws_recv_responses)
-        == undesired_responses_count
-    )
+    qsize = provider._request_processor._subscription_response_queue.qsize()
+    assert qsize == len(ws_messages) == undesired_responses_count
 
-    for cached_response in provider._request_processor._subscription_response_deque:
+    for i in range(qsize):
+        cached_response = (
+            await provider._request_processor._subscription_response_queue.get()
+        )
         # assert all cached responses are in the list of responses we received
-        assert to_bytes(text=json.dumps(cached_response)) in ws_recv_responses
-
-
-@pytest.mark.asyncio
-async def test_async_make_request_returns_cached_response_with_no_recv_if_cached():
-    provider = WebsocketProviderV2("ws://mocked")
-
-    method_under_test = provider.make_request
-
-    _mock_ws(provider)
-
-    # cache the response, so we should get it immediately & should never call `recv()`
-    desired_response = {"jsonrpc": "2.0", "id": 0, "result": "0x1337"}
-    provider._request_processor.cache_raw_response(desired_response)
-
-    response = await method_under_test(RPCEndpoint("some_method"), ["desired_params"])
-    assert response == desired_response
+        assert to_bytes(text=json.dumps(cached_response)) in ws_messages
 
+    assert provider._request_processor._subscription_response_queue.empty()
     assert len(provider._request_processor._request_response_cache) == 0
-    assert not provider._ws.recv.called  # type: ignore
+
+    await provider.disconnect()
 
 
 @pytest.mark.asyncio
@@ -88,10 +94,7 @@ async def test_async_make_request_times_out_of_while_loop_looking_for_response()
     provider = WebsocketProviderV2("ws://mocked", request_timeout=timeout)
 
     method_under_test = provider.make_request
-
     _mock_ws(provider)
-    # mock the websocket to never receive a response & sleep longer than the timeout
-    provider._ws.recv = lambda *args, **kwargs: asyncio.sleep(1)
 
     with pytest.raises(
         TimeExhausted,
@@ -99,3 +102,158 @@ async def test_async_make_request_times_out_of_while_loop_looking_for_response()
         rf"{timeout} second\(s\)",
     ):
         await method_under_test(RPCEndpoint("some_method"), ["desired_params"])
+
+
+@pytest.mark.asyncio
+async def test_msg_listener_task_starts_on_provider_connect_and_cancels_on_disconnect():
+    provider = WebsocketProviderV2("ws://mocked")
+    _mock_ws(provider)
+
+    assert provider._message_listener_task is None
+
+    with patch(
+        "web3.providers.websocket.websocket_v2.connect", new=lambda *_1, **_2: _coro()
+    ):
+        await provider.connect()  # connect
+
+    assert provider._message_listener_task is not None
+    assert not provider._message_listener_task.cancelled()
+
+    await provider.disconnect()  # disconnect
+
+    assert provider._message_listener_task.cancelled()
+    assert provider._message_listener_task.done()
+
+
+@pytest.mark.asyncio
+async def test_msg_listener_task_silences_exceptions_by_default_and_error_logs(caplog):
+    provider = WebsocketProviderV2("ws://mocked")
+    _mock_ws(provider)
+
+    with patch(
+        "web3.providers.websocket.websocket_v2.connect", new=lambda *_1, **_2: _coro()
+    ):
+        await provider.connect()
+        assert provider._message_listener_task is not None
+        assert provider.raise_listener_task_exceptions is False
+
+    provider._ws = WebsocketMessageStreamMock(
+        raise_exception=Exception("test exception")
+    )
+    await asyncio.sleep(0.05)
+
+    assert "test exception" in caplog.text
+    assert (
+        "Exception caught in listener, error logging and keeping listener background "
+        "task alive.\n    error=test exception"
+    ) in caplog.text
+
+    # assert is still running
+    assert not provider._message_listener_task.cancelled()
+
+    # proper cleanup
+    await provider.disconnect()
+
+
+@pytest.mark.asyncio
+async def test_msg_listener_task_raises_when_raise_listener_task_exceptions_is_true():
+    provider = WebsocketProviderV2("ws://mocked", raise_listener_task_exceptions=True)
+    _mock_ws(provider)
+
+    with patch(
+        "web3.providers.websocket.websocket_v2.connect", new=lambda *_1, **_2: _coro()
+    ):
+        await provider.connect()
+        assert provider._message_listener_task is not None
+
+    provider._ws = WebsocketMessageStreamMock(
+        raise_exception=Exception("test exception")
+    )
+    with pytest.raises(Exception, match="test exception"):
+        await provider._message_listener_task
+
+    assert provider._message_listener_task.done()
+
+
+@pytest.mark.asyncio
+async def test_listen_event_awaits_msg_processing_when_subscription_queue_is_full():
+    """
+    This test is to ensure that the `listen_event` method will wait for the
+    `process_subscriptions` method to process a message when the subscription queue
+    is full.
+    """
+    with patch(
+        "web3.providers.websocket.websocket_v2.connect", new=lambda *_1, **_2: _coro()
+    ):
+        async_w3 = await AsyncWeb3.persistent_websocket(
+            WebsocketProviderV2("ws://mocked")
+        )
+
+    _mock_ws(async_w3.provider)
+
+    assert async_w3.provider._message_listener_task is not None
+    assert not async_w3.provider._message_listener_task.cancelled()
+
+    # assert queue is instance of asyncio.Queue and replace with a new queue, maxsize=1
+    assert isinstance(
+        async_w3.provider._request_processor._subscription_response_queue,
+        type(asyncio.Queue()),
+    )
+    async_w3.provider._request_processor._subscription_response_queue = asyncio.Queue(
+        maxsize=1
+    )
+    assert not async_w3.provider._request_processor._subscription_response_queue.full()
+
+    # mock listen event
+    async_w3.provider._listen_event.wait = AsyncMock()
+    async_w3.provider._listen_event.set = Mock()
+
+    # mock subscription and add to active subscriptions
+    sub_id = "0x1"
+    sub_request_information = RequestInformation(
+        method=RPCEndpoint("eth_subscribe"),
+        params=["mock"],
+        response_formatters=(),
+        subscription_id=sub_id,
+    )
+    async_w3.provider._request_processor._request_information_cache.cache(
+        "", sub_request_information
+    )
+
+    mocked_sub = {
+        "jsonrpc": "2.0",
+        "method": "eth_subscription",
+        "params": {"subscription": sub_id, "result": "0x1337"},
+    }
+
+    # fill queue with one item so it is full
+    async_w3.provider._request_processor._subscription_response_queue.put_nowait(
+        mocked_sub
+    )
+    assert async_w3.provider._request_processor._subscription_response_queue.full()
+
+    # wait will be called on the _listen_event when the next message comes in
+    async_w3.provider._listen_event.wait.assert_not_called()
+
+    # mock the message stream with a single message
+    # the message listener task should then call the _listen_event.wait since the
+    # queue is full
+    async_w3.provider._ws = WebsocketMessageStreamMock(
+        messages=[to_bytes(text=json.dumps(mocked_sub))]
+    )
+    await asyncio.sleep(0.05)
+    async_w3.provider._listen_event.wait.assert_called_once()
+
+    # set is not called until we start consuming messages
+    async_w3.provider._listen_event.set.assert_not_called()
+
+    async for message in async_w3.ws.process_subscriptions():
+        # assert the very next message is the mocked subscription
+        assert message == mocked_sub
+        break
+
+    # assert we set the _listen_event after we consume the message
+    async_w3.provider._listen_event.set.assert_called_once()
+
+    # proper cleanup
+    await async_w3.provider.disconnect()
diff --git a/web3/_utils/module_testing/module_testing_utils.py b/web3/_utils/module_testing/module_testing_utils.py
index c4a9df2c..7cb0e0b2 100644
--- a/web3/_utils/module_testing/module_testing_utils.py
+++ b/web3/_utils/module_testing/module_testing_utils.py
@@ -1,7 +1,11 @@
+from collections import (
+    deque,
+)
 import time
 from typing import (
     TYPE_CHECKING,
     Any,
+    Collection,
     Dict,
     Generator,
     Sequence,
@@ -40,6 +44,9 @@ if TYPE_CHECKING:
     from requests import Response  # noqa: F401
 
     from web3 import Web3  # noqa: F401
+    from web3._utils.compat import (  # noqa: F401
+        Self,
+    )
 
 
 def mine_pending_block(w3: "Web3") -> None:
@@ -162,3 +169,31 @@ def async_mock_offchain_lookup_request_response(
     monkeypatch.setattr(
         f"aiohttp.ClientSession.{http_method.lower()}", _mock_specific_request
     )
+
+
+class WebsocketMessageStreamMock:
+    closed: bool = False
+
+    def __init__(
+        self, messages: Collection[bytes] = None, raise_exception: Exception = None
+    ) -> None:
+        self.messages = deque(messages) if messages else deque()
+        self.raise_exception = raise_exception
+
+    def __aiter__(self) -> "Self":
+        return self
+
+    async def __anext__(self) -> bytes:
+        if self.raise_exception:
+            raise self.raise_exception
+
+        elif len(self.messages) == 0:
+            raise StopAsyncIteration
+
+        return self.messages.popleft()
+
+    async def send(self, data: bytes) -> None:
+        pass
+
+    async def close(self) -> None:
+        pass
diff --git a/web3/_utils/module_testing/persistent_connection_provider.py b/web3/_utils/module_testing/persistent_connection_provider.py
index 4c6ca65e..e7ddd32d 100644
--- a/web3/_utils/module_testing/persistent_connection_provider.py
+++ b/web3/_utils/module_testing/persistent_connection_provider.py
@@ -1,5 +1,4 @@
 import asyncio
-import json
 import pytest
 from typing import (
     TYPE_CHECKING,
@@ -11,7 +10,6 @@ from typing import (
 
 from eth_utils import (
     is_hexstr,
-    to_bytes,
 )
 from hexbytes import (
     HexBytes,
@@ -33,13 +31,6 @@ if TYPE_CHECKING:
     )
 
 
-def _mocked_recv(sub_id: str, ws_subscription_response: Dict[str, Any]) -> bytes:
-    # Must be same subscription id, so we can know how to parse the message.
-    # We don't have this information when mocking the response.
-    ws_subscription_response["params"]["subscription"] = sub_id
-    return to_bytes(text=json.dumps(ws_subscription_response))
-
-
 class PersistentConnectionProviderTest:
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
@@ -295,16 +286,16 @@ class PersistentConnectionProviderTest:
         sub_id = await async_w3.eth.subscribe(*subscription_params)
         assert is_hexstr(sub_id)
 
-        async def _mocked_recv_coro() -> bytes:
-            return _mocked_recv(sub_id, ws_subscription_response)
+        # stub out the subscription id so we know how to process the response
+        ws_subscription_response["params"]["subscription"] = sub_id
 
-        actual_recv_fxn = async_w3.provider._ws.recv
-        async_w3.provider._ws.__setattr__(
-            "recv",
-            _mocked_recv_coro,
+        # add the response to the subscription response cache as if it came from the
+        # websocket connection
+        await async_w3.provider._request_processor.cache_raw_response(
+            ws_subscription_response, subscription=True
         )
 
-        async for msg in async_w3.ws.listen_to_websocket():
+        async for msg in async_w3.ws.process_subscriptions():
             response = cast(FormattedEthSubscriptionResponse, msg)
             assert response["subscription"] == sub_id
             assert response["result"] == expected_formatted_result
@@ -312,9 +303,6 @@ class PersistentConnectionProviderTest:
             # only testing one message, so break here
             break
 
-        # reset the mocked recv
-        async_w3.provider._ws.__setattr__("recv", actual_recv_fxn)
-
     @pytest.mark.asyncio
     async def test_async_extradata_to_poa_middleware_on_eth_subscription(
         self,
@@ -327,28 +315,23 @@ class PersistentConnectionProviderTest:
         sub_id = await async_w3.eth.subscribe("newHeads")
         assert is_hexstr(sub_id)
 
-        async def _mocked_recv_coro() -> bytes:
-            return _mocked_recv(
-                sub_id,
-                {
-                    "jsonrpc": "2.0",
-                    "method": "eth_subscription",
-                    "params": {
-                        "subscription": sub_id,
-                        "result": {
-                            "extraData": f"0x{'00' * 100}",
-                        },
+        # add the response to the subscription response cache as if it came from the
+        # websocket connection
+        await async_w3.provider._request_processor.cache_raw_response(
+            {
+                "jsonrpc": "2.0",
+                "method": "eth_subscription",
+                "params": {
+                    "subscription": sub_id,
+                    "result": {
+                        "extraData": f"0x{'00' * 100}",
                     },
                 },
-            )
-
-        actual_recv_fxn = async_w3.provider._ws.recv
-        async_w3.provider._ws.__setattr__(
-            "recv",
-            _mocked_recv_coro,
+            },
+            subscription=True,
         )
 
-        async for msg in async_w3.ws.listen_to_websocket():
+        async for msg in async_w3.ws.process_subscriptions():
             response = cast(FormattedEthSubscriptionResponse, msg)
             assert response.keys() == {"subscription", "result"}
             assert response["subscription"] == sub_id
@@ -356,10 +339,10 @@ class PersistentConnectionProviderTest:
                 f"0x{'00' * 100}"
             )
 
+            # only testing one message, so break here
             break
 
-        # reset the mocked recv
-        async_w3.provider._ws.__setattr__("recv", actual_recv_fxn)
+        # clean up
         async_w3.middleware_onion.remove("poa_middleware")
 
     @pytest.mark.asyncio