⚡️ Deduplicate buffered proposals, drop on restart (#1374)

Author: wraitii

crates/hyle-model/src/node/mempool.rs

@@ -142,6 +142,14 @@ impl Hashed<DataProposalHash> for DataProposal {
         hash
     }
 }
+
+// Warning: hashing DPs can be slow, so use with care
+impl std::hash::Hash for DataProposal {
+    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
+        self.hashed().hash(state);
+    }
+}
+
 impl Display for DataProposalHash {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}", self.0)

src/mempool.rs

@@ -26,6 +26,7 @@ use hyle_modules::{
     modules::Module, utils::static_type_map::Pick,
 };
 use hyle_net::{logged_task::logged_task, ordered_join_set::OrderedJoinSet};
+use indexmap::IndexSet;
 use metrics::MempoolMetrics;
 use serde::{Deserialize, Serialize};
 use staking::state::Staking;
@@ -107,7 +108,11 @@ pub struct MempoolStore {
     waiting_dissemination_txs: BorshableIndexMap<TxHash, Transaction>,
     #[borsh(skip)]
     own_data_proposal_in_preparation: JoinSet<(DataProposalHash, DataProposal)>,
-    buffered_proposals: BTreeMap<LaneId, Vec<DataProposal>>,
+    // Skipped to clear on reset
+    #[borsh(skip)]
+    buffered_proposals: BTreeMap<LaneId, IndexSet<DataProposal>>, // This is an indexSet just so we can pop by idx.
+    // Skipped to clear on reset
+    #[borsh(skip)]
     buffered_podas: BTreeMap<LaneId, BTreeMap<DataProposalHash, Vec<UnaggregatedPoDA>>>,
 
     // verify_tx.rs
@@ -680,17 +685,19 @@ impl Mempool {
         self.lanes
             .put_no_verification(lane_id.clone(), (metadata, data_proposal))?;
 
-        let mut waiting_proposals = match self.buffered_proposals.get_mut(lane_id) {
+        // Retry all buffered proposals in this lane.
+        // We'll re-buffer them in the on_data_proposal logic if they fail to be processed.
+        let waiting_proposals = match self.buffered_proposals.get_mut(lane_id) {
             Some(waiting_proposals) => std::mem::take(waiting_proposals),
-            None => vec![],
+            None => Default::default(),
         };
 
         // TODO: retry remaining wp when one succeeds to be processed
-        for wp in waiting_proposals.iter_mut() {
+        for wp in waiting_proposals.into_iter() {
             if self.lanes.contains(lane_id, &wp.hashed()) {
                 continue;
             }
-            self.on_data_proposal(lane_id, wp.hashed(), std::mem::take(wp))
+            self.on_data_proposal(lane_id, wp.hashed(), wp)
                 .context("Consuming waiting data proposal")?;
         }
 

src/mempool/verify_tx.rs

@@ -170,15 +170,15 @@ impl super::Mempool {
                 );
             }
             DataProposalVerdict::Wait => {
-                debug!("Buffering DataProposal");
+                debug!("Buffering DataProposal {}", data_proposal_hash);
                 // Push the data proposal in the waiting list
                 self.buffered_proposals
                     .entry(lane_id.clone())
                     .or_default()
-                    .push(data_proposal);
+                    .insert(data_proposal);
             }
             DataProposalVerdict::Refuse => {
-                debug!("Refuse vote for DataProposal");
+                debug!("Refuse vote for DataProposal {}", data_proposal.hashed());
             }
         }
         Ok(())
@@ -242,8 +242,10 @@ impl super::Mempool {
                         }
                     });
                     if let Some(child_idx) = child_idx {
-                        // We have a buffered proposal that is a child of this DP
-                        dp = Some(buffered_proposals.swap_remove(child_idx));
+                        // We have a buffered proposal that is a child of this DP, process it.
+                        // (I _would_ use a HashSet, but this requires https://github.com/rust-lang/rust/issues/59618
+                        // which is coming in 1.88)
+                        dp = buffered_proposals.swap_remove_index(child_idx);
                     }
                 }
                 if let Some(dp) = dp {