feat(l1): eth capability GetBlockBodies message (#1048)

**Motivation** Adds support for the eth p2p message "GetBlockBodies". Closes #1044 --------- Co-authored-by: Esteban Dimitroff Hodi <[email protected]>
lambdaclass · Nov 5, 2024 · a3c84f2 · a3c84f2
1 parent 9379be4
commit a3c84f2
Show file tree

Hide file tree

Showing 5 changed files with 79 additions and 8 deletions.
diff --git a/.github/workflows/hive_and_assertoor.yaml b/.github/workflows/hive_and_assertoor.yaml
@@ -70,6 +70,9 @@ jobs:
           - simulation: eth
             name: "Devp2p eth tests"
             run_command: make run-hive SIMULATION=devp2p TEST_PATTERN="eth/getblockheaders"
+          - simulation: eth
+            name: "Devp2p eth tests"
+            run_command: make run-hive SIMULATION=devp2p TEST_PATTERN="eth/getblockheaders|getblockbodies"
           - simulation: engine
             name: "Engine tests"
             run_command:  make run-hive-on-latest SIMULATION=ethereum/engine TEST_PATTERN="/Blob Transactions On Block 1, Cancun Genesis|Blob Transactions On Block 1, Shanghai Genesis|Blob Transaction Ordering, Single Account, Single Blob|Blob Transaction Ordering, Single Account, Dual Blob|Blob Transaction Ordering, Multiple Accounts|Replace Blob Transactions|Parallel Blob Transactions|ForkchoiceUpdatedV3 Modifies Payload ID on Different Beacon Root|NewPayloadV3 After Cancun|NewPayloadV3 Versioned Hashes|ForkchoiceUpdated Version on Payload Request"

diff --git a/cmd/ethereum_rust/ethereum_rust.rs b/cmd/ethereum_rust/ethereum_rust.rs
@@ -20,7 +20,7 @@ use std::{
     net::{Ipv4Addr, SocketAddr, ToSocketAddrs},
 };
 use tokio_util::task::TaskTracker;
-use tracing::{info, warn};
+use tracing::{error, info, warn};
 use tracing_subscriber::filter::Directive;
 use tracing_subscriber::{EnvFilter, FmtSubscriber};
 mod cli;
@@ -137,6 +137,20 @@ async fn main() {
                     block.header.number, hash, error
                 );
             }
+            if store
+                .update_latest_block_number(block.header.number)
+                .is_err()
+            {
+                error!("Fatal: added block {} but could not update the block number -- aborting block import", block.header.number);
+                break;
+            };
+            if store
+                .set_canonical_block(block.header.number, hash)
+                .is_err()
+            {
+                error!("Fatal: added block {} but could not set it as canonical -- aborting block import", block.header.number);
+                break;
+            };
         }
         if let Some(last_block) = blocks.last() {
             let hash = last_block.hash();

diff --git a/crates/networking/p2p/rlpx/connection.rs b/crates/networking/p2p/rlpx/connection.rs
@@ -1,6 +1,9 @@
 use crate::{
     rlpx::{
-        eth::{backend, blocks::BlockHeaders},
+        eth::{
+            backend,
+            blocks::{BlockBodies, BlockHeaders},
+        },
         handshake::encode_ack_message,
         message::Message,
         p2p,
@@ -165,6 +168,13 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
                             };
                             self.send(Message::BlockHeaders(response)).await;
                         }
+                        Message::GetBlockBodies(msg_data) => {
+                            let response = BlockBodies {
+                                id: msg_data.id,
+                                block_bodies: msg_data.fetch_blocks(&self.storage),
+                            };
+                            self.send(Message::BlockBodies(response)).await;
+                        }
                         Message::GetStorageRanges(req) => {
                             let response =
                                 process_storage_ranges_request(req, self.storage.clone())?;

diff --git a/crates/networking/p2p/rlpx/eth/blocks.rs b/crates/networking/p2p/rlpx/eth/blocks.rs
@@ -207,14 +207,41 @@ impl RLPxMessage for BlockHeaders {
 pub(crate) struct GetBlockBodies {
     // id is a u64 chosen by the requesting peer, the responding peer must mirror the value for the response
     // https://github.com/ethereum/devp2p/blob/master/caps/eth.md#protocol-messages
-    id: u64,
-    block_hashes: Vec<BlockHash>,
+    pub id: u64,
+    pub block_hashes: Vec<BlockHash>,
 }
 
+// Limit taken from here:
+// https://github.com/ethereum/go-ethereum/blob/a1093d98eb3260f2abf340903c2d968b2b891c11/eth/protocols/eth/handler.go#L45
+pub const BLOCK_BODY_LIMIT: usize = 1024;
+
 impl GetBlockBodies {
     pub fn new(id: u64, block_hashes: Vec<BlockHash>) -> Self {
         Self { block_hashes, id }
     }
+    pub fn fetch_blocks(&self, storage: &Store) -> Vec<BlockBody> {
+        let mut block_bodies = vec![];
+        for block_hash in &self.block_hashes {
+            match storage.get_block_body_by_hash(*block_hash) {
+                Ok(Some(block)) => {
+                    block_bodies.push(block);
+                    if block_bodies.len() >= BLOCK_BODY_LIMIT {
+                        break;
+                    }
+                }
+                Ok(None) => {
+                    continue;
+                }
+                Err(err) => {
+                    tracing::error!(
+                        "Error accessing DB while building block bodies response for peer: {err}"
+                    );
+                    return vec![];
+                }
+            }
+        }
+        block_bodies
+    }
 }
 
 impl RLPxMessage for GetBlockBodies {
@@ -244,11 +271,12 @@ impl RLPxMessage for GetBlockBodies {
 }
 
 // https://github.com/ethereum/devp2p/blob/master/caps/eth.md#blockbodies-0x06
+#[derive(Debug)]
 pub(crate) struct BlockBodies {
     // id is a u64 chosen by the requesting peer, the responding peer must mirror the value for the response
     // https://github.com/ethereum/devp2p/blob/master/caps/eth.md#protocol-messages
-    id: u64,
-    block_bodies: Vec<BlockBody>,
+    pub id: u64,
+    pub block_bodies: Vec<BlockBody>,
 }
 
 impl BlockBodies {

diff --git a/crates/networking/p2p/rlpx/message.rs b/crates/networking/p2p/rlpx/message.rs
@@ -2,7 +2,7 @@ use bytes::BufMut;
 use ethereum_rust_rlp::error::{RLPDecodeError, RLPEncodeError};
 use std::fmt::Display;
 
-use super::eth::blocks::{BlockHeaders, GetBlockHeaders};
+use super::eth::blocks::{BlockBodies, BlockHeaders, GetBlockBodies, GetBlockHeaders};
 use super::eth::status::StatusMessage;
 use super::p2p::{DisconnectMessage, HelloMessage, PingMessage, PongMessage};
 use super::snap::{
@@ -26,6 +26,8 @@ pub(crate) enum Message {
     // https://github.com/ethereum/devp2p/blob/5713591d0366da78a913a811c7502d9ca91d29a8/caps/eth.md#getblockheaders-0x03
     GetBlockHeaders(GetBlockHeaders),
     BlockHeaders(BlockHeaders),
+    GetBlockBodies(GetBlockBodies),
+    BlockBodies(BlockBodies),
     // snap capability
     GetAccountRange(GetAccountRange),
     AccountRange(AccountRange),
@@ -55,6 +57,7 @@ impl Message {
             0x10 => Ok(Message::Status(StatusMessage::decode(msg_data)?)),
             0x13 => Ok(Message::GetBlockHeaders(GetBlockHeaders::decode(msg_data)?)),
             0x14 => Ok(Message::BlockHeaders(BlockHeaders::decode(msg_data)?)),
+            0x15 => Ok(Message::GetBlockBodies(GetBlockBodies::decode(msg_data)?)),
             0x21 => Ok(Message::GetAccountRange(GetAccountRange::decode(msg_data)?)),
             0x22 => Ok(Message::AccountRange(AccountRange::decode(msg_data)?)),
             0x23 => Ok(Message::GetStorageRanges(GetStorageRanges::decode(
@@ -74,11 +77,22 @@ impl Message {
             Message::Ping(msg) => msg.encode(buf),
             Message::Pong(msg) => msg.encode(buf),
             Message::Status(msg) => msg.encode(buf),
-            Message::GetBlockHeaders(msg) => msg.encode(buf),
+            Message::GetBlockHeaders(msg) => {
+                0x13_u8.encode(buf);
+                msg.encode(buf)
+            }
             Message::BlockHeaders(msg) => {
                 0x14_u8.encode(buf);
                 msg.encode(buf)
             }
+            Message::GetBlockBodies(msg) => {
+                0x15_u8.encode(buf);
+                msg.encode(buf)
+            }
+            Message::BlockBodies(msg) => {
+                0x16_u8.encode(buf);
+                msg.encode(buf)
+            }
             Message::GetAccountRange(msg) => {
                 0x21_u8.encode(buf);
                 msg.encode(buf)
@@ -117,6 +131,8 @@ impl Display for Message {
             Message::Status(_) => "eth:Status".fmt(f),
             Message::GetBlockHeaders(_) => "eth:getBlockHeaders".fmt(f),
             Message::BlockHeaders(_) => "eth:BlockHeaders".fmt(f),
+            Message::BlockBodies(_) => "eth:BlockBodies".fmt(f),
+            Message::GetBlockBodies(_) => "eth:GetBlockBodies".fmt(f),
             Message::GetAccountRange(_) => "snap:GetAccountRange".fmt(f),
             Message::AccountRange(_) => "snap:AccountRange".fmt(f),
             Message::GetStorageRanges(_) => "snap:GetStorageRanges".fmt(f),