feat: bodies stage

[onbjerg]

Almost complete bodies stage, a few hanging todos + tests left. Also includes some minor touchups to documentation when I stumbled upon it.

The stage uses a Downloader like the headers stage, but this downloader is implemented using a stream. An iterator of block hashes for blocks we want to download is handed over to the downloader and the downloader will fetch the blocks. The blocks are fetched somewhat concurrently but emitted by the stream in order - so, if we are fetching 10 blocks (1-10) and block 3 finishes downloading first, then block 2 and then block 1, the order of emitted values from the stream will still be 1, 2, 3. This simplifies some of the stage logic

There are a few types we should probably consider consolidating wrt block bodies, I've marked them with todos for now

Edit: I've also moved the proofs module from the consensus crate to the primitives crate since the proofs are used by multiple modules. Some of the verification functions in the consensus crate that only operate on blocks/transactions/headers should probably also just live directly on those types in the primitives module, but I'll leave that for later

[codecov]

Codecov Report

Merging #190 (e38e193) into main (9575eb8) will increase coverage by 0.29%.
The diff coverage is 75.07%.

@@            Coverage Diff             @@
##             main     #190      +/-   ##
==========================================
+ Coverage   69.80%   70.10%   +0.29%     
==========================================
  Files         200      209       +9     
  Lines       16538    17375     +837     
==========================================
+ Hits        11544    12180     +636     
- Misses       4994     5195     +201     

Impacted Files	Coverage Δ
crates/consensus/src/consensus.rs	0.00% <0.00%> (ø)
crates/consensus/src/lib.rs	100.00% <ø> (ø)
crates/db/src/kv/mod.rs	94.83% <ø> (ø)
crates/interfaces/src/consensus.rs	100.00% <ø> (ø)
crates/interfaces/src/db/codecs/scale.rs	100.00% <ø> (ø)
crates/interfaces/src/db/mod.rs	94.73% <ø> (ø)
crates/interfaces/src/db/tables.rs	0.00% <ø> (ø)
crates/interfaces/src/p2p/bodies/client.rs	0.00% <0.00%> (ø)
crates/interfaces/src/p2p/error.rs	0.00% <0.00%> (ø)
crates/interfaces/src/test_utils/bodies.rs	0.00% <0.00%> (ø)
... and 37 more

📣 We’re building smart automated test selection to slash your CI/CD build times. Learn more

[rkrasiuk]

you can use unwind_table_by_num here

[onbjerg]

I can't since tx is already immutably borrowed by block_body_cursor

[mattsse]

the lifetime won't be necessary if you clone and move the client, but only useful depending on how it's called, so perhaps borrowing is sufficient here.

[onbjerg]

I tried cloning and moving the client but it didn't work :S

Edit: See the error here, maybe I am missing something

error: captured variable cannot escape `FnMut` closure body
--> crates/net/bodies-downloaders/src/concurrent.rs:48:21
|
44 |           let client = Arc::clone(&self.client);
|               ------ variable defined here
45 |           Box::pin(
46 |               stream::iter(headers.into_iter().map(move |header_hash| {
|                                                                     - inferred to be a `FnMut` closure
47 |                   {
48 |                       client
|                       ^-----
|                       |
|  _____________________variable captured here
| |
49 | |                         .get_block_body(header_hash)
50 | |                         .map_ok(move |body| (header_hash, body))
51 | |                         .map_err(|err| match err {
...  |
55 | |                             err => DownloadError::Client { source: err },
56 | |                         })
| |__________________________^ returns a reference to a captured variable which escapes the closure body
|
= note: `FnMut` closures only have access to their captured variables while they are executing...
= note: ...therefore, they cannot allow references to captured variables to escape

[mattsse]

this is not the batch size for GetBodies, instead, this is the number of concurrently active requests.

There should be different settings for both.

the batch_size is size for the headers chunks to divide via Vec::chunks(batch_size), and for buffured we need a separate value that limits the number of request we send.

[onbjerg]

It's unclear to me what the actual difference is, my assumption is that if we request multiple bodies in one GetBodies call they are downloaded concurrently but there is no order guarantee. The reason I only request one per call is because the order needs to be the same as the request order, otherwise we may skip bodies in the database

[mattsse]

there's no guarantee, only blocks that were found are returned.

So perhaps, if the BodiesStream needs to return them in order it probably needs to keep track of the next header to return and buffer the rest

[onbjerg]

That's what this is doing, just without requesting multiple bodies per call. Is there any tangible benefit to requesting multiple per call?

[mattsse]

this we should test, can then tune

[mattsse]

while it is possible to send a request for single block.

the GetBlockBodies request takes a Vec<H256> to fetch multiple bodies with one request

[onbjerg]

Is there an order guarantee? If not, see my above comment

[mattsse]

This is okay for now I think but I think we need to roll a custom impl of a buffered Stream that handles retries

[onbjerg]

Any particular reason to roll our own? Adding retries can be done in this loop w/o issue I think, we just need a generalized retry abstraction for futures

[gakonst]

https://docs.rs/futures-retry/latest/futures_retry/ have used this succesfully in the past

• https://github.com/interledger-rs/interledger-rs/blob/e15ae42d5ce80e45ace8c7bebd5875372b89dcfb/crates/interledger-settlement/src/core/settlement_client.rs#L37-L40
• https://github.com/interledger-rs/interledger-rs/blob/e15ae42d5ce80e45ace8c7bebd5875372b89dcfb/crates/interledger-settlement/src/core/settlement_client.rs#L157-L196

[onbjerg]

Pulling it out of draft so we can get some of the other tweaks I have stacked in this PR merged. Will continue to iterate on it (adding tests for the downloader, addressing some of the todos I left in the code etc) if it gets merged

[mattsse]

lgtm

[mattsse]

this we should test, can then tune

[onbjerg]

Merging and building on top