feat(l1): backward cap compatibility

crates/common/rlp/decode.rs

@@ -513,7 +513,7 @@ pub fn decode_bytes(data: &[u8]) -> Result<(&[u8], &[u8]), RLPDecodeError> {
 /// Pads a slice of bytes with zeros on the left to make it a fixed size slice.
 /// The size of the data must be less than or equal to the size of the output array.
 #[inline]
-pub(crate) fn static_left_pad<const N: usize>(data: &[u8]) -> Result<[u8; N], RLPDecodeError> {
+pub fn static_left_pad<const N: usize>(data: &[u8]) -> Result<[u8; N], RLPDecodeError> {
     let mut result = [0; N];
 
     if data.is_empty() {

crates/common/rlp/error.rs

@@ -15,6 +15,8 @@ pub enum RLPDecodeError {
     UnexpectedString,
     #[error("InvalidCompression")]
     InvalidCompression(#[from] snap::Error),
+    #[error("IncompatibleProtocol")]
+    IncompatibleProtocol,
     #[error("{0}")]
     Custom(String),
 }

crates/networking/p2p/Cargo.toml

@@ -11,7 +11,7 @@ ethrex-blockchain.workspace = true
 ethrex-rlp.workspace = true
 ethrex-storage.workspace = true
 ethrex-trie.workspace = true
-
+ethereum-types.workspace = true
 tracing.workspace = true
 tokio.workspace = true
 tokio-util.workspace = true

crates/networking/p2p/peer_handler.rs

@@ -17,7 +17,7 @@ use crate::{
             blocks::{
                 BlockBodies, BlockHeaders, GetBlockBodies, GetBlockHeaders, BLOCK_HEADER_LIMIT,
             },
-            receipts::{GetReceipts, Receipts},
+            receipts::GetReceipts,
         },
         message::Message as RLPxMessage,
         p2p::{Capability, SUPPORTED_ETH_CAPABILITIES, SUPPORTED_SNAP_CAPABILITIES},
@@ -207,10 +207,11 @@ impl PeerHandler {
             if let Some(receipts) = tokio::time::timeout(PEER_REPLY_TIMEOUT, async move {
                 loop {
                     match receiver.recv().await {
-                        Some(RLPxMessage::Receipts(Receipts { id, receipts }))
-                            if id == request_id =>
-                        {
-                            return Some(receipts)
+                        Some(RLPxMessage::Receipts(receipts)) => {
+                            if receipts.get_id() == request_id {
+                                return Some(receipts.get_receipts());
+                            }
+                            return None;
                         }
                         // Ignore replies that don't match the expected id (such as late responses)
                         Some(_) => continue,

crates/networking/p2p/rlpx/connection.rs

@@ -6,6 +6,7 @@ use crate::{
             backend,
             blocks::{BlockBodies, BlockHeaders},
             receipts::{GetReceipts, Receipts},
+            status::StatusMessage,
             transactions::{GetPooledTransactions, Transactions},
         },
         frame::RLPxCodec,
@@ -463,7 +464,7 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
             }
             Message::Status(msg_data) => {
                 if let Some(eth) = &self.negotiated_eth_capability {
-                    backend::validate_status(msg_data, &self.storage, eth.version).await?
+                    backend::validate_status(msg_data, &self.storage, eth).await?
                 };
             }
             Message::GetAccountRange(req) => {
@@ -499,12 +500,14 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
                 self.send(Message::BlockBodies(response)).await?;
             }
             Message::GetReceipts(GetReceipts { id, block_hashes }) if peer_supports_eth => {
-                let mut receipts = Vec::new();
-                for hash in block_hashes.iter() {
-                    receipts.push(self.storage.get_receipts_for_block(hash)?);
+                if let Some(eth) = &self.negotiated_eth_capability {
+                    let mut receipts = Vec::new();
+                    for hash in block_hashes.iter() {
+                        receipts.push(self.storage.get_receipts_for_block(hash)?);
+                    }
+                    let response = Receipts::new(id, receipts, eth)?;
+                    self.send(Message::Receipts(response)).await?;
                 }
-                let response = Receipts { id, receipts };
-                self.send(Message::Receipts(response)).await?;
             }
             Message::NewPooledTransactionHashes(new_pooled_transaction_hashes)
                 if peer_supports_eth =>
@@ -579,7 +582,7 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
     async fn init_peer_conn(&mut self) -> Result<(), RLPxError> {
         // Sending eth Status if peer supports it
         if let Some(eth) = self.negotiated_eth_capability.clone() {
-            let status = backend::get_status(&self.storage, eth.version).await?;
+            let status = StatusMessage::new(&self.storage, &eth).await?;
             log_peer_debug(&self.node, "Sending status");
             self.send(Message::Status(status)).await?;
             // The next immediate message in the ETH protocol is the
@@ -592,7 +595,7 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
             match msg {
                 Message::Status(msg_data) => {
                     log_peer_debug(&self.node, "Received Status");
-                    backend::validate_status(msg_data, &self.storage, eth.version).await?
+                    backend::validate_status(msg_data, &self.storage, &eth).await?
                 }
                 Message::Disconnect(disconnect) => {
                     return Err(RLPxError::HandshakeError(format!(

crates/networking/p2p/rlpx/error.rs

@@ -51,6 +51,8 @@ pub(crate) enum RLPxError {
     IoError(#[from] std::io::Error),
     #[error("Failed to decode message due to invalid frame: {0}")]
     InvalidMessageFrame(String),
+    #[error("IncompatibleProtocol")]
+    IncompatibleProtocol,
 }
 
 // tokio::sync::mpsc::error::SendError<Message> is too large to be part of the RLPxError enum directly

crates/networking/p2p/rlpx/eth/backend.rs

@@ -1,46 +1,14 @@
-use ethrex_common::{types::ForkId, U256};
+use ethrex_common::types::ForkId;
 use ethrex_storage::Store;
 
-use crate::rlpx::error::RLPxError;
+use crate::rlpx::{error::RLPxError, p2p::Capability};
 
 use super::status::StatusMessage;
 
-pub async fn get_status(storage: &Store, eth_version: u8) -> Result<StatusMessage, RLPxError> {
-    let chain_config = storage.get_chain_config()?;
-    let total_difficulty = U256::from(chain_config.terminal_total_difficulty.unwrap_or_default());
-    let network_id = chain_config.chain_id;
-
-    // These blocks must always be available
-    let genesis_header = storage
-        .get_block_header(0)?
-        .ok_or(RLPxError::NotFound("Genesis Block".to_string()))?;
-    let block_number = storage.get_latest_block_number().await?;
-    let block_header = storage
-        .get_block_header(block_number)?
-        .ok_or(RLPxError::NotFound(format!("Block {block_number}")))?;
-
-    let genesis = genesis_header.compute_block_hash();
-    let block_hash = block_header.compute_block_hash();
-    let fork_id = ForkId::new(
-        chain_config,
-        genesis_header,
-        block_header.timestamp,
-        block_number,
-    );
-    Ok(StatusMessage {
-        eth_version: eth_version as u32,
-        network_id,
-        total_difficulty,
-        block_hash,
-        genesis,
-        fork_id,
-    })
-}
-
 pub async fn validate_status(
     msg_data: StatusMessage,
     storage: &Store,
-    eth_version: u8,
+    eth_capability: &Capability,
 ) -> Result<(), RLPxError> {
     let chain_config = storage.get_chain_config()?;
 
@@ -61,26 +29,26 @@ pub async fn validate_status(
     );
 
     //Check networkID
-    if msg_data.network_id != chain_config.chain_id {
+    if msg_data.get_network_id() != chain_config.chain_id {
         return Err(RLPxError::HandshakeError(
             "Network Id does not match".to_string(),
         ));
     }
     //Check Protocol Version
-    if msg_data.eth_version as u8 != eth_version {
+    if msg_data.get_eth_version() != eth_capability.version {
         return Err(RLPxError::HandshakeError(
             "Eth protocol version does not match".to_string(),
         ));
     }
     //Check Genesis
-    if msg_data.genesis != genesis_hash {
+    if msg_data.get_genesis() != genesis_hash {
         return Err(RLPxError::HandshakeError(
             "Genesis does not match".to_string(),
         ));
     }
     // Check ForkID
     if !fork_id.is_valid(
-        msg_data.fork_id,
+        msg_data.get_fork_id(),
         latest_block_number,
         latest_block_header.timestamp,
         chain_config,
@@ -95,11 +63,14 @@ pub async fn validate_status(
 #[cfg(test)]
 mod tests {
     use super::validate_status;
+    use crate::rlpx::eth::eth68::status::StatusMessage68;
     use crate::rlpx::eth::status::StatusMessage;
+    use crate::rlpx::p2p::Capability;
     use ethrex_common::{
         types::{ForkId, Genesis},
         H256, U256,
     };
+
     use ethrex_storage::{EngineType, Store};
     use std::{fs::File, io::BufReader};
 
@@ -125,16 +96,16 @@ mod tests {
         let genesis_hash = genesis_header.compute_block_hash();
         let fork_id = ForkId::new(config, genesis_header, 2707305664, 123);
 
-        let eth_version = 68;
-        let message = StatusMessage {
-            eth_version: eth_version as u32,
+        let eth = Capability::eth(68);
+        let message = StatusMessage::StatusMessage68(StatusMessage68 {
+            eth_version: eth.version,
             network_id: 3503995874084926,
             total_difficulty,
             block_hash: H256::random(),
             genesis: genesis_hash,
             fork_id,
-        };
-        let result = validate_status(message, &storage, eth_version).await;
+        });
+        let result = validate_status(message, &storage, &eth).await;
         assert!(result.is_ok());
     }
 }

crates/networking/p2p/rlpx/eth/eth68/mod.rs

@@ -0,0 +1,2 @@
+pub mod receipt;
+pub mod status;

crates/networking/p2p/rlpx/eth/eth68/receipt.rs

@@ -0,0 +1,49 @@
+use crate::rlpx::{
+    message::RLPxMessage,
+    utils::{snappy_compress, snappy_decompress},
+};
+use bytes::BufMut;
+use ethrex_common::types::Receipt;
+use ethrex_rlp::{
+    error::{RLPDecodeError, RLPEncodeError},
+    structs::{Decoder, Encoder},
+};
+
+#[derive(Debug)]
+pub(crate) struct Receipts68 {
+    // 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
+    pub id: u64,
+    pub receipts: Vec<Vec<Receipt>>,
+}
+
+impl Receipts68 {
+    pub fn new(id: u64, receipts: Vec<Vec<Receipt>>) -> Self {
+        Self { receipts, id }
+    }
+}
+
+impl RLPxMessage for Receipts68 {
+    const CODE: u8 = 0x1F;
+
+    fn encode(&self, buf: &mut dyn BufMut) -> Result<(), RLPEncodeError> {
+        let mut encoded_data = vec![];
+        Encoder::new(&mut encoded_data)
+            .encode_field(&self.id)
+            .encode_field(&self.receipts)
+            .finish();
+
+        let msg_data = snappy_compress(encoded_data)?;
+        buf.put_slice(&msg_data);
+        Ok(())
+    }
+
+    fn decode(msg_data: &[u8]) -> Result<Self, RLPDecodeError> {
+        let decompressed_data = snappy_decompress(msg_data)?;
+        let decoder = Decoder::new(&decompressed_data)?;
+        let (id, decoder): (u64, _) = decoder.decode_field("request-id")?;
+        let (receipts, _): (Vec<Vec<Receipt>>, _) = decoder.decode_field("receipts")?;
+
+        Ok(Self::new(id, receipts))
+    }
+}

crates/networking/p2p/rlpx/eth/eth68/status.rs

@@ -0,0 +1,67 @@
+use crate::rlpx::{
+    message::RLPxMessage,
+    utils::{snappy_compress, snappy_decompress},
+};
+use bytes::BufMut;
+use ethrex_common::{
+    types::{BlockHash, ForkId},
+    U256,
+};
+use ethrex_rlp::{
+    error::{RLPDecodeError, RLPEncodeError},
+    structs::{Decoder, Encoder},
+};
+
+#[derive(Debug)]
+pub(crate) struct StatusMessage68 {
+    pub(crate) eth_version: u8,
+    pub(crate) network_id: u64,
+    pub(crate) total_difficulty: U256,
+    pub(crate) block_hash: BlockHash,
+    pub(crate) genesis: BlockHash,
+    pub(crate) fork_id: ForkId,
+}
+
+impl RLPxMessage for StatusMessage68 {
+    const CODE: u8 = 0x10;
+    fn encode(&self, buf: &mut dyn BufMut) -> Result<(), RLPEncodeError> {
+        let mut encoded_data = vec![];
+        Encoder::new(&mut encoded_data)
+            .encode_field(&self.eth_version)
+            .encode_field(&self.network_id)
+            .encode_field(&self.total_difficulty)
+            .encode_field(&self.block_hash)
+            .encode_field(&self.genesis)
+            .encode_field(&self.fork_id)
+            .finish();
+
+        let msg_data = snappy_compress(encoded_data)?;
+        buf.put_slice(&msg_data);
+        Ok(())
+    }
+
+    fn decode(msg_data: &[u8]) -> Result<Self, RLPDecodeError> {
+        let decompressed_data = snappy_decompress(msg_data)?;
+        let decoder = Decoder::new(&decompressed_data)?;
+        let (eth_version, decoder): (u32, _) = decoder.decode_field("protocolVersion")?;
+
+        assert_eq!(eth_version, 68, "only eth version 68 is supported");
+
+        let (network_id, decoder): (u64, _) = decoder.decode_field("networkId")?;
+        let (total_difficulty, decoder): (U256, _) = decoder.decode_field("totalDifficulty")?;
+        let (block_hash, decoder): (BlockHash, _) = decoder.decode_field("blockHash")?;
+        let (genesis, decoder): (BlockHash, _) = decoder.decode_field("genesis")?;
+        let (fork_id, decoder): (ForkId, _) = decoder.decode_field("forkId")?;
+        // Implementations must ignore any additional list elements
+        let _padding = decoder.finish_unchecked();
+
+        Ok(Self {
+            eth_version: eth_version as u8,
+            network_id,
+            total_difficulty,
+            block_hash,
+            genesis,
+            fork_id,
+        })
+    }
+}

crates/networking/p2p/rlpx/eth/eth69/mod.rs

@@ -0,0 +1 @@
+pub mod status;