merge main, fix conflicts

.github/workflows/ci_l1.yaml

@@ -157,7 +157,7 @@ jobs:
             test_pattern: /AccountRange|StorageRanges|ByteCodes|TrieNodes
           - name: "Devp2p eth tests"
             simulation: devp2p
-            test_pattern: eth/Status|GetBlockHeaders|SimultaneousRequests|SameRequestID|ZeroRequestID|GetBlockBodies|MaliciousHandshake|MaliciousStatus|Transaction|InvalidTxs
+            test_pattern: eth/Status|GetBlockHeaders|SimultaneousRequests|SameRequestID|ZeroRequestID|GetBlockBodies|MaliciousHandshake|MaliciousStatus|Transaction|InvalidTxs|NewPooledTxs
           - name: "Engine Auth and EC tests"
             simulation: ethereum/engine
             test_pattern: engine-(auth|exchange-capabilities)/

crates/networking/p2p/rlpx/connection.rs

@@ -22,21 +22,23 @@ use crate::{
 
 use super::{
     error::RLPxError,
+    eth::transactions::GetPooledTransactions,
     frame,
     handshake::{decode_ack_message, decode_auth_message, encode_auth_message},
     message::{self as rlpx},
     p2p::Capability,
     utils::{ecdh_xchng, pubkey2id},
 };
 use aes::cipher::KeyIvInit;
-use ethrex_blockchain::mempool;
+use ethrex_blockchain::mempool::{self};
 use ethrex_core::{H256, H512};
 use ethrex_rlp::decode::RLPDecode;
 use ethrex_storage::Store;
 use k256::{
     ecdsa::{RecoveryId, Signature, SigningKey, VerifyingKey},
     PublicKey, SecretKey,
 };
+use rand::random;
 use sha3::{Digest, Keccak256};
 use tokio::{
     io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
@@ -337,7 +339,7 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
         match message {
             Message::Disconnect(msg_data) => {
                 debug!("Received Disconnect: {:?}", msg_data.reason);
-                // Returning a Disonnect error to be handled later at the call stack
+                // Returning a Disconnect error to be handled later at the call stack
                 return Err(RLPxError::Disconnect());
             }
             Message::Ping(_) => {
@@ -377,6 +379,17 @@ impl<S: AsyncWrite + AsyncRead + std::marker::Unpin> RLPxConnection<S> {
                 };
                 self.send(Message::BlockBodies(response)).await?;
             }
+            Message::NewPooledTransactionHashes(new_pooled_transaction_hashes)
+                if peer_supports_eth =>
+            {
+                //TODO(#1415): evaluate keeping track of requests to avoid sending the same twice.
+                let hashes =
+                    new_pooled_transaction_hashes.get_transactions_to_request(&self.storage)?;
+
+                //TODO(#1416): Evaluate keeping track of the request-id.
+                let request = GetPooledTransactions::new(random(), hashes);
+                self.send(Message::GetPooledTransactions(request)).await?;
+            }
             Message::GetStorageRanges(req) => {
                 let response = process_storage_ranges_request(req, self.storage.clone())?;
                 self.send(Message::StorageRanges(response)).await?

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

@@ -1,9 +1,11 @@
 use bytes::BufMut;
+use bytes::Bytes;
 use ethrex_core::{types::Transaction, H256};
 use ethrex_rlp::{
     error::{RLPDecodeError, RLPEncodeError},
     structs::{Decoder, Encoder},
 };
+use ethrex_storage::{error::StoreError, Store};
 
 use crate::rlpx::{
     message::RLPxMessage,
@@ -64,7 +66,7 @@ impl RLPxMessage for Transactions {
 // Broadcast message
 #[derive(Debug)]
 pub(crate) struct NewPooledTransactionHashes {
-    transaction_types: Vec<u8>,
+    transaction_types: Bytes,
     transaction_sizes: Vec<usize>,
     transaction_hashes: Vec<H256>,
 }
@@ -88,11 +90,15 @@ impl NewPooledTransactionHashes {
             transaction_hashes.push(transaction_hash);
         }
         Self {
-            transaction_types,
+            transaction_types: transaction_types.into(),
             transaction_sizes,
             transaction_hashes,
         }
     }
+
+    pub fn get_transactions_to_request(&self, storage: &Store) -> Result<Vec<H256>, StoreError> {
+        storage.filter_unknown_transactions(&self.transaction_hashes)
+    }
 }
 
 impl RLPxMessage for NewPooledTransactionHashes {
@@ -112,8 +118,7 @@ impl RLPxMessage for NewPooledTransactionHashes {
     fn decode(msg_data: &[u8]) -> Result<Self, RLPDecodeError> {
         let decompressed_data = snappy_decompress(msg_data)?;
         let decoder = Decoder::new(&decompressed_data)?;
-        let (transaction_types, decoder): (Vec<u8>, _) =
-            decoder.decode_field("transactionTypes")?;
+        let (transaction_types, decoder): (Bytes, _) = decoder.decode_field("transactionTypes")?;
         let (transaction_sizes, decoder): (Vec<usize>, _) =
             decoder.decode_field("transactionSizes")?;
         let (transaction_hashes, _): (Vec<H256>, _) = decoder.decode_field("transactionHashes")?;

crates/networking/p2p/rlpx/message.rs

@@ -5,7 +5,7 @@ use std::fmt::Display;
 use super::eth::blocks::{BlockBodies, BlockHeaders, GetBlockBodies, GetBlockHeaders};
 use super::eth::receipts::Receipts;
 use super::eth::status::StatusMessage;
-use super::eth::transactions::Transactions;
+use super::eth::transactions::{GetPooledTransactions, NewPooledTransactionHashes, Transactions};
 use super::p2p::{DisconnectMessage, HelloMessage, PingMessage, PongMessage};
 use super::snap::{
     AccountRange, ByteCodes, GetAccountRange, GetByteCodes, GetStorageRanges, GetTrieNodes,
@@ -32,6 +32,8 @@ pub(crate) enum Message {
     Transactions(Transactions),
     GetBlockBodies(GetBlockBodies),
     BlockBodies(BlockBodies),
+    NewPooledTransactionHashes(NewPooledTransactionHashes),
+    GetPooledTransactions(GetPooledTransactions),
     Receipts(Receipts),
     // snap capability
     GetAccountRange(GetAccountRange),
@@ -67,6 +69,12 @@ impl Message {
             0x14 => Ok(Message::BlockHeaders(BlockHeaders::decode(msg_data)?)),
             0x15 => Ok(Message::GetBlockBodies(GetBlockBodies::decode(msg_data)?)),
             0x16 => Ok(Message::BlockBodies(BlockBodies::decode(msg_data)?)),
+            0x18 => Ok(Message::NewPooledTransactionHashes(
+                NewPooledTransactionHashes::decode(msg_data)?,
+            )),
+            0x19 => Ok(Message::GetPooledTransactions(
+                GetPooledTransactions::decode(msg_data)?,
+            )),
             0x20 => Ok(Message::Receipts(Receipts::decode(msg_data)?)),
             0x21 => Ok(Message::GetAccountRange(GetAccountRange::decode(msg_data)?)),
             0x22 => Ok(Message::AccountRange(AccountRange::decode(msg_data)?)),
@@ -124,6 +132,14 @@ impl Message {
                 0x16_u8.encode(buf);
                 msg.encode(buf)
             }
+            Message::NewPooledTransactionHashes(msg) => {
+                0x18_u8.encode(buf);
+                msg.encode(buf)
+            }
+            Message::GetPooledTransactions(msg) => {
+                0x19_u8.encode(buf);
+                msg.encode(buf)
+            }
             Message::Receipts(msg) => {
                 0x20_u8.encode(buf);
                 msg.encode(buf)
@@ -175,6 +191,8 @@ impl Display for Message {
             Message::GetBlockHeaders(_) => "eth:getBlockHeaders".fmt(f),
             Message::BlockHeaders(_) => "eth:BlockHeaders".fmt(f),
             Message::BlockBodies(_) => "eth:BlockBodies".fmt(f),
+            Message::NewPooledTransactionHashes(_) => "eth:NewPooledTransactionHashes".fmt(f),
+            Message::GetPooledTransactions(_) => "eth::GetPooledTransactions".fmt(f),
             Message::Transactions(_) => "eth:TransactionsMessage".fmt(f),
             Message::GetBlockBodies(_) => "eth:GetBlockBodies".fmt(f),
             Message::Receipts(_) => "eth:Receipts".fmt(f),

crates/storage/store/storage.rs

@@ -17,7 +17,7 @@ use ethrex_rlp::encode::RLPEncode;
 use ethrex_trie::Trie;
 use serde::{Deserialize, Serialize};
 use sha3::{Digest as _, Keccak256};
-use std::collections::HashMap;
+use std::collections::{HashMap, HashSet};
 use std::fmt::Debug;
 use std::sync::{Arc, Mutex};
 use tracing::info;
@@ -325,6 +325,24 @@ impl Store {
         Ok(txs_by_sender)
     }
 
+    /// Gets hashes from possible_hashes that are not already known in the mempool.
+    pub fn filter_unknown_transactions(
+        &self,
+        possible_hashes: &[H256],
+    ) -> Result<Vec<H256>, StoreError> {
+        let mempool = self
+            .mempool
+            .lock()
+            .map_err(|error| StoreError::Custom(error.to_string()))?;
+
+        let tx_set: HashSet<_> = mempool.iter().map(|(hash, _)| hash).collect();
+        Ok(possible_hashes
+            .iter()
+            .filter(|hash| !tx_set.contains(hash))
+            .copied()
+            .collect())
+    }
+
     fn add_account_code(&self, code_hash: H256, code: Bytes) -> Result<(), StoreError> {
         self.engine.add_account_code(code_hash, code)
     }

crates/vm/levm/src/errors.rs

@@ -78,6 +78,8 @@ pub enum VMError {
     Internal(#[from] InternalError),
     #[error("Transaction validation error: {0}")]
     TxValidation(#[from] TxValidationError),
+    #[error("Offset out of bounds")]
+    OutOfOffset,
 }
 
 impl VMError {