eth: request id dispatcher and direct req/reply APIs

[karalabe]

With all non-request-id eth protocol version dropped (<= eth/65), we can finally properly match up inbound replies to pending requests (e.g. inbound ReceiptsMsg to pending GetReceiptsMsg). This is a very good thing, as is greatly simplifies internal code... after a huge rewrite, this PR.

Background

A huge complexity in our network packet handling code is that there can be concurrent requests of the same type sent to the same peer. These happen because there are multiple subsystems that run concurrently, and which may want to fetch something:

• The downloader requests headers, bodies, receipts and states.
• The fetcher requests headers and bodies.
• The eth package does header challenges.

To make matters worse, if a peer disconnects and quickly reconnects (Parity liked to do this), it can happen that there are still some reply handling propagating through the system from the old connection, whilst the new connection is hit with requests and starts delivering replies too.

The net effect is that when a reply packet arrives (header, body, receipt, state delivery), there can be many requests waiting for it all over the place. Since there's no way in <= eth/65 to match the arriving packets up with the requests, our code had a filtering mechanism implemented: a reply packet is first given to the eth challenge handler; it it doesn't consume it, it is handed to the fetcher; if it is still not consumed, it is handed to the downloader.

This quickly becomes a horrible thing when we also allow partial deliveries or altogether empty replies (missing data); as subsystems will start to consume each others' data, leading to inconsistencies (timeouts and weird behavior) in the starved subsystem. Further complexity arises when peers send spam data; or when a reply reaches a subsystem where the peer already disconnected/reconnected and was issues a new request.

Geth does handle all this complexity currently. It's just insanely complicated and with The Merge coming and requiring heavy protocol changes; it's extremely brittle to keep hacking the existing code, hoping it won't blow up. The only sane way out, is to rework the entire reply consumption mechanism; or formally put, the reply dispatching.

Dispatcher

As detailed above, Geth's reply handling is insanely complicated, forced by the eth protocol not supporting matching replies to requests. eth/66, however, introduced a request ID in all request packets that get echoed back in the replies to those packets.

Using these ping-ponged IDs, we could ensure that subsystems don't consume each other's messages; that unrequested data gets correctly dropped; or that deliveries across reconnects don't influence each other. Still, keeping the old architecture of each subsystem filtering through all the inbound messages - just because in the legacy protocols they had to - becomes useless complexity that still hinders work on The Merge.

Instead of having an architecture where network replies enter the system through the protocol handlers and then bubble through everything to reach their final destination, this PR introduces a request/reply dispatcher that delivers a reply message directly to the point of request. This simplifies packet handling a lot:

• If a subsystem does a single request to a peer and wants to wait for the response (e.g. the eth chain challenges; or the downloader ancestor lookup); it can simply block - inline in the code, like a function call waiting for the return - until the reply arrives or the request times out. No more waiting on channels that might deliver junk.
• If a subsystem does concurrent requests to many peers (e.g. the downloader receipt chain retrieval), it can multiplex the replies into a single sink channel, but still have the guarantee that only one reply arrives to a request and that every request either gets serviced or times out. This moves the error handling complexity out of the higher level subsystem.

This dispatcher is implemented within eth/protocols/eth and is part of the eth.Peer (i.e. each peer gets its own dispatcher to avoid starvation issues). The dispatcher in essence is a goroutine that receives eth.Request objects to send over the wire and tracks these until a response-type message arrives bundled up as an eth.Response. After a successful match, the response payload is delivered on the sink channel of the request.

The eth.Request objects aren't constructed publicly, they are created by eth.Peer.RequestXYZ, but they are returned to the caller since they allow cancelling a pending request if the caller deems so (e.g. timeout, disconnect, etc). It is essential to note that the dispatcher does not untrack requests on disconnect and does not untrack requests on timeouts. It doesn't even have a notion of untracking requests itself. This is important, as it creates a deterministic workflow:

• Caller makes network request; if successful, gets an eth.Request instance
• Caller waits until it doesn't want to any more:
  • A. Response arrived, request is considered done.
  • B. Caller doesn't want to wait (e.g. too slow vs other peers), request is considered cancelled.
  • C. Caller has to abort (e.g. download cycle failed or shutdown), request is considered cancelled.
• Caller notifies the dispatcher to untrack the request.

This workflow is very easy to reason about as response delivery has it's dedicated flow/direction (dispatcher to caller via sink channel) and cleanup handling has it's own dedicated flow/direction (caller to dispatcher via method calls (internally channels)). There's no dirtying of flows by sending lifecycle events on the data channel.

API

Although the dispatcher is part of an eth.Peer, it is not publicly callable. Since the eth.Request types need to be assembled - which is half protocol dependent, half dispatcher dependent - the requests are created and dispatched by the peer.RequestXYZ methods. Similarly, the eth.Responses are assembled and dispatched internally in the handlers.

However, since requests do need to be tracked and potentially cancelled, the eth.Request is public and returned via the peer.RequestXYZ methods. Similarly, it's simpler to also deliver eth.Response as a public type to the caller (via the request sink) since it allows bundling up a lot of useful metadata. The exact content was derived by what was necessary to support the fetchers and downloaders.

One downside of working with public eth.Request and eth.Response types is that the downloader and fetcher got locked in to the eth protocol, whereas previously the same code was used for les too. This is a curious issue, because on the surface it might be considered a downside - fixable with introducing interfaces (currently the old code was duplicated into les). But thinking about it mid-term, it's not really a good idea to complicate the codebase back again to support multiple protocols:

• The fetcher becomes useless after The Merge, since block propagation gets removed, so even the eth one becomes dead code.
• les becomes completely different after The Merge, since the chain head is available via the consensus node; and there's no reason to download entire past chain segments (currently we need it to sync).

As such, this PR assumes that both the duplicated fetcher and downloader in les will be altogether deleted after the merge. There's very little point in hacking the entire codebase into something abstract that has a half a year death sentence anyway.

Refactors

The above described dispatcher which delivers responses directly to the callsite has the potential to simplify everything, but it does entail a significant refactor since everything was built on the bubble-through approach (les even does some accounting on top 😱).

The PR does a full and clean refactor of the downloader to support the new model. Since fast sync is deprecated, that part of the code was deleted altogether (since refactoring it is painful), completely relying on snap sync from now on. Snap sync was not previously tested within the context of the downloader, so the PR also completely overhauls the downloader tests to use the real core.BlockChain code and eth.Handlers code as we can't meaningfully mock out the snapshot tree handlig. This is a huge change in the downloader, but it cleans everything up in preparation for the reverse syncer work for The Merge.

The PR did a murky-er refactor within the fetcher. Instead of rewriting the whole thing - at significant time cost - I've just did the bare minimum to hack in the new model. The thinking is that the fetcher is going away anyway after the merge, so don't waste time on fixing stuff on the death row.

The PR also cleanly integrated the dispatcher model into the eth header challenges.

[holiman]

With this change, I think we should make the request tracker more strict -- so we can rely on it for incoming messages. If we do that, we can do many checks earlier in the pipeline, e.g. checking block body before delivering upstream.

[holiman]

Sorry, you're 51 lines overdraft

[holiman]

Either the clause can be removed entirely or this should be uncommented again?

[karalabe]

There's no fast sync in les, so either way it doesn't matter. I think I just needed to remove references to the main downloader.SyncProcess types, which have no notion of fast sync any more. I could have rewritten les's downloader too, but figured that it's gonna happen anyway from scratch so better just keep the changes minimal and don't try to make actual "real" changes. I guess I left as a comment to keep te original code intact if we want to see what was there.

[MariusVanDerWijden]

Suggested change

	// fetcher, unpacking the body data and delivering it to the downloader's queue.
	// fetcher, unpacking the receipt data and delivering it to the downloader's queue.

[holiman]

I ran the downloader-tests with -race -- I suspect this test is a bit timing-sensitive?

--- PASS: TestMultiSynchronisation66Full (8.37s)
panic: failed to synchronise blocks: peer is stalling

goroutine 280419 [running]:
github.com/ethereum/go-ethereum/eth/downloader.testForkedSyncProgress.func3()
        /home/user/go/src/github.com/ethereum/go-ethereum/eth/downloader/downloader_test.go:1112 +0x17a
created by github.com/ethereum/go-ethereum/eth/downloader.testForkedSyncProgress
        /home/user/go/src/github.com/ethereum/go-ethereum/eth/downloader/downloader_test.go:1109 +0x7e5
FAIL    github.com/ethereum/go-ethereum/eth/downloader  246.671s

[karalabe]

I ran the downloader-tests with -race -- I suspect this test is a bit timing-sensitive?

--- PASS: TestMultiSynchronisation66Full (8.37s)
panic: failed to synchronise blocks: peer is stalling

goroutine 280419 [running]:
github.com/ethereum/go-ethereum/eth/downloader.testForkedSyncProgress.func3()
        /home/user/go/src/github.com/ethereum/go-ethereum/eth/downloader/downloader_test.go:1112 +0x17a
created by github.com/ethereum/go-ethereum/eth/downloader.testForkedSyncProgress
        /home/user/go/src/github.com/ethereum/go-ethereum/eth/downloader/downloader_test.go:1109 +0x7e5
FAIL    github.com/ethereum/go-ethereum/eth/downloader  246.671s

Please see #23576 (comment)

If you enable the race detector, everything gets significantly slower, so yeah, kind of to be expected. The "solution" could be to abstract out the notion of time from the downloader and use a fake clock to drive it. We have 2-3 packages doing it, it's just a large rework and you'd need to mock out time in every submodule that the downloader depends on.

[MariusVanDerWijden]

Here's still a todo

[karalabe]

Yes, I'll delete this after a few benchmark syncs. Just want to make sue these don't end up in live code.

[MariusVanDerWijden]

Tried my best to review it, I didn't properly check the blockchain_test.go and I could not super meaningfully review fetchers_concurrent.go The rest looks good to me. We should create some fuzzers for Snap packages btw

[MariusVanDerWijden]

It looks like you check the error beforehand, so you wouldn't send back anything in case of error

[MariusVanDerWijden]

I think you need to add then new methods here so they can be called by graphql

[holiman]

Might be that this change causes a lot of existing setups to refuse to start. Maybe we shoud deprecate fast and just internally translate it to snap for a while?

[holiman]

Suggested change

	// of picking on channel or the other. To avoid such cases delivering
	// of picking one channel or the other. To avoid such cases delivering

[holiman]

With this change here, afaict the comment is now misleading. We don't cancel it so it's not considered in-flight anymore, since we no longer call req.Close() on it. I suppose the caller keeps waiting for the reply, even though this internally has been moved to stale? And after a little while longer, the peer will be rejected, prompting the stale request to be Closed and rescheduled?

[holiman]

s/pendign/pending

[holiman]

What happens if the response to a stale request comes in -- we just drop it, right? I don't quite follow the logic. So we shuffle it over from pending to stale, but we still don't (immediately) reschedule it, but neither do we handle a late response?
I must have missed something, but it seems like moving something to stale unconditionally means it will time out a bit later, leading to the peer being rejected.

[holiman]

Ah, my bad. The pending here is dispatcher-local pending, it has no relation to the pending/stales in the downloader.

[holiman]

LGTM