Websockets v2 update

[fselmo]

What was wrong?

The main reason WebsocketProviderV2 is in beta is that we have to asynchronously process a response from the original request that was made. This isn't so straightforward with the way websockets work because the socket will give back the next item it needs to, whatever that may be, when we call recv(). If a user has subscribed to an eth_subscribe subscription, say something like "newPendingTransactions" where it will return many transactions almost nearly instantly as soon as a user has subscribed to it, one (or many) of these subscription messages may return before the original request was made for which a user may be expecting a response. Consider the following:

async def wsV2_example():
    async with AsyncWeb3.persistent_websocket(
        WebsocketProviderV2(f"wss://mainnet.infura.io/ws/v3/{INFURA_PROJECT_ID}")
    ) as w3:
        # -- get pending transactions subscription -- #
        
        pending_full_txs = True
        pending_tx_sub_id = await w3.eth.subscribe("newPendingTransactions", pending_full_txs)
        print(f"pending_tx_sub_id: {pending_tx_sub_id}\n")

        # -- logs subscription -- #

        logs_sub_id = await w3.eth.subscribe("logs", {"address": ETHEREUM_WETH_CONTRACT})
        # we will likely get one of the "newPendingTransactions" subscriptions returned here 
        # before we get the response to our second `eth_subscribe` request with the `logs_sub_id` value.
        print(f"logs_sub_id: {logs_sub_id}\n")

        while True:
            async for response in w3.listen_to_websocket():
                # when we actually listen to the websocket for our subscriptions, we will eventually get the response
                # we needed to the second subscription request with our `logs_sub_id` value
                print(response)

asyncio.run(wsV2_example())

How was it fixed?

Thanks to JSON-RPC 2.0 specs, responses are required to return the id of the request that asked for the response. If we keep the request id and pass it to the method we use to retrieve and process a response, we can wait until this particular response is received and return that. It should take no longer than a call_timeout. While we look for the response, we need to store every response that came in before it in a transient cache that will then be emptied on each __anext__() call before we pull anything from the socket again.

• Created a RequestProcessor class that is strapped to PersistentConnectionProvider classes to process two types of caches now (both transient, in the sense that they only store information until it's needed and that time window shouldn't be too long):
  • The RequestInformation cache that already existed. This cache is responsible for storing a request's information so we can have that context and middleware methods to pipe the answer to the request through once we process it asynchronously.
  • A cache that stores raw responses from the open websocket connection that may come in before the response for the original request comes in. This cache gets emptied in a FIFO manner before we listen to any new messages from the websocket so that we don't lose any response that were received within that window.
• Refactored matching responses to request ids so that it exists within the make_request() method of the WebsocketProviderV2 class. This allows us to actually return a response that matches the request id within the same method, keeping types happy throughout the code base.
• Internalize the provider.ws websocket connection (as provider._ws) and add a public API class (WebsocketConnection) for interacting with the websocket on the _PersistentConnectionWeb3 class with more predictable behavior and formatting support, etc.
• Other refactoring along the way.

Todo:

• Add entry to the release notes
• Make provider.ws internal (provider._ws) and have any "listen" to ws method that is public look in the raw response cache first and process those responses in FIFO before actually calling ws.recv() again. This will be important to document since a user might want to directly call recv() on the websocket connection without knowing there may be cached responses ahead of any new responses.
• Missing a cute animal picture below 😱
• Manual testing (for now) of the new request processor transient caches and the logic therein

Cute Animal Picture

[reedsa]

Nothing blocking here, I haven't yet tested though. Looks great, love the separate commits and clean messages 💯

[pacrob]

lgtm!