Propagate listener task exceptions to the main loop.

fselmo commented 5 months ago

What was wrong?

Closes #3375 (along with #3387)

How was it fixed?

As I understand it, the way asyncio works is each task operates under its own context. Directly communicating with another context and trying to immediately propagate an exception to the main loop, for example, does not seem to be a very straightforward thing to do surprisingly. For better or worse, I think when we poll for messages we should check if the listener task is done and, if it is, if an exception was set. If an exception was set then we can raise it in the main loop where we are polling for messages.

When we poll for messages, poll to see if the background task is finished. If it is, handle any exceptions in the main loop. On subscription streams, since we hang awaiting a message to appear in the subscription queue, we need to signal when the background task is done since no messages will be coming in and we will hang indefinitely. Push a new exception TaskNotRunning to the queue. Revamp the queue to raise when this is pushed to it. Handle this exception in the subscription polling.
Refactor some of the logic. When disconnecting persistent connection providers, shut down the listener task before closing the connection to avoid any sort of errors when the task is still trying to read from the closing connection.

I added some nice logging that really shouldn't be too noisy unless the connection is reconnecting all the time (which I forced for the example here).

INFO:web3.providers.WebSocketProvider:Connecting to: wss://mainnet.infura.io/ws/v3/c364...
INFO:web3.providers.WebSocketProvider:Successfully connected to: wss://mainnet.infura.io/ws/v3/c364...
INFO:web3.providers.WebSocketProvider:WebSocketProvider listener background task started. Storing all messages in appropriate request processor queues / caches to be processed.
newHeads sub_id: 0xad65c6e2aefa417dafda6999d0bd774b
INFO:web3.providers.WebSocketProvider:Subscription response queue synced with websocket message stream.
blocknum: 19829151
blocknum: 19829152
ERROR:web3.providers.WebSocketProvider:Connection interrupted, attempting to reconnect...
INFO:web3.providers.WebSocketProvider:Message listener background task successfully shut down.
INFO:web3.providers.WebSocketProvider:Successfully disconnected from: wss://mainnet.infura.io/ws/v3/c364...
INFO:web3.providers.WebSocketProvider:Connecting to: wss://mainnet.infura.io/ws/v3/c364...
INFO:web3.providers.WebSocketProvider:Successfully connected to: wss://mainnet.infura.io/ws/v3/c364...
INFO:web3.providers.WebSocketProvider:WebSocketProvider listener background task started. Storing all messages in appropriate request processor queues / caches to be processed.
newHeads sub_id: 0x718852a668ed437fbdc86243db929405
blocknum: 19829153
blocknum: 19829154
ERROR:web3.providers.WebSocketProvider:Connection interrupted, attempting to reconnect...
INFO:web3.providers.WebSocketProvider:Message listener background task successfully shut down.
INFO:web3.providers.WebSocketProvider:Successfully disconnected from: wss://mainnet.infura.io/ws/v3/c364...
INFO:web3.providers.WebSocketProvider:Connecting to: wss://mainnet.infura.io/ws/v3/c364...
INFO:web3.providers.WebSocketProvider:Successfully connected to: wss://mainnet.infura.io/ws/v3/c364...
INFO:web3.providers.WebSocketProvider:WebSocketProvider listener background task started. Storing all messages in appropriate request processor queues / caches to be processed.
newHeads sub_id: 0xf46690d2853c4afba527ac71dd1915e0
blocknum: 19829155
...

Bonus:

I've been meaning to do this for some time and I think this is an appropriate time. Refactor the commonality of the listener task down to the base PersistentConnectionProvider class and define provider-specific methods that either need to be implemented or can be overridden to fine-tune logic specific to each persistent connection provider. This keeps things quite a bit DRYer.

Todo:

[x] Clean up commit history
[x] Add or update documentation related to these changes
[x] Add entry to the release notes

Cute Animal Picture

31205

fselmo commented 5 months ago

I'm actually going to re-structure this a bit. Will ping for review when ready again.

fselmo commented 5 months ago

Just kidding this is still in progress 😔

fselmo commented 4 months ago

@kclowes @pacrob @reedsa Hey it's me again. This is ready now 😄. We were missing a good test for subscription polling with the iterator pattern. This would've failed or hanged forever. That should be addressed now.

fselmo commented 4 months ago

Pushing None to the queue is the only thing that stands out as something that might be a little foot-gun-y.

I think you're absolutely right. I don't see None being pushed to the cache in another way but I think we can isolate this case to be more unique. If I can't, I think we can still use None to un-brick the awaiting of a message and we can make sure we make the proper checks of whether or not the task really did finish and if it didn't we just move on to the next iteration of the subscription messages - essentially causing a no-op in that scenario.

fselmo commented 4 months ago

@kclowes I came up with a more elegant solution for the queue to operate correctly as long as the listener task is running

In my head, it seems like it would be more descriptive and maybe less error prone if we could raise a custom exception there

taking this as inspiration, lmk what you think: a4790d6

...other comments addressed in 8ed6df9

kclowes commented 4 months ago

I came up with a more elegant solution for the queue to operate correctly as long as the listener task is running

super elegant 🤩 🚀

ethereum / web3.py