Fix: snapshot with smaller last_log_id than last_applied should not be installed

Otherwise, the state machine will revert to an older state, while the
in-memory `last_applied` is unchanged.
This finally causes logs that falls in range `(snapshot.last_log_id, last_applied]` can not be applied.

Author: drmingdrmer

openraft/snapshot_le_last_applied.rs

@@ -0,0 +1,92 @@
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::Result;
+use maplit::btreeset;
+use openraft::raft::InstallSnapshotRequest;
+use openraft::Config;
+use openraft::LogId;
+use openraft::RaftStorage;
+use openraft::SnapshotMeta;
+
+use crate::fixtures::RaftRouter;
+
+/// Snapshot with smaller last_log_id than local last_applied should not be installed.
+/// Otherwise state machine will revert to a older state.
+#[tokio::test(flavor = "multi_thread", worker_threads = 10)]
+async fn snapshot_le_last_applied() -> Result<()> {
+    let (_log_guard, ut_span) = init_ut!();
+    let _ent = ut_span.enter();
+
+    let config = Arc::new(Config { ..Default::default() }.validate()?);
+    let router = Arc::new(RaftRouter::new(config.clone()));
+
+    let mut log_index = router.new_nodes_from_single(btreeset! {0}, btreeset! {}).await?;
+
+    tracing::info!("--- send logs to increase last_applied");
+    {
+        router.client_request_many(0, "0", 10).await;
+        log_index += 10;
+
+        router
+            .wait_for_log(
+                &btreeset![0],
+                Some(log_index),
+                timeout(),
+                "send log to trigger snapshot",
+            )
+            .await?;
+
+        router.assert_stable_cluster(Some(1), Some(log_index)).await;
+    }
+
+    tracing::info!("--- it should fail to install a snapshot with smaller last_log_id");
+    {
+        assert!(log_index > 3);
+        let req = InstallSnapshotRequest {
+            term: 100,
+            leader_id: 1,
+            meta: SnapshotMeta {
+                snapshot_id: "ss1".into(),
+                last_log_id: Some(LogId { term: 1, index: 3 }),
+            },
+            offset: 0,
+            data: vec![1, 2, 3],
+            done: true,
+        };
+
+        let n = router.remove_node(0).await.unwrap();
+        n.0.install_snapshot(req).await?;
+        let st = n.1.last_applied_state().await?;
+        assert_eq!(
+            Some(LogId {
+                term: 1,
+                index: log_index
+            }),
+            st.0,
+            "last_applied is not affected"
+        );
+
+        let wait_rs =
+            n.0.wait(timeout())
+                .metrics(
+                    |x| {
+                        x.last_applied
+                            != Some(LogId {
+                                term: 1,
+                                index: log_index,
+                            })
+                    },
+                    "in-memory last_applied is not affected",
+                )
+                .await;
+
+        assert!(wait_rs.is_err());
+    }
+
+    Ok(())
+}
+
+fn timeout() -> Option<Duration> {
+    Some(Duration::from_millis(1_000))
+}

openraft/src/core/install_snapshot.rs

@@ -26,11 +26,13 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
     /// Leaders always send chunks in order. It is important to note that, according to the Raft spec,
     /// a log may only have one snapshot at any time. As snapshot contents are application specific,
     /// the Raft log will only store a pointer to the snapshot file along with the index & term.
-    #[tracing::instrument(level = "debug", skip(self, req), fields(req=%req.summary()))]
+    #[tracing::instrument(level = "debug", skip_all)]
     pub(super) async fn handle_install_snapshot_request(
         &mut self,
         req: InstallSnapshotRequest,
     ) -> RaftResult<InstallSnapshotResponse> {
+        tracing::info!("handle_install_snapshot_request: req: {}", req.summary());
+
         // If message's term is less than most recent term, then we do not honor the request.
         if req.term < self.current_term {
             return Ok(InstallSnapshotResponse {
@@ -193,7 +195,14 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         // --------------------------------------------------------------------> time
         // ```
 
-        // TODO(xp): do not install if self.last_applied >= snapshot.meta.last_applied
+        if req.meta.last_log_id <= self.last_applied {
+            tracing::info!(
+                "Skip install_snapshot because snapshot.last_log_id({:?}) <= last_applied({:?})",
+                req.meta.last_log_id,
+                self.last_applied
+            );
+            return Ok(());
+        }
 
         let changes = self
             .storage