Repository: ethereum/web3.py

Issue:
SubscriptionManager.handle_subscriptions hangs on PersistentConnectionClosedOK exception
### What happened?

  I'm recieving `Message listener background task for the provider has stopped unexpectedly. Stopping subscription handling.` in my logs like once in a week or so. And in that case `SubscriptionManager.handle_subscriptions` do not returns nor raises - which stops WebSocketProvider iteratior from reconnecting.
  When `PersistentConnectionProvider._message_listener` raises `PersistentConnectionClosedOK` exceptions, it returns without exceptions and `_message_listener_callback` places `TaskNotRunning` on the subscription queue.
  So `SubscriptionManager.handle_subscriptions` in `except TaskNotRunning` block just logs error message and continues the loop. Then `await length.get()` hangs in the next iteration because there is no more active `message_listener`.
  It seems to me that `return` or `break` is missing or something. I believe `SubscriptionManager.handle_subscriptions` should stop when the connection is closed gracefully by the server.

### Code that produced the error

```python
async for w3_websocket in AsyncWeb3(WebSocketProvider(some_rpc_ws_url)):
    sm = w3_websocket.subscription_manager
    await sm.subscribe([some_subscription])
    with suppress(ConnectionClosed):
        await sm.handle_subscriptions()
        # hangs when PersistentConnectionClosedOK occures with log message
        # logger.error(
        #     "Message listener background task for the provider has stopped "
        #     "unexpectedly. Stopping subscription handling."
        # )

    # cleanup
    for s in sm.subscriptions:
        sm._remove_subscription(s)
```

### Full error output

```shell

```

### Fill this section in if you know how this could or should be fixed

_No response_

### web3 Version

7.11.0

### Python Version

3.13.2

### Operating System

linux | ubuntu 24.04.2 LTS

### Output from `pip freeze`

```shell

```

Discussion hints:

Solution patch:
diff --git a/newsfragments/3690.bugfix.rst b/newsfragments/3690.bugfix.rst
new file mode 100644
index 00000000..e587bde8
--- /dev/null
+++ b/newsfragments/3690.bugfix.rst
@@ -0,0 +1,1 @@
+Properly handle ``PersistentConnectionClosedOK`` for ``subscription_manager.handle_subscriptions()``.
diff --git a/web3/providers/persistent/async_ipc.py b/web3/providers/persistent/async_ipc.py
index 583c9ae4..ad182382 100644
--- a/web3/providers/persistent/async_ipc.py
+++ b/web3/providers/persistent/async_ipc.py
@@ -110,7 +110,9 @@ class AsyncIPCProvider(PersistentConnectionProvider):
             raise
 
         if not data:
-            raise PersistentConnectionClosedOK("Socket reader received end of stream.")
+            raise PersistentConnectionClosedOK(
+                user_message="Socket reader received end of stream."
+            )
         return self.decode_rpc_response(data)
 
     # -- private methods -- #
diff --git a/web3/providers/persistent/subscription_manager.py b/web3/providers/persistent/subscription_manager.py
index e3305e4b..c5879f22 100644
--- a/web3/providers/persistent/subscription_manager.py
+++ b/web3/providers/persistent/subscription_manager.py
@@ -298,6 +298,7 @@ class SubscriptionManager:
                     "Message listener background task for the provider has stopped "
                     "unexpectedly. Stopping subscription handling."
                 )
+                break
 
         # no active handler subscriptions, clear the handler subscription queue
         self._provider._request_processor._reset_handler_subscription_queue()


Test patch:
diff --git a/tests/core/providers/test_websocket_provider.py b/tests/core/providers/test_websocket_provider.py
index 10f16d42..e20555a8 100644
--- a/tests/core/providers/test_websocket_provider.py
+++ b/tests/core/providers/test_websocket_provider.py
@@ -397,7 +397,7 @@ async def test_async_iterator_pattern_exception_handling_for_subscriptions():
 
 
 @pytest.mark.asyncio
-async def test_connection_closed_ok_breaks_message_iteration():
+async def test_connection_closed_ok_breaks_process_subscriptions_iteration():
     with patch(
         "web3.providers.persistent.websocket.connect",
         new=lambda *_1, **_2: WebSocketMessageStreamMock(
@@ -409,6 +409,22 @@ async def test_connection_closed_ok_breaks_message_iteration():
             pytest.fail("Should not reach this point.")
 
 
+@pytest.mark.asyncio
+async def test_connection_closed_ok_breaks_handle_subscriptions_iteration():
+    with patch(
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: WebSocketMessageStreamMock(
+            raise_exception=ConnectionClosedOK(None, None)
+        ),
+    ):
+        w3 = await AsyncWeb3(WebSocketProvider("ws://mocked"))
+        # would fail with a ``TimeoutError`` if the iteration did not break properly
+        # on ``ConnectionClosedOK``
+        await asyncio.wait_for(
+            w3.subscription_manager.handle_subscriptions(run_forever=True), timeout=1
+        )
+
+
 @pytest.mark.asyncio
 async def test_listener_task_breaks_out_of_stream_when_cancelled():
     with patch(