feat(storage-manager): add assertion on Replication Log in Debug

This commit introduces an assertion test every time an Event in inserted
in the Replication Log ONLY IN DEBUG.

The purpose is to make sure that the invariant of the Replication Log is
upheld: it should only contain one Event per key expression.

* plugins/zenoh-plugin-storage-manager/src/replication/classification.rs:
  added the method `assert_only_one_event_per_key_expr` to the
  `Interval` and `SubInterval` structures.

* plugins/zenoh-plugin-storage-manager/src/replication/log.rs: added the
  method `assert_only_one_event_per_key_expr` to the `LogLatest`
  structure.

Signed-off-by: Julien Loudet <[email protected]>

plugins/zenoh-plugin-storage-manager/src/replication/classification.rs

@@ -13,7 +13,7 @@
 //
 
 use std::{
-    collections::HashMap,
+    collections::{HashMap, HashSet},
     ops::{Deref, Sub},
 };
 
@@ -91,6 +91,26 @@ impl<const N: usize> From<[(SubIntervalIdx, SubInterval); N]> for Interval {
 }
 
 impl Interval {
+    /// Returns true if the Replication Log only contains a single Event for each key expression.
+    ///
+    /// To perform that check a HashSet is constructed by visiting each Interval and each
+    /// SubInterval, filling the HashSet with the key expression of all the Events contained.
+    ///
+    /// ⚠️ This method will only be called if Zenoh is compiled in Debug mode.
+    #[cfg(debug_assertions)]
+    pub(crate) fn assert_only_one_event_per_key_expr(
+        &self,
+        events: &mut HashSet<Option<OwnedKeyExpr>>,
+    ) -> bool {
+        for sub_interval in self.sub_intervals.values() {
+            if !sub_interval.assert_only_one_event_per_key_expr(events) {
+                return false;
+            }
+        }
+
+        true
+    }
+
     /// Returns the [Fingerprint] of this [Interval].
     ///
     /// The [Fingerprint] of an [Interval] is equal to the XOR (exclusive or) of the fingerprints
@@ -229,6 +249,30 @@ impl<const N: usize> From<[Event; N]> for SubInterval {
 }
 
 impl SubInterval {
+    /// Returns true if the Replication Log only contains a single Event for each key expression.
+    ///
+    /// To perform that check a HashSet is constructed by visiting each Interval and each
+    /// SubInterval, filling the HashSet with the key expression of all the Events contained.
+    ///
+    /// ⚠️ This method will only be called if Zenoh is compiled in Debug mode.
+    #[cfg(debug_assertions)]
+    fn assert_only_one_event_per_key_expr(
+        &self,
+        events: &mut HashSet<Option<OwnedKeyExpr>>,
+    ) -> bool {
+        for event_ke in self.events.keys() {
+            if !events.insert(event_ke.clone()) {
+                tracing::error!(
+                    "FATAL ERROR, REPLICATION LOG INVARIANT VIOLATED, KEY APPEARS MULTIPLE TIMES: \
+                     < {event_ke:?} >"
+                );
+                return false;
+            }
+        }
+
+        true
+    }
+
     /// Inserts the [Event], regardless of its [Timestamp].
     ///
     /// This method also updates the fingerprint of the [SubInterval].

plugins/zenoh-plugin-storage-manager/src/replication/log.rs

@@ -12,7 +12,7 @@
 //   ZettaScale Zenoh Team, <[email protected]>
 //
 
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 
 use bloomfilter::Bloom;
 use serde::{Deserialize, Serialize};
@@ -170,6 +170,24 @@ pub struct LogLatest {
 }
 
 impl LogLatest {
+    /// Returns true if the Replication Log only contains a single Event for each key expression.
+    ///
+    /// To perform that check a HashSet is constructed by visiting each Interval and each
+    /// SubInterval, filling the HashSet with the key expression of all the Events contained.
+    ///
+    /// ⚠️ This method will only be called if Zenoh is compiled in Debug mode.
+    #[cfg(debug_assertions)]
+    pub(crate) fn assert_only_one_event_per_key_expr(&self) -> bool {
+        let mut hash_set = HashSet::new();
+        for interval in self.intervals.values() {
+            if !interval.assert_only_one_event_per_key_expr(&mut hash_set) {
+                return false;
+            }
+        }
+
+        true
+    }
+
     /// Creates a new [LogLatest] configured with the provided [ReplicaConfig].
     pub fn new(
         storage_key_expr: OwnedKeyExpr,
@@ -270,6 +288,9 @@ impl LogLatest {
             .entry(interval_idx)
             .or_default()
             .insert_unchecked(sub_interval_idx, event);
+
+        #[cfg(debug_assertions)]
+        assert!(self.assert_only_one_event_per_key_expr());
     }
 
     /// Removes, if there is one, the previous event from the Replication Log for the provided key