Snap sync tx broadcaster (#4077)

**Motivation**

<!-- Why does this pull request exist? What are its goals? -->

**Description**

<!-- A clear and concise general description of the changes this PR
introduces -->

<!-- Link to issues: Resolves #111, Resolves #222 -->

Closes #issue_number

Author: gianbelinche

crates/blockchain/mempool.rs

@@ -19,6 +19,7 @@ use ethrex_storage::error::StoreError;
 
 #[derive(Debug, Default)]
 pub struct Mempool {
+    broadcast_pool: RwLock<HashSet<H256>>,
     transaction_pool: RwLock<HashMap<H256, MempoolTransaction>>,
     blobs_bundle_pool: Mutex<HashMap<H256, BlobsBundle>>,
     txs_by_sender_nonce: RwLock<BTreeMap<(H160, u64), H256>>,
@@ -42,10 +43,38 @@ impl Mempool {
             .write()
             .map_err(|error| StoreError::MempoolWriteLock(error.to_string()))?
             .insert(hash, transaction);
+        self.broadcast_pool
+            .write()
+            .map_err(|error| StoreError::MempoolWriteLock(error.to_string()))?
+            .insert(hash);
 
         Ok(())
     }
 
+    pub fn get_txs_for_broadcast(&self) -> Result<Vec<MempoolTransaction>, StoreError> {
+        let txs = self.transaction_pool.read()
+            .map_err(|error| StoreError::MempoolReadLock(error.to_string()))
+            .and_then(|pool| {
+                Ok(pool.iter()
+                    .filter_map(|(hash, tx)| {
+                        if !self.broadcast_pool.read().ok()?.contains(hash) {
+                            None
+                        } else {
+                            Some(tx.clone())
+                        }
+                    }).collect::<Vec<_>>())
+            })?;
+            Ok(txs)
+    }
+
+    pub fn clear_broadcasted_txs(&self) {
+        self.broadcast_pool
+            .write()
+            .map_err(|error| StoreError::MempoolWriteLock(error.to_string()))
+            .map(|mut pool| pool.clear())
+            .unwrap_or_default();
+    }
+
     /// Add a blobs bundle to the pool by its blob transaction hash
     pub fn add_blobs_bundle(
         &self,
@@ -75,6 +104,10 @@ impl Mempool {
             .transaction_pool
             .write()
             .map_err(|error| StoreError::MempoolWriteLock(error.to_string()))?;
+        let mut broadcast_pool = self
+            .broadcast_pool
+            .write()
+            .map_err(|error| StoreError::MempoolWriteLock(error.to_string()))?;
         if let Some(tx) = tx_pool.get(hash) {
             if matches!(tx.tx_type(), TxType::EIP4844) {
                 self.blobs_bundle_pool
@@ -88,6 +121,7 @@ impl Mempool {
                 .map_err(|error| StoreError::MempoolWriteLock(error.to_string()))?
                 .remove(&(tx.sender(), tx.nonce()));
             tx_pool.remove(hash);
+            broadcast_pool.remove(hash);
         };
 
         Ok(())

crates/networking/p2p/network.rs

@@ -2,16 +2,12 @@ use crate::{
     discv4::{
         server::{DiscoveryServer, DiscoveryServerError},
         side_car::{DiscoverySideCar, DiscoverySideCarError},
-    },
-    kademlia::Kademlia,
-    metrics::METRICS,
-    rlpx::{
+    }, kademlia::Kademlia, metrics::METRICS, rlpx::{
         connection::server::{RLPxConnBroadcastSender, RLPxConnection},
         initiator::{RLPxInitiator, RLPxInitiatorError},
         l2::l2_connection::P2PBasedContext,
         message::Message,
-    },
-    types::{Node, NodeRecord},
+    }, tx_broadcaster::{TxBroadcaster, TxBroadcasterError}, types::{Node, NodeRecord}
 };
 use ethrex_blockchain::Blockchain;
 use ethrex_storage::Store;
@@ -86,6 +82,8 @@ pub enum NetworkError {
     DiscoverySideCarError(#[from] DiscoverySideCarError),
     #[error("Failed to start RLPx Initiator: {0}")]
     RLPxInitiatorError(#[from] RLPxInitiatorError),
+    #[error("Failed to start Tx Broadcaster: {0}")]
+    TxBroadcasterError(#[from] TxBroadcasterError),
 }
 
 pub fn peer_table() -> Kademlia {
@@ -128,6 +126,12 @@ pub async fn start_network(context: P2PContext, bootnodes: Vec<Node>) -> Result<
             error!("Failed to start RLPx Initiator: {e}");
         })?;
 
+    TxBroadcaster::spawn(context.table.clone(), context.blockchain.clone())
+        .await
+        .inspect_err(|e| {
+            error!("Failed to start Tx Broadcaster: {e}");
+        })?;
+
     context.tracker.spawn(serve_p2p_requests(context.clone()));
 
     Ok(())

crates/networking/p2p/p2p.rs

@@ -9,6 +9,7 @@ pub mod sync;
 pub mod sync_manager;
 pub mod types;
 pub mod utils;
+pub mod tx_broadcaster;
 
 pub use network::periodically_show_peer_stats;
 pub use network::start_network;

crates/networking/p2p/rlpx/connection/server.rs

@@ -69,7 +69,7 @@ use crate::{
 };
 
 const PING_INTERVAL: Duration = Duration::from_secs(10);
-const TX_BROADCAST_INTERVAL: Duration = Duration::from_millis(500);
+const TX_BROADCAST_INTERVAL: Duration = Duration::from_millis(20000);
 const BLOCK_RANGE_UPDATE_INTERVAL: Duration = Duration::from_secs(60);
 // Soft limit for the number of transaction hashes sent in a single NewPooledTransactionHashes message as per [the spec](https://github.com/ethereum/devp2p/blob/master/caps/eth.md#newpooledtransactionhashes-0x080)
 const NEW_POOLED_TRANSACTION_HASHES_SOFT_LIMIT: usize = 4096;
@@ -172,7 +172,7 @@ pub enum CastMessage {
     PeerMessage(Message),
     BackendMessage(Message),
     SendPing,
-    SendNewPooledTxHashes,
+    SendNewPooledTxHashes(Vec<MempoolTransaction>),
     BlockRangeUpdate,
     BroadcastMessage(task::Id, Arc<Message>),
     L2(L2Cast),
@@ -296,8 +296,8 @@ impl GenServer for RLPxConnection {
                 Self::CastMsg::SendPing => {
                     send(&mut established_state, Message::Ping(PingMessage {})).await
                 }
-                Self::CastMsg::SendNewPooledTxHashes => {
-                    send_new_pooled_tx_hashes(&mut established_state).await
+                Self::CastMsg::SendNewPooledTxHashes(txs) => {
+                    send_new_pooled_tx_hashes(&mut established_state, txs).await
                 }
                 Self::CastMsg::BroadcastMessage(id, msg) => {
                     log_peer_debug(
@@ -418,14 +418,14 @@ where
     log_peer_debug(&state.node, "Peer connection initialized.");
 
     // Send transactions transaction hashes from mempool at connection start
-    send_new_pooled_tx_hashes(state).await?;
+    send_all_pooled_tx_hashes(state).await?;
 
     // Periodic broadcast check repeated events.
-    send_interval(
+    /*send_interval(
         TX_BROADCAST_INTERVAL,
         handle.clone(),
         CastMessage::SendNewPooledTxHashes,
-    );
+    );*/
 
     // Periodic Pings repeated events.
     send_interval(PING_INTERVAL, handle.clone(), CastMessage::SendPing);
@@ -467,7 +467,39 @@ where
     Ok(())
 }
 
-async fn send_new_pooled_tx_hashes(state: &mut Established) -> Result<(), RLPxError> {
+async fn send_new_pooled_tx_hashes(state: &mut Established, txs: Vec<MempoolTransaction>) -> Result<(), RLPxError> {
+    if SUPPORTED_ETH_CAPABILITIES
+        .iter()
+        .any(|cap| state.capabilities.contains(cap))
+    {
+        if !txs.is_empty() {
+            for tx_chunk in txs.chunks(NEW_POOLED_TRANSACTION_HASHES_SOFT_LIMIT) {
+                let tx_count = tx_chunk.len();
+                let mut txs_to_send = Vec::with_capacity(tx_count);
+                for tx in tx_chunk {
+                    txs_to_send.push((**tx).clone());
+                    state.broadcasted_txs.insert(tx.hash());
+                }
+
+                send(
+                    state,
+                    Message::NewPooledTransactionHashes(NewPooledTransactionHashes::new(
+                        txs_to_send,
+                        &state.blockchain,
+                    )?),
+                )
+                .await?;
+                log_peer_debug(
+                    &state.node,
+                    &format!("Sent {tx_count} transaction hashes to peer"),
+                );
+            }
+        }
+    }
+    Ok(())
+}
+
+async fn send_all_pooled_tx_hashes(state: &mut Established) -> Result<(), RLPxError> {
     if SUPPORTED_ETH_CAPABILITIES
         .iter()
         .any(|cap| state.capabilities.contains(cap))

crates/networking/p2p/tx_broadcaster.rs

@@ -0,0 +1,91 @@
+use std::{sync::Arc, time::Duration};
+
+use ethrex_blockchain::Blockchain;
+use spawned_concurrency::{messages::Unused, tasks::{send_interval, CastResponse, GenServer}};
+use tracing::{debug, error, info};
+
+use crate::{kademlia::Kademlia, rlpx::connection::server::CastMessage};
+
+#[derive(Debug, Clone)]
+pub struct TxBroadcaster {
+    kademlia: Kademlia,
+    blockchain: Arc<Blockchain>,
+}
+
+#[derive(Debug, Clone)]
+pub enum InMessage {
+    BroadcastTxs
+}
+
+#[derive(Debug, Clone)]
+pub enum OutMessage {
+    Done,
+}
+
+
+impl TxBroadcaster {
+    pub async fn spawn(
+        kademlia: Kademlia,
+        blockchain: Arc<Blockchain>,
+    ) -> Result<(), TxBroadcasterError> {
+        info!("Starting Transaction Broadcaster");
+
+        let state = TxBroadcaster {
+            kademlia,
+            blockchain,
+        };
+
+        let mut server = state.clone().start();
+
+        send_interval(
+            Duration::from_secs(1),
+            server.clone(),
+            InMessage::BroadcastTxs,
+        );
+
+        Ok(())
+    }
+
+    async fn broadcast_txs(&self) -> Result<(), TxBroadcasterError> {
+        let txs_to_broadcast = self.blockchain.mempool.get_txs_for_broadcast().map_err(|_| TxBroadcasterError::Broadcast)?;
+        let peers = self.kademlia.get_peer_channels(&[]).await;
+        for (peer_id, mut peer_channels) in peers {
+            peer_channels.connection.cast(CastMessage::SendNewPooledTxHashes(
+                txs_to_broadcast.clone(),
+            )).await.unwrap_or_else(|err| {
+                error!(peer_id = %format!("{:#x}", peer_id), err = ?err, "Failed to send new pooled tx hashes");
+            });
+        }
+        self.blockchain.mempool.clear_broadcasted_txs();
+        Ok(())
+    }
+}
+
+impl GenServer for TxBroadcaster {
+    type CallMsg = Unused;
+    type CastMsg = InMessage;
+    type OutMsg = OutMessage;
+    type Error = TxBroadcasterError;
+
+    async fn handle_cast(
+        &mut self,
+        message: Self::CastMsg,
+        handle: &spawned_concurrency::tasks::GenServerHandle<Self>,
+    ) -> CastResponse {
+        match message {
+            Self::CastMsg::BroadcastTxs => {
+                debug!(received = "BroadcastTxs");
+
+                self.broadcast_txs().await;
+
+                CastResponse::NoReply
+            }
+        }
+    }
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum TxBroadcasterError {
+    #[error("Failed to broadcast transactions")]
+    Broadcast
+}