Merge pull request #141 from dusk-network/removal-ping

Change maintainer to ping nodes while removal

CHANGELOG.md

@@ -15,6 +15,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - Change 'need_bootstrappers' to have dinamically threshold [#135]
 - Change `BinaryID::from_nonce` to return result [#136]
+- Change maintainer to ping nodes while removal [#138]
 
 ## [0.6.0] - 2023-11-01
 
@@ -137,6 +138,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 [#123]: https://github.com/dusk-network/kadcast/issues/123
 [#135]: https://github.com/dusk-network/kadcast/issues/135
 [#136]: https://github.com/dusk-network/kadcast/issues/136
+[#138]: https://github.com/dusk-network/kadcast/issues/138
 
 <!-- Releases -->
 

src/kbucket.rs

@@ -106,7 +106,7 @@ impl<V> Tree<V> {
         })
     }
 
-    //pick at most Alpha nodes for each idle bucket
+    // pick at most Alpha nodes for each idle bucket
     pub(crate) fn idle_buckets(
         &self,
     ) -> impl Iterator<Item = (BucketHeight, impl Iterator<Item = &Node<V>>)>
@@ -134,6 +134,10 @@ impl<V> Tree<V> {
         })
     }
 
+    pub(crate) fn idle_nodes(&self) -> impl Iterator<Item = &Node<V>> {
+        self.buckets.iter().flat_map(|(_, b)| b.idle_nodes())
+    }
+
     pub(crate) fn remove_idle_nodes(&mut self) {
         self.buckets
             .iter_mut()

src/kbucket/bucket.rs

@@ -231,6 +231,12 @@ impl<V> Bucket<V> {
         self.nodes.iter().filter(move |&n| n.is_alive(ttl))
     }
 
+    /// Get idle nodes from the bucket.
+    pub(crate) fn idle_nodes(&self) -> impl Iterator<Item = &Node<V>> {
+        let ttl = self.bucket_config.node_ttl;
+        self.nodes.iter().filter(move |n| n.is_alive(ttl))
+    }
+
     /// Checks if the bucket contains a node with the given peer key.
     pub(crate) fn has_node(&self, peer: &BinaryKey) -> bool {
         self.nodes.iter().any(|n| n.id().as_binary() == peer)

src/maintainer.rs

@@ -91,7 +91,7 @@ impl TableMaintainer {
 
     /// This is the main function of this utility class. It's responsible to:
     /// 1. Contact bootstrappers (if needed)
-    /// 2. Ping idle buckets
+    /// 2. Find new node for idle buckets
     /// 3. Remove idles nodes from buckets
     async fn monitor_buckets(&self, idle_time: Duration, min_peers: usize) {
         info!("TableMaintainer::monitor_buckets started");
@@ -102,16 +102,30 @@ impl TableMaintainer {
             tokio::time::sleep(idle_time).await;
 
             info!("TableMaintainer::monitor_buckets woke up");
-            self.ping_idle_buckets().await;
+            self.find_new_nodes().await;
 
             info!("TableMaintainer::monitor_buckets removing idle nodes");
-            self.ktable.write().await.remove_idle_nodes();
+            self.ping_and_remove_idles().await;
         }
     }
 
     /// Search for idle buckets (no message received) and try to contact some of
     /// the belonging nodes
-    async fn ping_idle_buckets(&self) {
+    async fn ping_and_remove_idles(&self) {
+        let idles = self
+            .ktable
+            .read()
+            .await
+            .idle_nodes()
+            .map(|n| *n.value().address())
+            .collect();
+        self.send((Message::Ping(self.header), idles)).await;
+        self.ktable.write().await.remove_idle_nodes();
+    }
+
+    /// Search for idle buckets (no message received) and try to contact some of
+    /// the belonging nodes
+    async fn find_new_nodes(&self) {
         let table_lock_read = self.ktable.read().await;
 
         let find_node_messages = table_lock_read