forget previous generation

chitchat-test/src/main.rs

@@ -1,6 +1,6 @@
 use std::net::SocketAddr;
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, SystemTime};
 
 use chitchat::transport::UdpTransport;
 use chitchat::{spawn_chitchat, Chitchat, ChitchatConfig, ChitchatId, FailureDetectorConfig};
@@ -84,15 +84,22 @@ async fn main() -> anyhow::Result<()> {
     let node_id = opt
         .node_id
         .unwrap_or_else(|| generate_server_id(public_addr));
-    let chitchat_id = ChitchatId::new(node_id, 0, public_addr);
+    let generation = SystemTime::now()
+        .duration_since(SystemTime::UNIX_EPOCH)
+        .unwrap()
+        .as_secs();
+    let chitchat_id = ChitchatId::new(node_id, generation, public_addr);
     let config = ChitchatConfig {
         cluster_id: "testing".to_string(),
         chitchat_id,
         gossip_interval: Duration::from_millis(opt.interval),
         listen_addr: opt.listen_addr,
         seed_nodes: opt.seeds.clone(),
-        failure_detector_config: FailureDetectorConfig::default(),
-        marked_for_deletion_grace_period: 10_000,
+        failure_detector_config: FailureDetectorConfig {
+            dead_node_grace_period: Duration::from_secs(10),
+            ..FailureDetectorConfig::default()
+        },
+        marked_for_deletion_grace_period: 60,
     };
     let chitchat_handler = spawn_chitchat(config, Vec::new(), &UdpTransport).await?;
     let chitchat = chitchat_handler.chitchat();

chitchat/src/failure_detector.rs

@@ -48,6 +48,19 @@ impl FailureDetector {
         heartbeat_window.report_heartbeat();
     }
 
+    pub fn report_unknown(&mut self, chitchat_id: &ChitchatId) {
+        debug!(node_id=%chitchat_id.node_id, "reporting unknown node heartbeat.");
+        self.node_samples
+            .entry(chitchat_id.clone())
+            .or_insert_with(|| {
+                SamplingWindow::new(
+                    self.config.sampling_window_size,
+                    self.config.max_interval,
+                    self.config.initial_interval,
+                )
+            });
+    }
+
     /// Marks the node as dead or alive based on the current phi value.
     pub fn update_node_liveness(&mut self, chitchat_id: &ChitchatId) {
         if let Some(phi) = self.phi(chitchat_id) {
@@ -181,10 +194,14 @@ impl SamplingWindow {
 
     /// Computes the sampling window's phi value.
     pub fn phi(&self) -> f64 {
-        // Ensure we don't call before any sample arrival.
-        assert!(self.intervals.mean() > 0.0 && self.last_heartbeat.is_some());
-        let elapsed_time = self.last_heartbeat.unwrap().elapsed().as_secs_f64();
-        elapsed_time / self.intervals.mean()
+        if self.last_heartbeat.is_none() {
+            // if we phi is called before we have a sample, we assume the node isn't really alive.
+            f64::INFINITY
+        } else {
+            assert!(self.intervals.mean() > 0.0);
+            let elapsed_time = self.last_heartbeat.unwrap().elapsed().as_secs_f64();
+            elapsed_time / self.intervals.mean()
+        }
     }
 }
 

chitchat/src/lib.rs

@@ -166,6 +166,8 @@ impl Chitchat {
                 {
                     self.failure_detector.report_heartbeat(chitchat_id);
                 }
+            } else {
+                self.failure_detector.report_unknown(chitchat_id);
             }
         }
     }