feat(lattica): infra reconnect and beetswap not request all peers (#26)

Author: JasonOE

bindings/python/Cargo.lock

@@ -14,6 +14,15 @@ def main():
             .with_bootstraps(bootstrap_nodes) \
             .build()
 
+        lattica.configure_bitswap_peer_selection(
+            top_n=3,
+            enabled=True,
+            min_peers=2,
+            enable_randomness=True,
+            have_wait_window_ms=100,
+            min_candidate_ratio=0.3
+        )
+
         # wait for connection
         time.sleep(1)
 
@@ -26,6 +35,8 @@ def main():
             data = lattica.get_block(request_cid)
             print(f"data: {data}")
 
+            lattica.print_bitswap_stats()
+
         else:
             # put block
             cid = lattica.put_block(b'hello')

bindings/python/examples/bitswap.py

@@ -253,6 +253,103 @@ def is_symmetric_nat(self):
         except Exception as e:
             raise RuntimeError(f"Failed to check is_symmetric_nat error: {e}")
 
+    def configure_bitswap_peer_selection(
+        self,
+        top_n: int = 3,
+        enabled: bool = True,
+        min_peers: int = 2,
+        enable_randomness: bool = True,
+        have_wait_window_ms: int = 100,
+        min_candidate_ratio: float = 0.3
+    ) -> None:
+        """Configure Bitswap peer selection strategy.
+        
+        Args:
+            top_n: Number of top peers to select
+            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, 
+                have_wait_window_ms, min_candidate_ratio
+            )
+        except Exception as e:
+            raise RuntimeError(f"Failed to configure bitswap peer selection: {e}")
+
+    def get_bitswap_global_stats(self) -> dict:
+        """Get Bitswap global statistics.
+        
+        Returns:
+            dict with total_requests, successful_requests, failed_requests, total_bytes_received
+        """
+        try:
+            return self._lattica_instance.get_bitswap_global_stats()
+        except Exception as e:
+            raise RuntimeError(f"Failed to get bitswap global stats: {e}")
+
+    def get_bitswap_peer_rankings(self) -> List[dict]:
+        """Get Bitswap peer rankings with detailed metrics.
+        
+        Returns:
+            List of dicts sorted by score descending, each containing:
+            - peer_id: Peer ID string
+            - score: Composite score (0-100)
+            - blocks_received: Number of successfully received blocks
+            - failures: Number of failures
+            - success_rate: Success rate (0.0-1.0)
+            - avg_speed: Average speed in bytes/sec
+        """
+        try:
+            return self._lattica_instance.get_bitswap_peer_rankings()
+        except Exception as e:
+            raise RuntimeError(f"Failed to get bitswap peer rankings: {e}")
+
+    def print_bitswap_stats(self) -> None:
+        """Print Bitswap stats report to stdout.
+        
+        This method combines get_bitswap_global_stats() and get_bitswap_peer_rankings()
+        to display a formatted statistics report in Python.
+        """
+        try:
+            # Get global stats
+            stats = self.get_bitswap_global_stats()
+            total_requests = stats.get('successful_requests', 0) + stats.get('failed_requests', 0)
+            success_rate = (stats.get('successful_requests', 0) / total_requests * 100) if total_requests > 0 else 0.0
+            bytes_mb = stats.get('total_bytes_received', 0) / (1024 * 1024)
+            
+            print(f"\n{'='*100}")
+            print("Bitswap Statistics Report")
+            print(f"{'='*100}")
+            print(f"  Total Requests:      {total_requests}")
+            print(f"  Successful:          {stats.get('successful_requests', 0)}")
+            print(f"  Failed:              {stats.get('failed_requests', 0)}")
+            print(f"  Success Rate:        {success_rate:.2f}%")
+            print(f"  Total Received:      {bytes_mb:.2f} MB")
+            
+            # Get peer rankings with details
+            rankings = self.get_bitswap_peer_rankings()
+            if rankings:
+                print(f"\nTop {min(len(rankings), 10)} Peers:")
+                print(f"  {'#':<3} {'Peer ID':<54} {'Score':>7} {'Success':>8} {'Fail':>5} {'Rate':>7} {'Speed':>12}")
+                print(f"  {'-'*3} {'-'*54} {'-'*7} {'-'*8} {'-'*5} {'-'*7} {'-'*12}")
+                for i, peer in enumerate(rankings[:10], 1):
+                    peer_id = peer['peer_id']
+                    display_id = peer_id[:52] + ".." if len(peer_id) > 54 else peer_id
+                    speed_mb = peer['avg_speed'] / (1024 * 1024)
+                    rate_pct = peer['success_rate'] * 100
+                    print(f"  {i:<3} {display_id:<54} {peer['score']:>7.2f} {peer['blocks_received']:>8} {peer['failures']:>5} {rate_pct:>6.1f}% {speed_mb:>9.2f}MB/s")
+            else:
+                print("\n  No peer data available.")
+            print(f"{'='*100}\n")
+        except Exception as e:
+            raise RuntimeError(f"Failed to print bitswap stats: {e}")
+
     def __enter__(self):
         self._ensure_initialized()
         return self

bindings/python/py_src/lattica/client.py

@@ -4,7 +4,7 @@ build-backend = "maturin"
 
 [project]
 name = "lattica"
-version = "1.0.18"
+version = "1.0.19"
 description = "A unified Python SDK for P2P networking with integrated DHT and NAT capabilities"
 readme = "README.md"
 license = {text = "MIT"}
@@ -127,4 +127,4 @@ exclude_lines = [
     "if __name__ == .__main__.:",
     "class .*\\bProtocol\\):",
     "@(abc\\.)?abstractmethod",
-]
+]

bindings/python/pyproject.toml

@@ -1,7 +1,7 @@
 use lattica::{network, rpc, common};
 use std::sync::{Arc};
 use tokio::sync::{Mutex};
-use pyo3::{prelude::*, types::PyDict, IntoPyObjectExt};
+use pyo3::{prelude::*, types::{PyDict, PyList}, IntoPyObjectExt};
 use tokio::runtime::Runtime;
 use libp2p::{Multiaddr, PeerId};
 use async_trait::async_trait;
@@ -557,6 +557,87 @@ impl LatticaSDK {
     fn is_symmetric_nat(&self) -> PyResult<Option<bool>> {
         self.lattica.is_symmetric_nat().map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!("{:?}", e)))
     }
+
+    /// Configure Bitswap peer selection strategy
+    /// 
+    /// 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 = lattica::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)))
+                })
+            })
+        })
+    }
+
+    /// Get Bitswap global statistics
+    fn get_bitswap_global_stats(&self, py: Python) -> PyResult<PyObject> {
+        py.allow_threads(|| {
+            self.runtime.block_on(async move {
+                let stats = self.lattica.get_bitswap_global_stats().await
+                    .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!("{:?}", e)))?;
+
+                Python::with_gil(|py| {
+                    let dict = PyDict::new(py);
+                    dict.set_item("total_requests", stats.total_requests)?;
+                    dict.set_item("successful_requests", stats.successful_requests)?;
+                    dict.set_item("failed_requests", stats.failed_requests)?;
+                    dict.set_item("total_bytes_received", stats.total_bytes_received)?;
+                    Ok(dict.into_py_any(py)?)
+                })
+            })
+        })
+    }
+
+    /// Get Bitswap peer rankings with detailed metrics
+    fn get_bitswap_peer_rankings(&self, py: Python) -> PyResult<PyObject> {
+        py.allow_threads(|| {
+            self.runtime.block_on(async move {
+                let details = self.lattica.get_bitswap_peer_rankings().await
+                    .map_err(|e| PyErr::new::<pyo3::exceptions::PyRuntimeError, _>(format!("{:?}", e)))?;
+                
+                Python::with_gil(|py| {
+                    let list = PyList::empty(py);
+                    for detail in details {
+                        let dict = PyDict::new(py);
+                        dict.set_item("peer_id", detail.peer_id)?;
+                        dict.set_item("score", detail.score)?;
+                        dict.set_item("blocks_received", detail.blocks_received)?;
+                        dict.set_item("failures", detail.failures)?;
+                        dict.set_item("success_rate", detail.success_rate)?;
+                        dict.set_item("avg_speed", detail.avg_speed)?;
+                        list.append(dict)?;
+                    }
+                    Ok(list.into_py_any(py)?)
+                })
+            })
+        })
+    }
 }
 
 #[pymethods]

bindings/python/src/core.rs

@@ -1,4 +1,7 @@
 pub mod common;
 pub mod network;
 pub mod rpc;
-pub mod fs_blockstore;
+pub mod fs_blockstore;
+
+// Re-export beetswap types for external use
+pub use beetswap::PeerSelectionConfig;

lattica/Cargo.lock

@@ -271,4 +271,26 @@ impl LatticaBehaviour{
             tx.send(Err(anyhow!("Kad is not enabled"))).ok();
         }
     }
+
+    /// Configure Bitswap peer selection strategy
+    pub fn configure_bitswap_peer_selection(&mut self, config: beetswap::PeerSelectionConfig) {
+        if let Some(bitswap) = self.bitswap.as_mut() {
+            bitswap.set_peer_selection_config(config);
+        }
+    }
+
+    /// Get Bitswap peer selection config
+    pub fn get_bitswap_peer_selection_config(&self) -> Option<beetswap::PeerSelectionConfig> {
+        self.bitswap.as_ref().map(|b| b.get_peer_selection_config().clone())
+    }
+
+    /// Get Bitswap global stats
+    pub fn get_bitswap_global_stats(&self) -> Option<beetswap::GlobalStats> {
+        self.bitswap.as_ref().map(|b| b.get_global_stats().clone())
+    }
+
+    /// Get Bitswap peer rankings with detailed metrics
+    pub fn get_bitswap_peer_rankings(&self) -> Vec<beetswap::PeerDetail> {
+        self.bitswap.as_ref().map(|b| b.get_peer_rankings()).unwrap_or_default()
+    }
 }