Merge branch 'main' into feat/instrument_fjall_calls

Author: hhalex

src/mempool/dissemination.rs

@@ -38,6 +38,7 @@ use crate::model::SharedRunContext;
 const REQUEST_BACKOFF_INITIAL: Duration = Duration::from_secs(1);
 const REQUEST_BACKOFF_MAX: Duration = Duration::from_secs(8);
 const HOUSEKEEPING_INTERVAL: Duration = Duration::from_millis(400);
+const RESEND_MIN_INTERVAL: Duration = Duration::from_secs(2);
 #[cfg(not(test))]
 const SYNC_REPLY_DEBOUNCE: Duration = Duration::from_secs(2);
 #[cfg(test)]
@@ -847,7 +848,11 @@ impl DisseminationManager {
             return Ok(false);
         }
 
-        let filtered_targets = candidate_targets;
+        let now = TimestampMsClock::now();
+        let filtered_targets: HashSet<ValidatorPublicKey> = candidate_targets
+            .into_iter()
+            .filter(|peer| self.can_send_to_peer(lane_id, dp_hash, peer, now.clone()))
+            .collect();
         if filtered_targets.is_empty() {
             return Ok(false);
         }
@@ -991,6 +996,33 @@ impl DisseminationManager {
             .insert((lane_id.clone(), dp_hash.clone()), now);
     }
 
+    fn can_send_to_peer(
+        &self,
+        lane_id: &LaneId,
+        dp_hash: &DataProposalHash,
+        peer: &ValidatorPublicKey,
+        now: TimestampMs,
+    ) -> bool {
+        let key = (lane_id.clone(), dp_hash.clone(), peer.clone());
+        let evidence = self.knowledge.by_dp.get(&key);
+        let should_rate_limit = matches!(
+            evidence,
+            Some(EvidenceState::WeakHas)
+                | Some(EvidenceState::StrongHas)
+                | Some(EvidenceState::ObservedHas)
+        );
+
+        if !should_rate_limit {
+            return true;
+        }
+
+        self.knowledge
+            .by_peer
+            .get(peer)
+            .and_then(|state| state.last_dp_sent.get(&(lane_id.clone(), dp_hash.clone())))
+            .is_none_or(|last_sent| now - last_sent.clone() >= RESEND_MIN_INTERVAL)
+    }
+
     fn should_debounce_sync_reply(
         &self,
         lane_id: &LaneId,
@@ -1010,6 +1042,19 @@ impl DisseminationManager {
         self.owned_lanes.insert(lane_id);
     }
 
+    #[cfg(test)]
+    pub(crate) fn clear_last_dp_sent_for_test(
+        &mut self,
+        lane_id: &LaneId,
+        dp_hash: &DataProposalHash,
+    ) {
+        for peer_state in self.knowledge.by_peer.values_mut() {
+            peer_state
+                .last_dp_sent
+                .remove(&(lane_id.clone(), dp_hash.clone()));
+        }
+    }
+
     #[cfg(test)]
     pub(crate) async fn process_sync_requests_and_replies_for_test(&mut self) -> Result<()> {
         self.process_sync_requests_and_replies().await

src/mempool/own_lane.rs

@@ -939,27 +939,27 @@ pub mod test {
         ctx1.submit_tx(&tx2);
         ctx1.timer_tick().await?;
 
-        // This ends up disseminating 4 DPs for now - when we receive it, then on tick, then the other, then on tick.
+        // With resend gating, each new DP is disseminated once on creation in this flow.
         let mut dps = vec![];
-        for _ in 0..4 {
+        for _ in 0..2 {
             match ctx1.assert_broadcast("DataProposal").await.msg {
                 MempoolNetMessage::DataProposal(_, hash, dp, _) => dps.push((hash, dp)),
                 _ => panic!("Expected DataProposal message"),
             }
         }
 
-        assert!(dps.len() == 4, "Should have 4 DataProposals");
+        assert!(dps.len() == 2, "Should have 2 DataProposals");
         assert_eq!(dps[0].1.txs, vec![tx1.clone()]);
-        assert_eq!(dps[1].1.txs, vec![tx1.clone()]);
-        assert_eq!(dps[2].1.txs, vec![tx2.clone()]);
-        assert_eq!(dps[3].1.txs, vec![tx1.clone()]);
+        assert_eq!(dps[1].1.txs, vec![tx2.clone()]);
 
         // Redisseminate the oldest pending DataProposal
         // TODO: implement this as more of an integration test?
         let (oldest_hash, _) = dps
             .iter()
             .find(|(_, dp)| dp.parent_data_proposal_hash.is_lane_root())
             .expect("oldest dp should exist");
+        ctx1.dissemination_manager
+            .clear_last_dp_sent_for_test(&ctx1.own_lane(), oldest_hash);
         ctx1.maybe_disseminate_dp(&ctx1.own_lane(), oldest_hash)
             .expect("disseminate");
 

src/mempool/tests/mod.rs

@@ -604,6 +604,10 @@ async fn test_redisseminate_owned_lanes_sends_oldest_first() -> Result<()> {
     // Drain any prior outbound messages
     while ctx.out_receiver.try_recv().is_ok() {}
 
+    // Allow an immediate resend in this deterministic test.
+    ctx.dissemination_manager
+        .clear_last_dp_sent_for_test(&lane_id, &dp1_hash);
+
     ctx.dissemination_manager
         .redisseminate_owned_lanes()
         .await?;