Key deletion logic bugfix

Closes #121

chitchat-test/src/main.rs

@@ -99,7 +99,7 @@ async fn main() -> anyhow::Result<()> {
             dead_node_grace_period: Duration::from_secs(10),
             ..FailureDetectorConfig::default()
         },
-        marked_for_deletion_grace_period: 60,
+        marked_for_deletion_grace_period: Duration::from_secs(60),
     };
     let chitchat_handler = spawn_chitchat(config, Vec::new(), &UdpTransport).await?;
     let chitchat = chitchat_handler.chitchat();

chitchat/src/configuration.rs

@@ -15,14 +15,14 @@ pub struct ChitchatConfig {
     pub failure_detector_config: FailureDetectorConfig,
     // Marked for deletion grace period expressed as a number of hearbeats.
     // Chitchat ensures a key marked for deletion is eventually deleted by three mechanisms:
-    // - Garbage collection: each heartbeat, marked for deletion keys with `tombstone +
-    //   marked_for_deletion_grace_period > node.heartbeat` are deleted.
+    // - Garbage collection: each heartbeat, marked for deletion keys with `deletion now > instant +
+    //   marked_for_deletion_grace_period` are deleted.
     // - Compute delta: for a given node digest, if `node_digest.heartbeat +
     //   marked_for_deletion_grace_period < node_state.heartbeat` the node is flagged "to be reset"
     //   and the delta is populated with all keys and values.
     // - Apply delta: for a node flagged "to be reset", Chitchat will remove the node state and
     //   populate a fresh new node state with the keys and values present in the delta.
-    pub marked_for_deletion_grace_period: usize,
+    pub marked_for_deletion_grace_period: Duration,
 }
 
 impl ChitchatConfig {
@@ -37,7 +37,7 @@ impl ChitchatConfig {
             listen_addr,
             seed_nodes: Vec::new(),
             failure_detector_config: Default::default(),
-            marked_for_deletion_grace_period: 10_000,
+            marked_for_deletion_grace_period: Duration::from_secs(10_000),
         }
     }
 }
@@ -56,7 +56,8 @@ impl Default for ChitchatConfig {
             failure_detector_config: Default::default(),
             // Each heartbeat increments the version, with one heartbeat each second
             // 86400 ~ 24h.
-            marked_for_deletion_grace_period: 86400,
+            // TODO set that to something much lower.
+            marked_for_deletion_grace_period: Duration::from_secs(86_400),
         }
     }
 }

chitchat/src/delta.rs

@@ -1,5 +1,7 @@
 use std::collections::HashSet;
 
+use tokio::time::Instant;
+
 use crate::serialize::*;
 use crate::{ChitchatId, Heartbeat, VersionedValue};
 
@@ -116,7 +118,12 @@ impl Deserializable for DeltaOp {
                 let key = String::deserialize(buf)?;
                 let value = String::deserialize(buf)?;
                 let version = u64::deserialize(buf)?;
-                let tombstone = Option::<u64>::deserialize(buf)?;
+                let deleted = bool::deserialize(buf)?;
+                let tombstone = if deleted {
+                    Some(Instant::now())
+                } else {
+                    None
+                };
                 let versioned_value: VersionedValue = VersionedValue {
                     value,
                     version,
@@ -182,7 +189,7 @@ impl<'a> Serializable for DeltaOpRef<'a> {
                 key.serialize(buf);
                 versioned_value.value.serialize(buf);
                 versioned_value.version.serialize(buf);
-                versioned_value.tombstone.serialize(buf);
+                versioned_value.tombstone.is_some().serialize(buf);
             }
             Self::NodeToReset(chitchat_id) => {
                 buf.push(DeltaOpTag::NodeToReset.into());
@@ -204,7 +211,7 @@ impl<'a> Serializable for DeltaOpRef<'a> {
                 key.serialized_len()
                     + versioned_value.value.serialized_len()
                     + versioned_value.version.serialized_len()
-                    + versioned_value.tombstone.serialized_len()
+                    + 1
             }
             Self::NodeToReset(chitchat_id) => chitchat_id.serialized_len(),
         }
@@ -267,13 +274,18 @@ impl Delta {
         key: &str,
         value: &str,
         version: crate::Version,
-        tombstone: Option<u64>,
+        deleted: bool,
     ) {
         let node_delta = self
             .node_deltas
             .iter_mut()
             .find(|node_delta| &node_delta.chitchat_id == chitchat_id)
             .unwrap();
+        let tombstone = if deleted {
+            Some(Instant::now())
+        } else {
+            None
+        };
         node_delta.key_values.push((
             key.to_string(),
             VersionedValue {
@@ -488,7 +500,7 @@ mod tests {
             VersionedValue {
                 value: "".to_string(),
                 version: 2,
-                tombstone: Some(0),
+                tombstone: Some(Instant::now()),
             },
         ));
 
@@ -515,7 +527,7 @@ mod tests {
                 tombstone: None,
             },
         ));
-        test_aux_delta_writer(delta_writer, 99);
+        test_aux_delta_writer(delta_writer, 98);
     }
 
     #[test]

chitchat/src/lib.rs

@@ -122,7 +122,6 @@ impl Chitchat {
                     &digest,
                     delta_mtu,
                     &scheduled_for_deletion,
-                    self.config.marked_for_deletion_grace_period as u64,
                 );
                 Some(ChitchatMessage::SynAck {
                     digest: self_digest,
@@ -137,7 +136,6 @@ impl Chitchat {
                     &digest,
                     MAX_UDP_DATAGRAM_PAYLOAD_SIZE - 1,
                     &scheduled_for_deletion,
-                    self.config.marked_for_deletion_grace_period as u64,
                 );
                 Some(ChitchatMessage::Ack { delta })
             }
@@ -155,7 +153,7 @@ impl Chitchat {
     fn gc_keys_marked_for_deletion(&mut self) {
         let dead_nodes = self.dead_nodes().cloned().collect::<HashSet<_>>();
         self.cluster_state.gc_keys_marked_for_deletion(
-            self.config.marked_for_deletion_grace_period as u64,
+            self.config.marked_for_deletion_grace_period,
             &dead_nodes,
         );
     }
@@ -402,7 +400,7 @@ mod tests {
                 initial_interval: Duration::from_millis(100),
                 ..Default::default()
             },
-            marked_for_deletion_grace_period: 10_000,
+            marked_for_deletion_grace_period: Duration::from_secs(3_600),
         };
         let initial_kvs: Vec<(String, String)> = Vec::new();
         spawn_chitchat(config, initial_kvs, transport)

chitchat/src/message.rs

@@ -165,12 +165,12 @@ mod tests {
             // +37 bytes = 27 bytes (ChitchatId) + 2 bytes (node delta len) + 8 bytes (heartbeat).
             delta.add_node(node.clone(), Heartbeat(0));
             // +29 bytes.
-            delta.add_kv(&node, "key", "value", 0, Some(5));
-            delta.set_serialized_len(70);
+            delta.add_kv(&node, "key", "value", 0, true);
+            delta.set_serialized_len(62);
 
             let syn_ack = ChitchatMessage::SynAck { digest, delta };
             // 1 bytes (syn ack message) + 45 bytes (digest) + 69 bytes (delta).
-            test_serdeser_aux(&syn_ack, 116);
+            test_serdeser_aux(&syn_ack, 108);
         }
     }
 
@@ -188,10 +188,10 @@ mod tests {
             // +37 bytes = 27 bytes (ChitchatId) + 2 bytes (node delta len) + 8 bytes (heartbeat).
             delta.add_node(node.clone(), Heartbeat(0));
             // +29 bytes.
-            delta.add_kv(&node, "key", "value", 0, Some(5));
-            delta.set_serialized_len(70);
+            delta.add_kv(&node, "key", "value", 0, true);
+            delta.set_serialized_len(62);
             let ack = ChitchatMessage::Ack { delta };
-            test_serdeser_aux(&ack, 71);
+            test_serdeser_aux(&ack, 63);
         }
     }