Repository: ethereum/web3.py
Pull request #3554 (2024-12-09 23:27:34)

Subscription management implementation
### What was wrong?

Related to Issue #3397

Working with many subscriptions is not ideal at the moment. In web3.py v7, we added significant support for `eth_subscribe` via the new `PersistentConnectionProvider` implementation. Though we've added support with parity to how `websockets` works, this still only left us with a raw API to handle messages as they come through the socket. We have JSON-RPC specifications of how subscriptions should be formed and the client returns a unique `id` per subscription. Since `PersistentConenctionProvider` classes are fully asynchronous, as they should be, we should implement an API that can leverage all of this in order to handle subscriptions asynchronously as they come in.

### How was it fixed?

- Implement a subscription manager class to manage all subscriptions under-the-hood. This manager can be accessed directly to deal with subscriptions as objects, rather than free-floating ids that are tracked. Each `EthSubscription` class can, and should be encouraged to, have a `handler` function passed to it so that it can properly handle the subscription as it comes in, rather than having the burden of listening and parsing based on message `id` be on the user.
- `w3.eth.subscribe` can now also accept a `handler` kwarg since all subscriptions are managed by the manager under-the-hood. It can also accept an `event` for "logs" subscriptions so that it may be easily parsed by the handler.
- Subscriptions can have descriptive labels for clearer log messages, if desired, instead of relying on an abstract subscription `id` to parse logs.
- The manager counts how many total handler calls have been issued and each subscription counts how many times its handler has been called.

---

In a separate PR we should add a significant docs overhaul for clearer communication on `PersistentConnectionProvider` usage as well as subscription management usage

### Todo:

- [x] Clean up commit history
- [x] Early documentation before a significant docs overhaul to highlight this usage more appropriately
- [x] Add tests
- [x] Add entry to the [release notes](https://github.com/ethereum/web3.py/blob/main/newsfragments/README.md)

#### Cute Animal Picture

![1000029633](https://github.com/user-attachments/assets/3d8d09ae-14d6-46c6-8129-44607e0d51c8)



fselmo: This is ready for a review :)
fselmo: I moved to a context-based args management for subscription handlers in order to facilitate custom args to be used within handlers. This is back on a freeze for reviews.
fselmo: > This doesn't break the existing API, right?

@kclowes, always good to ask 😅. The relevant API changes only [introduce more customization](https://github.com/ethereum/web3.py/pull/3554/files#diff-745353d44729acfd441365c5fe774d4b66affab2319ef3e1ab683dbed1885e38R722-R738). This shouldn't break the API. And if no handlers are passed in (new feature), you can still listen to free-flowing messages directly from the socket.

The typing in the socket stream was already `Optional` meaning a `None` could technically be returned as a "message". If a message has a handler a None is returned so that the iterator can move on to the next iteration. This is the only friction I see this PR introducing. I didn't see a clean way to move a users `async for message in socket` to the next message, if it was handled in a handler, without yielding a `None` but perhaps we can investigate that a bit more.

Thoughts on that?
fselmo: > I like this! I didn't have time to go through with a fine tooth comb, but if you feel good about it and the tests, don't let me block your merge 👍

It has been pretty battle tested but it's causing some irrelevant flaky replace transaction tests to fail pretty consistently. I checked against `main` and it is specific to this branch. I'll wait a little longer to debug it and try to figure out what's causing it. The frustrating thing is that locally it's hard to reproduce, as the flaky tests have been all along. So I might push a commit on top of these until I can sniff it out and try to assess, roll the commit back, and handle appropriately.

My suspicion is it's not directly related to this PR. Ultimately those tests need to be properly fixed, so that might be what I end up doing in a separate PR.

Either way I'll be sure to preserve the current state of the new code (no force pushing on top of the previous commits) so that nothing that you all have reviewed changes.
reedsa: Your docs are great and I wonder if they would be worth referencing in the [Eth.subscribe](https://web3py.readthedocs.io/en/stable/web3.eth.html#web3.eth.Eth.subscribe) section and/or the Providers pages? Might help guide more advanced usage there.
fselmo: > Your docs are great and I wonder if they would be worth referencing in the [Eth.subscribe](https://web3py.readthedocs.io/en/stable/web3.eth.html#web3.eth.Eth.subscribe) section and/or the Providers pages? Might help guide more advanced usage there.

Good thought. We're going to do a big overhaul of these docs so this is just some basic stuff on usage for now. Don't want to put too much into this since it may change quite a bit. Thoughts?
reedsa: > > Your docs are great and I wonder if they would be worth referencing in the [Eth.subscribe](https://web3py.readthedocs.io/en/stable/web3.eth.html#web3.eth.Eth.subscribe) section and/or the Providers pages? Might help guide more advanced usage there.
> 
> Good thought. We're going to do a big overhaul of these docs so this is just some basic stuff on usage for now. Don't want to put too much into this since it may change quite a bit. Thoughts?

SGTM! Looks good to go from my perspective.

Patch:
diff --git a/docs/internals.rst b/docs/internals.rst
index 2f6182c6..bae72f4c 100644
--- a/docs/internals.rst
+++ b/docs/internals.rst
@@ -430,10 +430,139 @@ 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
+user on the same line. The many responses this call will produce can be handled in one
+of a few ways.
+
+The recommended way to handle one-to-many responses is to use the subscription manager
+API. The subscription manager API is a public API on the ``AsyncWeb3`` class, when
+connected to a ``PersistentConnectionProvider`` instance, that allows the user to
+subscribe to a subscription and handle the many responses asynchronously. The
+``subscription_manager`` instance is responsible for handling the many responses that
+come in over the socket connection, as long as handlers are passed to each subscription
+call. The subscription manager can also be used to unsubscribe from a subscription when
+the user is done with it.
+
+.. code-block:: python
+
+    >>> async def new_heads_handler(
+    ...     handler_context: NewHeadsSubscriptionContext,
+    ... ) -> None:
+    ...     result = handler_context.result
+    ...     print(f"New block header: {result}\n")
+    ...     if result["number"] > 1234567:
+    ...         await handler_context.subscription.unsubscribe()
+
+    >>> async def ws_subscription_example():
+    ...     async with AsyncWeb3(WebSocketProvider(f"ws://127.0.0.1:8546")) as w3:
+    ...         # Subscribe to new block headers and receive the subscription_id.
+    ...         # A one-to-one call with a trigger for many responses
+    ...         subscription_id = await w3.eth.subscribe("newHeads", handler=new_heads_handler)
+    ...
+    ...         # Handle the subscription messages asynchronously using the subscription
+    ...         # manager. This will continue until no more subscriptions are present in
+    ...         # the subscription manager, or indefinitely if the `run_forever` flag
+    ...         # is set to `True`.
+    ...         await w3.subscription_manager.handle_subscriptions(run_forever=False)
+    >>> asyncio.run(ws_subscription_example())
+
+The manager can also subscribe to many subscriptions at one time. The
+``EthSubscription`` classes, available via ``web3.utils.subscriptions``, provide a
+friendly API for managing subscriptions. Since each connection and provider instance
+has its own message listener task and subscription manager instance, you can subscribe
+to many subscriptions at once and handle the many responses that come in over the socket
+connections via handlers. The handlers contain:
+
+- ``async_w3``: The ``AsyncWeb3`` instance that the subscription was made on.
+- ``subscription``: The subscription instance that the handler is attached to.
+- ``result``: The response that came in over the socket connection for the subscription.
+
+Subscriptions also accept a ``handler_context`` argument that can be used to pass
+additional information to the handler when subscribing to a subscription. This can be
+used to pass in an event object, for example, that can be used to parse a log event
+when it comes in.
+
+
+.. code-block:: python
+
+    >>> from web3 import (
+    ...     AsyncWeb3,
+    ...     WebSocketProvider,
+    ...     AsyncIPCProvider,
+    ... )
+    >>> from web3.utils.subscriptions import (
+    ...     EthSubscription,
+    ...     NewHeadsSubscription,
+    ...     NewHeadsSubscriptionContext,
+    ...     PendingTxSubscription,
+    ...     PendingTxSubscriptionContext,
+    ...     LogsSubscription,
+    ...     LogsSubscriptionContext,
+    ... )
+
+    >>> async def new_heads_handler(
+    ...     handler_context: NewHeadsSubscriptionContext,
+    ... ) -> None:
+    ...     header = handler_context.result
+    ...     print(f"New block header: {header}\n")
+    ...     if header["number"] > 1234567:
+    ...         await handler_context.subscription.unsubscribe()
+
+    >>> async def pending_txs_handler(
+    ...     handler_context: PendingTxSubscriptionContext,
+    ... ) -> None:
+    ...     ...
+
+    >>> async def log_handler(
+    ...     handler_context: LogsSubscriptionContext,
+    ... ) -> None:
+    ...     log_receipt = handler_context.result
+    ...     # event is now available in the handler context, because we pass it to in the
+    ...     # ``handler_context`` when subscribing to the log
+    ...     event_data = handler_context.event.process_log(log_receipt)
+    ...     print(f"Log event data: {event_data}\n")
+
+    >>> async def sub_manager():
+    ...     local_w3 = await AsyncWeb3(AsyncIPCProvider(LOCAL_IPC, label="mainnet-ipc"))
+    ...     # subscribe to many subscriptions via the subscription manager with handlers
+    ...     await local_w3.subscription_manager.subscribe(
+    ...         [
+    ...             NewHeadsSubscription(label="new-heads-mainnet", handler=new_heads_handler),
+    ...             PendingTxSubscription(
+    ...                 label="pending-tx-mainnet",  # optional label
+    ...                 full_transactions=True,
+    ...                 handler=pending_tx_handler,
+    ...             ),
+    ...             LogsSubscription(
+    ...                 label="WETH transfers",  # optional label
+    ...                 address=local_w3.to_checksum_address("0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"),
+    ...                 topics=[HexStr("0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef")],
+    ...                 handler=log_handler,
+    ...                 # optional ``handler_context`` args to help parse a response
+    ...                 handler_context={"event": my_event},
+    ...             ),
+    ...         ]
+    ...     )
+    ...
+    ...     public_w3 = await AsyncWeb3(WebSocketProvider(PUBLIC_PROVIDER_WS, label="public-ws"))
+    ...     # subscribe via eth_subscribe, with handler and label (optional)
+    ...     await public_w3.eth.subscribe("public_newHeads", handler=pending_tx_handler, label="new-heads-public-ws")
+
+    >>>     # This will handle all subscriptions until no more subscriptions are present
+    ...     # in either subscription manager instance. If the `run_forever` flag is set
+    ...     # to `True` on any manager instance, this will run indefinitely.
+    >>>     await asyncio.gather(
+    ...         public_w3.subscription_manager.handle_subscriptions(),
+    ...         local_w3.subscription_manager.handle_subscriptions(),
+    ...     )
+
+    >>> asyncio.run(sub_manager())
+
+
+The ``process_subscriptions()`` method on the
 :class:`~web3.providers.persistent.PersistentConnection` class, the public API for
-interacting with the active persistent socket connection, is set up to receive
-``eth_subscription`` responses over an asynchronous interator pattern.
+interacting with the active persistent socket connection, is also set up to receive
+``eth_subscription`` responses over an asynchronous interator pattern. You can use this
+method to listen for raw messages and process them as they come in.
 
 .. code-block:: python
 
diff --git a/newsfragments/3554.feature.rst b/newsfragments/3554.feature.rst
new file mode 100644
index 00000000..2f7090b9
--- /dev/null
+++ b/newsfragments/3554.feature.rst
@@ -0,0 +1,1 @@
+Add a subscription manager to persistent connection providers, with support for handler methods for ``eth_subscribe`` subscriptions.
diff --git a/web3/_utils/caching/caching_utils.py b/web3/_utils/caching/caching_utils.py
index fe031742..0dd5e85b 100644
--- a/web3/_utils/caching/caching_utils.py
+++ b/web3/_utils/caching/caching_utils.py
@@ -64,10 +64,12 @@ if TYPE_CHECKING:
     from web3.providers import (  # noqa: F401
         AsyncBaseProvider,
         BaseProvider,
+        PersistentConnectionProvider,
     )
     from web3.types import (  # noqa: F401
         AsyncMakeRequestFn,
         MakeRequestFn,
+        RPCRequest,
         RPCResponse,
     )
 
@@ -367,3 +369,65 @@ def async_handle_request_caching(
     # save a reference to the decorator on the wrapped function
     wrapper._decorator = async_handle_request_caching  # type: ignore
     return wrapper
+
+
+def async_handle_send_caching(
+    func: Callable[
+        [ASYNC_PROVIDER_TYPE, RPCEndpoint, Any],
+        Coroutine[Any, Any, "RPCRequest"],
+    ],
+) -> Callable[..., Coroutine[Any, Any, "RPCRequest"]]:
+    async def wrapper(
+        provider: ASYNC_PROVIDER_TYPE, method: RPCEndpoint, params: Any
+    ) -> "RPCRequest":
+        if is_cacheable_request(provider, method, params):
+            request_cache = provider._request_cache
+            cache_key = generate_cache_key(
+                f"{threading.get_ident()}:{(method, params)}"
+            )
+            cached_response = request_cache.get_cache_entry(cache_key)
+            if cached_response is not None:
+                # The request data isn't used, this just prevents a cached request from
+                # being sent - return an empty request object
+                return {"id": -1, "method": RPCEndpoint(""), "params": []}
+        return await func(provider, method, params)
+
+    # save a reference to the decorator on the wrapped function
+    wrapper._decorator = async_handle_send_caching  # type: ignore
+    return wrapper
+
+
+def async_handle_recv_caching(
+    func: Callable[
+        ["PersistentConnectionProvider", "RPCRequest"],
+        Coroutine[Any, Any, "RPCResponse"],
+    ]
+) -> Callable[..., Coroutine[Any, Any, "RPCResponse"]]:
+    async def wrapper(
+        provider: "PersistentConnectionProvider",
+        rpc_request: "RPCRequest",
+    ) -> "RPCResponse":
+        method = rpc_request["method"]
+        params = rpc_request["params"]
+        if is_cacheable_request(provider, method, params):
+            request_cache = provider._request_cache
+            cache_key = generate_cache_key(
+                f"{threading.get_ident()}:{(method, params)}"
+            )
+            cache_result = request_cache.get_cache_entry(cache_key)
+            if cache_result is not None:
+                return cache_result
+            else:
+                response = await func(provider, rpc_request)
+                if await _async_should_cache_response(
+                    provider, method, params, response
+                ):
+                    async with provider._request_cache_lock:
+                        request_cache.cache(cache_key, response)
+                return response
+        else:
+            return await func(provider, rpc_request)
+
+    # save a reference to the decorator on the wrapped function
+    wrapper._decorator = async_handle_recv_caching  # type: ignore
+    return wrapper
diff --git a/web3/_utils/caching/request_caching_validation.py b/web3/_utils/caching/request_caching_validation.py
index 4f904c72..8ebd7916 100644
--- a/web3/_utils/caching/request_caching_validation.py
+++ b/web3/_utils/caching/request_caching_validation.py
@@ -19,6 +19,7 @@ if TYPE_CHECKING:
     from web3.providers import (  # noqa: F401
         AsyncBaseProvider,
         BaseProvider,
+        PersistentConnectionProvider,
     )
 
 UNCACHEABLE_BLOCK_IDS = {"finalized", "safe", "latest", "pending"}
diff --git a/web3/contract/base_contract.py b/web3/contract/base_contract.py
index 26476496..ef3605e5 100644
--- a/web3/contract/base_contract.py
+++ b/web3/contract/base_contract.py
@@ -117,6 +117,7 @@ from web3.types import (
     BlockIdentifier,
     EventData,
     FilterParams,
+    LogReceipt,
     TContractFn,
     TxParams,
     TxReceipt,
@@ -253,7 +254,7 @@ class BaseContractEvent:
             yield rich_log
 
     @combomethod
-    def process_log(self, log: HexStr) -> EventData:
+    def process_log(self, log: LogReceipt) -> EventData:
         return get_event_data(self.w3.codec, self.abi, log)
 
     @combomethod
diff --git a/web3/eth/async_eth.py b/web3/eth/async_eth.py
index 26993d39..252a24a9 100644
--- a/web3/eth/async_eth.py
+++ b/web3/eth/async_eth.py
@@ -99,11 +99,16 @@ from web3.types import (
     _Hash32,
 )
 from web3.utils import (
+    EthSubscription,
     async_handle_offchain_lookup,
 )
+from web3.utils.subscriptions import (
+    EthSubscriptionHandler,
+)
 
 if TYPE_CHECKING:
     from web3 import AsyncWeb3  # noqa: F401
+    from web3.contract.async_contract import AsyncContractEvent  # noqa: F401
 
 
 class AsyncEth(BaseEth):
@@ -714,6 +719,9 @@ class AsyncEth(BaseEth):
                 bool,  # newPendingTransactions, full_transactions
             ]
         ] = None,
+        handler: Optional[EthSubscriptionHandler] = None,
+        handler_context: Optional[Dict[str, Any]] = None,
+        label: Optional[str] = None,
     ) -> HexStr:
         if not isinstance(self.w3.provider, PersistentConnectionProvider):
             raise MethodNotSupported(
@@ -721,10 +729,13 @@ class AsyncEth(BaseEth):
                 "persistent connections."
             )
 
-        if subscription_arg is None:
-            return await self._subscribe(subscription_type)
-
-        return await self._subscribe_with_args(subscription_type, subscription_arg)
+        sub = EthSubscription._create_type_aware_subscription(
+            subscription_params=(subscription_type, subscription_arg),
+            handler=handler,
+            handler_context=handler_context or {},
+            label=label,
+        )
+        return await self.w3.subscription_manager.subscribe(sub)
 
     _unsubscribe: Method[Callable[[HexStr], Awaitable[bool]]] = Method(
         RPC.eth_unsubscribe,
@@ -738,7 +749,14 @@ class AsyncEth(BaseEth):
                 "persistent connections."
             )
 
-        return await self._unsubscribe(subscription_id)
+        for sub in self.w3.subscription_manager.subscriptions:
+            if sub._id == subscription_id:
+                return await sub.unsubscribe()
+
+        raise Web3ValueError(
+            f"Cannot unsubscribe subscription with id `{subscription_id}`. "
+            "Subscription not found."
+        )
 
     # -- contract methods -- #
 
diff --git a/web3/exceptions.py b/web3/exceptions.py
index dd5afaa1..05c9f188 100644
--- a/web3/exceptions.py
+++ b/web3/exceptions.py
@@ -353,6 +353,13 @@ class PersistentConnectionClosedOK(PersistentConnectionError):
     """
 
 
+class SubscriptionProcessingFinished(Web3Exception):
+    """
+    Raised to alert the subscription manager that the processing of subscriptions
+    has finished.
+    """
+
+
 class Web3RPCError(Web3Exception):
     """
     Raised when a JSON-RPC response contains an error field.
diff --git a/web3/main.py b/web3/main.py
index e74c4946..bdfc8c20 100644
--- a/web3/main.py
+++ b/web3/main.py
@@ -148,6 +148,9 @@ from web3.tracing import (
 from web3.types import (
     Wei,
 )
+from web3.providers.persistent.subscription_manager import (
+    SubscriptionManager,
+)
 
 if TYPE_CHECKING:
     from web3._utils.batching import RequestBatcher  # noqa: F401
@@ -508,12 +511,27 @@ class AsyncWeb3(BaseWeb3):
             new_ens.w3 = self  # set self object reference for ``AsyncENS.w3``
         self._ens = new_ens
 
-    # -- persistent connection methods -- #
+    # -- persistent connection settings -- #
+
+    _subscription_manager: Optional[SubscriptionManager] = None
+    _persistent_connection: Optional["PersistentConnection"] = None
+
+    @property
+    @persistent_connection_provider_method()
+    def subscription_manager(self) -> SubscriptionManager:
+        """
+        Access the subscription manager for the current PersistentConnectionProvider.
+        """
+        if not self._subscription_manager:
+            self._subscription_manager = SubscriptionManager(self)
+        return self._subscription_manager
 
     @property
     @persistent_connection_provider_method()
     def socket(self) -> PersistentConnection:
-        return PersistentConnection(self)
+        if self._persistent_connection is None:
+            self._persistent_connection = PersistentConnection(self)
+        return self._persistent_connection
 
     # w3 = await AsyncWeb3(PersistentConnectionProvider(...))
     @persistent_connection_provider_method(
diff --git a/web3/manager.py b/web3/manager.py
index f273eba9..8e2f2b64 100644
--- a/web3/manager.py
+++ b/web3/manager.py
@@ -6,6 +6,7 @@ from typing import (
     AsyncGenerator,
     Callable,
     Coroutine,
+    Dict,
     List,
     Optional,
     Sequence,
@@ -69,7 +70,9 @@ from web3.providers.async_base import (
     AsyncJSONBaseProvider,
 )
 from web3.types import (
+    FormattedEthSubscriptionResponse,
     RPCEndpoint,
+    RPCRequest,
     RPCResponse,
 )
 
@@ -292,9 +295,6 @@ class RequestManager:
             provider = cast(PersistentConnectionProvider, self.provider)
             self._request_processor: RequestProcessor = provider._request_processor
 
-    w3: Union["AsyncWeb3", "Web3"] = None
-    _provider = None
-
     @property
     def provider(self) -> Union["BaseProvider", "AsyncBaseProvider"]:
         return self._provider
@@ -474,7 +474,10 @@ class RequestManager:
 
         if isinstance(self.provider, PersistentConnectionProvider):
             # call _process_response for each response in the batch
-            return [await self._process_response(resp) for resp in responses]
+            return [
+                cast(RPCResponse, await self._process_response(resp))
+                for resp in responses
+            ]
 
         formatted_responses = [
             self._format_batched_response(info, resp)
@@ -500,32 +503,65 @@ class RequestManager:
 
     # -- persistent connection -- #
 
-    async def socket_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
+    async def socket_request(
+        self,
+        method: RPCEndpoint,
+        params: Any,
+        response_formatters: Optional[
+            Tuple[Dict[str, Callable[..., Any]], Callable[..., Any], Callable[..., Any]]
+        ] = None,
+    ) -> RPCResponse:
         provider = cast(PersistentConnectionProvider, self._provider)
-        request_func = await provider.request_func(
-            cast("AsyncWeb3", self.w3), cast("MiddlewareOnion", self.middleware_onion)
-        )
         self.logger.debug(
             "Making request to open socket connection and waiting for response: "
-            f"{provider.get_endpoint_uri_or_ipc_path()}, method: {method}"
+            f"{provider.get_endpoint_uri_or_ipc_path()},\n    method: {method},\n"
+            f"    params: {params}"
         )
-        response = await request_func(method, params)
-        return await self._process_response(response)
+        rpc_request = await self.send(method, params)
+        provider._request_processor.cache_request_information(
+            rpc_request["id"],
+            rpc_request["method"],
+            rpc_request["params"],
+            response_formatters=response_formatters or ((), (), ()),
+        )
+        return await self.recv_for_request(rpc_request)
 
-    async def send(self, method: RPCEndpoint, params: Any) -> None:
+    async def send(self, method: RPCEndpoint, params: Any) -> RPCRequest:
         provider = cast(PersistentConnectionProvider, self._provider)
-        # run through the request processors of the middleware
-        for mw_class in self.middleware_onion.as_tuple_of_middleware():
-            mw = mw_class(self.w3)
-            method, params = mw.request_processor(method, params)
-
+        async_w3 = cast("AsyncWeb3", self.w3)
+        middleware_onion = cast("MiddlewareOnion", self.middleware_onion)
+        send_func = await provider.send_func(
+            async_w3,
+            middleware_onion,
+        )
         self.logger.debug(
             "Sending request to open socket connection: "
-            f"{provider.get_endpoint_uri_or_ipc_path()}, method: {method}"
+            f"{provider.get_endpoint_uri_or_ipc_path()},\n    method: {method},\n"
+            f"    params: {params}"
         )
-        await provider.socket_send(provider.encode_rpc_request(method, params))
+        return await send_func(method, params)
 
-    async def recv(self) -> RPCResponse:
+    async def recv_for_request(self, rpc_request: RPCRequest) -> RPCResponse:
+        provider = cast(PersistentConnectionProvider, self._provider)
+        async_w3 = cast("AsyncWeb3", self.w3)
+        middleware_onion = cast("MiddlewareOnion", self.middleware_onion)
+        recv_func = await provider.recv_func(
+            async_w3,
+            middleware_onion,
+        )
+        self.logger.debug(
+            "Getting response for request from open socket connection:\n"
+            f"    request: {rpc_request}"
+        )
+        response = await recv_func(rpc_request)
+        try:
+            return cast(RPCResponse, await self._process_response(response))
+        except Exception:
+            response_id_key = generate_cache_key(response["id"])
+            provider._request_processor._request_information_cache.pop(response_id_key)
+            raise
+
+    async def recv(self) -> Union[RPCResponse, FormattedEthSubscriptionResponse]:
         provider = cast(PersistentConnectionProvider, self._provider)
         self.logger.debug(
             "Getting next response from open socket connection: "
@@ -543,15 +579,18 @@ class RequestManager:
     def _persistent_message_stream(self) -> "_AsyncPersistentMessageStream":
         return _AsyncPersistentMessageStream(self)
 
-    async def _get_next_message(self) -> RPCResponse:
+    async def _get_next_message(self) -> FormattedEthSubscriptionResponse:
         return await self._message_stream().__anext__()
 
-    async def _message_stream(self) -> AsyncGenerator[RPCResponse, None]:
+    async def _message_stream(
+        self,
+    ) -> AsyncGenerator[FormattedEthSubscriptionResponse, None]:
         if not isinstance(self._provider, PersistentConnectionProvider):
             raise Web3TypeError(
                 "Only providers that maintain an open, persistent connection "
                 "can listen to streams."
             )
+        async_w3 = cast("AsyncWeb3", self.w3)
 
         if self._provider._message_listener_task is None:
             raise ProviderConnectionError(
@@ -563,13 +602,21 @@ class RequestManager:
                 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)
+                # if the subscription was unsubscribed from, we won't have a formatted
+                # response because we lost the request information.
+                sub_id = response.get(
+                    "subscription", response.get("params", {}).get("subscription")
+                )
+                if async_w3.subscription_manager.get_by_id(sub_id):
+                    # if active subscription, process and yield the formatted response
+                    formatted_sub_response = cast(
+                        FormattedEthSubscriptionResponse,
+                        await self._process_response(response),
+                    )
+                    yield formatted_sub_response
+                else:
+                    # if not an active sub, skip processing and continue
+                    continue
             except TaskNotRunning:
                 await asyncio.sleep(0)
                 self._provider._handle_listener_task_exceptions()
@@ -579,7 +626,9 @@ class RequestManager:
                 )
                 return
 
-    async def _process_response(self, response: RPCResponse) -> RPCResponse:
+    async def _process_response(
+        self, response: RPCResponse
+    ) -> Union[RPCResponse, FormattedEthSubscriptionResponse]:
         provider = cast(PersistentConnectionProvider, self._provider)
         request_info = self._request_processor.get_request_information_for_response(
             response
@@ -642,5 +691,5 @@ class _AsyncPersistentMessageStream:
     def __aiter__(self) -> Self:
         return self
 
-    async def __anext__(self) -> RPCResponse:
+    async def __anext__(self) -> FormattedEthSubscriptionResponse:
         return await self.manager._get_next_message()
diff --git a/web3/method.py b/web3/method.py
index a664f17d..f67828f9 100644
--- a/web3/method.py
+++ b/web3/method.py
@@ -203,7 +203,7 @@ class Method(Generic[TFunc]):
     def process_params(
         self, module: "Module", *args: Any, **kwargs: Any
     ) -> Tuple[
-        Tuple[Union[RPCEndpoint, Callable[..., RPCEndpoint]], Tuple[Any, ...]],
+        Tuple[Union[RPCEndpoint, Callable[..., RPCEndpoint]], Tuple[RPCEndpoint, ...]],
         Tuple[
             Union[TReturn, Dict[str, Callable[..., Any]]],
             Callable[..., Any],
diff --git a/web3/module.py b/web3/module.py
index 8165f381..0f9d85d7 100644
--- a/web3/module.py
+++ b/web3/module.py
@@ -32,6 +32,7 @@ from web3.providers.persistent import (
     PersistentConnectionProvider,
 )
 from web3.types import (
+    FormattedEthSubscriptionResponse,
     RPCEndpoint,
     RPCResponse,
 )
@@ -74,7 +75,7 @@ def retrieve_request_information_for_batching(
         )
         if isinstance(w3.provider, PersistentConnectionProvider):
             w3.provider._request_processor.cache_request_information(
-                cast(RPCEndpoint, method_str), params, response_formatters
+                None, cast(RPCEndpoint, method_str), params, response_formatters
             )
         return (cast(RPCEndpoint, method_str), params), response_formatters
 
@@ -121,8 +122,17 @@ def retrieve_async_method_call_fn(
     async_w3: "AsyncWeb3",
     module: "Module",
     method: Method[Callable[..., Any]],
-) -> Callable[..., Coroutine[Any, Any, Optional[Union[RPCResponse, AsyncLogFilter]]]]:
-    async def caller(*args: Any, **kwargs: Any) -> Union[RPCResponse, AsyncLogFilter]:
+) -> Callable[
+    ...,
+    Coroutine[
+        Any,
+        Any,
+        Optional[Union[RPCResponse, FormattedEthSubscriptionResponse, AsyncLogFilter]],
+    ],
+]:
+    async def caller(
+        *args: Any, **kwargs: Any
+    ) -> Union[RPCResponse, FormattedEthSubscriptionResponse, AsyncLogFilter]:
         try:
             (method_str, params), response_formatters = method.process_params(
                 module, *args, **kwargs
@@ -131,31 +141,17 @@ def retrieve_async_method_call_fn(
             return AsyncLogFilter(eth_module=module, filter_id=err.filter_id)
 
         if isinstance(async_w3.provider, PersistentConnectionProvider):
-            provider = async_w3.provider
-            cache_key = provider._request_processor.cache_request_information(
-                cast(RPCEndpoint, method_str), params, response_formatters
+            return await async_w3.manager.socket_request(
+                cast(RPCEndpoint, method_str),
+                params,
+                response_formatters=response_formatters,
             )
-
-            try:
-                method_str = cast(RPCEndpoint, method_str)
-                return await async_w3.manager.socket_request(method_str, params)
-            except Exception as e:
-                if (
-                    cache_key is not None
-                    and cache_key
-                    in provider._request_processor._request_information_cache
-                ):
-                    provider._request_processor.pop_cached_request_information(
-                        cache_key
-                    )
-                raise e
         else:
             (
                 result_formatters,
                 error_formatters,
                 null_result_formatters,
             ) = response_formatters
-
             result = await async_w3.manager.coro_request(
                 method_str, params, error_formatters, null_result_formatters
             )
diff --git a/web3/providers/async_base.py b/web3/providers/async_base.py
index eb8a7426..55a4a8ee 100644
--- a/web3/providers/async_base.py
+++ b/web3/providers/async_base.py
@@ -6,6 +6,7 @@ from typing import (
     Any,
     Callable,
     Coroutine,
+    Dict,
     List,
     Optional,
     Set,
@@ -43,6 +44,7 @@ from web3.middleware.base import (
 )
 from web3.types import (
     RPCEndpoint,
+    RPCRequest,
     RPCResponse,
 )
 from web3.utils import (
@@ -83,10 +85,6 @@ class AsyncBaseProvider:
     global_ccip_read_enabled: bool = True
     ccip_read_max_redirects: int = 4
 
-    # request caching
-    _request_cache: SimpleCache
-    _request_cache_lock: asyncio.Lock = asyncio.Lock()
-
     def __init__(
         self,
         cache_allowed_requests: bool = False,
@@ -96,6 +94,8 @@ class AsyncBaseProvider:
         ] = empty,
     ) -> None:
         self._request_cache = SimpleCache(1000)
+        self._request_cache_lock: asyncio.Lock = asyncio.Lock()
+
         self.cache_allowed_requests = cache_allowed_requests
         self.cacheable_requests = cacheable_requests or CACHEABLE_REQUESTS
         self.request_cache_validation_threshold = request_cache_validation_threshold
@@ -172,23 +172,31 @@ class AsyncBaseProvider:
 
 
 class AsyncJSONBaseProvider(AsyncBaseProvider):
-    logger = logging.getLogger("web3.providers.async_base.AsyncJSONBaseProvider")
-
     def __init__(self, **kwargs: Any) -> None:
-        self.request_counter = itertools.count()
         super().__init__(**kwargs)
+        self.request_counter = itertools.count()
 
-    def encode_rpc_request(self, method: RPCEndpoint, params: Any) -> bytes:
+    def form_request(self, method: RPCEndpoint, params: Any = None) -> RPCRequest:
         request_id = next(self.request_counter)
         rpc_dict = {
+            "id": request_id,
             "jsonrpc": "2.0",
             "method": method,
             "params": params or [],
-            "id": request_id,
         }
-        encoded = FriendlyJsonSerde().json_encode(rpc_dict, cls=Web3JsonEncoder)
+        return cast(RPCRequest, rpc_dict)
+
+    @staticmethod
+    def encode_rpc_dict(rpc_dict: RPCRequest) -> bytes:
+        encoded = FriendlyJsonSerde().json_encode(
+            cast(Dict[str, Any], rpc_dict), cls=Web3JsonEncoder
+        )
         return to_bytes(text=encoded)
 
+    def encode_rpc_request(self, method: RPCEndpoint, params: Any) -> bytes:
+        rpc_dict = self.form_request(method, params)
+        return self.encode_rpc_dict(rpc_dict)
+
     @staticmethod
     def decode_rpc_response(raw_response: bytes) -> RPCResponse:
         text_response = str(
diff --git a/web3/providers/base.py b/web3/providers/base.py
index 3d9249a8..e1606967 100644
--- a/web3/providers/base.py
+++ b/web3/providers/base.py
@@ -66,10 +66,6 @@ class BaseProvider:
     global_ccip_read_enabled: bool = True
     ccip_read_max_redirects: int = 4
 
-    # request caching
-    _request_cache: SimpleCache
-    _request_cache_lock: threading.Lock = threading.Lock()
-
     def __init__(
         self,
         cache_allowed_requests: bool = False,
@@ -79,6 +75,8 @@ class BaseProvider:
         ] = empty,
     ) -> None:
         self._request_cache = SimpleCache(1000)
+        self._request_cache_lock: threading.Lock = threading.Lock()
+
         self.cache_allowed_requests = cache_allowed_requests
         self.cacheable_requests = cacheable_requests or CACHEABLE_REQUESTS
         self.request_cache_validation_threshold = request_cache_validation_threshold
@@ -124,8 +122,8 @@ class JSONBaseProvider(BaseProvider):
     ] = (None, None)
 
     def __init__(self, **kwargs: Any) -> None:
-        self.request_counter = itertools.count()
         super().__init__(**kwargs)
+        self.request_counter = itertools.count()
 
     def encode_rpc_request(self, method: RPCEndpoint, params: Any) -> bytes:
         rpc_dict = {
diff --git a/web3/providers/ipc.py b/web3/providers/ipc.py
index 349cb1a4..6b81669a 100644
--- a/web3/providers/ipc.py
+++ b/web3/providers/ipc.py
@@ -146,6 +146,7 @@ class IPCProvider(JSONBaseProvider):
         timeout: int = 30,
         **kwargs: Any,
     ) -> None:
+        super().__init__(**kwargs)
         if ipc_path is None:
             self.ipc_path = get_default_ipc_path()
         elif isinstance(ipc_path, str) or isinstance(ipc_path, Path):
@@ -156,7 +157,6 @@ class IPCProvider(JSONBaseProvider):
         self.timeout = timeout
         self._lock = threading.Lock()
         self._socket = PersistantSocket(self.ipc_path)
-        super().__init__(**kwargs)
 
     def __str__(self) -> str:
         return f"<{self.__class__.__name__} {self.ipc_path}>"
diff --git a/web3/providers/legacy_websocket.py b/web3/providers/legacy_websocket.py
index 5790850a..fd0dbb2c 100644
--- a/web3/providers/legacy_websocket.py
+++ b/web3/providers/legacy_websocket.py
@@ -102,6 +102,7 @@ class LegacyWebSocketProvider(JSONBaseProvider):
         websocket_timeout: int = DEFAULT_WEBSOCKET_TIMEOUT,
         **kwargs: Any,
     ) -> None:
+        super().__init__(**kwargs)
         self.endpoint_uri = URI(endpoint_uri)
         self.websocket_timeout = websocket_timeout
         if self.endpoint_uri is None:
@@ -120,7 +121,6 @@ class LegacyWebSocketProvider(JSONBaseProvider):
                     f"in websocket_kwargs, found: {found_restricted_keys}"
                 )
         self.conn = PersistentWebSocket(self.endpoint_uri, websocket_kwargs)
-        super().__init__(**kwargs)
 
     def __str__(self) -> str:
         return f"WS connection {self.endpoint_uri}"
diff --git a/web3/providers/persistent/async_ipc.py b/web3/providers/persistent/async_ipc.py
index d9a18628..583c9ae4 100644
--- a/web3/providers/persistent/async_ipc.py
+++ b/web3/providers/persistent/async_ipc.py
@@ -14,6 +14,7 @@ from typing import (
 )
 
 from web3.types import (
+    RPCEndpoint,
     RPCResponse,
 )
 
@@ -59,15 +60,15 @@ class AsyncIPCProvider(PersistentConnectionProvider):
         # `PersistentConnectionProvider` kwargs can be passed through
         **kwargs: Any,
     ) -> None:
+        # initialize the ipc_path before calling the super constructor
         if ipc_path is None:
             self.ipc_path = get_default_ipc_path()
         elif isinstance(ipc_path, str) or isinstance(ipc_path, Path):
             self.ipc_path = str(Path(ipc_path).expanduser().resolve())
         else:
             raise Web3TypeError("ipc_path must be of type string or pathlib.Path")
-
-        self.read_buffer_limit = read_buffer_limit
         super().__init__(**kwargs)
+        self.read_buffer_limit = read_buffer_limit
 
     def __str__(self) -> str:
         return f"<{self.__class__.__name__} {self.ipc_path}>"
@@ -77,7 +78,7 @@ class AsyncIPCProvider(PersistentConnectionProvider):
             return False
 
         try:
-            await self.make_request("web3_clientVersion", [])
+            await self.make_request(RPCEndpoint("web3_clientVersion"), [])
             return True
         except (OSError, ProviderConnectionError) as e:
             if show_traceback:
@@ -139,6 +140,7 @@ class AsyncIPCProvider(PersistentConnectionProvider):
         )
 
     async def _provider_specific_disconnect(self) -> None:
+        # this should remain idempotent
         if self._writer and not self._writer.is_closing():
             self._writer.close()
             await self._writer.wait_closed()
diff --git a/web3/providers/persistent/persistent.py b/web3/providers/persistent/persistent.py
index 280ac160..5a5a2245 100644
--- a/web3/providers/persistent/persistent.py
+++ b/web3/providers/persistent/persistent.py
@@ -3,10 +3,13 @@ from abc import (
     abstractmethod,
 )
 import asyncio
-import json
 import logging
+import signal
 from typing import (
+    TYPE_CHECKING,
     Any,
+    Callable,
+    Coroutine,
     List,
     Optional,
     Tuple,
@@ -24,9 +27,12 @@ from web3._utils.batching import (
     sort_batch_response_by_response_ids,
 )
 from web3._utils.caching import (
-    async_handle_request_caching,
     generate_cache_key,
 )
+from web3._utils.caching.caching_utils import (
+    async_handle_recv_caching,
+    async_handle_send_caching,
+)
 from web3.exceptions import (
     PersistentConnectionClosedOK,
     ProviderConnectionError,
@@ -43,9 +49,15 @@ from web3.providers.persistent.request_processor import (
 from web3.types import (
     RPCEndpoint,
     RPCId,
+    RPCRequest,
     RPCResponse,
 )
 
+if TYPE_CHECKING:
+    from web3 import AsyncWeb3  # noqa: F401
+    from web3.middleware.base import MiddlewareOnion  # noqa: F401
+
+
 DEFAULT_PERSISTENT_CONNECTION_TIMEOUT = 30.0
 
 
@@ -53,11 +65,14 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
     logger = logging.getLogger("web3.providers.PersistentConnectionProvider")
     has_persistent_connection = True
 
-    _request_processor: RequestProcessor
-    _message_listener_task: Optional["asyncio.Task[None]"] = None
-    _listen_event: asyncio.Event = asyncio.Event()
-
-    _batch_request_counter: Optional[int] = None
+    _send_func_cache: Tuple[int, Callable[..., Coroutine[Any, Any, RPCRequest]]] = (
+        None,
+        None,
+    )
+    _recv_func_cache: Tuple[int, Callable[..., Coroutine[Any, Any, RPCResponse]]] = (
+        None,
+        None,
+    )
 
     def __init__(
         self,
@@ -72,9 +87,63 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
             self,
             subscription_response_queue_size=subscription_response_queue_size,
         )
+        self._message_listener_task: Optional["asyncio.Task[None]"] = None
+        self._batch_request_counter: Optional[int] = None
+        self._listen_event: asyncio.Event = asyncio.Event()
+        self._max_connection_retries = max_connection_retries
+
         self.request_timeout = request_timeout
         self.silence_listener_task_exceptions = silence_listener_task_exceptions
-        self._max_connection_retries = max_connection_retries
+
+    async def send_func(
+        self, async_w3: "AsyncWeb3", middleware_onion: "MiddlewareOnion"
+    ) -> Callable[..., Coroutine[Any, Any, RPCRequest]]:
+        """
+        Cache the middleware chain for `send`.
+        """
+        middleware = middleware_onion.as_tuple_of_middleware()
+        cache_key = hash(tuple(id(mw) for mw in middleware))
+
+        if cache_key != self._send_func_cache[0]:
+
+            async def send_function(method: RPCEndpoint, params: Any) -> RPCRequest:
+                for mw in middleware:
+                    initialized = mw(async_w3)
+                    method, params = await initialized.async_request_processor(
+                        method, params
+                    )
+
+                return await self.send_request(method, params)
+
+            self._send_func_cache = (cache_key, send_function)
+
+        return self._send_func_cache[1]
+
+    async def recv_func(
+        self, async_w3: "AsyncWeb3", middleware_onion: "MiddlewareOnion"
+    ) -> Any:
+        """
+        Cache and compose the middleware stack for `recv`.
+        """
+        middleware = middleware_onion.as_tuple_of_middleware()
+        cache_key = hash(tuple(id(mw) for mw in middleware))
+
+        if cache_key != self._recv_func_cache[0]:
+
+            async def recv_function(rpc_request: RPCRequest) -> RPCResponse:
+                # first, retrieve the response
+                response = await self.recv_for_request(rpc_request)
+                method = rpc_request["method"]
+                for mw in reversed(middleware):
+                    initialized = mw(async_w3)
+                    response = await initialized.async_response_processor(
+                        method, response
+                    )
+                return response
+
+            self._recv_func_cache = (cache_key, recv_function)
+
+        return self._recv_func_cache[1]
 
     def get_endpoint_uri_or_ipc_path(self) -> str:
         if hasattr(self, "endpoint_uri"):
@@ -108,6 +177,7 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
                 self.logger.info(
                     f"Successfully connected to: {self.get_endpoint_uri_or_ipc_path()}"
                 )
+                self._set_signal_handlers()
                 break
             except (WebSocketException, OSError) as e:
                 if _connection_attempts == self._max_connection_retries:
@@ -124,6 +194,7 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
                 _backoff_time *= _backoff_rate_change
 
     async def disconnect(self) -> None:
+        # this should remain idempotent
         try:
             if self._message_listener_task:
                 self._message_listener_task.cancel()
@@ -140,15 +211,23 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
             f"Successfully disconnected from: {self.get_endpoint_uri_or_ipc_path()}"
         )
 
-    @async_handle_request_caching
-    async def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
-        request_data = self.encode_rpc_request(method, params)
-        await self.socket_send(request_data)
+    @async_handle_send_caching
+    async def send_request(self, method: RPCEndpoint, params: Any) -> RPCRequest:
+        request_dict = self.form_request(method, params)
+        await self.socket_send(self.encode_rpc_dict(request_dict))
+        return request_dict
 
-        current_request_id = json.loads(request_data)["id"]
-        response = await self._get_response_for_request_id(current_request_id)
+    @async_handle_recv_caching
+    async def recv_for_request(self, rpc_request: RPCRequest) -> RPCResponse:
+        return await self._get_response_for_request_id(rpc_request["id"])
 
-        return response
+    async def make_request(
+        self,
+        method: RPCEndpoint,
+        params: Any,
+    ) -> RPCResponse:
+        rpc_request = await self.send_request(method, params)
+        return await self.recv_for_request(rpc_request)
 
     async def make_batch_request(
         self, requests: List[Tuple[RPCEndpoint, Any]]
@@ -184,19 +263,46 @@ class PersistentConnectionProvider(AsyncJSONBaseProvider, ABC):
         raise NotImplementedError("Must be implemented by subclasses")
 
     async def _provider_specific_disconnect(self) -> None:
+        # this method should be idempotent
         raise NotImplementedError("Must be implemented by subclasses")
 
     async def _provider_specific_socket_reader(self) -> RPCResponse:
         raise NotImplementedError("Must be implemented by subclasses")
 
+    def _set_signal_handlers(self) -> None:
+        loop = asyncio.get_event_loop()
+
+        def extended_handler(sig: int, frame: Any, existing_handler: Any) -> None:
+            loop.create_task(self.disconnect())
+
+            # invoke the existing handler, if callable
+            if callable(existing_handler):
+                existing_handler(sig, frame)
+
+        existing_sigint_handler = signal.getsignal(signal.SIGINT)
+        existing_sigterm_handler = signal.getsignal(signal.SIGTERM)
+
+        # extend the existing signal handlers to include the disconnect method
+        signal.signal(
+            signal.SIGINT,
+            lambda sig, frame: extended_handler(sig, frame, existing_sigint_handler),
+        )
+        signal.signal(
+            signal.SIGTERM,
+            lambda sig, frame: extended_handler(sig, frame, existing_sigterm_handler),
+        )
+
     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.
+        # Puts a `TaskNotRunning` in appropriate queues to signal the end of the
+        # listener task to any listeners relying on the queues.
         self._request_processor._subscription_response_queue.put_nowait(
             TaskNotRunning(message_listener_task)
         )
+        self._request_processor._handler_subscription_queue.put_nowait(
+            TaskNotRunning(message_listener_task)
+        )
 
     async def _message_listener(self) -> None:
         self.logger.info(
diff --git a/web3/providers/persistent/persistent_connection.py b/web3/providers/persistent/persistent_connection.py
index b60a698a..e5f9cf6a 100644
--- a/web3/providers/persistent/persistent_connection.py
+++ b/web3/providers/persistent/persistent_connection.py
@@ -2,9 +2,12 @@ from typing import (
     TYPE_CHECKING,
     Any,
     Dict,
+    Union,
+    cast,
 )
 
 from web3.types import (
+    FormattedEthSubscriptionResponse,
     RPCEndpoint,
     RPCResponse,
 )
@@ -16,6 +19,9 @@ if TYPE_CHECKING:
     from web3.manager import (  # noqa: F401
         _AsyncPersistentMessageStream,
     )
+    from web3.providers.persistent import (  # noqa: F401
+        PersistentConnectionProvider,
+    )
 
 
 class PersistentConnection:
@@ -26,6 +32,7 @@ class PersistentConnection:
 
     def __init__(self, w3: "AsyncWeb3"):
         self._manager = w3.manager
+        self.provider = cast("PersistentConnectionProvider", self._manager.provider)
 
     @property
     def subscriptions(self) -> Dict[str, Any]:
@@ -47,10 +54,10 @@ class PersistentConnection:
         :param method: The RPC method, e.g. `eth_getBlockByNumber`.
         :param params: The RPC method parameters, e.g. `["0x1337", False]`.
 
-        :return: The processed response from the persistent connection.
+        :return: The unprocessed response from the persistent connection.
         :rtype: RPCResponse
         """
-        return await self._manager.socket_request(method, params)
+        return await self.provider.make_request(method, params)
 
     async def send(self, method: RPCEndpoint, params: Any) -> None:
         """
@@ -63,14 +70,14 @@ class PersistentConnection:
         """
         await self._manager.send(method, params)
 
-    async def recv(self) -> RPCResponse:
+    async def recv(self) -> Union[RPCResponse, FormattedEthSubscriptionResponse]:
         """
         Receive the next unprocessed response for a request from the persistent
         connection.
 
         :return: The next unprocessed response for a request from the persistent
                  connection.
-        :rtype: RPCResponse
+        :rtype: Union[RPCResponse, FormattedEthSubscriptionResponse]
         """
         return await self._manager.recv()
 
diff --git a/web3/providers/persistent/request_processor.py b/web3/providers/persistent/request_processor.py
index 1625ce33..9d5bf30a 100644
--- a/web3/providers/persistent/request_processor.py
+++ b/web3/providers/persistent/request_processor.py
@@ -1,7 +1,4 @@
 import asyncio
-from copy import (
-    copy,
-)
 import sys
 from typing import (
     TYPE_CHECKING,
@@ -27,11 +24,16 @@ from web3._utils.caching import (
     generate_cache_key,
 )
 from web3.exceptions import (
+    SubscriptionProcessingFinished,
     TaskNotRunning,
     Web3ValueError,
 )
+from web3.providers.persistent.subscription_manager import (
+    SubscriptionContainer,
+)
 from web3.types import (
     RPCEndpoint,
+    RPCId,
     RPCResponse,
 )
 from web3.utils import (
@@ -75,18 +77,23 @@ class TaskReliantQueue(_TaskReliantQueue[T]):
 class RequestProcessor:
     _subscription_queue_synced_with_ws_stream: bool = False
 
+    # set by the subscription manager when it is initialized
+    _subscription_container: Optional[SubscriptionContainer] = None
+
     def __init__(
         self,
         provider: "PersistentConnectionProvider",
         subscription_response_queue_size: int = 500,
     ) -> None:
         self._provider = provider
-
         self._request_information_cache: SimpleCache = SimpleCache(500)
         self._request_response_cache: SimpleCache = SimpleCache(500)
         self._subscription_response_queue: TaskReliantQueue[
             Union[RPCResponse, TaskNotRunning]
         ] = TaskReliantQueue(maxsize=subscription_response_queue_size)
+        self._handler_subscription_queue: TaskReliantQueue[
+            Union[RPCResponse, TaskNotRunning, SubscriptionProcessingFinished]
+        ] = TaskReliantQueue(maxsize=subscription_response_queue_size)
 
     @property
     def active_subscriptions(self) -> Dict[str, Any]:
@@ -100,6 +107,7 @@ class RequestProcessor:
 
     def cache_request_information(
         self,
+        request_id: Optional[RPCId],
         method: RPCEndpoint,
         params: Any,
         response_formatters: Tuple[
@@ -120,25 +128,23 @@ class RequestProcessor:
                 )
                 return None
 
-        if self._provider._is_batching:
+        if request_id is None:
+            if not self._provider._is_batching:
+                raise Web3ValueError(
+                    "Request id must be provided when not batching requests."
+                )
             # the _batch_request_counter is set when entering the context manager
-            current_request_id = self._provider._batch_request_counter
+            request_id = self._provider._batch_request_counter
             self._provider._batch_request_counter += 1
-        else:
-            # copy the request counter and find the next request id without incrementing
-            # since this is done when / if the request is successfully sent
-            current_request_id = next(copy(self._provider.request_counter))
-        cache_key = generate_cache_key(current_request_id)
-
-        self._bump_cache_if_key_present(cache_key, current_request_id)
 
+        cache_key = generate_cache_key(request_id)
         request_info = RequestInformation(
             method,
             params,
             response_formatters,
         )
         self._provider.logger.debug(
-            f"Caching request info:\n    request_id={current_request_id},\n"
+            f"Caching request info:\n    request_id={request_id},\n"
             f"    cache_key={cache_key},\n    request_info={request_info.__dict__}"
         )
         self._request_information_cache.cache(
@@ -147,30 +153,6 @@ class RequestProcessor:
         )
         return cache_key
 
-    def _bump_cache_if_key_present(self, cache_key: str, request_id: int) -> None:
-        """
-        If the cache key is present in the cache, bump the cache key and request id
-        by one to make room for the new request. This behavior is necessary when a
-        request is made but inner requests, say to `eth_estimateGas` if the `gas` is
-        missing, are made before the original request is sent.
-        """
-        if cache_key in self._request_information_cache:
-            original_request_info = self._request_information_cache.get_cache_entry(
-                cache_key
-            )
-            bump = generate_cache_key(request_id + 1)
-
-            # recursively bump the cache if the new key is also present
-            self._bump_cache_if_key_present(bump, request_id + 1)
-
-            self._provider.logger.debug(
-                "Caching internal request. Bumping original request in cache:\n"
-                f"    request_id=[{request_id}] -> [{request_id + 1}],\n"
-                f"    cache_key=[{cache_key}] -> [{bump}],\n"
-                f"    request_info={original_request_info.__dict__}"
-            )
-            self._request_information_cache.cache(bump, original_request_info)
-
     def pop_cached_request_information(
         self, cache_key: str
     ) -> Optional[RequestInformation]:
@@ -303,7 +285,17 @@ class RequestProcessor:
             self._provider.logger.debug(
                 f"Caching subscription response:\n    response={raw_response}"
             )
-            await self._subscription_response_queue.put(raw_response)
+            subscription_id = raw_response.get("params", {}).get("subscription")
+            sub_container = self._subscription_container
+            if sub_container and sub_container.get_handler_subscription_by_id(
+                subscription_id
+            ):
+                # if the subscription has a handler, put it in the handler queue
+                await self._handler_subscription_queue.put(raw_response)
+            else:
+                # otherwise, put it in the subscription response queue so a response
+                # can be yielded by the message stream
+                await self._subscription_response_queue.put(raw_response)
         elif isinstance(raw_response, list):
             # Since only one batch should be in the cache at all times, we use a
             # constant cache key for the batch response.
@@ -367,7 +359,12 @@ class RequestProcessor:
 
         return raw_response
 
-    # request processor class methods
+    # cache methods
+
+    def _reset_handler_subscription_queue(self) -> None:
+        self._handler_subscription_queue = TaskReliantQueue(
+            maxsize=self._handler_subscription_queue.maxsize
+        )
 
     def clear_caches(self) -> None:
         """Clear the request processor caches."""
@@ -376,3 +373,4 @@ class RequestProcessor:
         self._subscription_response_queue = TaskReliantQueue(
             maxsize=self._subscription_response_queue.maxsize
         )
+        self._reset_handler_subscription_queue()
diff --git a/web3/providers/persistent/subscription_container.py b/web3/providers/persistent/subscription_container.py
new file mode 100644
index 00000000..57db2640
--- /dev/null
+++ b/web3/providers/persistent/subscription_container.py
@@ -0,0 +1,56 @@
+from typing import (
+    Any,
+    Dict,
+    Iterator,
+    List,
+    Optional,
+)
+
+from eth_typing import (
+    HexStr,
+)
+
+from web3.utils import (
+    EthSubscription,
+)
+
+
+class SubscriptionContainer:
+    def __init__(self) -> None:
+        self.subscriptions: List[EthSubscription[Any]] = []
+        self.subscriptions_by_id: Dict[HexStr, EthSubscription[Any]] = {}
+        self.subscriptions_by_label: Dict[str, EthSubscription[Any]] = {}
+
+    def __len__(self) -> int:
+        return len(self.subscriptions)
+
+    def __iter__(self) -> Iterator[EthSubscription[Any]]:
+        return iter(self.subscriptions)
+
+    def add_subscription(self, subscription: EthSubscription[Any]) -> None:
+        self.subscriptions.append(subscription)
+        self.subscriptions_by_id[subscription.id] = subscription
+        self.subscriptions_by_label[subscription.label] = subscription
+
+    def remove_subscription(self, subscription: EthSubscription[Any]) -> None:
+        self.subscriptions.remove(subscription)
+        self.subscriptions_by_id.pop(subscription.id)
+        self.subscriptions_by_label.pop(subscription.label)
+
+    def get_by_id(self, sub_id: HexStr) -> EthSubscription[Any]:
+        return self.subscriptions_by_id.get(sub_id)
+
+    def get_by_label(self, label: str) -> EthSubscription[Any]:
+        return self.subscriptions_by_label.get(label)
+
+    @property
+    def handler_subscriptions(self) -> List[EthSubscription[Any]]:
+        return [sub for sub in self.subscriptions if sub._handler is not None]
+
+    def get_handler_subscription_by_id(
+        self, sub_id: HexStr
+    ) -> Optional[EthSubscription[Any]]:
+        sub = self.get_by_id(sub_id)
+        if sub and sub._handler:
+            return sub
+        return None
diff --git a/web3/providers/persistent/subscription_manager.py b/web3/providers/persistent/subscription_manager.py
new file mode 100644
index 00000000..c6efdcec
--- /dev/null
+++ b/web3/providers/persistent/subscription_manager.py
@@ -0,0 +1,233 @@
+import asyncio
+import logging
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    List,
+    Sequence,
+    Union,
+    cast,
+    overload,
+)
+
+from eth_typing import (
+    HexStr,
+)
+
+from web3.exceptions import (
+    SubscriptionProcessingFinished,
+    TaskNotRunning,
+    Web3TypeError,
+    Web3ValueError,
+)
+from web3.providers.persistent.subscription_container import (
+    SubscriptionContainer,
+)
+from web3.types import (
+    FormattedEthSubscriptionResponse,
+    RPCResponse,
+)
+from web3.utils.subscriptions import (
+    EthSubscription,
+    EthSubscriptionContext,
+)
+
+if TYPE_CHECKING:
+    from web3 import AsyncWeb3  # noqa: F401
+    from web3.providers.persistent import (  # noqa: F401
+        PersistentConnectionProvider,
+        RequestProcessor,
+    )
+
+
+class SubscriptionManager:
+    """
+    The ``SubscriptionManager`` is responsible for subscribing, unsubscribing, and
+    managing all active subscriptions for an ``AsyncWeb3`` instance. It is also
+    used for processing all subscriptions that have handler functions.
+    """
+
+    logger: logging.Logger = logging.getLogger(
+        "web3.providers.persistent.subscription_manager"
+    )
+    total_handler_calls: int = 0
+
+    def __init__(self, w3: "AsyncWeb3") -> None:
+        self._w3 = w3
+        self._provider = cast("PersistentConnectionProvider", w3.provider)
+        self._subscription_container = SubscriptionContainer()
+
+        # share the subscription container with the request processor so it can separate
+        # subscriptions into different queues based on ``sub._handler`` presence
+        self._provider._request_processor._subscription_container = (
+            self._subscription_container
+        )
+
+    def _add_subscription(self, subscription: EthSubscription[Any]) -> None:
+        self._subscription_container.add_subscription(subscription)
+
+    def _remove_subscription(self, subscription: EthSubscription[Any]) -> None:
+        self._subscription_container.remove_subscription(subscription)
+
+    @property
+    def subscriptions(self) -> List[EthSubscription[Any]]:
+        return self._subscription_container.subscriptions
+
+    def get_by_id(self, sub_id: HexStr) -> EthSubscription[Any]:
+        return self._subscription_container.get_by_id(sub_id)
+
+    def get_by_label(self, label: str) -> EthSubscription[Any]:
+        return self._subscription_container.get_by_label(label)
+
+    @overload
+    async def subscribe(self, subscriptions: EthSubscription[Any]) -> HexStr:
+        ...
+
+    @overload
+    async def subscribe(
+        self, subscriptions: Sequence[EthSubscription[Any]]
+    ) -> List[HexStr]:
+        ...
+
+    async def subscribe(
+        self, subscriptions: Union[EthSubscription[Any], Sequence[EthSubscription[Any]]]
+    ) -> Union[HexStr, List[HexStr]]:
+        """
+        Used to subscribe to a single or multiple subscriptions.
+
+        :param subscriptions: A single subscription or a sequence of subscriptions.
+        :type subscriptions: Union[EthSubscription, Sequence[EthSubscription]]
+        :return:
+        """
+        if isinstance(subscriptions, EthSubscription):
+            if (
+                subscriptions.label
+                in self._subscription_container.subscriptions_by_label
+            ):
+                raise Web3ValueError(
+                    "Subscription label already exists. Subscriptions must have "
+                    f"unique labels.\n    label: {subscriptions.label}"
+                )
+
+            subscriptions.manager = self
+            sub_id = await self._w3.eth._subscribe(*subscriptions.subscription_params)
+            subscriptions._id = sub_id
+            self._add_subscription(subscriptions)
+            self.logger.info(
+                "Successfully subscribed to subscription:\n    "
+                f"label: {subscriptions.label}\n    id: {sub_id}"
+            )
+            return sub_id
+        elif isinstance(subscriptions, Sequence):
+            if len(subscriptions) == 0:
+                raise Web3ValueError("No subscriptions provided.")
+
+            sub_ids: List[HexStr] = []
+            for sub in subscriptions:
+                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:
+        """
+        Used to unsubscribe from a subscription.
+
+        :param subscription: The subscription to unsubscribe from.
+        :type subscription: EthSubscription
+        :return: ``True`` if unsubscribing was successful, ``False`` otherwise.
+        :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
+        return False
+
+    async def unsubscribe_all(self) -> bool:
+        """
+        Used to unsubscribe from all subscriptions that are being managed by the
+        subscription manager.
+
+        :return: ``True`` if unsubscribing was successful, ``False`` otherwise.
+        :rtype: bool
+        """
+        unsubscribed = [
+            await self.unsubscribe(sub) for sub in self.subscriptions.copy()
+        ]
+        if all(unsubscribed):
+            self.logger.info("Successfully unsubscribed from all subscriptions.")
+            return True
+        else:
+            if len(self.subscriptions) > 0:
+                self.logger.warning(
+                    "Failed to unsubscribe from all subscriptions. Some subscriptions "
+                    f"are still active.\n    subscriptions={self.subscriptions}"
+                )
+            return False
+
+    async def handle_subscriptions(self, run_forever: bool = False) -> None:
+        """
+        Used to handle all subscriptions that have handlers. The method will run until
+        all subscriptions that have handler functions are unsubscribed from or, if
+        ``run_forever`` is set to ``True``, it will run indefinitely.
+
+        :param run_forever: If ``True``, the method will run indefinitely.
+        :type run_forever: bool
+        :return: None
+        """
+        if not self._subscription_container.handler_subscriptions:
+            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:
+                response = cast(RPCResponse, await queue.get())
+                formatted_sub_response = cast(
+                    FormattedEthSubscriptionResponse,
+                    await self._w3.manager._process_response(response),
+                )
+
+                # if the subscription was unsubscribed from, the response won't be
+                # formatted because we lost the request information
+                sub_id = formatted_sub_response.get("subscription")
+                sub = self._subscription_container.get_handler_subscription_by_id(
+                    sub_id
+                )
+                if sub:
+                    await sub._handler(
+                        EthSubscriptionContext(
+                            self._w3,
+                            sub,
+                            formatted_sub_response["result"],
+                            **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."
+            )
+
+        # no active handler subscriptions, clear the handler subscription queue
+        self._provider._request_processor._reset_handler_subscription_queue()
diff --git a/web3/providers/persistent/websocket.py b/web3/providers/persistent/websocket.py
index 258ce223..0a598049 100644
--- a/web3/providers/persistent/websocket.py
+++ b/web3/providers/persistent/websocket.py
@@ -59,8 +59,6 @@ class WebSocketProvider(PersistentConnectionProvider):
     logger = logging.getLogger("web3.providers.WebSocketProvider")
     is_async: bool = True
 
-    _ws: Optional[WebSocketClientProtocol] = None
-
     def __init__(
         self,
         endpoint_uri: Optional[Union[URI, str]] = None,
@@ -68,9 +66,12 @@ class WebSocketProvider(PersistentConnectionProvider):
         # `PersistentConnectionProvider` kwargs can be passed through
         **kwargs: Any,
     ) -> None:
+        # initialize the endpoint_uri before calling the super constructor
         self.endpoint_uri = (
             URI(endpoint_uri) if endpoint_uri is not None else get_default_endpoint()
         )
+        super().__init__(**kwargs)
+        self._ws: Optional[WebSocketClientProtocol] = None
 
         if not any(
             self.endpoint_uri.startswith(prefix)
@@ -93,8 +94,6 @@ class WebSocketProvider(PersistentConnectionProvider):
 
         self.websocket_kwargs = merge(DEFAULT_WEBSOCKET_KWARGS, websocket_kwargs or {})
 
-        super().__init__(**kwargs)
-
     def __str__(self) -> str:
         return f"WebSocket connection: {self.endpoint_uri}"
 
@@ -133,6 +132,7 @@ class WebSocketProvider(PersistentConnectionProvider):
         self._ws = await connect(self.endpoint_uri, **self.websocket_kwargs)
 
     async def _provider_specific_disconnect(self) -> None:
+        # this should remain idempotent
         if self._ws is not None and not self._ws.closed:
             await self._ws.close()
             self._ws = None
diff --git a/web3/providers/rpc/rpc.py b/web3/providers/rpc/rpc.py
index 23593e84..7dde0416 100644
--- a/web3/providers/rpc/rpc.py
+++ b/web3/providers/rpc/rpc.py
@@ -71,6 +71,7 @@ class HTTPProvider(JSONBaseProvider):
         ] = empty,
         **kwargs: Any,
     ) -> None:
+        super().__init__(**kwargs)
         self._request_session_manager = HTTPSessionManager()
 
         if endpoint_uri is None:
@@ -88,8 +89,6 @@ class HTTPProvider(JSONBaseProvider):
                 self.endpoint_uri, session
             )
 
-        super().__init__(**kwargs)
-
     def __str__(self) -> str:
         return f"RPC connection {self.endpoint_uri}"
 
diff --git a/web3/types.py b/web3/types.py
index 7b2a227e..f7d0acc7 100644
--- a/web3/types.py
+++ b/web3/types.py
@@ -262,6 +262,13 @@ EthSubscriptionParams = Union[
 RPCId = Optional[Union[int, str]]
 
 
+class RPCRequest(TypedDict, total=False):
+    id: RPCId
+    jsonrpc: Literal["2.0"]
+    method: RPCEndpoint
+    params: Any
+
+
 class RPCResponse(TypedDict, total=False):
     error: RPCError
     id: RPCId
@@ -273,11 +280,19 @@ class RPCResponse(TypedDict, total=False):
     params: EthSubscriptionParams
 
 
+EthSubscriptionResult = Union[
+    BlockData,  # newHeads
+    TxData,  # newPendingTransactions, full_transactions=True
+    HexBytes,  # newPendingTransactions, full_transactions=False
+    LogReceipt,  # logs
+    SyncProgress,  # syncing
+    GethSyncingSubscriptionResult,  # geth syncing
+]
+
+
 class FormattedEthSubscriptionResponse(TypedDict):
     subscription: HexStr
-    result: Union[
-        BlockData, TxData, LogReceipt, SyncProgress, GethSyncingSubscriptionResult
-    ]
+    result: EthSubscriptionResult
 
 
 class CreateAccessListResponse(TypedDict):
diff --git a/web3/utils/__init__.py b/web3/utils/__init__.py
index d8cffb99..f603daea 100644
--- a/web3/utils/__init__.py
+++ b/web3/utils/__init__.py
@@ -42,6 +42,9 @@ from .caching import (
 from .exception_handling import (
     handle_offchain_lookup,
 )
+from .subscriptions import (
+    EthSubscription,
+)
 
 __all__ = [
     "abi_to_signature",
@@ -70,5 +73,6 @@ __all__ = [
     "async_handle_offchain_lookup",
     "RequestCacheValidationThreshold",
     "SimpleCache",
+    "EthSubscription",
     "handle_offchain_lookup",
 ]
diff --git a/web3/utils/subscriptions.py b/web3/utils/subscriptions.py
new file mode 100644
index 00000000..6fa28b95
--- /dev/null
+++ b/web3/utils/subscriptions.py
@@ -0,0 +1,285 @@
+from typing import (
+    TYPE_CHECKING,
+    Any,
+    Callable,
+    Coroutine,
+    Dict,
+    Generic,
+    List,
+    Optional,
+    Sequence,
+    TypeVar,
+    Union,
+)
+
+from eth_typing import (
+    Address,
+    ChecksumAddress,
+    HexStr,
+)
+from hexbytes import (
+    HexBytes,
+)
+
+from web3.exceptions import (
+    Web3AttributeError,
+    Web3ValueError,
+)
+from web3.types import (
+    BlockData,
+    FilterParams,
+    LogReceipt,
+    SyncProgress,
+    TxData,
+)
+
+if TYPE_CHECKING:
+    from web3 import (
+        AsyncWeb3,
+    )
+    from web3.providers.persistent.subscription_manager import (
+        SubscriptionManager,
+    )
+    from web3.types import EthSubscriptionResult  # noqa: F401
+
+
+TSubscriptionResult = TypeVar("TSubscriptionResult", bound="EthSubscriptionResult")
+TSubscription = TypeVar("TSubscription", bound="EthSubscription[Any]")
+
+
+class EthSubscriptionContext(Generic[TSubscription, TSubscriptionResult]):
+    def __init__(
+        self,
+        async_w3: "AsyncWeb3",
+        subscription: TSubscription,
+        result: TSubscriptionResult,
+        **kwargs: Any,
+    ) -> None:
+        self.async_w3 = async_w3
+        self.subscription = subscription
+        self.result = result
+        self.__dict__.update(kwargs)
+
+    def __getattr__(self, item: str) -> Any:
+        if item in self.__dict__:
+            return self.__dict__[item]
+        raise Web3AttributeError(
+            f"'{self.__class__.__name__}' object has no attribute '{item}'"
+        )
+
+
+EthSubscriptionHandler = Callable[
+    [EthSubscriptionContext[Any, Any]], Coroutine[Any, Any, None]
+]
+
+
+def handler_wrapper(
+    handler: Optional[EthSubscriptionHandler],
+) -> Optional[EthSubscriptionHandler]:
+    """Wrap the handler to add bookkeeping and context creation."""
+    if handler is None:
+        return None
+
+    async def wrapped_handler(
+        context: EthSubscriptionContext[TSubscription, TSubscriptionResult],
+    ) -> None:
+        sub = context.subscription
+        sub.handler_call_count += 1
+        sub.manager.total_handler_calls += 1
+        sub.manager.logger.debug(
+            f"Subscription handler called.\n"
+            f"    label: {sub.label}\n"
+            f"    call count: {sub.handler_call_count}\n"
+            f"    total handler calls: {sub.manager.total_handler_calls}"
+        )
+        await handler(context)
+
+    return wrapped_handler
+
+
+class EthSubscription(Generic[TSubscriptionResult]):
+    _id: HexStr = None
+    manager: "SubscriptionManager" = None
+
+    def __init__(
+        self: TSubscription,
+        subscription_params: Optional[Sequence[Any]] = None,
+        handler: Optional[EthSubscriptionHandler] = None,
+        handler_context: Optional[Dict[str, Any]] = None,
+        label: Optional[str] = None,
+    ) -> None:
+        self._subscription_params = subscription_params
+        self._handler = handler_wrapper(handler)
+        self._handler_context = handler_context or {}
+        self._label = label
+        self.handler_call_count = 0
+
+    @classmethod
+    def _create_type_aware_subscription(
+        cls,
+        subscription_params: Optional[Sequence[Any]],
+        handler: Optional[EthSubscriptionHandler] = None,
+        handler_context: Optional[Dict[str, Any]] = None,
+        label: Optional[str] = None,
+    ) -> "EthSubscription[Any]":
+        subscription_type = subscription_params[0]
+        subscription_arg = (
+            subscription_params[1] if len(subscription_params) > 1 else None
+        )
+        if subscription_type == "newHeads":
+            return NewHeadsSubscription(
+                handler=handler, handler_context=handler_context, label=label
+            )
+        elif subscription_type == "logs":
+            subscription_arg = subscription_arg or {}
+            return LogsSubscription(
+                **subscription_arg,
+                handler=handler,
+                handler_context=handler_context,
+                label=label,
+            )
+        elif subscription_type == "newPendingTransactions":
+            subscription_arg = subscription_arg or False
+            return PendingTxSubscription(
+                full_transactions=subscription_arg,
+                handler=handler,
+                handler_context=handler_context,
+                label=label,
+            )
+        elif subscription_type == "syncing":
+            return SyncingSubscription(
+                handler=handler, handler_context=handler_context, label=label
+            )
+        else:
+            params = (
+                (subscription_type, subscription_arg)
+                if subscription_arg
+                else (subscription_type,)
+            )
+            return cls(
+                params,
+                handler=handler,
+                handler_context=handler_context,
+                label=label,
+            )
+
+    @property
+    def subscription_params(self) -> Sequence[Any]:
+        return self._subscription_params
+
+    @property
+    def label(self) -> str:
+        if not self._label:
+            self._label = f"{self.__class__.__name__}{self.subscription_params}"
+        return self._label
+
+    @property
+    def id(self) -> HexStr:
+        if not self._id:
+            raise Web3ValueError("No `id` found for subscription.")
+        return self._id
+
+    async def unsubscribe(self) -> bool:
+        return await self.manager.unsubscribe(self)
+
+
+LogsSubscriptionContext = EthSubscriptionContext[
+    "LogsSubscription", "EthSubscriptionResult"
+]
+LogsSubscriptionHandler = Callable[[LogsSubscriptionContext], Coroutine[Any, Any, None]]
+
+
+class LogsSubscription(EthSubscription[LogReceipt]):
+    def __init__(
+        self,
+        address: Optional[
+            Union[Address, ChecksumAddress, List[Address], List[ChecksumAddress]]
+        ] = None,
+        topics: Optional[List[HexStr]] = None,
+        handler: LogsSubscriptionHandler = None,
+        handler_context: Optional[Dict[str, Any]] = None,
+        label: Optional[str] = None,
+    ) -> None:
+        self.address = address
+        self.topics = topics
+
+        logs_filter: FilterParams = {}
+        if address:
+            logs_filter["address"] = address
+        if topics:
+            logs_filter["topics"] = topics
+        self.logs_filter = logs_filter
+
+        super().__init__(
+            subscription_params=("logs", logs_filter),
+            handler=handler,
+            handler_context=handler_context,
+            label=label,
+        )
+
+
+NewHeadsSubscriptionContext = EthSubscriptionContext["NewHeadsSubscription", BlockData]
+NewHeadsSubscriptionHandler = Callable[
+    [NewHeadsSubscriptionContext], Coroutine[Any, Any, None]
+]
+
+
+class NewHeadsSubscription(EthSubscription[BlockData]):
+    def __init__(
+        self,
+        label: Optional[str] = None,
+        handler: Optional[NewHeadsSubscriptionHandler] = None,
+        handler_context: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        super().__init__(
+            subscription_params=("newHeads",),
+            handler=handler,
+            handler_context=handler_context,
+            label=label,
+        )
+
+
+PendingTxSubscriptionContext = EthSubscriptionContext[
+    "PendingTxSubscription", Union[HexBytes, TxData]
+]
+PendingTxSubscriptionHandler = Callable[
+    [PendingTxSubscriptionContext], Coroutine[Any, Any, None]
+]
+
+
+class PendingTxSubscription(EthSubscription[Union[HexBytes, TxData]]):
+    def __init__(
+        self,
+        full_transactions: bool = False,
+        label: Optional[str] = None,
+        handler: Optional[PendingTxSubscriptionHandler] = None,
+        handler_context: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        self.full_transactions = full_transactions
+        super().__init__(
+            subscription_params=("newPendingTransactions", full_transactions),
+            handler=handler,
+            handler_context=handler_context,
+            label=label,
+        )
+
+
+SyncingSubscriptionContext = EthSubscriptionContext["SyncingSubscription", SyncProgress]
+SyncingSubscriptionHandler = Callable[
+    [SyncingSubscriptionContext], Coroutine[Any, Any, None]
+]
+
+
+class SyncingSubscription(EthSubscription[SyncProgress]):
+    def __init__(
+        self,
+        label: Optional[str] = None,
+        handler: Optional[SyncingSubscriptionHandler] = None,
+        handler_context: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        super().__init__(
+            subscription_params=("syncing",),
+            handler=handler,
+            handler_context=handler_context,
+            label=label,
+        )


Test patch:
diff --git a/tests/core/providers/test_async_ipc_provider.py b/tests/core/providers/test_async_ipc_provider.py
index 845d1403..c559a445 100644
--- a/tests/core/providers/test_async_ipc_provider.py
+++ b/tests/core/providers/test_async_ipc_provider.py
@@ -180,9 +180,11 @@ async def test_disconnect_cleanup(
     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"})
+    provider._request_processor._handler_subscription_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
+    assert provider._request_processor._handler_subscription_queue.qsize() == 1
 
     await w3.provider.disconnect()
 
@@ -192,6 +194,7 @@ async def test_disconnect_cleanup(
     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()
+    assert provider._request_processor._handler_subscription_queue.empty()
 
 
 async def _raise_connection_closed(*_args, **_kwargs):
@@ -361,5 +364,5 @@ async def test_async_ipc_provider_write_messages_end_with_new_line_delimiter(
 
         await w3.provider.make_request("method", [])
 
-        request_data = b'{"jsonrpc": "2.0", "method": "method", "params": [], "id": 0}'
+        request_data = b'{"id": 0, "jsonrpc": "2.0", "method": "method", "params": []}'
         w3.provider._writer.write.assert_called_with(request_data + b"\n")
diff --git a/tests/core/providers/test_websocket_provider.py b/tests/core/providers/test_websocket_provider.py
index 88fbc9a6..7df7fbd8 100644
--- a/tests/core/providers/test_websocket_provider.py
+++ b/tests/core/providers/test_websocket_provider.py
@@ -20,6 +20,7 @@ from web3 import (
 )
 from web3._utils.caching import (
     RequestInformation,
+    generate_cache_key,
 )
 from web3._utils.module_testing.module_testing_utils import (
     WebSocketMessageStreamMock,
@@ -33,6 +34,9 @@ from web3.providers.persistent import (
 from web3.types import (
     RPCEndpoint,
 )
+from web3.utils import (
+    EthSubscription,
+)
 
 
 def _mock_ws(provider):
@@ -79,9 +83,11 @@ async def test_disconnect_cleanup():
     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"})
+    provider._request_processor._handler_subscription_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
+    assert provider._request_processor._handler_subscription_queue.qsize() == 1
 
     await provider.disconnect()
 
@@ -89,6 +95,7 @@ async def test_disconnect_cleanup():
     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()
+    assert provider._request_processor._handler_subscription_queue.empty()
 
 
 @pytest.mark.asyncio
@@ -270,12 +277,16 @@ async def test_listen_event_awaits_msg_processing_when_subscription_queue_is_ful
     sub_request_information = RequestInformation(
         method=RPCEndpoint("eth_subscribe"),
         params=["mock"],
-        response_formatters=(),
+        response_formatters=[[], [], []],
         subscription_id=sub_id,
     )
     async_w3.provider._request_processor._request_information_cache.cache(
-        "", sub_request_information
+        generate_cache_key(sub_id),
+        sub_request_information,
     )
+    sub = EthSubscription()
+    sub._id = sub_id
+    async_w3.subscription_manager._add_subscription(sub)
 
     mocked_sub = {
         "jsonrpc": "2.0",
@@ -305,8 +316,8 @@ async def test_listen_event_awaits_msg_processing_when_subscription_queue_is_ful
     async_w3.provider._listen_event.set.assert_not_called()
 
     async for message in async_w3.socket.process_subscriptions():
-        # assert the very next message is the mocked subscription
-        assert message == mocked_sub
+        # assert the very next message is the formatted mocked subscription
+        assert message == mocked_sub["params"]
         break
 
     # assert we set the _listen_event after we consume the message
@@ -379,3 +390,45 @@ async def test_connection_closed_ok_breaks_message_iteration():
         w3 = await AsyncWeb3(WebSocketProvider("ws://mocked"))
         async for _ in w3.socket.process_subscriptions():
             pytest.fail("Should not reach this point.")
+
+
+@pytest.mark.asyncio
+async def test_listener_task_breaks_out_of_stream_when_cancelled():
+    with patch(
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
+    ):
+        async_w3 = await AsyncWeb3(WebSocketProvider("ws://mocked"))
+
+    async_w3.provider._message_listener_task.cancel()
+    sub = EthSubscription()
+    sub._id = "0x1"
+    async_w3.subscription_manager._add_subscription(sub)
+    # this should hang indefinitely if the listener task does not put a
+    # ``TaskNotRunning`` in the ``_subscription_response_queue`` to break out of
+    # listening. The call to ``provider._handle_listener_task_exceptions`` bubbles up
+    # the exception.
+    with pytest.raises(asyncio.CancelledError):
+        async for _ in async_w3.socket.process_subscriptions():
+            ...
+
+
+@pytest.mark.asyncio
+async def test_listener_task_breaks_out_of_handle_subscriptions_when_cancelled():
+    with patch(
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
+    ):
+        async_w3 = await AsyncWeb3(WebSocketProvider("ws://mocked"))
+
+    async_w3.provider._message_listener_task.cancel()
+    sub = EthSubscription(handler=AsyncMock())
+    sub._id = "0x1"
+
+    async_w3.subscription_manager._add_subscription(sub)
+    # this should hang indefinitely if the listener task does not put a
+    # ``TaskNotRunning`` in the ``_handler_subscription_queue`` to break out of
+    # listening. The call to ``provider._handle_listener_task_exceptions`` bubbles
+    # up the exception.
+    with pytest.raises(asyncio.CancelledError):
+        await async_w3.subscription_manager.handle_subscriptions(run_forever=True)
diff --git a/tests/core/subscriptions/test_subscription_manager.py b/tests/core/subscriptions/test_subscription_manager.py
new file mode 100644
index 00000000..341d38f2
--- /dev/null
+++ b/tests/core/subscriptions/test_subscription_manager.py
@@ -0,0 +1,112 @@
+import itertools
+import pytest
+from unittest.mock import (
+    AsyncMock,
+)
+
+import pytest_asyncio
+
+from web3 import (
+    AsyncWeb3,
+    PersistentConnectionProvider,
+)
+from web3.exceptions import (
+    Web3ValueError,
+)
+from web3.providers.persistent.subscription_manager import (
+    SubscriptionManager,
+)
+from web3.utils.subscriptions import (
+    LogsSubscription,
+    NewHeadsSubscription,
+    PendingTxSubscription,
+)
+
+
+class MockProvider(PersistentConnectionProvider):
+    socket_recv = AsyncMock()
+    socket_send = AsyncMock()
+
+
+@pytest_asyncio.fixture
+async def subscription_manager():
+    countr = itertools.count()
+    _w3 = AsyncWeb3(MockProvider())
+    _w3.eth._subscribe = AsyncMock()
+    _w3.eth._subscribe.side_effect = lambda *_: f"0x{str(next(countr))}"
+    _w3.eth._unsubscribe = AsyncMock()
+    _w3.eth._unsubscribe.return_value = True
+    yield SubscriptionManager(_w3)
+
+
+@pytest.mark.asyncio
+async def test_subscription_manager_raises_for_sub_with_the_same_label(
+    subscription_manager,
+):
+    sub1 = NewHeadsSubscription(label="foo")
+    await subscription_manager.subscribe(sub1)
+
+    with pytest.raises(
+        Web3ValueError,
+        match="Subscription label already exists. Subscriptions must have unique "
+        "labels.\n    label: foo",
+    ):
+        sub2 = LogsSubscription(label="foo")
+        await subscription_manager.subscribe(sub2)
+
+    # make sure the subscription was subscribed to and not added to the manager
+    assert subscription_manager.subscriptions == [sub1]
+    sub_container = subscription_manager._subscription_container
+    assert len(sub_container) == 1
+    assert sub_container.subscriptions == [sub1]
+    assert sub_container.subscriptions_by_id == {"0x0": sub1}
+    assert sub_container.subscriptions_by_label == {"foo": sub1}
+
+
+@pytest.mark.asyncio
+async def test_subscription_manager_get_by_id(subscription_manager):
+    sub = NewHeadsSubscription(label="foo")
+    await subscription_manager.subscribe(sub)
+    assert subscription_manager.get_by_id("0x0") == sub
+    assert subscription_manager.get_by_id("0x1") is None
+
+
+@pytest.mark.asyncio
+async def test_subscription_manager_get_by_label(subscription_manager):
+    sub = NewHeadsSubscription(label="foo")
+    await subscription_manager.subscribe(sub)
+    assert subscription_manager.get_by_label("foo") == sub
+    assert subscription_manager.get_by_label("bar") is None
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_one_by_one_clears_all_subscriptions(
+    subscription_manager,
+):
+    sub1 = NewHeadsSubscription(label="foo")
+    sub2 = PendingTxSubscription(label="bar")
+    await subscription_manager.subscribe(sub1)
+    await subscription_manager.subscribe(sub2)
+
+    await subscription_manager.unsubscribe(sub1)
+    assert subscription_manager.subscriptions == [sub2]
+
+    await subscription_manager.unsubscribe(sub2)
+    assert subscription_manager.subscriptions == []
+
+
+@pytest.mark.asyncio
+async def test_unsubscribe_all_clears_all_subscriptions(subscription_manager):
+    sub1 = NewHeadsSubscription(label="foo")
+    sub2 = PendingTxSubscription(label="bar")
+    await subscription_manager.subscribe([sub1, sub2])
+    assert subscription_manager.subscriptions == [sub1, sub2]
+
+    await subscription_manager.unsubscribe_all()
+    assert subscription_manager.subscriptions == []
+
+    sub_container = subscription_manager._subscription_container
+    assert len(sub_container) == 0
+    assert sub_container.subscriptions == []
+    assert sub_container.subscriptions_by_id == {}
+    assert sub_container.subscriptions_by_label == {}
diff --git a/tests/core/subscriptions/test_subscriptions.py b/tests/core/subscriptions/test_subscriptions.py
new file mode 100644
index 00000000..84284e86
--- /dev/null
+++ b/tests/core/subscriptions/test_subscriptions.py
@@ -0,0 +1,83 @@
+import pytest
+
+from web3.utils.subscriptions import (
+    LogsSubscription,
+    NewHeadsSubscription,
+    PendingTxSubscription,
+    SyncingSubscription,
+)
+
+
+@pytest.fixture
+def handler():
+    pass
+
+
+def test_new_heads_subscription_properties(handler):
+    new_heads_subscription = NewHeadsSubscription(
+        handler=handler, label="new heads label"
+    )
+    assert new_heads_subscription._handler is handler
+    assert new_heads_subscription.label == "new heads label"
+    assert new_heads_subscription.subscription_params == ("newHeads",)
+
+
+def test_pending_tx_subscription_properties_default(handler):
+    pending_tx_subscription = PendingTxSubscription(
+        handler=handler, label="pending tx label"
+    )
+    assert pending_tx_subscription._handler is handler
+    assert pending_tx_subscription.label == "pending tx label"
+    assert pending_tx_subscription.full_transactions is False
+    assert pending_tx_subscription.subscription_params == (
+        "newPendingTransactions",
+        False,
+    )
+
+
+def test_pending_tx_subscription_properties_full_transactions(handler):
+    pending_tx_subscription = PendingTxSubscription(
+        full_transactions=True, handler=handler, label="pending tx label"
+    )
+    assert pending_tx_subscription._handler is handler
+    assert pending_tx_subscription.label == "pending tx label"
+    assert pending_tx_subscription.full_transactions is True
+    assert pending_tx_subscription.subscription_params == (
+        "newPendingTransactions",
+        True,
+    )
+
+
+def test_logs_subscription_properties_default(handler):
+    logs_subscription = LogsSubscription(handler=handler, label="logs label")
+    assert logs_subscription._handler is handler
+    assert logs_subscription.label == "logs label"
+    assert logs_subscription.subscription_params == ("logs", {})
+    assert logs_subscription.address is None
+    assert logs_subscription.topics is None
+
+
+def test_logs_subscription_properties(handler):
+    address = "0x1234567890123456789012345678901234567890"
+    topics = [
+        "0x0000000000000000000000000000000000000000000000000000000000000001",
+        "0x0000000000000000000000000000000000000000000000000000000000000002",
+    ]
+    logs_subscription = LogsSubscription(
+        address=address, topics=topics, handler=handler, label="logs label"
+    )
+    assert logs_subscription._handler is handler
+    assert logs_subscription.label == "logs label"
+    assert logs_subscription.subscription_params == (
+        "logs",
+        {"address": address, "topics": topics},
+    )
+    assert logs_subscription.address == address
+    assert logs_subscription.topics == topics
+
+
+def test_syncing_subscription_properties(handler):
+    syncing_subscription = SyncingSubscription(handler=handler, label="syncing label")
+    assert syncing_subscription._handler is handler
+    assert syncing_subscription.label == "syncing label"
+    assert syncing_subscription.subscription_params == ("syncing",)
diff --git a/tests/integration/go_ethereum/conftest.py b/tests/integration/go_ethereum/conftest.py
index feb1e6dd..e1666203 100644
--- a/tests/integration/go_ethereum/conftest.py
+++ b/tests/integration/go_ethereum/conftest.py
@@ -94,7 +94,7 @@ def base_geth_command_arguments(geth_binary, datadir):
         datadir,
         "--dev",
         "--dev.period",
-        "1",
+        "2",
         "--password",
         os.path.join(datadir, "keystore", "pw.txt"),
     )
diff --git a/tests/integration/go_ethereum/test_goethereum_http.py b/tests/integration/go_ethereum/test_goethereum_http.py
index 42b7959b..393a526e 100644
--- a/tests/integration/go_ethereum/test_goethereum_http.py
+++ b/tests/integration/go_ethereum/test_goethereum_http.py
@@ -1,5 +1,8 @@
 import pytest
 
+from aiohttp import (
+    ClientTimeout,
+)
 import pytest_asyncio
 
 from tests.utils import (
@@ -70,7 +73,7 @@ def geth_command_arguments(rpc_port, base_geth_command_arguments, get_geth_versi
 @pytest.fixture(scope="module")
 def w3(geth_process, endpoint_uri):
     wait_for_http(endpoint_uri)
-    return Web3(Web3.HTTPProvider(endpoint_uri))
+    return Web3(Web3.HTTPProvider(endpoint_uri, request_kwargs={"timeout": 10}))
 
 
 class TestGoEthereumWeb3ModuleTest(GoEthereumWeb3ModuleTest):
@@ -119,7 +122,9 @@ class TestGoEthereumTxPoolModuleTest(GoEthereumTxPoolModuleTest):
 @pytest_asyncio.fixture(scope="module")
 async def async_w3(geth_process, endpoint_uri):
     await wait_for_aiohttp(endpoint_uri)
-    _w3 = AsyncWeb3(AsyncHTTPProvider(endpoint_uri))
+    _w3 = AsyncWeb3(
+        AsyncHTTPProvider(endpoint_uri, request_kwargs={"timeout": ClientTimeout(10)})
+    )
     return _w3
 
 
diff --git a/tests/integration/go_ethereum/test_goethereum_ipc.py b/tests/integration/go_ethereum/test_goethereum_ipc.py
index ec708d8b..2be3b503 100644
--- a/tests/integration/go_ethereum/test_goethereum_ipc.py
+++ b/tests/integration/go_ethereum/test_goethereum_ipc.py
@@ -56,7 +56,7 @@ def geth_ipc_path(datadir):
 @pytest.fixture(scope="module")
 def w3(geth_process, geth_ipc_path):
     wait_for_socket(geth_ipc_path)
-    return Web3(Web3.IPCProvider(geth_ipc_path, timeout=30))
+    return Web3(Web3.IPCProvider(geth_ipc_path, timeout=10))
 
 
 class TestGoEthereumWeb3ModuleTest(GoEthereumWeb3ModuleTest):
@@ -101,7 +101,7 @@ class TestGoEthereumAdminModuleTest(GoEthereumAdminModuleTest):
 @pytest_asyncio.fixture(scope="module")
 async def async_w3(geth_process, geth_ipc_path):
     await wait_for_async_socket(geth_ipc_path)
-    async with AsyncWeb3(AsyncIPCProvider(geth_ipc_path)) as _aw3:
+    async with AsyncWeb3(AsyncIPCProvider(geth_ipc_path, request_timeout=10)) as _aw3:
         yield _aw3
 
 
diff --git a/tests/integration/go_ethereum/test_goethereum_ws/test_async_await_w3.py b/tests/integration/go_ethereum/test_goethereum_ws/test_async_await_w3.py
index a04c56d1..e491d92f 100644
--- a/tests/integration/go_ethereum/test_goethereum_ws/test_async_await_w3.py
+++ b/tests/integration/go_ethereum/test_goethereum_ws/test_async_await_w3.py
@@ -27,9 +27,8 @@ from ..utils import (
 @pytest_asyncio.fixture(scope="module")
 async def async_w3(geth_process, endpoint_uri):
     await wait_for_aiohttp(endpoint_uri)
-
     # await the persistent connection itself
-    return await AsyncWeb3(WebSocketProvider(endpoint_uri))
+    return await AsyncWeb3(WebSocketProvider(endpoint_uri, request_timeout=10))
 
 
 class TestGoEthereumAsyncWeb3ModuleTest(GoEthereumAsyncWeb3ModuleTest):
diff --git a/tests/integration/go_ethereum/test_goethereum_ws/test_async_ctx_manager_w3.py b/tests/integration/go_ethereum/test_goethereum_ws/test_async_ctx_manager_w3.py
index 127627fb..7187dbf4 100644
--- a/tests/integration/go_ethereum/test_goethereum_ws/test_async_ctx_manager_w3.py
+++ b/tests/integration/go_ethereum/test_goethereum_ws/test_async_ctx_manager_w3.py
@@ -27,9 +27,8 @@ from ..utils import (
 @pytest_asyncio.fixture(scope="module")
 async def async_w3(geth_process, endpoint_uri):
     await wait_for_aiohttp(endpoint_uri)
-
     # async context manager pattern
-    async with AsyncWeb3(WebSocketProvider(endpoint_uri)) as w3:
+    async with AsyncWeb3(WebSocketProvider(endpoint_uri, request_timeout=10)) as w3:
         yield w3
 
 
diff --git a/tests/integration/go_ethereum/test_goethereum_ws/test_async_iterator_w3.py b/tests/integration/go_ethereum/test_goethereum_ws/test_async_iterator_w3.py
index d10bad9c..5e80702d 100644
--- a/tests/integration/go_ethereum/test_goethereum_ws/test_async_iterator_w3.py
+++ b/tests/integration/go_ethereum/test_goethereum_ws/test_async_iterator_w3.py
@@ -27,9 +27,8 @@ from ..utils import (
 @pytest_asyncio.fixture(scope="module")
 async def async_w3(geth_process, endpoint_uri):
     await wait_for_aiohttp(endpoint_uri)
-
     # async iterator pattern
-    async for w3 in AsyncWeb3(WebSocketProvider(endpoint_uri)):
+    async for w3 in AsyncWeb3(WebSocketProvider(endpoint_uri, request_timeout=10)):
         return w3
 
 
diff --git a/web3/_utils/module_testing/eth_module.py b/web3/_utils/module_testing/eth_module.py
index c2f16658..7ccd1c75 100644
--- a/web3/_utils/module_testing/eth_module.py
+++ b/web3/_utils/module_testing/eth_module.py
@@ -58,7 +58,6 @@ from web3._utils.module_testing.module_testing_utils import (
     assert_contains_log,
     async_mock_offchain_lookup_request_response,
     flaky_geth_dev_mining,
-    flaky_with_xfail_on_exception,
     mock_offchain_lookup_request_response,
 )
 from web3._utils.module_testing.utils import (
@@ -77,7 +76,6 @@ from web3.exceptions import (
     MultipleFailedRequests,
     NameNotFound,
     OffchainLookup,
-    RequestTimedOut,
     TimeExhausted,
     TooManyRequests,
     TransactionNotFound,
@@ -1991,9 +1989,8 @@ class AsyncEthModuleTest:
         # Test with None overflowing
         filter_params: FilterParams = {
             "fromBlock": BlockNumber(0),
-            "topics": [None, None, None],
+            "topics": [None, None, None, None],
         }
-
         result = await async_w3.eth.get_logs(filter_params)
         assert len(result) == 0
 
@@ -2391,10 +2388,6 @@ class AsyncEthModuleTest:
         with pytest.raises(Web3ValueError):
             await async_w3.eth.replace_transaction(txn_hash, txn_params)
 
-    @flaky_with_xfail_on_exception(
-        reason="Very flaky on CI runs, hard to reproduce locally.",
-        exception=RequestTimedOut,
-    )
     @pytest.mark.asyncio
     async def test_async_eth_replace_transaction_gas_price_defaulting_minimum(
         self, async_w3: "AsyncWeb3", async_keyfile_account_address: ChecksumAddress
@@ -2418,10 +2411,6 @@ class AsyncEthModuleTest:
             gas_price * 1.125
         )  # minimum gas price
 
-    @flaky_with_xfail_on_exception(
-        reason="Very flaky on CI runs, hard to reproduce locally.",
-        exception=RequestTimedOut,
-    )
     @pytest.mark.asyncio
     async def test_async_eth_replace_transaction_gas_price_defaulting_strategy_higher(
         self, async_w3: "AsyncWeb3", async_keyfile_account_address: ChecksumAddress
@@ -2437,7 +2426,7 @@ class AsyncEthModuleTest:
 
         two_gwei_in_wei = async_w3.to_wei(2, "gwei")
 
-        def higher_gas_price_strategy(async_w3: "AsyncWeb3", txn: TxParams) -> Wei:
+        def higher_gas_price_strategy(_async_w3: "AsyncWeb3", _txn: TxParams) -> Wei:
             return two_gwei_in_wei
 
         async_w3.eth.set_gas_price_strategy(higher_gas_price_strategy)
@@ -2450,16 +2439,11 @@ class AsyncEthModuleTest:
         )  # Strategy provides higher gas price
         async_w3.eth.set_gas_price_strategy(None)  # reset strategy
 
-    @flaky_with_xfail_on_exception(
-        reason="Very flaky on CI runs, hard to reproduce locally.",
-        exception=RequestTimedOut,
-    )
     @pytest.mark.asyncio
     async def test_async_eth_replace_transaction_gas_price_defaulting_strategy_lower(
         self, async_w3: "AsyncWeb3", async_keyfile_account_address: ChecksumAddress
     ) -> None:
         gas_price = async_w3.to_wei(2, "gwei")
-
         txn_params: TxParams = {
             "from": async_keyfile_account_address,
             "to": async_keyfile_account_address,
diff --git a/web3/_utils/module_testing/module_testing_utils.py b/web3/_utils/module_testing/module_testing_utils.py
index a15fea72..6a19fe56 100644
--- a/web3/_utils/module_testing/module_testing_utils.py
+++ b/web3/_utils/module_testing/module_testing_utils.py
@@ -106,7 +106,7 @@ def assert_contains_log(
     emitter_contract_address: ChecksumAddress,
     txn_hash_with_log: HexStr,
 ) -> None:
-    assert len(result) == 1
+    assert len(result) > 0
     log_entry = result[0]
     assert log_entry["blockNumber"] == block_with_txn_with_log["number"]
     assert log_entry["blockHash"] == block_with_txn_with_log["hash"]
diff --git a/web3/_utils/module_testing/persistent_connection_provider.py b/web3/_utils/module_testing/persistent_connection_provider.py
index 41fe8fbc..2eec3111 100644
--- a/web3/_utils/module_testing/persistent_connection_provider.py
+++ b/web3/_utils/module_testing/persistent_connection_provider.py
@@ -1,13 +1,23 @@
 import asyncio
+from dataclasses import (
+    dataclass,
+)
 import pytest
 from typing import (
     TYPE_CHECKING,
     Any,
     Dict,
+    Generator,
+    List,
     Tuple,
+    Union,
     cast,
 )
 
+from eth_typing import (
+    ChecksumAddress,
+    HexStr,
+)
 from eth_utils import (
     is_hexstr,
 )
@@ -15,6 +25,13 @@ from hexbytes import (
     HexBytes,
 )
 
+from web3 import (
+    AsyncWeb3,
+    PersistentConnectionProvider,
+)
+from web3.beacon import (
+    AsyncBeacon,
+)
 from web3.datastructures import (
     AttributeDict,
 )
@@ -22,16 +39,43 @@ from web3.middleware import (
     ExtraDataToPOAMiddleware,
 )
 from web3.types import (
+    BlockData,
     FormattedEthSubscriptionResponse,
+    LogReceipt,
+    Nonce,
     RPCEndpoint,
+    TxData,
+    Wei,
+)
+from web3.utils import (
+    EthSubscription,
+)
+from web3.utils.subscriptions import (
+    LogsSubscription,
+    LogsSubscriptionContext,
+    NewHeadsSubscription,
+    NewHeadsSubscriptionContext,
+    PendingTxSubscription,
+    PendingTxSubscriptionContext,
 )
 
 if TYPE_CHECKING:
-    from web3.main import (
-        AsyncWeb3,
+    from web3.contract.async_contract import (
+        AsyncContract,
+        AsyncContractFunction,
+    )
+    from web3.providers.persistent.subscription_manager import (
+        SubscriptionContainer,
     )
 
 
+# LogIndexedAndNotIndexed event args
+INDEXED_ADDR = "0xdEad000000000000000000000000000000000000"
+INDEXED_UINT256 = 1337
+NON_INDEXED_ADDR = "0xbeeF000000000000000000000000000000000000"
+NON_INDEXED_UINT256 = 1999
+NON_INDEXED_STRING = "test logs subscriptions"
+
 SOME_BLOCK_KEYS = [
     "number",
     "hash",
@@ -39,214 +83,182 @@ SOME_BLOCK_KEYS = [
     "transactionsRoot",
     "stateRoot",
     "receiptsRoot",
-    "size",
     "gasLimit",
     "gasUsed",
     "timestamp",
-    "transactions",
     "baseFeePerGas",
+    "withdrawalsRoot",
 ]
 
 
+@dataclass
+class SubscriptionHandlerTest:
+    passed: bool = False
+
+
+async def new_heads_handler(
+    handler_context: NewHeadsSubscriptionContext,
+) -> None:
+    w3 = handler_context.async_w3
+    sub = handler_context.subscription
+    assert isinstance(w3, AsyncWeb3)
+    provider = cast(PersistentConnectionProvider, w3.provider)
+    assert isinstance(provider.get_endpoint_uri_or_ipc_path(), str)
+
+    assert isinstance(sub, EthSubscription)
+
+    block = handler_context.result
+    assert block is not None
+    assert all(k in block.keys() for k in SOME_BLOCK_KEYS)
+
+    assert handler_context.new_heads_handler_test.passed is False
+    handler_context.new_heads_handler_test.passed = True
+    assert await sub.unsubscribe()
+
+
+async def pending_tx_handler(
+    handler_context: PendingTxSubscriptionContext,
+) -> None:
+    w3 = handler_context.async_w3
+    sub = handler_context.subscription
+    tx = handler_context.result
+
+    assert w3 is not None
+    provider = cast(PersistentConnectionProvider, w3.provider)
+    assert isinstance(provider.get_endpoint_uri_or_ipc_path(), str)
+
+    assert isinstance(sub, PendingTxSubscription)
+
+    assert tx is not None
+    tx = cast(TxData, tx)
+    accts = await w3.eth.accounts
+    assert tx["from"] == accts[0]
+    await w3.eth.wait_for_transaction_receipt(tx["hash"])
+
+    assert handler_context.pending_tx_handler_test.passed is False
+    handler_context.pending_tx_handler_test.passed = True
+    assert await sub.unsubscribe()
+
+
+async def logs_handler(
+    handler_context: LogsSubscriptionContext,
+) -> None:
+    w3 = handler_context.async_w3
+    sub = handler_context.subscription
+    log_receipt = handler_context.result
+
+    assert w3 is not None
+    provider = cast(PersistentConnectionProvider, w3.provider)
+    assert isinstance(provider.get_endpoint_uri_or_ipc_path(), str)
+
+    assert isinstance(sub, LogsSubscription)
+    event_data = handler_context.event.process_log(log_receipt)
+    assert event_data.args.indexedAddress == INDEXED_ADDR
+    assert event_data.args.indexedUint256 == INDEXED_UINT256
+    assert event_data.args.nonIndexedAddress == NON_INDEXED_ADDR
+    assert event_data.args.nonIndexedUint256 == NON_INDEXED_UINT256
+    assert event_data.args.nonIndexedString == NON_INDEXED_STRING
+
+    assert handler_context.logs_handler_test.passed is False
+    handler_context.logs_handler_test.passed = True
+    assert await sub.unsubscribe()
+
+
+async def idle_handler(
+    _handler_context: Any,
+) -> None:
+    pass
+
+
+async def emit_contract_event(
+    async_w3: AsyncWeb3,
+    acct: ChecksumAddress,
+    contract_function: "AsyncContractFunction",
+    args: Any = (),
+    delay: float = 0.25,
+) -> None:
+    await asyncio.sleep(delay)
+    tx_hash = await contract_function(*args).transact({"from": acct})
+    receipt = await async_w3.eth.wait_for_transaction_receipt(tx_hash)
+    assert receipt["status"] == 1
+
+
+async def log_indexed_and_non_indexed_args_task(
+    async_w3: AsyncWeb3,
+    async_emitter_contract: "AsyncContract",
+    acct: ChecksumAddress,
+    delay: float = 0.1,
+) -> "asyncio.Task[None]":
+    return asyncio.create_task(
+        emit_contract_event(
+            async_w3,
+            acct,
+            async_emitter_contract.functions.logIndexedAndNotIndexedArgs,
+            args=(
+                INDEXED_ADDR,
+                INDEXED_UINT256,
+                NON_INDEXED_ADDR,
+                NON_INDEXED_UINT256,
+                NON_INDEXED_STRING,
+            ),
+            delay=delay,
+        )
+    )
+
+
+def assert_no_subscriptions_left(sub_container: "SubscriptionContainer") -> None:
+    assert len(sub_container) == 0
+    assert len(sub_container.subscriptions) == 0
+    assert len(sub_container.subscriptions_by_id) == 0
+    assert len(sub_container.subscriptions_by_label) == 0
+    assert len(sub_container.handler_subscriptions) == 0
+
+
+async def clean_up_task(task: "asyncio.Task[Any]") -> None:
+    task.cancel()
+    try:
+        await task
+    except asyncio.CancelledError:
+        pass
+
+
 class PersistentConnectionProviderTest:
+    @pytest.fixture(autouse=True)
+    def clear_caches(self, async_w3: AsyncWeb3) -> Generator[None, None, None]:
+        yield
+        async_w3.provider._request_processor.clear_caches()
+        async_w3.subscription_manager.total_handler_calls = 0
+
+    @staticmethod
+    async def seed_transactions_to_geth(
+        async_w3: AsyncWeb3,
+        acct: ChecksumAddress,
+        num_txs: int = 1,
+        delay: float = 0.1,
+    ) -> None:
+        nonce = int(await async_w3.eth.get_transaction_count(acct))
+
+        async def send_tx() -> None:
+            nonlocal nonce
+            await async_w3.eth.send_transaction(
+                {
+                    "from": acct,
+                    "to": acct,
+                    "value": Wei(nonce),
+                    "gas": 21000,
+                    "nonce": Nonce(nonce),
+                }
+            )
+            nonce += 1
+
+        for _ in range(num_txs):
+            await asyncio.sleep(delay)
+            await send_tx()
+
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
         "subscription_params,ws_subscription_response,expected_formatted_result",
         (
-            (
-                ("newHeads",),
-                {
-                    "jsonrpc": "2.0",
-                    "method": "eth_subscription",
-                    "params": {
-                        "subscription": "THIS_WILL_BE_REPLACED_IN_THE_TEST",
-                        "result": {
-                            "number": "0x539",
-                            "hash": "0xb46b85928f2c2264c2bf7ad5c6d6985664f1527e744193ef990cc0d3da5afc5e",  # noqa: E501
-                            "parentHash": "0xb46b85928f2c2264c2bf7ad5c6d6985664f1527e744193ef990cc0d3da5afc5e",  # noqa: E501
-                            "sha3Uncles": "0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347",  # noqa: E501
-                            "logsBloom": "0x00",
-                            "transactionsRoot": "0x56260fe8298aff6d360e3a68fa855693f25dcb2708d8a7e509e8519b265d3988",  # noqa: E501
-                            "stateRoot": "0x56260fe8298aff6d360e3a68fa855693f25dcb2708d8a7e509e8519b265d3988",  # noqa: E501
-                            "receiptsRoot": "0x56260fe8298aff6d360e3a68fa855693f25dcb2708d8a7e509e8519b265d3988",  # noqa: E501
-                            "miner": "0x0000000000000000000000000000000000000000",
-                            "difficulty": "0x0",
-                            "extraData": "0x496c6c756d696e61746520446d6f63726174697a6520447374726962757465",  # noqa: E501
-                            "gasLimit": "0x1c9c380",
-                            "gasUsed": "0xd1ce44",
-                            "timestamp": "0x539",
-                            "baseFeePerGas": "0x26f93fef9",
-                            "withdrawalsRoot": "0x56260fe8298aff6d360e3a68fa855693f25dcb2708d8a7e509e8519b265d3988",  # noqa: E501
-                            "nonce": "0x0000000000000000",
-                            "mixHash": "0x73e9e036ec894047f29954571d4b6d9e8717de7304269c263cbf150caa4e0768",  # noqa: E501
-                        },
-                    },
-                },
-                AttributeDict(
-                    {
-                        "number": 1337,
-                        "hash": HexBytes(
-                            "0xb46b85928f2c2264c2bf7ad5c6d6985664f1527e744193ef990cc0d3da5afc5e"  # noqa: E501
-                        ),
-                        "parentHash": HexBytes(
-                            "0xb46b85928f2c2264c2bf7ad5c6d6985664f1527e744193ef990cc0d3da5afc5e"  # noqa: E501
-                        ),
-                        "sha3Uncles": HexBytes(
-                            "0x1dcc4de8dec75d7aab85b567b6ccd41ad312451b948a7413f0a142fd40d49347"  # noqa: E501
-                        ),
-                        "logsBloom": HexBytes("0x00"),
-                        "transactionsRoot": HexBytes(
-                            "0x56260fe8298aff6d360e3a68fa855693f25dcb2708d8a7e509e8519b265d3988"  # noqa: E501
-                        ),
-                        "stateRoot": HexBytes(
-                            "0x56260fe8298aff6d360e3a68fa855693f25dcb2708d8a7e509e8519b265d3988"  # noqa: E501
-                        ),
-                        "receiptsRoot": HexBytes(
-                            "0x56260fe8298aff6d360e3a68fa855693f25dcb2708d8a7e509e8519b265d3988"  # noqa: E501
-                        ),
-                        "miner": "0x0000000000000000000000000000000000000000",
-                        "difficulty": 0,
-                        "extraData": HexBytes(
-                            "0x496c6c756d696e61746520446d6f63726174697a6520447374726962757465"  # noqa: E501
-                        ),
-                        "gasLimit": 30000000,
-                        "gasUsed": 13749828,
-                        "timestamp": 1337,
-                        "baseFeePerGas": 10461904633,
-                        "withdrawalsRoot": HexBytes(
-                            "0x56260fe8298aff6d360e3a68fa855693f25dcb2708d8a7e509e8519b265d3988"  # noqa: E501
-                        ),
-                        "nonce": HexBytes("0x0000000000000000"),
-                        "mixHash": HexBytes(
-                            "0x73e9e036ec894047f29954571d4b6d9e8717de7304269c263cbf150caa4e0768"  # noqa: E501
-                        ),
-                    }
-                ),
-            ),
-            (
-                ("newPendingTransactions", True),
-                {
-                    "jsonrpc": "2.0",
-                    "method": "eth_subscription",
-                    "params": {
-                        "subscription": "THIS_WILL_BE_REPLACED_IN_THE_TEST",
-                        "result": {
-                            "blockHash": None,
-                            "blockNumber": None,
-                            "from": "0x0000000000000000000000000000000000000000",
-                            "gas": "0xf2f4",
-                            "gasPrice": "0x29035f36f",
-                            "maxFeePerGas": "0x29035f36f",
-                            "maxPriorityFeePerGas": "0x3b9aca00",
-                            "hash": "0xb46b85928f2c2264c2bf7ad5c6d6985664f1527e744193ef990cc0d3da5afc5e",  # noqa: E501
-                            "input": "0x00",
-                            "nonce": "0x2013",
-                            "to": "0x0000000000000000000000000000000000000000",
-                            "transactionIndex": None,
-                            "value": "0x0",
-                            "type": "0x2",
-                            "accessList": [],
-                            "chainId": "0x1",
-                            "v": "0x1",
-                            "r": "0x3c144a7c00ed3118d55445cd5be2ae4620ca377f7c685e9c5f3687671d4dece1",  # noqa: E501
-                            "s": "0x284de67cbf75fec8a9edb368dee3a37cf6faba87f0af4413b2f869ebfa87d002",  # noqa: E501
-                            "yParity": "0x1",
-                        },
-                    },
-                },
-                AttributeDict(
-                    {
-                        "blockHash": None,
-                        "blockNumber": None,
-                        "from": "0x0000000000000000000000000000000000000000",
-                        "gas": 62196,
-                        "gasPrice": 11009389423,
-                        "maxFeePerGas": 11009389423,
-                        "maxPriorityFeePerGas": 1000000000,
-                        "hash": HexBytes(
-                            "0xb46b85928f2c2264c2bf7ad5c6d6985664f1527e744193ef990cc0d3da5afc5e"  # noqa: E501
-                        ),
-                        "input": HexBytes("0x00"),
-                        "nonce": 8211,
-                        "to": "0x0000000000000000000000000000000000000000",
-                        "transactionIndex": None,
-                        "value": 0,
-                        "type": 2,
-                        "accessList": [],
-                        "chainId": 1,
-                        "v": 1,
-                        "r": HexBytes(
-                            "0x3c144a7c00ed3118d55445cd5be2ae4620ca377f7c685e9c5f3687671d4dece1"  # noqa: E501
-                        ),
-                        "s": HexBytes(
-                            "0x284de67cbf75fec8a9edb368dee3a37cf6faba87f0af4413b2f869ebfa87d002"  # noqa: E501
-                        ),
-                        "yParity": 1,
-                    }
-                ),
-            ),
-            (
-                ("newPendingTransactions", False),
-                {
-                    "jsonrpc": "2.0",
-                    "method": "eth_subscription",
-                    "params": {
-                        "subscription": "THIS_WILL_BE_REPLACED_IN_THE_TEST",
-                        "result": "0xb46b85928f2c2264c2bf7ad5c6d6985664f1527e744193ef990cc0d3da5afc5e",  # noqa: E501
-                    },
-                },
-                HexBytes(
-                    "0xb46b85928f2c2264c2bf7ad5c6d6985664f1527e744193ef990cc0d3da5afc5e"
-                ),
-            ),
-            (
-                ("logs", {"address": "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"}),
-                {
-                    "jsonrpc": "2.0",
-                    "method": "eth_subscription",
-                    "params": {
-                        "subscription": "THIS_WILL_BE_REPLACED_IN_THE_TEST",
-                        "result": {
-                            "removed": False,
-                            "logIndex": "0x0",
-                            "transactionIndex": "0x0",
-                            "transactionHash": "0x56260fe8298aff6d360e3a68fa855693f25dcb2708d8a7e509e8519b265d3988",  # noqa: E501
-                            "blockHash": "0xb46b85928f2c2264c2bf7ad5c6d6985664f1527e744193ef990cc0d3da5afc5e",  # noqa: E501
-                            "blockNumber": "0x539",
-                            "address": "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2",
-                            "data": "0x00",
-                            "topics": [
-                                "0xe1fffdd4923d04f559f4d29e8bfc6cda04eb5b0d3c460751c2402c5c5cc9105c",  # noqa: E501
-                                "0x00000000000000000000000016250d5630b4cf539739df2c5dacb4c659f2482d",  # noqa: E501
-                            ],
-                        },
-                    },
-                },
-                AttributeDict(
-                    {
-                        "removed": False,
-                        "logIndex": 0,
-                        "transactionIndex": 0,
-                        "transactionHash": HexBytes(
-                            "0x56260fe8298aff6d360e3a68fa855693f25dcb2708d8a7e509e8519b265d3988"  # noqa: E501
-                        ),
-                        "blockHash": HexBytes(
-                            "0xb46b85928f2c2264c2bf7ad5c6d6985664f1527e744193ef990cc0d3da5afc5e"  # noqa: E501
-                        ),
-                        "blockNumber": 1337,
-                        "address": "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2",
-                        "data": HexBytes("0x00"),
-                        "topics": [
-                            HexBytes(
-                                "0xe1fffdd4923d04f559f4d29e8bfc6cda04eb5b0d3c460751c2402c5c5cc9105c"  # noqa: E501
-                            ),
-                            HexBytes(
-                                "0x00000000000000000000000016250d5630b4cf539739df2c5dacb4c659f2482d"  # noqa: E501
-                            ),
-                        ],
-                    }
-                ),
-            ),
             (
                 ("syncing",),
                 {
@@ -285,17 +297,13 @@ class PersistentConnectionProviderTest:
             ),
         ),
         ids=[
-            "newHeads",
-            "newPendingTransactions-FullTxs",
-            "newPendingTransactions-TxHashes",
-            "logs",
             "syncing-False",
             "syncing-True",
         ],
     )
-    async def test_async_eth_subscribe_mocked(
+    async def test_async_eth_subscribe_syncing_mocked(
         self,
-        async_w3: "AsyncWeb3",
+        async_w3: AsyncWeb3,
         subscription_params: Tuple[Any, ...],
         ws_subscription_response: Dict[str, Any],
         expected_formatted_result: Any,
@@ -321,10 +329,262 @@ class PersistentConnectionProviderTest:
             await async_w3.eth.unsubscribe(sub_id)
             break
 
+        assert_no_subscriptions_left(
+            async_w3.subscription_manager._subscription_container
+        )
+
+    @pytest.mark.asyncio
+    async def test_async_eth_subscribe_new_heads(self, async_w3: AsyncWeb3) -> None:
+        sub_id = await async_w3.eth.subscribe("newHeads")
+        assert is_hexstr(sub_id)
+
+        async for msg in async_w3.socket.process_subscriptions():
+            response = cast(FormattedEthSubscriptionResponse, msg)
+            assert response["subscription"] == sub_id
+            result = cast(BlockData, response["result"])
+            assert all(k in result.keys() for k in SOME_BLOCK_KEYS)
+            break
+
+        assert await async_w3.eth.unsubscribe(sub_id)
+        assert_no_subscriptions_left(
+            async_w3.subscription_manager._subscription_container
+        )
+
+    @pytest.mark.asyncio
+    async def test_async_eth_subscribe_creates_and_handles_new_heads_subscription_type(
+        self,
+        async_w3: AsyncWeb3,
+    ) -> None:
+        sub_manager = async_w3.subscription_manager
+        new_heads_handler_test = SubscriptionHandlerTest()
+
+        sub_id = await async_w3.eth.subscribe(
+            "newHeads",
+            handler=new_heads_handler,
+            handler_context={"new_heads_handler_test": new_heads_handler_test},
+        )
+        assert is_hexstr(sub_id)
+
+        assert len(sub_manager.subscriptions) == 1
+        sub = sub_manager.subscriptions[0]
+        assert isinstance(sub, NewHeadsSubscription)
+
+        await sub_manager.handle_subscriptions()
+
+        assert new_heads_handler_test.passed
+        assert len(sub_manager.subscriptions) == 0
+
+        assert sub_manager.total_handler_calls == 1
+        assert sub.handler_call_count == 1
+
+    @pytest.mark.asyncio
+    async def test_async_eth_subscribe_process_pending_tx_true(
+        self,
+        async_w3: AsyncWeb3,
+    ) -> None:
+        sub_id = await async_w3.eth.subscribe("newPendingTransactions", True)
+        assert is_hexstr(sub_id)
+
+        accts = await async_w3.eth.accounts
+        acct = accts[0]
+
+        num_txs = 2
+        original_nonce = await async_w3.eth.get_transaction_count(acct)
+        tx_seeder_task = asyncio.create_task(
+            self.seed_transactions_to_geth(async_w3, acct, num_txs=num_txs)
+        )
+
+        nonce = int(original_nonce)
+        tx_hash = None
+        async for msg in async_w3.socket.process_subscriptions():
+            response = cast(FormattedEthSubscriptionResponse, msg)
+            assert response["subscription"] == sub_id
+            result = cast(TxData, response["result"])
+            assert result["gas"] == 21000
+            assert result["from"] == acct
+            assert result["to"] == acct
+            assert int(result["value"]) == int(nonce)
+            tx_hash = result["hash"]
+            assert tx_hash is not None
+
+            nonce += 1
+            if nonce == int(original_nonce) + num_txs:
+                break
+
+        # cleanup
+        assert await async_w3.eth.unsubscribe(sub_id)
+        assert_no_subscriptions_left(
+            async_w3.subscription_manager._subscription_container
+        )
+        async_w3.provider._request_processor.clear_caches()
+        await async_w3.eth.wait_for_transaction_receipt(tx_hash)
+        await clean_up_task(tx_seeder_task)
+
+    @pytest.mark.asyncio
+    async def test_async_eth_subscribe_and_process_pending_tx_false(
+        self,
+        async_w3: AsyncWeb3,
+    ) -> None:
+        sub_id = await async_w3.eth.subscribe("newPendingTransactions")
+        assert is_hexstr(sub_id)
+
+        accts = await async_w3.eth.accounts
+        acct = accts[0]
+        await async_w3.eth.get_transaction_count(acct)
+
+        num_txs = 2
+        tx_seeder_task = asyncio.create_task(
+            self.seed_transactions_to_geth(async_w3, acct, num_txs=num_txs)
+        )
+
+        tx_hash = None
+        i = 0
+        async for msg in async_w3.socket.process_subscriptions():
+            response = cast(FormattedEthSubscriptionResponse, msg)
+            assert response["subscription"] == sub_id
+            assert isinstance(response["result"], HexBytes)
+            tx_hash = response["result"]
+
+            i += 1
+            if i == num_txs:
+                break
+
+        # cleanup
+        await async_w3.eth.unsubscribe(sub_id)
+        assert_no_subscriptions_left(
+            async_w3.subscription_manager._subscription_container
+        )
+        await async_w3.eth.wait_for_transaction_receipt(tx_hash)
+        await clean_up_task(tx_seeder_task)
+
+    @pytest.mark.asyncio
+    async def test_async_eth_subscribe_creates_and_handles_pending_tx_subscription_type(
+        self,
+        async_w3: AsyncWeb3,
+    ) -> None:
+        sub_manager = async_w3.subscription_manager
+        pending_tx_handler_test = SubscriptionHandlerTest()
+
+        sub_id = await async_w3.eth.subscribe(
+            "newPendingTransactions",
+            True,
+            handler=pending_tx_handler,
+            handler_context={"pending_tx_handler_test": pending_tx_handler_test},
+        )
+        assert is_hexstr(sub_id)
+
+        assert len(sub_manager.subscriptions) == 1
+        sub = sub_manager.subscriptions[0]
+        assert isinstance(sub, PendingTxSubscription)
+
+        # seed transactions to geth
+        accts = await async_w3.eth.accounts
+        acct = accts[0]
+        tx_seeder_task = asyncio.create_task(
+            self.seed_transactions_to_geth(async_w3, acct)
+        )
+        await sub_manager.handle_subscriptions()
+
+        assert pending_tx_handler_test.passed
+        assert len(sub_manager.subscriptions) == 0
+
+        assert sub_manager.total_handler_calls == 1
+        assert sub.handler_call_count == 1
+
+        # cleanup
+        sub_manager.total_handler_calls = 0
+        await clean_up_task(tx_seeder_task)
+
+    @pytest.mark.asyncio
+    async def test_async_eth_subscribe_and_process_logs(
+        self, async_w3: AsyncWeb3, async_emitter_contract: "AsyncContract"
+    ) -> None:
+        event = async_emitter_contract.events.LogIndexedAndNotIndexed
+        event_topic = async_w3.keccak(text=event.abi_element_identifier).to_0x_hex()
+
+        sub_id = await async_w3.eth.subscribe(
+            "logs",
+            {
+                "address": async_emitter_contract.address,
+                "topics": [HexStr(event_topic)],
+            },
+        )
+        assert is_hexstr(sub_id)
+
+        accts = await async_w3.eth.accounts
+        acct = accts[0]
+        emit_event_task = await log_indexed_and_non_indexed_args_task(
+            async_w3, async_emitter_contract, acct
+        )
+
+        async for msg in async_w3.socket.process_subscriptions():
+            response = cast(FormattedEthSubscriptionResponse, msg)
+            assert response["subscription"] == sub_id
+            log_receipt = cast(LogReceipt, response["result"])
+            event_data = event.process_log(log_receipt)
+            assert event_data.args.indexedAddress == INDEXED_ADDR
+            assert event_data.args.indexedUint256 == INDEXED_UINT256
+            assert event_data.args.nonIndexedAddress == NON_INDEXED_ADDR
+            assert event_data.args.nonIndexedUint256 == NON_INDEXED_UINT256
+            assert event_data.args.nonIndexedString == NON_INDEXED_STRING
+            break
+
+        assert await async_w3.eth.unsubscribe(sub_id)
+        assert len(async_w3.subscription_manager.subscriptions) == 0
+        await clean_up_task(emit_event_task)
+
+    @pytest.mark.asyncio
+    async def test_async_eth_subscribe_creates_and_handles_logs_subscription_type(
+        self,
+        async_w3: AsyncWeb3,
+        async_emitter_contract: "AsyncContract",
+    ) -> None:
+        sub_manager = async_w3.subscription_manager
+
+        event = async_emitter_contract.events.LogIndexedAndNotIndexed
+        event_topic = async_w3.keccak(text=event.abi_element_identifier).to_0x_hex()
+
+        logs_handler_test = SubscriptionHandlerTest()
+        sub_id = await async_w3.eth.subscribe(
+            "logs",
+            {
+                "address": async_emitter_contract.address,
+                "topics": [HexStr(event_topic)],
+            },
+            handler=logs_handler,
+            handler_context={
+                "logs_handler_test": logs_handler_test,
+                "event": event,
+            },
+        )
+        assert is_hexstr(sub_id)
+
+        assert len(sub_manager.subscriptions) == 1
+        sub = sub_manager.subscriptions[0]
+        assert isinstance(sub, LogsSubscription)
+
+        accts = await async_w3.eth.accounts
+        acct = accts[0]
+        emit_event_task = await log_indexed_and_non_indexed_args_task(
+            async_w3, async_emitter_contract, acct
+        )
+
+        await sub_manager.handle_subscriptions()
+
+        assert logs_handler_test.passed
+        assert len(sub_manager.subscriptions) == 0
+
+        assert sub_manager.total_handler_calls == 1
+        assert sub.handler_call_count == 1
+
+        # cleanup
+        sub_manager.total_handler_calls = 0
+        await clean_up_task(emit_event_task)
+
     @pytest.mark.asyncio
     async def test_async_extradata_poa_middleware_on_eth_subscription(
         self,
-        async_w3: "AsyncWeb3",
+        async_w3: AsyncWeb3,
     ) -> None:
         async_w3.middleware_onion.inject(
             ExtraDataToPOAMiddleware, "poa_middleware", layer=0
@@ -361,12 +621,13 @@ class PersistentConnectionProviderTest:
             break
 
         # clean up
+        assert await async_w3.eth.unsubscribe(sub_id)
         async_w3.middleware_onion.remove("poa_middleware")
 
     @pytest.mark.asyncio
     async def test_asyncio_gather_for_multiple_requests_matches_the_responses(
         self,
-        async_w3: "AsyncWeb3",
+        async_w3: AsyncWeb3,
     ) -> None:
         (
             latest,
@@ -402,7 +663,10 @@ class PersistentConnectionProviderTest:
         assert isinstance(chain_id3, int)
 
     @pytest.mark.asyncio
-    async def test_public_socket_api(self, async_w3: "AsyncWeb3") -> None:
+    async def test_async_public_socket_api(self, async_w3: AsyncWeb3) -> None:
+        # clear all caches and queues
+        async_w3.provider._request_processor.clear_caches()
+
         # send a request over the socket
         await async_w3.socket.send(
             RPCEndpoint("eth_getBlockByNumber"), ["latest", True]
@@ -425,3 +689,191 @@ class PersistentConnectionProviderTest:
         assert "result" in response, "Expected 'result' key in response."
         assert all(k in response["result"].keys() for k in SOME_BLOCK_KEYS)
         assert not isinstance(response["result"]["number"], int)  # assert not processed
+
+    @pytest.mark.asyncio
+    async def test_async_subscription_manager_subscribes_to_many_subscriptions(
+        self, async_w3: AsyncWeb3, async_emitter_contract: "AsyncContract"
+    ) -> None:
+        sub_manager = async_w3.subscription_manager
+
+        event = async_emitter_contract.events.LogIndexedAndNotIndexed
+        event_topic = async_w3.keccak(text=event.abi_element_identifier).to_0x_hex()
+
+        new_heads_handler_test = SubscriptionHandlerTest()
+        pending_tx_handler_test = SubscriptionHandlerTest()
+        logs_handler_test = SubscriptionHandlerTest()
+
+        await sub_manager.subscribe(
+            [
+                NewHeadsSubscription(
+                    handler=new_heads_handler,
+                    handler_context={"new_heads_handler_test": new_heads_handler_test},
+                ),
+                PendingTxSubscription(
+                    full_transactions=True,
+                    handler=pending_tx_handler,
+                    handler_context={
+                        "pending_tx_handler_test": pending_tx_handler_test
+                    },
+                ),
+                LogsSubscription(
+                    address=async_emitter_contract.address,
+                    topics=[HexStr(event_topic)],
+                    handler=logs_handler,
+                    handler_context={
+                        "logs_handler_test": logs_handler_test,
+                        "event": event,
+                    },
+                ),
+            ]
+        )
+
+        # emit contract event for `logs` subscription
+        accts = await async_w3.eth.accounts
+        acct = accts[0]
+        emit_event_task = await log_indexed_and_non_indexed_args_task(
+            async_w3, async_emitter_contract, acct
+        )
+
+        # get subscriptions before they are unsubscribed and removed
+        subs = sub_manager.subscriptions
+
+        await sub_manager.handle_subscriptions()
+
+        # assert unsubscribed and removed subscriptions
+        assert len(sub_manager.subscriptions) == 0
+
+        assert sub_manager.total_handler_calls == 3
+        assert all(sub.handler_call_count == 1 for sub in subs)
+
+        assert new_heads_handler_test.passed
+        assert pending_tx_handler_test.passed
+        assert logs_handler_test.passed
+
+        # cleanup
+        sub_manager.total_handler_calls = 0
+        await clean_up_task(emit_event_task)
+
+    @pytest.mark.asyncio
+    async def test_subscription_handler_context(self, async_w3: AsyncWeb3) -> None:
+        base_url = "http://localhost:1337"
+        async_beacon = AsyncBeacon(base_url)
+        handler_test = SubscriptionHandlerTest()
+
+        async def test_sub_handler(
+            handler_context: NewHeadsSubscriptionContext,
+        ) -> None:
+            beacon = handler_context.beacon
+            assert isinstance(beacon, AsyncBeacon)
+            assert beacon.base_url == base_url
+            assert handler_context.int1 == 1337
+            assert handler_context.str1 == "foo"
+            assert handler_context.int2 == 1999
+            assert handler_context.str2 == "bar"
+
+            handler_context.handler_test.passed = True
+            unsubscribed = await handler_context.subscription.unsubscribe()
+            assert unsubscribed
+
+        subscribed = await async_w3.eth.subscribe(
+            "newHeads",
+            label="foo",
+            handler=test_sub_handler,
+            handler_context={
+                "beacon": async_beacon,
+                "int1": 1337,
+                "str1": "foo",
+                "int2": 1999,
+                "str2": "bar",
+                "handler_test": handler_test,
+            },
+        )
+        assert is_hexstr(subscribed)
+
+        sub_manager = async_w3.subscription_manager
+
+        await sub_manager.handle_subscriptions()
+
+        assert len(sub_manager.subscriptions) == 0
+        assert sub_manager.total_handler_calls == 1
+        assert handler_test.passed
+
+        # cleanup
+        sub_manager.total_handler_calls = 0
+
+    @pytest.mark.asyncio
+    async def test_subscriptions_with_handler_and_without(
+        self, async_w3: AsyncWeb3
+    ) -> None:
+        handler_test = SubscriptionHandlerTest()
+        stream_passed = False
+
+        async def test_sub_handler(
+            handler_context: NewHeadsSubscriptionContext,
+        ) -> None:
+            handler_context.handler_test.passed = True
+            await handler_context.subscription.unsubscribe()
+
+        async def handle_subscription_stream() -> None:
+            nonlocal stream_passed
+            async for msg in async_w3.socket.process_subscriptions():
+                response = cast(FormattedEthSubscriptionResponse, msg)
+                assert sub_manager.get_by_id(response["subscription"]) is not None
+                assert response["result"] is not None
+                # wait for the handler to unsubscribe:
+                stream_passed = True
+                await async_w3.eth.unsubscribe(response["subscription"])
+                break
+
+        await async_w3.eth.subscribe(
+            "newHeads",
+            handler=test_sub_handler,
+            label="managed",
+            handler_context={"handler_test": handler_test},
+        )
+        await async_w3.eth.subscribe("newHeads", label="streamed")
+
+        sub_manager = async_w3.subscription_manager
+        assert len(sub_manager.subscriptions) == 2
+
+        await asyncio.gather(
+            sub_manager.handle_subscriptions(),
+            handle_subscription_stream(),
+        )
+
+        assert len(sub_manager.subscriptions) == 0
+        assert sub_manager.total_handler_calls == 1
+        assert handler_test.passed
+        assert stream_passed
+
+        # cleanup
+        sub_manager.total_handler_calls = 0
+
+    @pytest.mark.asyncio
+    async def test_handle_subscriptions_breaks_on_unsubscribe(
+        self,
+        async_w3: AsyncWeb3,
+    ) -> None:
+        async def unsubscribe_subs(
+            subs: List[Union[NewHeadsSubscription, LogsSubscription]]
+        ) -> None:
+            for sub in subs:
+                await sub.unsubscribe()
+
+        sub_manager = async_w3.subscription_manager
+        sub1 = NewHeadsSubscription(label="foo", handler=idle_handler)
+        sub2 = LogsSubscription(label="bar", handler=idle_handler)
+
+        await sub_manager.subscribe([sub1, sub2])
+
+        assert sub_manager.subscriptions == [sub1, sub2]
+
+        unsubscribe_task = asyncio.create_task(unsubscribe_subs([sub1, sub2]))
+        # With no subscriptions in the queue, ``handle_subscriptions`` should hang
+        # indefinitely. Test that when the last subscription is unsubscribed from,
+        # the method breaks out of the loop. This is done via a raised
+        # ``SubscriptionProcessingFinished`` within the ``TaskReliantQueue``.
+        await sub_manager.handle_subscriptions()
+
+        assert_no_subscriptions_left(sub_manager._subscription_container)
+        await clean_up_task(unsubscribe_task)
diff --git a/web3/_utils/module_testing/utils.py b/web3/_utils/module_testing/utils.py
index 4bce464e..c6c59874 100644
--- a/web3/_utils/module_testing/utils.py
+++ b/web3/_utils/module_testing/utils.py
@@ -10,10 +10,14 @@ from typing import (
     cast,
 )
 
-from toolz import (
+from eth_utils.toolz import (
     merge,
 )
 
+from web3.providers.persistent import (
+    PersistentConnectionProvider,
+)
+
 if TYPE_CHECKING:
     from web3 import (  # noqa: F401
         AsyncWeb3,
@@ -26,6 +30,7 @@ if TYPE_CHECKING:
         AsyncMakeRequestFn,
         MakeRequestFn,
         RPCEndpoint,
+        RPCRequest,
         RPCResponse,
     )
 
@@ -101,9 +106,21 @@ class RequestMocker:
         self.mock_results = mock_results or {}
         self.mock_errors = mock_errors or {}
         self.mock_responses = mock_responses or {}
-        self._make_request: Union[
-            "AsyncMakeRequestFn", "MakeRequestFn"
-        ] = w3.provider.make_request
+        if isinstance(w3.provider, PersistentConnectionProvider):
+            self._send_request = w3.provider.send_request
+            self._recv_for_request = w3.provider.recv_for_request
+        else:
+            self._make_request: Union[
+                "AsyncMakeRequestFn", "MakeRequestFn"
+            ] = w3.provider.make_request
+
+    def _build_request_id(self) -> int:
+        request_id = (
+            next(copy.deepcopy(self.w3.provider.request_counter))
+            if hasattr(self.w3.provider, "request_counter")
+            else 1
+        )
+        return request_id
 
     def __enter__(self) -> "Self":
         # mypy error: Cannot assign to a method
@@ -131,11 +148,7 @@ class RequestMocker:
         ):
             return self._make_request(method, params)
 
-        request_id = (
-            next(copy.deepcopy(self.w3.provider.request_counter))
-            if hasattr(self.w3.provider, "request_counter")
-            else 1
-        )
+        request_id = self._build_request_id()
         response_dict = {"jsonrpc": "2.0", "id": request_id}
 
         if method in self.mock_responses:
@@ -176,35 +189,34 @@ class RequestMocker:
     # -- async -- #
 
     async def __aenter__(self) -> "Self":
-        # mypy error: Cannot assign to a method
-        self.w3.provider.make_request = self._async_mock_request_handler  # type: ignore[method-assign]  # noqa: E501
-        # reset request func cache to re-build request_func with mocked make_request
-        self.w3.provider._request_func_cache = (None, None)
+        if not isinstance(self.w3.provider, PersistentConnectionProvider):
+            # mypy error: Cannot assign to a method
+            self.w3.provider.make_request = self._async_mock_request_handler  # type: ignore[method-assign]  # noqa: E501
+            # reset request func cache to re-build request_func w/ mocked make_request
+            self.w3.provider._request_func_cache = (None, None)
+        else:
+            self.w3.provider.send_request = self._async_mock_send_handler  # type: ignore[method-assign]  # noqa: E501
+            self.w3.provider.recv_for_request = self._async_mock_recv_handler  # type: ignore[method-assign]  # noqa: E501
+            self.w3.provider._send_func_cache = (None, None)
+            self.w3.provider._recv_func_cache = (None, None)
         return self
 
     async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
-        # mypy error: Cannot assign to a method
-        self.w3.provider.make_request = self._make_request  # type: ignore[assignment]
-        # reset request func cache to re-build request_func with original make_request
-        self.w3.provider._request_func_cache = (None, None)
+        if not isinstance(self.w3.provider, PersistentConnectionProvider):
+            # mypy error: Cannot assign to a method
+            self.w3.provider.make_request = self._make_request  # type: ignore[assignment]  # noqa: E501
+            # reset request func cache to re-build request_func w/ original make_request
+            self.w3.provider._request_func_cache = (None, None)
+        else:
+            self.w3.provider.send_request = self._send_request  # type: ignore[method-assign]  # noqa: E501
+            self.w3.provider.recv_for_request = self._recv_for_request  # type: ignore[method-assign]  # noqa: E501
+            self.w3.provider._send_func_cache = (None, None)
+            self.w3.provider._recv_func_cache = (None, None)
 
-    async def _async_mock_request_handler(
-        self, method: "RPCEndpoint", params: Any
+    async def _async_build_mock_result(
+        self, method: "RPCEndpoint", params: Any, request_id: int = None
     ) -> "RPCResponse":
-        self.w3 = cast("AsyncWeb3", self.w3)
-        self._make_request = cast("AsyncMakeRequestFn", self._make_request)
-
-        if all(
-            method not in mock_dict
-            for mock_dict in (self.mock_errors, self.mock_results, self.mock_responses)
-        ):
-            return await self._make_request(method, params)
-
-        request_id = (
-            next(copy.deepcopy(self.w3.provider.request_counter))
-            if hasattr(self.w3.provider, "request_counter")
-            else 1
-        )
+        request_id = request_id if request_id else self._build_request_id()
         response_dict = {"jsonrpc": "2.0", "id": request_id}
 
         if method in self.mock_responses:
@@ -244,6 +256,19 @@ class RequestMocker:
         else:
             raise Exception("Invariant: unreachable code path")
 
+        return mocked_result
+
+    async def _async_mock_request_handler(
+        self, method: "RPCEndpoint", params: Any
+    ) -> "RPCResponse":
+        self.w3 = cast("AsyncWeb3", self.w3)
+        self._make_request = cast("AsyncMakeRequestFn", self._make_request)
+        if all(
+            method not in mock_dict
+            for mock_dict in (self.mock_errors, self.mock_results, self.mock_responses)
+        ):
+            return await self._make_request(method, params)
+        mocked_result = await self._async_build_mock_result(method, params)
         decorator = getattr(self._make_request, "_decorator", None)
         if decorator is not None:
             # If the original make_request was decorated, we need to re-apply
@@ -259,6 +284,47 @@ class RequestMocker:
         else:
             return mocked_result
 
+    async def _async_mock_send_handler(
+        self, method: "RPCEndpoint", params: Any
+    ) -> "RPCRequest":
+        if all(
+            method not in mock_dict
+            for mock_dict in (self.mock_errors, self.mock_results, self.mock_responses)
+        ):
+            return await self._send_request(method, params)
+        else:
+            request_id = self._build_request_id()
+            return {"id": request_id, "method": method, "params": params}
+
+    async def _async_mock_recv_handler(
+        self, rpc_request: "RPCRequest"
+    ) -> "RPCResponse":
+        self.w3 = cast("AsyncWeb3", self.w3)
+        method = rpc_request["method"]
+        request_id = rpc_request["id"]
+        if all(
+            method not in mock_dict
+            for mock_dict in (self.mock_errors, self.mock_results, self.mock_responses)
+        ):
+            return await self._recv_for_request(request_id)
+        mocked_result = await self._async_build_mock_result(
+            method, rpc_request["params"], request_id=int(request_id)
+        )
+        decorator = getattr(self._recv_for_request, "_decorator", None)
+        if decorator is not None:
+            # If the original recv_for_request was decorated, we need to re-apply
+            # the decorator to the mocked recv_for_request. This is necessary for
+            # the request caching decorator to work properly.
+
+            async def _coro(
+                _provider: Any, _rpc_request: "RPCRequest"
+            ) -> "RPCResponse":
+                return mocked_result
+
+            return await decorator(_coro)(self.w3.provider, rpc_request)
+        else:
+            return mocked_result
+
     @staticmethod
     def _create_error_object(error: Dict[str, Any]) -> Dict[str, Any]:
         code = error.get("code", -32000)