Repository: ethereum/web3.py

Issue:
Async iteration of AsyncWeb3 doesn't work as expected
### What happened?

I am using the latest stable release - 6.18.0

I'm trying out the AsyncWeb3 class following the usage as described in the docs (https://web3py.readthedocs.io/en/stable/providers.html#usage). I would like to maintain an indefinite subscription, automatically reconnecting where necessary due to connection drops. From my understanding this should be possible via the second example using the `async for w3 in AsyncWeb3.persistent_websocket(...` pattern, however this does not work as expected for me.

Using that example, I am able to connect and retrieve some values, however after a short while when receiving a ConnectionClosedError the connection is not automatically retried. Instead, the main asyncio task is cancelled meaning that I am unable to retry by catching `websockets.ConnectionClosed`.

### Code that produced the error

```python
import asyncio
import logging

from websockets.exceptions import ConnectionClosed
from web3 import AsyncWeb3
from web3.providers import WebsocketProviderV2


logging.basicConfig(level=logging.INFO)


async def main():
    async for w3 in AsyncWeb3.persistent_websocket(
        WebsocketProviderV2(...)
    ):
        try:
            subscription_id = await w3.eth.subscribe("newHeads")

            async for response in w3.ws.process_subscriptions():
                block_number = response["result"]["number"]
                logging.info("block number: %d", block_number)
        except ConnectionClosed:
            logging.info("connection closed, retrying")


if __name__ == "__main__":
    asyncio.run(main())
```


### Full error output

```shell
File "/Users/ryancollingham/code/block-scholes/defiDataGrabber/main.py", line 24, in <module>
    asyncio.run(main())
  File "/Users/ryancollingham/.pyenv/versions/3.11.8/lib/python3.11/asyncio/runners.py", line 190, in run
    return runner.run(main)
           ^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/3.11.8/lib/python3.11/asyncio/runners.py", line 118, in run
    return self._loop.run_until_complete(task)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/3.11.8/lib/python3.11/asyncio/base_events.py", line 654, in run_until_complete
    return future.result()
           ^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/code/block-scholes/defiDataGrabber/main.py", line 19, in main
    except ConnectionClosed:
           ^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/code/block-scholes/defiDataGrabber/repro.py", line 21, in main
    async for response in w3.ws.process_subscriptions():
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/web3/manager.py", line 454, in __anext__
    return await self.manager._get_next_ws_message()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/web3/manager.py", line 367, in _get_next_ws_message
    return await self._ws_message_stream().__anext__()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/web3/manager.py", line 380, in _ws_message_stream
    response = await self._request_processor.pop_raw_response(subscription=True)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/web3/providers/websocket/request_processor.py", line 230, in pop_raw_response
    raw_response = await self._subscription_response_queue.get()
                   ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/3.11.8/lib/python3.11/asyncio/queues.py", line 158, in get
    await getter
asyncio.exceptions.CancelledError
ERROR:asyncio:Task exception was never retrieved
future: <Task finished name='Task-5' coro=<WebsocketProviderV2._ws_message_listener() done, defined at /Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/web3/providers/websocket/websocket_v2.py:212> exception=ConnectionClosedError(None, Close(code=<CloseCode.INTERNAL_ERROR: 1011>, reason='keepalive ping timeout'), None)>
Traceback (most recent call last):
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/websockets/legacy/protocol.py", line 1301, in close_connection
    await self.transfer_data_task
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/websockets/legacy/protocol.py", line 963, in transfer_data
    message = await self.read_message()
              ^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/websockets/legacy/protocol.py", line 1033, in read_message
    frame = await self.read_data_frame(max_size=self.max_size)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/websockets/legacy/protocol.py", line 1108, in read_data_frame
    frame = await self.read_frame(max_size)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/websockets/legacy/protocol.py", line 1165, in read_frame
    frame = await Frame.read(
            ^^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/websockets/legacy/framing.py", line 68, in read
    data = await reader(2)
           ^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/3.11.8/lib/python3.11/asyncio/streams.py", line 750, in readexactly
    await self._wait_for_data('readexactly')
  File "/Users/ryancollingham/.pyenv/versions/3.11.8/lib/python3.11/asyncio/streams.py", line 543, in _wait_for_data
    await self._waiter
asyncio.exceptions.CancelledError

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/web3/providers/websocket/websocket_v2.py", line 236, in _ws_message_listener
    raise e
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/web3/providers/websocket/websocket_v2.py", line 223, in _ws_message_listener
    async for raw_message in self._ws:
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/websockets/legacy/protocol.py", line 498, in __aiter__
    yield await self.recv()
          ^^^^^^^^^^^^^^^^^
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/websockets/legacy/protocol.py", line 568, in recv
    await self.ensure_open()
  File "/Users/ryancollingham/.pyenv/versions/defiDataGrabber/lib/python3.11/site-packages/websockets/legacy/protocol.py", line 948, in ensure_open
    raise self.connection_closed_exc()
websockets.exceptions.ConnectionClosedError: sent 1011 (internal error) keepalive ping timeout; no close frame received
```


### Fill this section in if you know how this could or should be fixed

It seems like the error handling here is problematic: https://github.com/ethereum/web3.py/blob/08665fcd54e0e593ba6d19a884b4ba8c9b5753d0/web3/providers/persistent/websocket.py#L198-L209

Since the listener is run in a separate background task, when it cancels all tasks this prevents the ConnectionClosed error from being caught and handled by reconnecting.

### web3 Version

6.18.0

### Python Version

3.11.8

### Operating System

osx

### Output from `pip freeze`

```shell
aiohttp==3.9.5
aiosignal==1.3.1
attrs==23.2.0
bitarray==2.9.2
black==24.4.2
certifi==2024.2.2
charset-normalizer==3.3.2
ckzg==1.0.1
click==8.1.7
cytoolz==0.12.3
eth-account==0.11.2
eth-hash==0.7.0
eth-keyfile==0.8.1
eth-keys==0.5.1
eth-rlp==1.0.1
eth-typing==4.2.2
eth-utils==4.1.0
eth_abi==5.1.0
frozenlist==1.4.1
greenlet==3.0.3
hexbytes==0.3.1
idna==3.7
iniconfig==2.0.0
jsonschema==4.22.0
jsonschema-specifications==2023.12.1
lru-dict==1.2.0
msgpack==1.0.8
multidict==6.0.5
mypy==1.10.0
mypy-extensions==1.0.0
neovim==0.3.1
packaging==24.0
parsimonious==0.10.0
pathspec==0.12.1
platformdirs==4.2.1
pluggy==1.5.0
protobuf==5.26.1
pycryptodome==3.20.0
pynvim==0.5.0
pytest==8.2.0
pyunormalize==15.1.0
referencing==0.35.1
regex==2024.4.28
requests==2.31.0
rlp==4.0.1
rpds-py==0.18.0
ruff==0.4.2
toolz==0.12.1
typing_extensions==4.11.0
urllib3==2.2.1
web3==6.18.0
websockets==12.0
yarl==1.9.4
```


Discussion:
fselmo: I'm actually going to re-structure this a bit. Will ping for review when ready again.
fselmo: @kclowes @pacrob @reedsa This is ready for review again
fselmo: Just kidding this is still in progress 😔 
fselmo: @kclowes @pacrob @reedsa Hey it's me again. This is ready now 😄. We were missing a good test for subscription polling with the iterator pattern. This would've failed or hanged forever. That should be addressed now.
fselmo: > Pushing None to the queue is the only thing that stands out as something that might be a little foot-gun-y.

I think you're absolutely right. I don't see `None` being pushed to the cache in another way but I think we can make this a lot more unique. If I can't, I think we can still use `None` to un-brick the awaiting of a message and we can make sure we make the proper checks of where or not the task really did finish or not and if it didn't we just move on to the next iteration of the subscription messages - essentially causing a no-op in that scenario. 
kclowes: > I came up with a more elegant solution for the queue to operate correctly as long as the listener task is running

super elegant 🤩  🚀 



Patch:
diff --git a/newsfragments/3378.bugfix.rst b/newsfragments/3378.bugfix.rst
new file mode 100644
index 00000000..87597b9e
--- /dev/null
+++ b/newsfragments/3378.bugfix.rst
@@ -0,0 +1,1 @@
+Properly propagate exceptions from the message listener task to the main loop for persistent connection providers.
diff --git a/web3/exceptions.py b/web3/exceptions.py
index a51c324a..1dd23852 100644
--- a/web3/exceptions.py
+++ b/web3/exceptions.py
@@ -1,6 +1,7 @@
 import datetime
 import time
 from typing import (
+    TYPE_CHECKING,
     Any,
     Dict,
     Optional,
@@ -16,6 +17,9 @@ from web3.types import (
     RPCResponse,
 )
 
+if TYPE_CHECKING:
+    import asyncio
+
 
 class Web3Exception(Exception):
     """
@@ -307,6 +311,22 @@ class BadResponseFormat(Web3Exception):
     """
 
 
+class TaskNotRunning(Web3Exception):
+    """
+    Used to signal between asyncio contexts that a task that is being awaited
+    is not currently running.
+    """
+
+    def __init__(
+        self, task: "asyncio.Task[Any]", message: Optional[str] = None
+    ) -> None:
+        self.task = task
+        if message is None:
+            message = f"Task {task} is not running."
+        self.message = message
+        super().__init__(message)
+
+
 class Web3RPCError(Web3Exception):
     """
     Raised when a JSON-RPC response contains an error field.
diff --git a/web3/main.py b/web3/main.py
index 7bfe7dc1..c12b140d 100644
--- a/web3/main.py
+++ b/web3/main.py
@@ -144,6 +144,7 @@ from web3.types import (
 
 if TYPE_CHECKING:
     from web3._utils.empty import Empty  # noqa: F401
+    from web3.providers.persistent import PersistentConnectionProvider  # noqa: F401
 
 
 def get_async_default_modules() -> Dict[str, Union[Type[Module], Sequence[Any]]]:
@@ -511,12 +512,11 @@ class AsyncWeb3(BaseWeb3):
         "when instantiating via ``async for``."
     )
     async def __aiter__(self) -> AsyncIterator[Self]:
-        if not await self.provider.is_connected():
-            await self.provider.connect()
-
+        provider = self.provider
         while True:
-            try:
-                yield self
-            except Exception:
-                # provider should handle connection / reconnection
-                continue
+            await provider.connect()
+            yield self
+            cast("PersistentConnectionProvider", provider).logger.error(
+                "Connection interrupted, attempting to reconnect..."
+            )
+            await provider.disconnect()
diff --git a/web3/manager.py b/web3/manager.py
index b1870624..59f7ef97 100644
--- a/web3/manager.py
+++ b/web3/manager.py
@@ -35,6 +35,7 @@ from web3.exceptions import (
     BadResponseFormat,
     MethodUnavailable,
     ProviderConnectionError,
+    TaskNotRunning,
     Web3RPCError,
     Web3TypeError,
 )
@@ -408,14 +409,24 @@ class RequestManager:
             )
 
         while True:
-            response = await self._request_processor.pop_raw_response(subscription=True)
-            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_response(response)
+            try:
+                response = await self._request_processor.pop_raw_response(
+                    subscription=True
+                )
+                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_response(response)
+            except TaskNotRunning:
+                self._provider._handle_listener_task_exceptions()
+                self.logger.error(
+                    "Message listener background task has stopped unexpectedly. "
+                    "Stopping message stream."
+                )
+                raise StopAsyncIteration
 
     async def _process_response(self, response: RPCResponse) -> RPCResponse:
         provider = cast(PersistentConnectionProvider, self._provider)
diff --git a/web3/providers/persistent/async_ipc.py b/web3/providers/persistent/async_ipc.py
index 67989bcc..c9b81b90 100644
--- a/web3/providers/persistent/async_ipc.py
+++ b/web3/providers/persistent/async_ipc.py
@@ -59,11 +59,12 @@ class AsyncIPCProvider(PersistentConnectionProvider):
 
     _reader: Optional[asyncio.StreamReader] = None
     _writer: Optional[asyncio.StreamWriter] = None
+    _decoder: json.JSONDecoder = json.JSONDecoder()
+    _raw_message: str = ""
 
     def __init__(
         self,
         ipc_path: Optional[Union[str, Path]] = None,
-        max_connection_retries: int = 5,
         # `PersistentConnectionProvider` kwargs can be passed through
         **kwargs: Any,
     ) -> None:
@@ -74,7 +75,6 @@ class AsyncIPCProvider(PersistentConnectionProvider):
         else:
             raise Web3TypeError("ipc_path must be of type string or pathlib.Path")
 
-        self._max_connection_retries = max_connection_retries
         super().__init__(**kwargs)
 
     def __str__(self) -> str:
@@ -99,48 +99,16 @@ class AsyncIPCProvider(PersistentConnectionProvider):
                 )
             return False
 
-    async def connect(self) -> None:
-        _connection_attempts = 0
-        _backoff_rate_change = 1.75
-        _backoff_time = 1.75
-
-        while _connection_attempts != self._max_connection_retries:
-            try:
-                _connection_attempts += 1
-                self._reader, self._writer = await async_get_ipc_socket(self.ipc_path)
-                self._message_listener_task = asyncio.create_task(
-                    self._message_listener()
-                )
-                break
-            except OSError as e:
-                if _connection_attempts == self._max_connection_retries:
-                    raise ProviderConnectionError(
-                        f"Could not connect to: {self.ipc_path}. "
-                        f"Retries exceeded max of {self._max_connection_retries}."
-                    ) from e
-                self.logger.info(
-                    f"Could not connect to: {self.ipc_path}. Retrying in "
-                    f"{round(_backoff_time, 1)} seconds.",
-                    exc_info=True,
-                )
-                await asyncio.sleep(_backoff_time)
-                _backoff_time *= _backoff_rate_change
+    async def _provider_specific_connect(self) -> None:
+        self._reader, self._writer = await async_get_ipc_socket(self.ipc_path)
 
-    async def disconnect(self) -> None:
+    async def _provider_specific_disconnect(self) -> None:
         if self._writer and not self._writer.is_closing():
             self._writer.close()
             await self._writer.wait_closed()
             self._writer = None
-            self.logger.debug(f'Successfully disconnected from : "{self.ipc_path}')
-
-        try:
-            self._message_listener_task.cancel()
-            await self._message_listener_task
+        if self._reader:
             self._reader = None
-        except (asyncio.CancelledError, StopAsyncIteration):
-            pass
-
-        self._request_processor.clear_caches()
 
     async def _reset_socket(self) -> None:
         self._writer.close()
@@ -172,43 +140,22 @@ class AsyncIPCProvider(PersistentConnectionProvider):
 
         return response
 
-    async def _message_listener(self) -> None:
-        self.logger.info(
-            "IPC socket listener background task started. Storing all messages in "
-            "appropriate request processor queues / caches to be processed."
-        )
-        raw_message = ""
-        decoder = json.JSONDecoder()
-
-        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)
+    async def _provider_specific_message_listener(self) -> None:
+        self._raw_message += to_text(await self._reader.read(4096)).lstrip()
 
+        while self._raw_message:
             try:
-                raw_message += to_text(await self._reader.read(4096)).lstrip()
-
-                while raw_message:
-                    try:
-                        response, pos = decoder.raw_decode(raw_message)
-                    except JSONDecodeError:
-                        break
-
-                    is_subscription = response.get("method") == "eth_subscription"
-                    await self._request_processor.cache_raw_response(
-                        response, subscription=is_subscription
-                    )
-                    raw_message = raw_message[pos:].lstrip()
-            except Exception as e:
-                if not self.silence_listener_task_exceptions:
-                    loop = asyncio.get_event_loop()
-                    for task in asyncio.all_tasks(loop=loop):
-                        task.cancel()
-                    raise e
-
-                self.logger.error(
-                    "Exception caught in listener, error logging and keeping listener "
-                    f"background task alive.\n    error={e}"
-                )
-                # if only error logging, reset the ``raw_message`` buffer and continue
-                raw_message = ""
+                response, pos = self._decoder.raw_decode(self._raw_message)
+            except JSONDecodeError:
+                break
+
+            is_subscription = response.get("method") == "eth_subscription"
+            await self._request_processor.cache_raw_response(
+                response, subscription=is_subscription
+            )
+            self._raw_message = self._raw_message[pos:].lstrip()
+
+    def _error_log_listener_task_exception(self, e: Exception) -> None:
+        super()._error_log_listener_task_exception(e)
+        # reset the raw message buffer on exception when error logging
+        self._raw_message = ""
diff --git a/web3/providers/persistent/persistent.py b/web3/providers/persistent/persistent.py
index 6eb1d936..d7b75d03 100644
--- a/web3/providers/persistent/persistent.py
+++ b/web3/providers/persistent/persistent.py
@@ -7,10 +7,17 @@ from typing import (
     Optional,
 )
 
+from websockets import (
+    ConnectionClosed,
+    WebSocketException,
+)
+
 from web3._utils.caching import (
     generate_cache_key,
 )
 from web3.exceptions import (
+    ProviderConnectionError,
+    TaskNotRunning,
     TimeExhausted,
 )
 from web3.providers.async_base import (
@@ -40,6 +47,7 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
         request_timeout: float = DEFAULT_PERSISTENT_CONNECTION_TIMEOUT,
         subscription_response_queue_size: int = 500,
         silence_listener_task_exceptions: bool = False,
+        max_connection_retries: int = 5,
     ) -> None:
         super().__init__()
         self._request_processor = RequestProcessor(
@@ -48,6 +56,7 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
         )
         self.request_timeout = request_timeout
         self.silence_listener_task_exceptions = silence_listener_task_exceptions
+        self._max_connection_retries = max_connection_retries
 
     def get_endpoint_uri_or_ipc_path(self) -> str:
         if hasattr(self, "endpoint_uri"):
@@ -61,14 +70,122 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
             )
 
     async def connect(self) -> None:
-        raise NotImplementedError("Must be implemented by subclasses")
+        _connection_attempts = 0
+        _backoff_rate_change = 1.75
+        _backoff_time = 1.75
+
+        while _connection_attempts != self._max_connection_retries:
+            try:
+                _connection_attempts += 1
+                self.logger.info(
+                    f"Connecting to: {self.get_endpoint_uri_or_ipc_path()}"
+                )
+                await self._provider_specific_connect()
+                self._message_listener_task = asyncio.create_task(
+                    self._message_listener()
+                )
+                self._message_listener_task.add_done_callback(
+                    self._message_listener_callback
+                )
+                self.logger.info(
+                    f"Successfully connected to: {self.get_endpoint_uri_or_ipc_path()}"
+                )
+                break
+            except (WebSocketException, OSError) as e:
+                if _connection_attempts == self._max_connection_retries:
+                    raise ProviderConnectionError(
+                        f"Could not connect to: {self.get_endpoint_uri_or_ipc_path()}. "
+                        f"Retries exceeded max of {self._max_connection_retries}."
+                    ) from e
+                self.logger.info(
+                    f"Could not connect to: {self.get_endpoint_uri_or_ipc_path()}. "
+                    f"Retrying in {round(_backoff_time, 1)} seconds.",
+                    exc_info=True,
+                )
+                await asyncio.sleep(_backoff_time)
+                _backoff_time *= _backoff_rate_change
 
     async def disconnect(self) -> None:
+        try:
+            if self._message_listener_task:
+                self._message_listener_task.cancel()
+                await self._message_listener_task
+        except (asyncio.CancelledError, StopAsyncIteration, ConnectionClosed):
+            pass
+        finally:
+            self._message_listener_task = None
+            self.logger.info("Message listener background task successfully shut down.")
+
+        await self._provider_specific_disconnect()
+        self._request_processor.clear_caches()
+        self.logger.info(
+            f"Successfully disconnected from: {self.get_endpoint_uri_or_ipc_path()}"
+        )
+
+    # -- private methods -- #
+
+    async def _provider_specific_connect(self) -> None:
         raise NotImplementedError("Must be implemented by subclasses")
 
-    async def _message_listener(self) -> None:
+    async def _provider_specific_disconnect(self) -> None:
         raise NotImplementedError("Must be implemented by subclasses")
 
+    async def _provider_specific_message_listener(self) -> None:
+        raise NotImplementedError("Must be implemented by subclasses")
+
+    def _message_listener_callback(
+        self, message_listener_task: "asyncio.Task[None]"
+    ) -> None:
+        # Puts a `TaskNotRunning` in the queue to signal the end of the listener task
+        # to any running subscription streams that are awaiting a response.
+        self._request_processor._subscription_response_queue.put_nowait(
+            TaskNotRunning(message_listener_task)
+        )
+
+    async def _message_listener(self) -> None:
+        self.logger.info(
+            f"{self.__class__.__qualname__} 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:
+                await self._provider_specific_message_listener()
+            except Exception as e:
+                if not self.silence_listener_task_exceptions:
+                    raise e
+                else:
+                    self._error_log_listener_task_exception(e)
+
+    def _error_log_listener_task_exception(self, e: Exception) -> None:
+        """
+        When silencing listener task exceptions, this method is used to log the
+        exception and keep the listener task alive. Override this method to fine-tune
+        error logging behavior for the implementation class.
+        """
+        self.logger.error(
+            "Exception caught in listener, error logging and keeping "
+            "listener background task alive."
+            f"\n    error={e.__class__.__name__}: {e}"
+        )
+
+    def _handle_listener_task_exceptions(self) -> None:
+        """
+        Should be called every time a `PersistentConnectionProvider` is polling for
+        messages in the main loop. If the message listener task has completed and an
+        exception was recorded, raise the exception in the main loop.
+        """
+        msg_listener_task = getattr(self, "_message_listener_task", None)
+        if (
+            msg_listener_task
+            and msg_listener_task.done()
+            and msg_listener_task.exception()
+        ):
+            raise msg_listener_task.exception()
+
     async def _get_response_for_request_id(
         self, request_id: RPCId, timeout: Optional[float] = None
     ) -> RPCResponse:
@@ -79,10 +196,9 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
             request_cache_key = generate_cache_key(request_id)
 
             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 in the
-                # queue.
-                await asyncio.sleep(0)
+                # 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(
@@ -92,11 +208,13 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
                         cache_key=request_cache_key,
                     )
                     return popped_response
+                else:
+                    await asyncio.sleep(0)
 
         try:
             # Add the request timeout around the while loop that checks the request
-            # cache and tried to recv(). If the request is neither in the cache, nor
-            # received within the request_timeout, raise ``TimeExhausted``.
+            # cache. If the request is not in the cache within the request_timeout,
+            # raise ``TimeExhausted``.
             return await asyncio.wait_for(_match_response_id_to_request_id(), timeout)
         except asyncio.TimeoutError:
             raise TimeExhausted(
diff --git a/web3/providers/persistent/request_processor.py b/web3/providers/persistent/request_processor.py
index e18049e6..5abbac56 100644
--- a/web3/providers/persistent/request_processor.py
+++ b/web3/providers/persistent/request_processor.py
@@ -2,13 +2,17 @@ import asyncio
 from copy import (
     copy,
 )
+import sys
 from typing import (
     TYPE_CHECKING,
     Any,
     Callable,
     Dict,
+    Generic,
     Optional,
     Tuple,
+    TypeVar,
+    Union,
 )
 
 from web3._utils.caching import (
@@ -16,6 +20,7 @@ from web3._utils.caching import (
     generate_cache_key,
 )
 from web3.exceptions import (
+    TaskNotRunning,
     Web3ValueError,
 )
 from web3.types import (
@@ -31,6 +36,34 @@ if TYPE_CHECKING:
         PersistentConnectionProvider,
     )
 
+T = TypeVar("T")
+
+# TODO: This is an ugly hack for python 3.8. Remove this after we drop support for it
+#  and use `asyncio.Queue[T]` type directly in the `TaskReliantQueue` class.
+if sys.version_info >= (3, 9):
+
+    class _TaskReliantQueue(asyncio.Queue[T], Generic[T]):
+        pass
+
+else:
+
+    class _TaskReliantQueue(asyncio.Queue, Generic[T]):  # type: ignore
+        pass
+
+
+class TaskReliantQueue(_TaskReliantQueue[T]):
+    """
+    A queue that relies on a task to be running to process items in the queue.
+    """
+
+    async def get(self) -> T:
+        item = await super().get()
+        if isinstance(item, Exception):
+            # if the item is an exception, raise it so the task can handle this case
+            # more gracefully
+            raise item
+        return item
+
 
 class RequestProcessor:
     _subscription_queue_synced_with_ws_stream: bool = False
@@ -44,9 +77,9 @@ class RequestProcessor:
 
         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
-        )
+        self._subscription_response_queue: TaskReliantQueue[
+            Union[RPCResponse, TaskNotRunning]
+        ] = TaskReliantQueue(maxsize=subscription_response_queue_size)
 
     @property
     def active_subscriptions(self) -> Dict[str, Any]:
@@ -218,7 +251,7 @@ class RequestProcessor:
     ) -> None:
         if subscription:
             if self._subscription_response_queue.full():
-                self._provider.logger.info(
+                self._provider.logger.debug(
                     "Subscription queue is full. Waiting for provider to consume "
                     "messages before caching."
                 )
@@ -289,6 +322,6 @@ class RequestProcessor:
         """Clear the request processor caches."""
         self._request_information_cache.clear()
         self._request_response_cache.clear()
-        self._subscription_response_queue = asyncio.Queue(
+        self._subscription_response_queue = TaskReliantQueue(
             maxsize=self._subscription_response_queue.maxsize
         )
diff --git a/web3/providers/persistent/websocket.py b/web3/providers/persistent/websocket.py
index 5b737e01..3854eaac 100644
--- a/web3/providers/persistent/websocket.py
+++ b/web3/providers/persistent/websocket.py
@@ -61,7 +61,6 @@ class WebSocketProvider(PersistentConnectionProvider):
     logger = logging.getLogger("web3.providers.WebSocketProvider")
     is_async: bool = True
 
-    _max_connection_retries: int = 5
     _ws: Optional[WebSocketClientProtocol] = None
 
     def __init__(
@@ -116,47 +115,13 @@ class WebSocketProvider(PersistentConnectionProvider):
                 ) from e
             return False
 
-    async def connect(self) -> None:
-        _connection_attempts = 0
-        _backoff_rate_change = 1.75
-        _backoff_time = 1.75
-
-        while _connection_attempts != self._max_connection_retries:
-            try:
-                _connection_attempts += 1
-                self._ws = await connect(self.endpoint_uri, **self.websocket_kwargs)
-                self._message_listener_task = asyncio.create_task(
-                    self._message_listener()
-                )
-                break
-            except WebSocketException as e:
-                if _connection_attempts == self._max_connection_retries:
-                    raise ProviderConnectionError(
-                        f"Could not connect to endpoint: {self.endpoint_uri}. "
-                        f"Retries exceeded max of {self._max_connection_retries}."
-                    ) from e
-                self.logger.info(
-                    f"Could not connect to endpoint: {self.endpoint_uri}. Retrying in "
-                    f"{round(_backoff_time, 1)} seconds.",
-                    exc_info=True,
-                )
-                await asyncio.sleep(_backoff_time)
-                _backoff_time *= _backoff_rate_change
+    async def _provider_specific_connect(self) -> None:
+        self._ws = await connect(self.endpoint_uri, **self.websocket_kwargs)
 
-    async def disconnect(self) -> None:
+    async def _provider_specific_disconnect(self) -> None:
         if self._ws is not None and not self._ws.closed:
             await self._ws.close()
             self._ws = None
-            self.logger.debug(
-                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
     async def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
@@ -176,34 +141,12 @@ class WebSocketProvider(PersistentConnectionProvider):
 
         return response
 
-    async def _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.
+    async def _provider_specific_message_listener(self) -> None:
+        async for raw_message in self._ws:
             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 not self.silence_listener_task_exceptions:
-                    loop = asyncio.get_event_loop()
-                    for task in asyncio.all_tasks(loop=loop):
-                        task.cancel()
-                    raise e
-
-                self.logger.error(
-                    "Exception caught in listener, error logging and keeping "
-                    "listener background task alive."
-                    f"\n    error={e.__class__.__name__}: {e}"
-                )
+            response = json.loads(raw_message)
+            subscription = response.get("method") == "eth_subscription"
+            await self._request_processor.cache_raw_response(
+                response, subscription=subscription
+            )


Test patch:
diff --git a/tests/core/providers/test_async_ipc_provider.py b/tests/core/providers/test_async_ipc_provider.py
index e9c94835..1312cafd 100644
--- a/tests/core/providers/test_async_ipc_provider.py
+++ b/tests/core/providers/test_async_ipc_provider.py
@@ -9,6 +9,10 @@ from threading import (
 )
 import time
 
+from websockets import (
+    ConnectionClosed,
+)
+
 from web3 import (
     AsyncWeb3,
 )
@@ -112,6 +116,50 @@ def test_ipc_tilde_in_path():
     assert AsyncIPCProvider(pathlib.Path("~/foo")).ipc_path == expected_path
 
 
+def test_get_endpoint_uri_or_ipc_path_returns_ipc_path():
+    provider = AsyncIPCProvider(pathlib.Path("/path/to/file"))
+    assert (
+        provider.get_endpoint_uri_or_ipc_path() == "/path/to/file" == provider.ipc_path
+    )
+
+
+# -- async -- #
+
+
+@pytest.mark.asyncio
+async def test_disconnect_cleanup(
+    simple_ipc_server,
+    jsonrpc_ipc_pipe_path,
+):
+    w3 = await AsyncWeb3(AsyncIPCProvider(pathlib.Path(jsonrpc_ipc_pipe_path)))
+    provider = w3.provider
+
+    assert provider._message_listener_task is not None
+    assert provider._reader is not None
+    assert provider._writer is not None
+
+    # put some items in each cache
+    provider._request_processor._request_response_cache.cache("0", "0x1337")
+    provider._request_processor._request_information_cache.cache("0", "0x1337")
+    provider._request_processor._subscription_response_queue.put_nowait({"id": "0"})
+    assert len(provider._request_processor._request_response_cache) == 1
+    assert len(provider._request_processor._request_information_cache) == 1
+    assert provider._request_processor._subscription_response_queue.qsize() == 1
+
+    await w3.provider.disconnect()
+
+    assert not provider._message_listener_task
+    assert not w3.provider._reader
+    assert not w3.provider._writer
+    assert len(provider._request_processor._request_response_cache) == 0
+    assert len(provider._request_processor._request_information_cache) == 0
+    assert provider._request_processor._subscription_response_queue.empty()
+
+
+async def _raise_connection_closed(*_args, **_kwargs):
+    raise ConnectionClosed(None, None)
+
+
 @pytest.mark.asyncio
 async def test_provider_is_connected(jsonrpc_ipc_pipe_path, serve_empty_result):
     w3 = await AsyncWeb3(AsyncIPCProvider(pathlib.Path(jsonrpc_ipc_pipe_path)))
@@ -170,8 +218,43 @@ async def test_eth_subscription(jsonrpc_ipc_pipe_path, serve_subscription_result
         await w3.provider.disconnect()
 
 
-def test_get_endpoint_uri_or_ipc_path_returns_ipc_path():
-    provider = AsyncIPCProvider(pathlib.Path("/path/to/file"))
-    assert (
-        provider.get_endpoint_uri_or_ipc_path() == "/path/to/file" == provider.ipc_path
-    )
+@pytest.mark.asyncio
+async def test_async_iterator_pattern_exception_handling_for_requests(
+    simple_ipc_server,
+    jsonrpc_ipc_pipe_path,
+):
+    exception_caught = False
+    async for w3 in AsyncWeb3(AsyncIPCProvider(pathlib.Path(jsonrpc_ipc_pipe_path))):
+        # patch the listener to raise `ConnectionClosed` on read
+        w3.provider._reader.read = _raise_connection_closed
+        try:
+            await w3.eth.block_number
+        except ConnectionClosed:
+            exception_caught = True
+            break
+
+        pytest.fail("Expected `ConnectionClosed` exception.")
+
+    assert exception_caught
+
+
+@pytest.mark.asyncio
+async def test_async_iterator_pattern_exception_handling_for_subscriptions(
+    simple_ipc_server,
+    jsonrpc_ipc_pipe_path,
+):
+    exception_caught = False
+    async for w3 in AsyncWeb3(AsyncIPCProvider(pathlib.Path(jsonrpc_ipc_pipe_path))):
+        # patch the listener to raise `ConnectionClosed` on read
+        w3.provider._reader.read = _raise_connection_closed
+        try:
+            async for _ in w3.socket.process_subscriptions():
+                # raises exception
+                pass
+        except ConnectionClosed:
+            exception_caught = True
+            break
+
+        pytest.fail("Expected `ConnectionClosed` exception.")
+
+    assert exception_caught
diff --git a/tests/core/providers/test_websocket_provider.py b/tests/core/providers/test_websocket_provider.py
index ebb92c8d..118236a4 100644
--- a/tests/core/providers/test_websocket_provider.py
+++ b/tests/core/providers/test_websocket_provider.py
@@ -10,6 +10,9 @@ from unittest.mock import (
 from eth_utils import (
     to_bytes,
 )
+from websockets import (
+    ConnectionClosed,
+)
 
 from web3 import (
     AsyncWeb3,
@@ -33,22 +36,67 @@ from web3.types import (
 
 def _mock_ws(provider):
     provider._ws = AsyncMock()
+    provider._ws.closed = False
 
 
-async def _coro():
-    return None
+async def _mocked_ws_conn():
+    _conn = AsyncMock()
+    _conn.closed = False
+    return _conn
 
 
 class WSException(Exception):
     pass
 
 
+def test_get_endpoint_uri_or_ipc_path_returns_endpoint_uri():
+    provider = WebSocketProvider("ws://mocked")
+    assert (
+        provider.get_endpoint_uri_or_ipc_path()
+        == "ws://mocked"
+        == provider.endpoint_uri
+    )
+
+
+# -- async -- #
+
+
+@pytest.mark.asyncio
+async def test_disconnect_cleanup():
+    provider = WebSocketProvider("ws://mocked")
+
+    with patch(
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
+    ):
+        await provider.connect()
+
+    assert provider._ws is not None
+    assert provider._message_listener_task is not None
+
+    # put some items in each cache
+    provider._request_processor._request_response_cache.cache("0", "0x1337")
+    provider._request_processor._request_information_cache.cache("0", "0x1337")
+    provider._request_processor._subscription_response_queue.put_nowait({"id": "0"})
+    assert len(provider._request_processor._request_response_cache) == 1
+    assert len(provider._request_processor._request_information_cache) == 1
+    assert provider._request_processor._subscription_response_queue.qsize() == 1
+
+    await provider.disconnect()
+
+    assert provider._ws is None
+    assert len(provider._request_processor._request_response_cache) == 0
+    assert len(provider._request_processor._request_information_cache) == 0
+    assert provider._request_processor._subscription_response_queue.empty()
+
+
 @pytest.mark.asyncio
 async def test_async_make_request_returns_desired_response():
     provider = WebSocketProvider("ws://mocked")
 
     with patch(
-        "web3.providers.persistent.websocket.connect", new=lambda *_1, **_2: _coro()
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
     ):
         await provider.connect()
 
@@ -109,14 +157,15 @@ async def test_async_make_request_times_out_of_while_loop_looking_for_response()
 
 
 @pytest.mark.asyncio
-async def test_msg_listener_task_starts_on_provider_connect_and_cancels_on_disconnect():
+async def test_msg_listener_task_starts_on_provider_connect_and_clears_on_disconnect():
     provider = WebSocketProvider("ws://mocked")
     _mock_ws(provider)
 
     assert provider._message_listener_task is None
 
     with patch(
-        "web3.providers.persistent.websocket.connect", new=lambda *_1, **_2: _coro()
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
     ):
         await provider.connect()  # connect
 
@@ -125,8 +174,7 @@ async def test_msg_listener_task_starts_on_provider_connect_and_cancels_on_disco
 
     await provider.disconnect()  # disconnect
 
-    assert provider._message_listener_task.cancelled()
-    assert provider._message_listener_task.done()
+    assert not provider._message_listener_task
 
 
 @pytest.mark.asyncio
@@ -135,7 +183,8 @@ async def test_msg_listener_task_raises_exceptions_by_default():
     _mock_ws(provider)
 
     with patch(
-        "web3.providers.persistent.websocket.connect", new=lambda *_1, **_2: _coro()
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
     ):
         await provider.connect()
         assert provider._message_listener_task is not None
@@ -158,7 +207,8 @@ async def test_msg_listener_task_silences_exceptions_and_error_logs_when_configu
     _mock_ws(provider)
 
     with patch(
-        "web3.providers.persistent.websocket.connect", new=lambda *_1, **_2: _coro()
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
     ):
         await provider.connect()
         assert provider._message_listener_task is not None
@@ -190,7 +240,8 @@ async def test_listen_event_awaits_msg_processing_when_subscription_queue_is_ful
     is full.
     """
     with patch(
-        "web3.providers.persistent.websocket.connect", new=lambda *_1, **_2: _coro()
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
     ):
         async_w3 = await AsyncWeb3(WebSocketProvider("ws://mocked"))
 
@@ -264,10 +315,53 @@ async def test_listen_event_awaits_msg_processing_when_subscription_queue_is_ful
     await async_w3.provider.disconnect()
 
 
-def test_get_endpoint_uri_or_ipc_path_returns_endpoint_uri():
-    provider = WebSocketProvider("ws://mocked")
-    assert (
-        provider.get_endpoint_uri_or_ipc_path()
-        == "ws://mocked"
-        == provider.endpoint_uri
-    )
+@pytest.mark.asyncio
+async def test_async_iterator_pattern_exception_handling_for_requests():
+    iterations = 1
+
+    with patch(
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: WebSocketMessageStreamMock(
+            raise_exception=ConnectionClosed(None, None)
+        ),
+    ):
+        async for w3 in AsyncWeb3(WebSocketProvider("ws://mocked")):
+            try:
+                await w3.eth.block_number
+            except ConnectionClosed:
+                if iterations == 3:
+                    break
+                else:
+                    iterations += 1
+                    continue
+
+            pytest.fail("Expected `ConnectionClosed` exception.")
+
+        assert iterations == 3
+
+
+@pytest.mark.asyncio
+async def test_async_iterator_pattern_exception_handling_for_subscriptions():
+    iterations = 1
+
+    with patch(
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: WebSocketMessageStreamMock(
+            raise_exception=ConnectionClosed(None, None)
+        ),
+    ):
+        async for w3 in AsyncWeb3(WebSocketProvider("ws://mocked")):
+            try:
+                async for _ in w3.socket.process_subscriptions():
+                    # raises exception
+                    pass
+            except ConnectionClosed:
+                if iterations == 3:
+                    break
+                else:
+                    iterations += 1
+                    continue
+
+            pytest.fail("Expected `ConnectionClosed` exception.")
+
+        assert iterations == 3
diff --git a/web3/_utils/module_testing/module_testing_utils.py b/web3/_utils/module_testing/module_testing_utils.py
index 8b87f455..a0e9b582 100644
--- a/web3/_utils/module_testing/module_testing_utils.py
+++ b/web3/_utils/module_testing/module_testing_utils.py
@@ -177,6 +177,12 @@ class WebSocketMessageStreamMock:
         self.messages = deque(messages) if messages else deque()
         self.raise_exception = raise_exception
 
+    def __await__(self) -> Generator[Any, Any, "Self"]:
+        async def __async_init__() -> "Self":
+            return self
+
+        return __async_init__().__await__()
+
     def __aiter__(self) -> "Self":
         return self
 
@@ -189,6 +195,13 @@ class WebSocketMessageStreamMock:
 
         return self.messages.popleft()
 
+    @staticmethod
+    async def pong() -> Literal[False]:
+        return False
+
+    async def connect(self) -> None:
+        pass
+
     async def send(self, data: bytes) -> None:
         pass