Fix: Do not report snapshot.last_log_id to metrics until snapshot is …

…finished building/installing

Before this commit `RaftMetrics.snapshot` contains the last log id of a
snapshot that is **going** to install. Therefore there is chance the
metrics is updated but the store does not.

In this commit, `RaftMetrics.snapshot` will only be updated when a
snapshot is finished building or installing, by adding a new field
`snpashot` to `IOState` for tracking persisted snapshot meta data.

- Fix: #912

openraft/src/core/raft_core.rs

@@ -512,7 +512,7 @@ where
             vote: *st.io_state().vote(),
             last_log_index: st.last_log_id().index(),
             last_applied: st.io_applied().copied(),
-            snapshot: st.snapshot_meta.last_log_id,
+            snapshot: st.io_snapshot_last_log_id().copied(),
             purged: st.io_purged().copied(),
 
             // --- cluster ---
@@ -1301,7 +1301,15 @@ where
                             meta.summary(),
                             func_name!()
                         );
+
+                        // Update in-memory state first, then the io state.
+                        // In-memory state should always be ahead or equal to the io state.
+
+                        let last_log_id = meta.last_log_id;
                         self.engine.finish_building_snapshot(meta);
+
+                        let st = self.engine.state.io_state_mut();
+                        st.update_snapshot(last_log_id);
                     }
                     sm::Response::ReceiveSnapshotChunk(_) => {
                         tracing::info!("sm::StateMachine command done: ReceiveSnapshotChunk: {}", func_name!());
@@ -1314,7 +1322,9 @@ where
                         );
 
                         if let Some(meta) = meta {
-                            self.engine.state.io_state_mut().update_applied(meta.last_log_id);
+                            let st = self.engine.state.io_state_mut();
+                            st.update_applied(meta.last_log_id);
+                            st.update_snapshot(meta.last_log_id);
                         }
                     }
                     sm::Response::Apply(res) => {

openraft/src/core/sm/mod.rs

@@ -227,6 +227,7 @@ where
             let cmd_res = CommandResult::new(seq, res);
             let _ = resp_tx.send(Notify::sm(cmd_res));
         });
+        tracing::info!("{} returning; spawned building snapshot task", func_name!());
     }
 
     #[tracing::instrument(level = "info", skip_all)]

openraft/src/progress/entry/tests.rs

@@ -121,6 +121,10 @@ impl LogStateReader<u64> for LogState {
         todo!()
     }
 
+    fn io_snapshot_last_log_id(&self) -> Option<&LogId<u64>> {
+        todo!()
+    }
+
     fn io_purged(&self) -> Option<&LogId<u64>> {
         todo!()
     }

openraft/src/raft_state/io_state.rs

@@ -17,14 +17,25 @@ pub(crate) struct LogIOId<NID: NodeId> {
 ///
 /// These states are updated only when the io complete and thus may fall behind to the state stored
 /// in [`RaftState`](`crate::RaftState`),.
+///
+/// The log ids that are tracked includes:
+///
+/// ```text
+/// | log ids
+/// | *------------+---------+---------+---------+------------------>
+/// |              |         |         |         `---> flushed
+/// |              |         |         `-------------> applied
+/// |              |         `-----------------------> snapshot
+/// |              `---------------------------------> purged
+/// ```
 #[derive(Debug, Clone, Copy)]
 #[derive(Default)]
 #[derive(PartialEq, Eq)]
 pub(crate) struct IOState<NID: NodeId> {
     /// Whether it is building a snapshot
     building_snapshot: bool,
 
-    // The last flushed vote.
+    /// The last flushed vote.
     pub(crate) vote: Vote<NID>,
 
     /// The last log id that has been flushed to storage.
@@ -33,6 +44,9 @@ pub(crate) struct IOState<NID: NodeId> {
     /// The last log id that has been applied to state machine.
     pub(crate) applied: Option<LogId<NID>>,
 
+    /// The last log id in the currently persisted snapshot.
+    pub(crate) snapshot: Option<LogId<NID>>,
+
     /// The last log id that has been purged from storage.
     ///
     /// `RaftState::last_purged_log_id()`
@@ -46,13 +60,15 @@ impl<NID: NodeId> IOState<NID> {
         vote: Vote<NID>,
         flushed: LogIOId<NID>,
         applied: Option<LogId<NID>>,
+        snapshot: Option<LogId<NID>>,
         purged: Option<LogId<NID>>,
     ) -> Self {
         Self {
             building_snapshot: false,
             vote,
             flushed,
             applied,
+            snapshot,
             purged,
         }
     }
@@ -83,6 +99,23 @@ impl<NID: NodeId> IOState<NID> {
         self.applied.as_ref()
     }
 
+    pub(crate) fn update_snapshot(&mut self, log_id: Option<LogId<NID>>) {
+        tracing::debug!(snapshot = display(DisplayOption(&log_id)), "{}", func_name!());
+
+        debug_assert!(
+            log_id > self.snapshot,
+            "snapshot log id should be monotonically increasing: current: {:?}, update: {:?}",
+            self.snapshot,
+            log_id
+        );
+
+        self.snapshot = log_id;
+    }
+
+    pub(crate) fn snapshot(&self) -> Option<&LogId<NID>> {
+        self.snapshot.as_ref()
+    }
+
     pub(crate) fn set_building_snapshot(&mut self, building: bool) {
         self.building_snapshot = building;
     }

openraft/src/raft_state/log_state_reader.rs

@@ -54,6 +54,11 @@ pub(crate) trait LogStateReader<NID: NodeId> {
     /// This is actually happened io-state which might fall behind committed log id.
     fn io_applied(&self) -> Option<&LogId<NID>>;
 
+    /// The last log id in the last persisted snapshot.
+    ///
+    /// This is actually happened io-state which might fall behind `Self::snapshot_last_log_id()`.
+    fn io_snapshot_last_log_id(&self) -> Option<&LogId<NID>>;
+
     /// The last known purged log id, inclusive.
     ///
     /// This is actually purged log id from storage.

openraft/src/raft_state/mod.rs

@@ -115,6 +115,10 @@ where
         self.io_state.applied()
     }
 
+    fn io_snapshot_last_log_id(&self) -> Option<&LogId<NID>> {
+        self.io_state.snapshot()
+    }
+
     fn io_purged(&self) -> Option<&LogId<NID>> {
         self.io_state.purged()
     }

openraft/src/storage/helper.rs

@@ -76,9 +76,6 @@ where
 
         let log_ids = LogIdList::load_log_ids(last_purged_log_id, last_log_id, self.log_store).await?;
 
-        // TODO: `flushed` is not set.
-        let io_state = IOState::new(vote, LogIOId::default(), last_applied, last_purged_log_id);
-
         let snapshot = self.state_machine.get_current_snapshot().await?;
 
         // If there is not a snapshot and there are logs purged, which means the snapshot is not persisted,
@@ -97,6 +94,15 @@ where
         };
         let snapshot_meta = snapshot.map(|x| x.meta).unwrap_or_default();
 
+        // TODO: `flushed` is not set.
+        let io_state = IOState::new(
+            vote,
+            LogIOId::default(),
+            last_applied,
+            snapshot_meta.last_log_id,
+            last_purged_log_id,
+        );
+
         let now = Instant::now();
 
         Ok(RaftState {