Detect and remove surplus events at start-up #332

Changelog.md

@@ -3,6 +3,15 @@
 Please see [here](https://github.com/NLnetLabs/krill/projects?query=is%3Aopen+sort%3Aname-asc)
 for planned releases.
 
+## 0.8.0
+
+
+Breaking API changes:
+
+The 
+
+
+
 ## 0.8.0-rc1 'Festina Lente'
 
 As of now we will use release candidates as part of the Krill release process. If no major issues

src/commons/api/history.rs

@@ -292,6 +292,8 @@ pub struct CommandHistoryCriteria {
     #[serde(skip_serializing_if = "Option::is_none")]
     after: Option<i64>,
     #[serde(skip_serializing_if = "Option::is_none")]
+    after_sequence: Option<u64>,
+    #[serde(skip_serializing_if = "Option::is_none")]
     label_includes: Option<Vec<String>>,
     #[serde(skip_serializing_if = "Option::is_none")]
     label_excludes: Option<Vec<String>>,
@@ -318,6 +320,10 @@ impl CommandHistoryCriteria {
         self.before = Some(timestamp);
     }
 
+    pub fn set_after_sequence(&mut self, sequence: u64) {
+        self.after_sequence = Some(sequence)
+    }
+
     pub fn set_rows(&mut self, rows: usize) {
         self.rows_limit = Some(rows);
     }
@@ -344,6 +350,13 @@ impl CommandHistoryCriteria {
         true
     }
 
+    pub fn matches_sequence(&self, sequence: u64) -> bool {
+        match self.after_sequence {
+            None => true,
+            Some(seq_crit) => sequence > seq_crit,
+        }
+    }
+
     #[allow(clippy::ptr_arg)]
     pub fn matches_label(&self, label: &Label) -> bool {
         if let Some(includes) = &self.label_includes {
@@ -374,6 +387,7 @@ impl Default for CommandHistoryCriteria {
         CommandHistoryCriteria {
             before: None,
             after: None,
+            after_sequence: None,
             label_includes: None,
             label_excludes: None,
             offset: 0,

src/commons/eventsourcing/store.rs

@@ -86,7 +86,9 @@ impl CommandKey {
     }
 
     pub fn matches_crit(&self, crit: &CommandHistoryCriteria) -> bool {
-        crit.matches_timestamp_secs(self.timestamp_secs) && crit.matches_label(&self.label)
+        crit.matches_timestamp_secs(self.timestamp_secs)
+            && crit.matches_label(&self.label)
+            && crit.matches_sequence(self.sequence)
     }
 }
 
@@ -183,7 +185,37 @@ where
         for handle in self.list()? {
             let _ = self
                 .get_latest(&handle)
-                .map_err(|e| AggregateStoreError::WarmupFailed(handle, e.to_string()))?;
+                .map_err(|e| AggregateStoreError::WarmupFailed(handle.clone(), e.to_string()))?;
+
+            // check that last command and event are consistent with
+            // the info, if not fail warmup and force recover
+            let info = self.get_info(&handle)?;
+
+            // for events we can just check if the next event, after
+            // the last event in the info exists
+            if self.get_event::<A::Event>(&handle, info.last_event + 1)?.is_some() {
+                return Err(AggregateStoreError::WarmupFailed(
+                    handle.clone(),
+                    format!(
+                        "Additional event(s) found after version: {}. Force recover.",
+                        info.last_event
+                    ),
+                ));
+            }
+
+            // Check if there are any commands with a sequence after the last
+            // recorded sequence in the info.
+            let mut crit = CommandHistoryCriteria::default();
+            crit.set_after_sequence(info.last_command);
+            if !self.command_keys_ascending(&handle, &crit)?.is_empty() {
+                return Err(AggregateStoreError::WarmupFailed(
+                    handle,
+                    format!(
+                        "Additional commands(s) found after version: {}. Force recover.",
+                        info.last_command
+                    ),
+                ));
+            }
         }
         Ok(())
     }

tests/history.rs

@@ -19,14 +19,11 @@ async fn history() {
     use krill::commons::crypto::KrillSigner;
     use krill::commons::util::file;
     use krill::daemon::ca::CaServer;
-    use krill::daemon::config::CONFIG;
     use krill::daemon::mq::EventQueueListener;
     use krill::test::*;
 
     const KRILL_HISTORY_JSON_GENERATE: &str = "KRILL_HISTORY_JSON_GENERATE";
 
-    CONFIG.init_logging().unwrap();
-
     // Run this test with ENV variable KRILL_HISTORY_JSON_GENERATE = 1 in order to generate json
     // for missing files
     assert_scenario("ca_embedded", &["ta", "child"]).await;