chore(l1): optimize the structure for tracking known txs sent to peers in TxBroadcaster (#4675)

**Motivation**

We were seeing spikes of 3~10Gb on post-sync execution, most of it was
due to `resize_rehash` during the `add_txs` of the `TxBroadcaster`.

**Description**

This PR replace the current hashmap of `(tx_hash, peer_id) -> Instant`
for a less duplicated `tx_hash -> (peer_bitmap, last_seen)` + another
`peer_id -> peer_index`. This removes orders of magnitud of entries,
going from 27M to less than 100k. Now all the memory taken from the
TxBroadcaster disappeared from our memory flamegraphs.

WIP: ~I'm right now analyzing the execution of the client throughout the
weekend, we still appear to have a memory issue.~ This was related to a
change in rocksdb that enabled transaction. It didn't affected this fix.

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

Closes #issue_number

Author: rodrigo-o

crates/networking/p2p/tx_broadcaster.rs

@@ -28,20 +28,75 @@ use crate::{
 // 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;
 
-// Amount of seconds after which we prune old entries from broadcasted_txs_per_peer (We should fine tune this)
-const PRUNE_WAIT_TIME_SECS: u64 = 300; // 5 minutes
+// Amount of seconds after which we prune broadcast records (We should fine tune this)
+const PRUNE_WAIT_TIME_SECS: u64 = 600; // 10 minutes
 
 // Amount of seconds between each prune
-const PRUNE_INTERVAL_SECS: u64 = 300; // 5 minutes
+const PRUNE_INTERVAL_SECS: u64 = 360; // 6 minutes
 
 // Amount of seconds between each broadcast
 const BROADCAST_INTERVAL_SECS: u64 = 1; // 1 second
 
+#[derive(Debug, Clone, Default)]
+struct PeerMask {
+    bits: Vec<u64>,
+}
+
+impl PeerMask {
+    #[inline]
+    // Ensure that the internal bit vector can hold the given index
+    // If not, resize the vector.
+    fn ensure(&mut self, idx: u32) {
+        let word = (idx as usize) / 64;
+        if self.bits.len() <= word {
+            self.bits.resize(word + 1, 0);
+        }
+    }
+
+    #[inline]
+    fn is_set(&self, idx: u32) -> bool {
+        let word = (idx as usize) / 64;
+        if word >= self.bits.len() {
+            return false;
+        }
+        let bit = (idx as usize) % 64;
+        (self.bits[word] >> bit) & 1 == 1
+    }
+
+    #[inline]
+    fn set(&mut self, idx: u32) {
+        self.ensure(idx);
+        let word = (idx as usize) / 64;
+        let bit = (idx as usize) % 64;
+        self.bits[word] |= 1u64 << bit;
+    }
+}
+
+#[derive(Debug, Clone)]
+struct BroadcastRecord {
+    peers: PeerMask,
+    last_sent: Instant,
+}
+
+impl Default for BroadcastRecord {
+    fn default() -> Self {
+        Self {
+            peers: PeerMask::default(),
+            last_sent: Instant::now(),
+        }
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct TxBroadcaster {
     kademlia: Kademlia,
     blockchain: Arc<Blockchain>,
-    broadcasted_txs_per_peer: HashMap<(H256, H256), Instant>, // (peer_id,tx_hash) -> timestamp
+    // tx_hash -> broadcast record (which peers know it and when it was last sent)
+    known_txs: HashMap<H256, BroadcastRecord>,
+    // Assign each peer_id (H256) a u32 index used by PeerMask entries
+    peer_indexer: HashMap<H256, u32>,
+    // Next index to assign to a new peer
+    next_peer_idx: u32,
 }
 
 #[derive(Debug, Clone)]
@@ -66,7 +121,9 @@ impl TxBroadcaster {
         let state = TxBroadcaster {
             kademlia,
             blockchain,
-            broadcasted_txs_per_peer: HashMap::new(),
+            known_txs: HashMap::new(),
+            peer_indexer: HashMap::new(),
+            next_peer_idx: 0,
         };
 
         let server = state.clone().start();
@@ -86,12 +143,36 @@ impl TxBroadcaster {
         Ok(server)
     }
 
+    // Get or assign a unique index to the peer_id
+    #[inline]
+    fn peer_index(&mut self, peer_id: H256) -> u32 {
+        if let Some(&idx) = self.peer_indexer.get(&peer_id) {
+            idx
+        } else {
+            // We are assigning indexes sequentially, so next_peer_idx is always the next available one.
+            // self.peer_indexer.len() could be used instead of next_peer_idx but avoided here if we ever
+            // remove entries from peer_indexer in the future.
+            let idx = self.next_peer_idx;
+            // In practice we won't exceed u32::MAX (~4.29 Billion) peers.
+            self.next_peer_idx += 1;
+            self.peer_indexer.insert(peer_id, idx);
+            idx
+        }
+    }
+
     fn add_txs(&mut self, txs: Vec<H256>, peer_id: H256) {
+        debug!(total = self.known_txs.len(), adding = txs.len(), peer_id = %format!("{:#x}", peer_id), "Adding transactions to known list");
+
+        if txs.is_empty() {
+            return;
+        }
+
         let now = Instant::now();
+        let peer_idx = self.peer_index(peer_id);
         for tx in txs {
-            self.broadcasted_txs_per_peer
-                .entry((peer_id, tx))
-                .insert_entry(now);
+            let record = self.known_txs.entry(tx).or_default();
+            record.peers.set(peer_idx);
+            record.last_sent = now;
         }
     }
 
@@ -127,12 +208,15 @@ impl TxBroadcaster {
             shuffled_peers.split_at(peer_sqrt.ceil() as usize);
 
         for (peer_id, mut peer_channels, capabilities) in peers_to_send_full_txs.iter().cloned() {
+            let peer_idx = self.peer_index(peer_id);
             let txs_to_send = full_txs
                 .iter()
                 .filter(|tx| {
+                    let hash = tx.hash();
                     !self
-                        .broadcasted_txs_per_peer
-                        .contains_key(&(peer_id, tx.hash()))
+                        .known_txs
+                        .get(&hash)
+                        .is_some_and(|record| record.peers.is_set(peer_idx))
                 })
                 .cloned()
                 .collect::<Vec<Transaction>>();
@@ -170,12 +254,15 @@ impl TxBroadcaster {
         peer_channels: &mut PeerChannels,
         peer_id: H256,
     ) -> Result<(), TxBroadcasterError> {
+        let peer_idx = self.peer_index(peer_id);
         let txs_to_send = txs
             .iter()
             .filter(|tx| {
+                let hash = tx.hash();
                 !self
-                    .broadcasted_txs_per_peer
-                    .contains_key(&(peer_id, tx.hash()))
+                    .known_txs
+                    .get(&hash)
+                    .is_some_and(|record| record.peers.is_set(peer_idx))
             })
             .cloned()
             .collect::<Vec<MempoolTransaction>>();
@@ -250,9 +337,16 @@ impl GenServer for TxBroadcaster {
             Self::CastMsg::PruneTxs => {
                 debug!(received = "PruneTxs");
                 let now = Instant::now();
-                self.broadcasted_txs_per_peer.retain(|_, &mut timestamp| {
-                    now.duration_since(timestamp) < Duration::from_secs(PRUNE_WAIT_TIME_SECS)
-                });
+                let before = self.known_txs.len();
+                let prune_window = Duration::from_secs(PRUNE_WAIT_TIME_SECS);
+
+                self.known_txs
+                    .retain(|_, record| now.duration_since(record.last_sent) < prune_window);
+                debug!(
+                    before = before,
+                    after = self.known_txs.len(),
+                    "Pruned old broadcasted transactions"
+                );
                 CastResponse::NoReply
             }
         }