Makes it possible to register batching listeners

Upon reset_node and apply_update, we want to be
able to react to a batch of changes rather than individual changes.

In quickwit for instance, we want to be able to react to the reception of a batch of deleted shard and group this reaction into a single metastore call.

Author: fulmicoton

chitchat-test/src/main.rs

@@ -37,8 +37,7 @@ impl Api {
     #[oai(path = "/set_kv/", method = "get")]
     async fn set_kv(&self, key: Query<String>, value: Query<String>) -> Json<serde_json::Value> {
         let mut chitchat_guard = self.chitchat.lock().await;
-
-        let cc_state = chitchat_guard.self_node_state();
+        let mut cc_state = chitchat_guard.self_node_state();
         cc_state.set(key.as_str(), value.as_str());
 
         Json(serde_json::to_value(&SetKeyValueResponse { status: true }).unwrap())
@@ -48,8 +47,7 @@ impl Api {
     #[oai(path = "/mark_for_deletion/", method = "get")]
     async fn mark_for_deletion(&self, key: Query<String>) -> Json<serde_json::Value> {
         let mut chitchat_guard = self.chitchat.lock().await;
-
-        let cc_state = chitchat_guard.self_node_state();
+        let mut cc_state = chitchat_guard.self_node_state();
         cc_state.delete(key.as_str());
         Json(serde_json::to_value(&SetKeyValueResponse { status: true }).unwrap())
     }

chitchat/src/delta.rs

@@ -330,6 +330,30 @@ impl NodeDelta {
     }
 }
 
+impl NodeDelta {
+    pub(crate) fn applicable_key_value_mutations(
+        self,
+        max_version: u64,
+        last_gc_version: u64,
+    ) -> impl Iterator<Item = KeyValueMutation> {
+        self.key_values
+            .into_iter()
+            .filter(move |key_value_mutation| {
+                if key_value_mutation.version <= max_version {
+                    // We already know about this KV.
+                    return false;
+                }
+                if key_value_mutation.status.scheduled_for_deletion() {
+                    // This KV has already been GCed.
+                    if key_value_mutation.version <= last_gc_version {
+                        return false;
+                    }
+                }
+                true
+            })
+    }
+}
+
 #[derive(Default)]
 struct DeltaBuilder {
     existing_nodes: HashSet<ChitchatId>,