Refactor shared logic for persistent cxn providers:

- Refactor shared logic for persistent connection providers back into the
  `PersistentConnectionProvider` base class.
- Refactor reading from and sending to the socket into their own methods.
  Add these as abstract methods on the base class and force subclasses to
  implement them.

Important:

- Refactor the public `socket` API (`PersistentConnection` class) to properly
  define a `send()`, `recv()`, and `make_request()` which do what they say
  they do. Previously, `send()` was actually a "send_and_recv()" method. This
  was confusing and not a good API design. Note, these methods don't and cannot
  process the response in the way that invoking a method via its module within
  the web3 class does.

Author: fselmo

newsfragments/3433.breaking.rst

@@ -0,0 +1 @@
+Refactor the public ``socket`` api for persistent connection providers to properly define ``send()``, ``recv()``, and ``make_request()`` (send and wait for response) methods for interacting with the open socket.

newsfragments/3433.feature.rst

@@ -0,0 +1 @@
+Add ``popitem()`` functionality to the ``SimpleCache`` class as well as an async utility method to wait for the next item, ``async_await_and_popitem()``.

newsfragments/3433.internal.rst

@@ -0,0 +1 @@
+Refactor some common logic for persistent connection providers back into the base ``PersistentConnectionProvider`` class to reduce code duplication and improve maintainability.

web3/_utils/module_testing/persistent_connection_provider.py

@@ -23,6 +23,7 @@
 )
 from web3.types import (
     FormattedEthSubscriptionResponse,
+    RPCEndpoint,
 )
 
 if TYPE_CHECKING:
@@ -31,6 +32,22 @@
     )
 
 
+SOME_BLOCK_KEYS = [
+    "number",
+    "hash",
+    "parentHash",
+    "transactionsRoot",
+    "stateRoot",
+    "receiptsRoot",
+    "size",
+    "gasLimit",
+    "gasUsed",
+    "timestamp",
+    "transactions",
+    "baseFeePerGas",
+]
+
+
 class PersistentConnectionProviderTest:
     @pytest.mark.asyncio
     @pytest.mark.parametrize(
@@ -346,11 +363,40 @@ async def test_async_extradata_poa_middleware_on_eth_subscription(
         # clean up
         async_w3.middleware_onion.remove("poa_middleware")
 
+    @pytest.mark.asyncio
+    async def test_public_socket_api(self, async_w3: "AsyncWeb3") -> None:
+        async_w3.provider._request_processor.clear_caches()
+
+        # send a request over the socket
+        await async_w3.socket.send(
+            RPCEndpoint("eth_getBlockByNumber"), ["latest", True]
+        )
+
+        # recv and validate the unprocessed response
+        response = await async_w3.socket.recv()
+        assert "id" in response, "Expected 'id' key in response."
+        assert "jsonrpc" in response, "Expected 'jsonrpc' key in response."
+        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
+
+        # make a request over the socket
+        response = await async_w3.socket.make_request(
+            RPCEndpoint("eth_getBlockByNumber"), ["latest", True]
+        )
+        assert "id" in response, "Expected 'id' key in response."
+        assert "jsonrpc" in response, "Expected 'jsonrpc' key in response."
+        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_asyncio_gather_for_multiple_requests_matches_the_responses(
         self,
         async_w3: "AsyncWeb3",
     ) -> None:
+        async_w3.provider._request_processor.clear_caches()
+
         (
             latest,
             chain_id,
@@ -372,22 +418,8 @@ async def test_asyncio_gather_for_multiple_requests_matches_the_responses(
         assert isinstance(pending, AttributeDict)
 
         # assert block values
-        some_block_keys = [
-            "number",
-            "hash",
-            "parentHash",
-            "transactionsRoot",
-            "stateRoot",
-            "receiptsRoot",
-            "size",
-            "gasLimit",
-            "gasUsed",
-            "timestamp",
-            "transactions",
-            "baseFeePerGas",
-        ]
-        assert all(k in latest.keys() for k in some_block_keys)
-        assert all(k in pending.keys() for k in some_block_keys)
+        assert all(k in latest.keys() for k in SOME_BLOCK_KEYS)
+        assert all(k in pending.keys() for k in SOME_BLOCK_KEYS)
 
         assert isinstance(block_num, int)
         assert latest["number"] == block_num

web3/manager.py

@@ -469,22 +469,50 @@ def _format_batched_response(
 
     # -- persistent connection -- #
 
-    async def send(self, method: RPCEndpoint, params: Any) -> RPCResponse:
+    async def socket_request(self, method: RPCEndpoint, params: Any) -> 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: "
+            "Making request to open socket connection and waiting for response: "
             f"{provider.get_endpoint_uri_or_ipc_path()}, method: {method}"
         )
         response = await request_func(method, params)
         return await self._process_response(response)
 
+    async def send(self, method: RPCEndpoint, params: Any) -> None:
+        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)
+
+        self.logger.debug(
+            "Sending request to open socket connection: "
+            f"{provider.get_endpoint_uri_or_ipc_path()}, method: {method}"
+        )
+        await provider.socket_send(provider.encode_rpc_request(method, params))
+
+    async def recv(self) -> RPCResponse:
+        provider = cast(PersistentConnectionProvider, self._provider)
+        self.logger.debug(
+            "Getting next response from open socket connection: "
+            f"{provider.get_endpoint_uri_or_ipc_path()}"
+        )
+        # pop from the queue since the listener task is responsible for reading
+        # directly from the socket
+        request_response_cache = self._request_processor._request_response_cache
+        _key, response = await request_response_cache.async_await_and_popitem(
+            last=False,
+            timeout=provider.request_timeout,
+        )
+        return await self._process_response(response)
+
     def _persistent_message_stream(self) -> "_AsyncPersistentMessageStream":
         return _AsyncPersistentMessageStream(self)
 
-    async def _get_next_message(self) -> Optional[RPCResponse]:
+    async def _get_next_message(self) -> RPCResponse:
         return await self._message_stream().__anext__()
 
     async def _message_stream(self) -> AsyncGenerator[RPCResponse, None]:

web3/module.py

@@ -138,7 +138,7 @@ async def caller(*args: Any, **kwargs: Any) -> Union[RPCResponse, AsyncLogFilter
 
             try:
                 method_str = cast(RPCEndpoint, method_str)
-                return await async_w3.manager.send(method_str, params)
+                return await async_w3.manager.socket_request(method_str, params)
             except Exception as e:
                 if (
                     cache_key is not None

web3/providers/persistent/async_ipc.py

@@ -11,32 +11,22 @@
 import sys
 from typing import (
     Any,
-    List,
     Optional,
     Tuple,
     Union,
-    cast,
 )
 
 from eth_utils import (
     to_text,
 )
 
 from web3.types import (
-    RPCEndpoint,
     RPCResponse,
 )
 
 from . import (
     PersistentConnectionProvider,
 )
-from ..._utils.batching import (
-    BATCH_REQUEST_ID,
-    sort_batch_response_by_response_ids,
-)
-from ..._utils.caching import (
-    async_handle_request_caching,
-)
 from ...exceptions import (
     ProviderConnectionError,
     Web3TypeError,
@@ -91,12 +81,7 @@ async def is_connected(self, show_traceback: bool = False) -> bool:
             return False
 
         try:
-            request_data = self.encode_rpc_request(
-                RPCEndpoint("web3_clientVersions"), []
-            )
-            self._writer.write(request_data)
-            current_request_id = json.loads(request_data)["id"]
-            await self._get_response_for_request_id(current_request_id, timeout=2)
+            await self.make_request("web3_clientVersion", [])
             return True
         except (OSError, ProviderConnectionError) as e:
             if show_traceback:
@@ -105,55 +90,33 @@ async def is_connected(self, show_traceback: bool = False) -> bool:
                 )
             return False
 
-    async def _provider_specific_connect(self) -> None:
-        self._reader, self._writer = await async_get_ipc_socket(self.ipc_path)
-
-    async def _provider_specific_disconnect(self) -> None:
-        if self._writer and not self._writer.is_closing():
-            self._writer.close()
-            await self._writer.wait_closed()
-            self._writer = None
-        if self._reader:
-            self._reader = None
-
-    async def _reset_socket(self) -> None:
-        self._writer.close()
-        await self._writer.wait_closed()
-        self._reader, self._writer = await async_get_ipc_socket(self.ipc_path)
-
-    @async_handle_request_caching
-    async def make_request(self, method: RPCEndpoint, params: Any) -> RPCResponse:
+    async def socket_send(self, request_data: bytes) -> None:
         if self._writer is None:
             raise ProviderConnectionError(
                 "Connection to ipc socket has not been initiated for the provider."
             )
 
-        request_data = self.encode_rpc_request(method, params)
-        try:
-            self._writer.write(request_data)
-            await self._writer.drain()
-        except OSError as e:
-            # Broken pipe
-            if e.errno == errno.EPIPE:
-                # one extra attempt, then give up
-                await self._reset_socket()
-                self._writer.write(request_data)
-                await self._writer.drain()
+        return await asyncio.wait_for(
+            self._socket_send(request_data), timeout=self.request_timeout
+        )
 
-        current_request_id = json.loads(request_data)["id"]
-        response = await self._get_response_for_request_id(current_request_id)
+    async def socket_recv(self) -> RPCResponse:
+        while True:
+            # yield to the event loop to allow other tasks to run
+            await asyncio.sleep(0)
 
-        return response
+            try:
+                response, pos = self._decoder.raw_decode(self._raw_message)
+                self._raw_message = self._raw_message[pos:].lstrip()
+                return response
+            except JSONDecodeError:
+                # read more data from the socket if the current raw message is
+                # incomplete
+                self._raw_message += to_text(await self._reader.read(4096)).lstrip()
 
-    async def make_batch_request(
-        self, requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
-        if self._writer is None:
-            raise ProviderConnectionError(
-                "Connection to ipc socket has not been initiated for the provider."
-            )
+    # -- private methods -- #
 
-        request_data = self.encode_batch_rpc_request(requests)
+    async def _socket_send(self, request_data: bytes) -> None:
         try:
             self._writer.write(request_data)
             await self._writer.drain()
@@ -165,32 +128,24 @@ async def make_batch_request(
                 self._writer.write(request_data)
                 await self._writer.drain()
 
-        response = cast(
-            List[RPCResponse], await self._get_response_for_request_id(BATCH_REQUEST_ID)
-        )
-        return response
-
-    async def _provider_specific_message_listener(self) -> None:
-        self._raw_message += to_text(await self._reader.read(4096)).lstrip()
+    async def _reset_socket(self) -> None:
+        self._writer.close()
+        await self._writer.wait_closed()
+        self._reader, self._writer = await async_get_ipc_socket(self.ipc_path)
 
-        while self._raw_message:
-            try:
-                response, pos = self._decoder.raw_decode(self._raw_message)
-            except JSONDecodeError:
-                break
+    async def _provider_specific_connect(self) -> None:
+        self._reader, self._writer = await async_get_ipc_socket(self.ipc_path)
 
-            if isinstance(response, list):
-                response = sort_batch_response_by_response_ids(response)
+    async def _provider_specific_disconnect(self) -> None:
+        if self._writer and not self._writer.is_closing():
+            self._writer.close()
+            await self._writer.wait_closed()
+            self._writer = None
+        if self._reader:
+            self._reader = None
 
-            is_subscription = (
-                response.get("method") == "eth_subscription"
-                if not isinstance(response, list)
-                else False
-            )
-            await self._request_processor.cache_raw_response(
-                response, subscription=is_subscription
-            )
-            self._raw_message = self._raw_message[pos:].lstrip()
+    async def _provider_specific_socket_reader(self) -> RPCResponse:
+        return await self.socket_recv()
 
     def _error_log_listener_task_exception(self, e: Exception) -> None:
         super()._error_log_listener_task_exception(e)