Repository: ethereum/web3.py
Pull request #3595 (2025-01-30 18:01:38)

Subscription manager bugfixes:
### What was wrong?

Some bugs shared via Discord conversation with @fubuloubu.

### How was it fixed?

- Fix ``run_forever`` logic so that it doesn't require subscriptions to start and doesn't end when no subscriptions are present.
- Fix the ``unsubscribe`` method so that it can accept subscription ids.
- Fix the ``unsubscribe`` method so it behaves like the ``subscribe`` method, accepting both single or multiple subscriptions or subscription ids. **Note this does change the signature from ``subscription`` to ``subscriptions``, but I think this can be a bugfix since the ``subscribe`` method accepts multiple subscriptions and ``unsubscribe`` should behave similarly.** Happy to revert back and change this in v8 if we don't want to break it here.

- Fix ``subscribe`` so that it returns a list of subscription ids when a list of subscriptions is provided.

Add tests:

- Add a test for ``run_forever`` to make sure it can start without subscriptions and that the task does not stop when no subscriptions are present.
- Add a test for ``unsubscribe`` to make sure it can accept multiple subscriptions as hexstr ids or as subscription objects.
- Update subscriptions tests to check that a list of ids is returned when multiple subscriptions are provided to the ``subscribe`` method.

### Todo:

- [x] Clean up commit history
- [x] Add entry to the [release notes](https://github.com/ethereum/web3.py/blob/main/newsfragments/README.md)

#### Cute Animal Picture

<img width="239" alt="Screenshot 2025-01-30 at 11 17 58" src="https://github.com/user-attachments/assets/8081c05a-0893-41ae-a7fa-345bcc961371" />


fselmo: Of note here is `subscription` -> `subscriptions`. If we feel that is too breaking of a change, even this early on in the manager's existence, I can put this back and we can change the signature in `v8`. I think it could be chalked up to a bugfix and would significantly reduce confusion if both `subscribe` and `unsubscribe` had the same arg name.
fselmo: > Can we easilly alias subscription/subscriptions? and then remove subscription in v8? If not, I think it's okay to break, but we should put it in the release notes and make sure docs are up to date!

If you're good with it I'm happy to take the blame on any backlash 😆 . I'll update the release notes with a breaking newsfragment. I don't think the docs event highlight this yet but I'll double check.

Patch:
diff --git a/newsfragments/3585.breaking.rst b/newsfragments/3585.breaking.rst
new file mode 100644
index 00000000..f427c092
--- /dev/null
+++ b/newsfragments/3585.breaking.rst
@@ -0,0 +1,1 @@
+The bugfix to match ``unsubscribe`` to ``subscribe`` for multiple subscriptions breaks the function signature for ``unsubscribe``, changing the ``subscription`` argument to ``subscriptions``.
diff --git a/newsfragments/3595.bugfix.rst b/newsfragments/3595.bugfix.rst
new file mode 100644
index 00000000..be85c5e7
--- /dev/null
+++ b/newsfragments/3595.bugfix.rst
@@ -0,0 +1,1 @@
+Fix bugs related to subscription manager: ``run_forever`` can start with ``0`` subscriptions and remains alive, ``unsubscribe`` accepts single or multiple subs as objects or hexstrs, ``subscribe`` for many subs returns a list of hexstr ids.
diff --git a/web3/providers/persistent/subscription_manager.py b/web3/providers/persistent/subscription_manager.py
index a079e1d8..cde51880 100644
--- a/web3/providers/persistent/subscription_manager.py
+++ b/web3/providers/persistent/subscription_manager.py
@@ -133,34 +133,88 @@ class SubscriptionManager:
 
             sub_ids: List[HexStr] = []
             for sub in subscriptions:
-                await self.subscribe(sub)
+                sub_ids.append(await self.subscribe(sub))
             return sub_ids
         raise Web3TypeError("Expected a Subscription or a sequence of Subscriptions.")
 
-    async def unsubscribe(self, subscription: EthSubscription[Any]) -> bool:
+    @overload
+    async def unsubscribe(self, subscriptions: EthSubscription[Any]) -> bool:
+        ...
+
+    @overload
+    async def unsubscribe(self, subscriptions: HexStr) -> bool:
+        ...
+
+    @overload
+    async def unsubscribe(
+        self,
+        subscriptions: Sequence[Union[EthSubscription[Any], HexStr]],
+    ) -> bool:
+        ...
+
+    async def unsubscribe(
+        self,
+        subscriptions: Union[
+            EthSubscription[Any],
+            HexStr,
+            Sequence[Union[EthSubscription[Any], HexStr]],
+        ],
+    ) -> bool:
         """
-        Used to unsubscribe from a subscription.
+        Used to unsubscribe from one or multiple subscriptions.
 
-        :param subscription: The subscription to unsubscribe from.
-        :type subscription: EthSubscription
-        :return: ``True`` if unsubscribing was successful, ``False`` otherwise.
+        :param subscriptions: The subscription(s) to unsubscribe from.
+        :type subscriptions: Union[EthSubscription, Sequence[EthSubscription], HexStr,
+            Sequence[HexStr]]
+        :return: ``True`` if unsubscribing to all was successful, ``False`` otherwise
+            with a warning.
         :rtype: bool
         """
-        if subscription not in self.subscriptions:
-            raise Web3ValueError(
-                "Subscription not found or is not being managed by the subscription "
-                f"manager.\n    label: {subscription.label}\n    id: {subscription._id}"
-            )
-        if await self._w3.eth._unsubscribe(subscription.id):
-            self._remove_subscription(subscription)
-            self.logger.info(
-                "Successfully unsubscribed from subscription:\n    "
-                f"label: {subscription.label}\n    id: {subscription.id}"
-            )
-            if len(self._subscription_container.handler_subscriptions) == 0:
-                queue = self._provider._request_processor._handler_subscription_queue
-                await queue.put(SubscriptionProcessingFinished())
-            return True
+        if isinstance(subscriptions, EthSubscription) or isinstance(subscriptions, str):
+            if isinstance(subscriptions, str):
+                subscription_id = subscriptions
+                subscriptions = self.get_by_id(subscription_id)
+                if subscriptions is None:
+                    raise Web3ValueError(
+                        "Subscription not found or is not being managed by the "
+                        f"subscription manager.\n    id: {subscription_id}"
+                    )
+
+            if subscriptions not in self.subscriptions:
+                raise Web3ValueError(
+                    "Subscription not found or is not being managed by the "
+                    "subscription manager.\n    "
+                    f"label: {subscriptions.label}\n    id: {subscriptions._id}"
+                )
+
+            if await self._w3.eth._unsubscribe(subscriptions.id):
+                self._remove_subscription(subscriptions)
+                self.logger.info(
+                    "Successfully unsubscribed from subscription:\n    "
+                    f"label: {subscriptions.label}\n    id: {subscriptions.id}"
+                )
+
+                if len(self._subscription_container.handler_subscriptions) == 0:
+                    queue = (
+                        self._provider._request_processor._handler_subscription_queue
+                    )
+                    await queue.put(SubscriptionProcessingFinished())
+                return True
+
+        elif isinstance(subscriptions, Sequence):
+            if len(subscriptions) == 0:
+                raise Web3ValueError("No subscriptions provided.")
+
+            unsubscribed: List[bool] = []
+            for sub in subscriptions:
+                if isinstance(sub, str):
+                    sub = HexStr(sub)
+                unsubscribed.append(await self.unsubscribe(sub))
+            return all(unsubscribed)
+
+        self.logger.warning(
+            f"Failed to unsubscribe from subscription\n    subscription={subscriptions}"
+        )
         return False
 
     async def unsubscribe_all(self) -> bool:
@@ -195,15 +249,15 @@ class SubscriptionManager:
         :type run_forever: bool
         :return: None
         """
-        if not self._subscription_container.handler_subscriptions:
+        if not self._subscription_container.handler_subscriptions and not run_forever:
             self.logger.warning(
                 "No handler subscriptions found. Subscription handler did not run."
             )
             return
 
         queue = self._provider._request_processor._handler_subscription_queue
-        try:
-            while run_forever or self._subscription_container.handler_subscriptions:
+        while run_forever or self._subscription_container.handler_subscriptions:
+            try:
                 response = cast(RPCResponse, await queue.get())
                 formatted_sub_response = cast(
                     FormattedEthSubscriptionResponse,
@@ -225,18 +279,20 @@ class SubscriptionManager:
                             **sub._handler_context,
                         )
                     )
-        except SubscriptionProcessingFinished:
-            self.logger.info(
-                "All handler subscriptions have been unsubscribed from. "
-                "Stopping subscription handling."
-            )
-        except TaskNotRunning:
-            await asyncio.sleep(0)
-            self._provider._handle_listener_task_exceptions()
-            self.logger.error(
-                "Message listener background task for the provider has stopped "
-                "unexpectedly. Stopping subscription handling."
-            )
+            except SubscriptionProcessingFinished:
+                if not run_forever:
+                    self.logger.info(
+                        "All handler subscriptions have been unsubscribed from. "
+                        "Stopping subscription handling."
+                    )
+                    break
+            except TaskNotRunning:
+                await asyncio.sleep(0)
+                self._provider._handle_listener_task_exceptions()
+                self.logger.error(
+                    "Message listener background task for the provider has stopped "
+                    "unexpectedly. Stopping subscription handling."
+                )
 
         # no active handler subscriptions, clear the handler subscription queue
         self._provider._request_processor._reset_handler_subscription_queue()


Test patch:
diff --git a/tests/core/subscriptions/test_subscription_manager.py b/tests/core/subscriptions/test_subscription_manager.py
index 648a5724..264e3bb3 100644
--- a/tests/core/subscriptions/test_subscription_manager.py
+++ b/tests/core/subscriptions/test_subscription_manager.py
@@ -4,6 +4,9 @@ from unittest.mock import (
     AsyncMock,
 )
 
+from eth_typing import (
+    HexStr,
+)
 import pytest_asyncio
 
 from web3 import (
@@ -46,7 +49,8 @@ async def test_subscription_default_labels_are_unique(subscription_manager):
     sub3 = NewHeadsSubscription()
     sub4 = NewHeadsSubscription()
 
-    await subscription_manager.subscribe([sub1, sub2, sub3, sub4])
+    sub_ids = await subscription_manager.subscribe([sub1, sub2, sub3, sub4])
+    assert sub_ids == ["0x0", "0x1", "0x2", "0x3"]
 
     assert sub1.label != sub2.label != sub3.label != sub4.label
     assert sub1.label == "NewHeadsSubscription('newHeads',)"
@@ -133,3 +137,54 @@ async def test_unsubscribe_all_clears_all_subscriptions(subscription_manager):
     assert sub_container.subscriptions == []
     assert sub_container.subscriptions_by_id == {}
     assert sub_container.subscriptions_by_label == {}
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_with_one_or_multiple(subscription_manager):
+    sub1 = NewHeadsSubscription()
+    sub2 = PendingTxSubscription()
+    sub3 = NewHeadsSubscription()
+    sub4 = LogsSubscription()
+    sub5 = NewHeadsSubscription()
+    sub6 = PendingTxSubscription()
+
+    (
+        sub_id1,
+        sub_id2,
+        sub_id3,
+        sub_id4,
+        sub_id5,
+        sub_id6,
+    ) = await subscription_manager.subscribe([sub1, sub2, sub3, sub4, sub5, sub6])
+    assert sub_id1 == "0x0"
+    assert sub_id2 == "0x1"
+    assert sub_id3 == "0x2"
+    assert sub_id4 == "0x3"
+    assert sub_id5 == "0x4"
+    assert sub_id6 == "0x5"
+
+    assert subscription_manager.subscriptions == [sub1, sub2, sub3, sub4, sub5, sub6]
+
+    # unsubscribe single by hex id
+    assert await subscription_manager.unsubscribe(sub_id1) is True
+    assert subscription_manager.subscriptions == [sub2, sub3, sub4, sub5, sub6]
+
+    # unsubscribe many by hex id
+    assert await subscription_manager.unsubscribe([sub_id2, sub_id3]) is True
+    assert subscription_manager.subscriptions == [sub4, sub5, sub6]
+
+    # unsubscribe non-existent hex id
+    with pytest.raises(Web3ValueError, match=f"Subscription not found|{0x7}"):
+        await subscription_manager.unsubscribe(HexStr("0x7"))
+
+    # unsubscribe by subscription object
+    assert await subscription_manager.unsubscribe(sub4) is True
+    assert subscription_manager.subscriptions == [sub5, sub6]
+
+    # unsubscribe many by subscription object
+    assert await subscription_manager.unsubscribe([sub5, sub6]) is True
+    assert subscription_manager.subscriptions == []
+
+    # unsubscribe non-existent subscription object
+    with pytest.raises(Web3ValueError, match=f"Subscription not found|{sub5.id}"):
+        await subscription_manager.unsubscribe(sub5)
diff --git a/web3/_utils/module_testing/persistent_connection_provider.py b/web3/_utils/module_testing/persistent_connection_provider.py
index 9860bff9..d1db9a91 100644
--- a/web3/_utils/module_testing/persistent_connection_provider.py
+++ b/web3/_utils/module_testing/persistent_connection_provider.py
@@ -876,3 +876,41 @@ class PersistentConnectionProviderTest:
 
         assert_no_subscriptions_left(sub_manager._subscription_container)
         await clean_up_task(unsubscribe_task)
+
+    @pytest.mark.asyncio
+    async def test_run_forever_starts_with_0_subs_and_runs_until_task_cancelled(
+        self, async_w3: AsyncWeb3
+    ) -> None:
+        sub_manager = async_w3.subscription_manager
+        assert_no_subscriptions_left(sub_manager._subscription_container)
+
+        run_forever_task = asyncio.create_task(
+            sub_manager.handle_subscriptions(run_forever=True)
+        )
+
+        await asyncio.sleep(0.1)
+        assert run_forever_task.done() is False
+        assert sub_manager.subscriptions == []
+
+        # subscribe to newHeads and validate it
+        new_heads_handler_test = SubscriptionHandlerTest()
+        sub1 = NewHeadsSubscription(
+            label="foo",
+            handler=new_heads_handler,
+            handler_context={"new_heads_handler_test": new_heads_handler_test},
+        )
+        sub_id = await sub_manager.subscribe(sub1)
+        assert is_hexstr(sub_id)
+        assert len(sub_manager.subscriptions) == 1
+        assert sub_manager.subscriptions[0] == sub1
+
+        # wait for the handler to unsubscribe
+        while sub_manager.subscriptions:
+            await asyncio.sleep(0.1)
+
+        assert new_heads_handler_test.passed
+        assert run_forever_task.done() is False
+        assert run_forever_task.cancelled() is False
+
+        # cleanup
+        await clean_up_task(run_forever_task)