wait a while , for better select

Author: JasonOE

bindings/python/py_src/lattica/client.py

@@ -258,7 +258,9 @@ def configure_bitswap_peer_selection(
         top_n: int = 3,
         enabled: bool = True,
         min_peers: int = 2,
-        enable_randomness: bool = True
+        enable_randomness: bool = True,
+        have_wait_window_ms: int = 100,
+        min_candidate_ratio: float = 0.3
     ) -> None:
         """Configure Bitswap peer selection strategy.
         
@@ -267,9 +269,16 @@ def configure_bitswap_peer_selection(
             enabled: Enable smart selection
             min_peers: Minimum peers threshold
             enable_randomness: Enable randomness in selection
+            have_wait_window_ms: Wait window in ms after first Have response before selecting peers.
+                This allows more peers to respond, ensuring better selection. Default: 100ms
+            min_candidate_ratio: Minimum candidate ratio (0.0-1.0) before starting selection.
+                Selection starts when candidates >= total_peers * min_candidate_ratio. Default: 0.3
         """
         try:
-            self._lattica_instance.configure_bitswap_peer_selection(top_n, enabled, min_peers, enable_randomness)
+            self._lattica_instance.configure_bitswap_peer_selection(
+                top_n, enabled, min_peers, enable_randomness, 
+                have_wait_window_ms, min_candidate_ratio
+            )
         except Exception as e:
             raise RuntimeError(f"Failed to configure bitswap peer selection: {e}")
 

bindings/python/src/core.rs

@@ -560,12 +560,35 @@ impl LatticaSDK {
     }
 
     /// Configure Bitswap peer selection strategy
-    #[pyo3(signature = (top_n = 3, enabled = true, min_peers = 2, enable_randomness = true))]
-    fn configure_bitswap_peer_selection(&self, top_n: usize, enabled: bool, min_peers: usize, enable_randomness: bool) -> PyResult<()> {
+    /// 
+    /// Args:
+    ///     top_n: Number of top peers to select (default: 3)
+    ///     enabled: Enable smart selection (default: true)
+    ///     min_peers: Minimum peers threshold (default: 2)
+    ///     enable_randomness: Enable randomness in selection (default: true)
+    ///     have_wait_window_ms: Wait window in ms after first Have response (default: 100)
+    ///     min_candidate_ratio: Minimum candidate ratio before selection (default: 0.3)
+    #[pyo3(signature = (top_n = 3, enabled = true, min_peers = 2, enable_randomness = true, have_wait_window_ms = 100, min_candidate_ratio = 0.3))]
+    fn configure_bitswap_peer_selection(
+        &self, 
+        top_n: usize, 
+        enabled: bool, 
+        min_peers: usize, 
+        enable_randomness: bool,
+        have_wait_window_ms: u64,
+        min_candidate_ratio: f64,
+    ) -> PyResult<()> {
         Python::with_gil(|py| {
             py.allow_threads(|| {
                 self.runtime.block_on(async move {
-                    let config = beetswap::PeerSelectionConfig { top_n, enabled, min_peers, enable_randomness };
+                    let config = beetswap::PeerSelectionConfig { 
+                        top_n, 
+                        enabled, 
+                        min_peers, 
+                        enable_randomness,
+                        have_wait_window_ms,
+                        min_candidate_ratio,
+                    };
                     self.lattica.configure_bitswap_peer_selection(config).await
                         .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!("{:?}", e)))
                 })

third_party/beetswap/src/client.rs

@@ -83,6 +83,8 @@ where
     peer_selection_config: PeerSelectionConfig,
     /// Global stats
     global_stats: GlobalStats,
+    /// Track the first Have response time for each CID (for wait window)
+    cid_first_have_time: FnvHashMap<CidGeneric<S>, Instant>,
 }
 
 #[derive(Debug)]
@@ -158,6 +160,7 @@ where
             new_blocks: Vec::new(),
             peer_selection_config: PeerSelectionConfig::default(),
             global_stats: GlobalStats::default(),
+            cid_first_have_time: FnvHashMap::default(),
         }
     }
 
@@ -276,6 +279,7 @@ where
                     let cid = cid.to_owned();
                     self.cid_to_queries.remove(&cid);
                     self.wantlist.remove(&cid);
+                    self.cid_first_have_time.remove(&cid);
                 }
 
                 break;
@@ -290,10 +294,14 @@ where
 
         let mut new_blocks = Vec::new();
 
-        // Update presence
+        // Update presence and track first Have time for wait window
         for (cid, block_presence) in msg.block_presences {
             match block_presence {
-                BlockPresenceType::Have => peer_state.wantlist.got_have(&cid),
+                BlockPresenceType::Have => {
+                    peer_state.wantlist.got_have(&cid);
+                    // Record first Have time for this CID (for wait window mechanism)
+                    self.cid_first_have_time.entry(cid).or_insert_with(Instant::now);
+                }
                 BlockPresenceType::DontHave => peer_state.wantlist.got_dont_have(&cid),
             }
         }
@@ -340,6 +348,9 @@ where
                 continue;
             }
 
+            // Clean up wait window tracking
+            self.cid_first_have_time.remove(&cid);
+
             peer_state.wantlist.got_block(&cid);
             new_blocks.push((cid, block.clone()));
 
@@ -392,8 +403,11 @@ where
             }
         }
 
-        // Select optimal peers for each CID
+        // Select optimal peers for each CID (with wait window mechanism)
         let mut selected_peers_for_cid: FnvHashMap<CidGeneric<S>, Vec<PeerId>> = FnvHashMap::default();
+        let total_peers = self.peers.len();
+        let wait_window = Duration::from_millis(self.peer_selection_config.have_wait_window_ms);
+        let min_candidate_ratio = self.peer_selection_config.min_candidate_ratio;
 
         for (cid, candidate_peers) in &cid_to_candidates {
             let already_sent = *cid_already_sent_count.get(cid).unwrap_or(&0);
@@ -404,6 +418,37 @@ where
                 continue;
             }
 
+            // Wait window check: ensure we have enough candidates before selecting
+            let should_select = if let Some(first_have_time) = self.cid_first_have_time.get(cid) {
+                let elapsed = first_have_time.elapsed();
+                let min_candidates = ((total_peers as f64) * min_candidate_ratio).ceil() as usize;
+                
+                // Start selection if:
+                // 1. Wait window has elapsed, OR
+                // 2. We have enough candidates (>= min_candidate_ratio), OR
+                // 3. We have enough candidates to fill remaining slots
+                let window_elapsed = elapsed >= wait_window;
+                let enough_candidates = candidate_peers.len() >= min_candidates;
+                let can_fill_slots = candidate_peers.len() >= (top_n - already_sent);
+                
+                if !window_elapsed && !enough_candidates && !can_fill_slots {
+                    debug!(
+                        "CID {} - waiting for more candidates: elapsed={:?}, candidates={}, min={}, window={:?}",
+                        cid, elapsed, candidate_peers.len(), min_candidates, wait_window
+                    );
+                    false
+                } else {
+                    true
+                }
+            } else {
+                // No first Have time recorded yet (shouldn't happen, but be safe)
+                true
+            };
+            
+            if !should_select {
+                continue;
+            }
+            
             let need_to_select = top_n - already_sent;
 
             let candidates_with_metrics: FnvHashMap<PeerId, &PeerMetrics> = candidate_peers
@@ -422,8 +467,9 @@ where
             let selected = PeerSelector::select_top_peers(&candidates_with_metrics, &temp_config);
 
             debug!(
-                "CID {} - sent: {}, candidates: {}, selected: {} (top_n={})",
-                cid, already_sent, candidates_with_metrics.len(), selected.len(), top_n
+                "CID {} - sent: {}, candidates: {}, selected: {} (top_n={}, waited: {:?})",
+                cid, already_sent, candidates_with_metrics.len(), selected.len(), top_n,
+                self.cid_first_have_time.get(cid).map(|t| t.elapsed())
             );
 
             selected_peers_for_cid.insert(*cid, selected);

third_party/beetswap/src/peer_selection.rs

@@ -147,6 +147,12 @@ pub struct PeerSelectionConfig {
     pub min_peers: usize,
     /// Enable randomness in selection
     pub enable_randomness: bool,
+    /// Wait window in milliseconds after first Have response before selecting peers
+    /// This allows more peers to respond with Have, ensuring better selection
+    pub have_wait_window_ms: u64,
+    /// Minimum candidate ratio before starting selection (0.0 - 1.0)
+    /// Selection starts when: candidates >= total_peers * min_candidate_ratio
+    pub min_candidate_ratio: f64,
 }
 
 impl Default for PeerSelectionConfig {
@@ -156,6 +162,8 @@ impl Default for PeerSelectionConfig {
             enabled: true,
             min_peers: 2,
             enable_randomness: true,
+            have_wait_window_ms: 100,  // Wait 100ms for more Have responses
+            min_candidate_ratio: 0.3,   // Or start when 30% of peers responded
         }
     }
 }