Port changes from #3424 to main / v7:

- Raise ``StopAsyncIteration`` if a ``ConnectionClosedOK`` is
  raised by the server in the websocket provider. Handle this
  appropriately instead of handling the exception directly.
- Add a sanity check for AsyncIPC as well to stop iteration when
  / if the reader does not exist.

Author: fselmo

newsfragments/3432.bugfix.rst

@@ -0,0 +1 @@
+Handle ``ConnectionClosedOK`` case for ``WebSocketProvider``. If the connection is closed gracefully, debug log and silently break out of the message iterator loop by raising ``StopAsyncIteration``.

web3/_utils/module_testing/module_testing_utils.py

@@ -1,6 +1,4 @@
-from collections import (
-    deque,
-)
+import asyncio
 from typing import (
     TYPE_CHECKING,
     Any,
@@ -179,7 +177,9 @@ class WebSocketMessageStreamMock:
     def __init__(
         self, messages: Collection[bytes] = None, raise_exception: Exception = None
     ) -> None:
-        self.messages = deque(messages) if messages else deque()
+        self.queue = asyncio.Queue()  # type: ignore  # py38 issue
+        for msg in messages or []:
+            self.queue.put_nowait(msg)
         self.raise_exception = raise_exception
 
     def __await__(self) -> Generator[Any, Any, "Self"]:
@@ -192,13 +192,12 @@ def __aiter__(self) -> "Self":
         return self
 
     async def __anext__(self) -> bytes:
+        return await self.queue.get()
+
+    async def recv(self) -> bytes:
         if self.raise_exception:
             raise self.raise_exception
-
-        elif len(self.messages) == 0:
-            raise StopAsyncIteration
-
-        return self.messages.popleft()
+        return await self.queue.get()
 
     @staticmethod
     async def pong() -> Literal[False]:

web3/manager.py

@@ -20,9 +20,6 @@
 from hexbytes import (
     HexBytes,
 )
-from websockets.exceptions import (
-    ConnectionClosedOK,
-)
 
 from web3._utils.batching import (
     RequestBatcher,
@@ -586,7 +583,4 @@ def __aiter__(self) -> Self:
         return self
 
     async def __anext__(self) -> RPCResponse:
-        try:
-            return await self.manager._get_next_message()
-        except ConnectionClosedOK:
-            raise StopAsyncIteration
+        return await self.manager._get_next_message()

web3/providers/persistent/async_ipc.py

@@ -171,8 +171,15 @@ async def make_batch_request(
         return response
 
     async def _provider_specific_message_listener(self) -> None:
-        self._raw_message += to_text(await self._reader.read(4096)).lstrip()
+        if self._reader is None:
+            # sanity check to ensure the reader is initialized
+            self.logger.debug(
+                "IPC reader is not initialized. If this was not expected, "
+                "check the connection status."
+            )
+            raise StopAsyncIteration
 
+        self._raw_message += to_text(await self._reader.read(4096)).lstrip()
         while self._raw_message:
             try:
                 response, pos = self._decoder.raw_decode(self._raw_message)

web3/providers/persistent/persistent.py

@@ -160,6 +160,8 @@ async def _message_listener(self) -> None:
             await asyncio.sleep(0)
             try:
                 await self._provider_specific_message_listener()
+            except StopAsyncIteration:
+                raise
             except Exception as e:
                 if not self.silence_listener_task_exceptions:
                     raise e
@@ -202,10 +204,6 @@ async def _match_response_id_to_request_id() -> RPCResponse:
             request_cache_key = generate_cache_key(request_id)
 
             while True:
-                # check if an exception was recorded in the listener task and raise it
-                # in the main loop if so
-                self._handle_listener_task_exceptions()
-
                 if request_cache_key in self._request_processor._request_response_cache:
                     self.logger.debug(
                         f"Popping response for id {request_id} from cache."
@@ -215,6 +213,9 @@ async def _match_response_id_to_request_id() -> RPCResponse:
                     )
                     return popped_response
                 else:
+                    # check if an exception was recorded in the listener task and raise
+                    # it in the main loop if so
+                    self._handle_listener_task_exceptions()
                     await asyncio.sleep(0)
 
         try:

web3/providers/persistent/websocket.py

@@ -25,6 +25,7 @@
     connect,
 )
 from websockets.exceptions import (
+    ConnectionClosedOK,
     WebSocketException,
 )
 
@@ -169,18 +170,23 @@ async def make_batch_request(
         return response
 
     async def _provider_specific_message_listener(self) -> None:
-        async for raw_message in self._ws:
-            await asyncio.sleep(0)
-
-            response = json.loads(raw_message)
-            if isinstance(response, list):
-                response = sort_batch_response_by_response_ids(response)
-
-            subscription = (
-                response.get("method") == "eth_subscription"
-                if not isinstance(response, list)
-                else False
-            )
-            await self._request_processor.cache_raw_response(
-                response, subscription=subscription
-            )
+        try:
+            while True:
+                raw_message = await self._ws.recv()
+                await asyncio.sleep(0)
+
+                response = json.loads(raw_message)
+                if isinstance(response, list):
+                    response = sort_batch_response_by_response_ids(response)
+
+                subscription = (
+                    response.get("method") == "eth_subscription"
+                    if not isinstance(response, list)
+                    else False
+                )
+                await self._request_processor.cache_raw_response(
+                    response, subscription=subscription
+                )
+        except ConnectionClosedOK:
+            self.logger.debug("WebSocket connection closed gracefully.")
+            raise StopAsyncIteration