fix: leader should re-create and send snapshot when threshold/2 < last_log_index - snapshot < threshold

drmingdrmer · drmingdrmer · commit cf4badd0d762 · 2021-07-08T16:16:43.000+08:00
The problem:

If `last_log_index` advances `snapshot.applied_index` too many, i.e.:
`threshold/2 &lt; last_log_index - snapshot &lt; threshold`
(e.g., `10/2 &lt; 16-10 &lt; 20` in the test that reproduce this bug), the leader
tries to re-create a new snapshot. But when
`last_log_index &lt; threshold`, it won't create, which result in a dead
loop.

Solution:

In such case, force to create a snapshot without considering the
threshold.
diff --git a/async-raft/src/core/client.rs b/async-raft/src/core/client.rs
@@ -339,7 +339,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
         }
 
         // Trigger log compaction if needed.
-        self.core.trigger_log_compaction_if_needed();
+        self.core.trigger_log_compaction_if_needed(false);
     }
 
     /// Apply the given log entry to the state machine.
diff --git a/async-raft/src/core/mod.rs b/async-raft/src/core/mod.rs
@@ -416,8 +416,9 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
     }
 
     /// Trigger a log compaction (snapshot) job if needed.
+    /// If force is True, it will skip the threshold check and start creating snapshot as demanded.
     #[tracing::instrument(level = "trace", skip(self))]
-    pub(self) fn trigger_log_compaction_if_needed(&mut self) {
+    pub(self) fn trigger_log_compaction_if_needed(&mut self, force: bool) {
         if self.snapshot_state.is_some() {
             return;
         }
@@ -426,11 +427,12 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
         if self.last_applied == 0 || self.last_applied < self.snapshot_index {
             return;
         }
-        // If we are below the threshold, then there is nothing to do.
-        let is_below_threshold =
-            self.last_applied.checked_sub(self.snapshot_index).map(|diff| diff < *threshold).unwrap_or(false);
-        if is_below_threshold {
-            return;
+
+        if !force {
+            // If we are below the threshold, then there is nothing to do.
+            if self.last_applied < self.snapshot_index + *threshold {
+                return;
+            }
         }
 
         // At this point, we are clear to begin a new compaction process.
@@ -473,7 +475,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra
             self.last_applied = last_applied;
         }
         self.report_metrics(Update::Ignore);
-        self.trigger_log_compaction_if_needed();
+        self.trigger_log_compaction_if_needed(false);
         Ok(())
     }
 
diff --git a/async-raft/src/core/replication.rs b/async-raft/src/core/replication.rs
@@ -296,6 +296,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
             .get_current_snapshot()
             .await
             .map_err(|err| self.core.map_fatal_storage_error(err))?;
+
         if let Some(snapshot) = current_snapshot_opt {
             // If snapshot exists, ensure its distance from the leader's last log index is <= half
             // of the configured snapshot threshold, else create a new snapshot.
@@ -325,7 +326,10 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>
         //
         // If this block is executed, and a snapshot is needed, the repl stream will submit another
         // request here shortly, and will hit the above logic where it will await the snapshot completion.
-        self.core.trigger_log_compaction_if_needed();
+        //
+        // If snapshot is too old, i.e., the distance from last_log_index is greater than half of snapshot threshold,
+        // always force a snapshot creation.
+        self.core.trigger_log_compaction_if_needed(true);
         Ok(())
     }
 }
diff --git a/async-raft/tests/snapshot_ge_half_threshold.rs b/async-raft/tests/snapshot_ge_half_threshold.rs
@@ -0,0 +1,103 @@
+mod fixtures;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use anyhow::Result;
+use async_raft::raft::MembershipConfig;
+use async_raft::Config;
+use async_raft::SnapshotPolicy;
+use async_raft::State;
+use fixtures::RaftRouter;
+use maplit::hashset;
+use tokio::time::sleep;
+
+/// A leader should create and send snapshot when snapshot is old and is not that old to trigger a snapshot, i.e.:
+/// `threshold/2 < leader.last_log_index - snapshot.applied_index < threshold`
+///
+/// What does this test do?
+///
+/// - build a stable single node cluster.
+/// - send enough requests to the node that log compaction will be triggered.
+/// - send some other log after snapshot created, to make the `leader.last_log_index - snapshot.applied_index` big
+///   enough.
+/// - add non-voter and assert that they receive the snapshot and logs.
+///
+/// export RUST_LOG=async_raft,memstore,snapshot_ge_half_threshold=trace
+/// cargo test -p async-raft --test snapshot_ge_half_threshold
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn snapshot_ge_half_threshold() -> Result<()> {
+    fixtures::init_tracing();
+
+    let snapshot_threshold: u64 = 10;
+    let log_cnt = snapshot_threshold + 6;
+
+    let config = Arc::new(
+        Config::build("test".into())
+            .snapshot_policy(SnapshotPolicy::LogsSinceLast(snapshot_threshold))
+            .validate()
+            .expect("failed to build Raft config"),
+    );
+    let router = Arc::new(RaftRouter::new(config.clone()));
+
+    let mut want = 0;
+
+    tracing::info!("--- initializing cluster");
+    {
+        router.new_raft_node(0).await;
+
+        router.wait_for_log(&hashset![0], want, None, "empty").await?;
+        router.wait_for_state(&hashset![0], State::NonVoter, None, "empty").await?;
+        router.initialize_from_single_node(0).await?;
+        want += 1;
+
+        router.wait_for_log(&hashset![0], want, None, "init leader").await?;
+        router.assert_stable_cluster(Some(1), Some(want)).await;
+    }
+
+    tracing::info!("--- send just enough logs to trigger snapshot");
+    {
+        router.client_request_many(0, "0", (snapshot_threshold - want) as usize).await;
+        want = snapshot_threshold;
+
+        router.wait_for_log(&hashset![0], want, None, "send log to trigger snapshot").await?;
+        router.assert_stable_cluster(Some(1), Some(want)).await;
+
+        // TODO: add snapshot info into metrics.
+        //       Then watch metrics instead of waiting.
+        sleep(Duration::from_secs(5)).await;
+        router
+            .assert_storage_state(
+                1,
+                want,
+                Some(0),
+                want,
+                Some((want.into(), 1, MembershipConfig {
+                    members: hashset![0],
+                    members_after_consensus: None,
+                })),
+            )
+            .await;
+    }
+
+    tracing::info!("--- send logs to make distance between snapshot index and last_log_index");
+    {
+        router.client_request_many(0, "0", (log_cnt - want) as usize).await;
+        want = log_cnt;
+    }
+
+    tracing::info!("--- add non-voter to receive snapshot and logs");
+    {
+        router.new_raft_node(1).await;
+        router.add_non_voter(0, 1).await.expect("failed to add new node as non-voter");
+
+        router.wait_for_log(&hashset![0, 1], want, None, "add non-voter").await?;
+        let expected_snap = Some((want.into(), 1, MembershipConfig {
+            members: hashset![0u64],
+            members_after_consensus: None,
+        }));
+        router.assert_storage_state(1, want, None /* non-voter does not vote */, want, expected_snap).await;
+    }
+
+    Ok(())
+}

Original file line number	Diff line number	Diff line change
`@@ -339,7 +339,7 @@ impl<'a, D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>`
`339`	`339`	`}`
`340`	`340`
`341`	`341`	`// Trigger log compaction if needed.`
`342`		`- self.core.trigger_log_compaction_if_needed();`
	`342`	`+ self.core.trigger_log_compaction_if_needed(false);`
`343`	`343`	`}`
`344`	`344`
`345`	`345`	`/// Apply the given log entry to the state machine.`
Original file line number	Diff line number	Diff line change
`@@ -416,8 +416,9 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra`
`416`	`416`	`}`
`417`	`417`
`418`	`418`	`/// Trigger a log compaction (snapshot) job if needed.`
	`419`	`+ /// If force is True, it will skip the threshold check and start creating snapshot as demanded.`
`419`	`420`	`#[tracing::instrument(level = "trace", skip(self))]`
`420`		`- pub(self) fn trigger_log_compaction_if_needed(&mut self) {`
	`421`	`+ pub(self) fn trigger_log_compaction_if_needed(&mut self, force: bool) {`
`421`	`422`	`if self.snapshot_state.is_some() {`
`422`	`423`	`return;`
`423`	`424`	`}`
`@@ -426,11 +427,12 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra`
`426`	`427`	`if self.last_applied == 0 \|\| self.last_applied < self.snapshot_index {`
`427`	`428`	`return;`
`428`	`429`	`}`
`429`		`- // If we are below the threshold, then there is nothing to do.`
`430`		`- let is_below_threshold =`
`431`		`- self.last_applied.checked_sub(self.snapshot_index).map(\|diff\| diff < *threshold).unwrap_or(false);`
`432`		`- if is_below_threshold {`
`433`		`- return;`
	`430`	`+`
	`431`	`+ if !force {`
	`432`	`+ // If we are below the threshold, then there is nothing to do.`
	`433`	`+ if self.last_applied < self.snapshot_index + *threshold {`
	`434`	`+ return;`
	`435`	`+ }`
`434`	`436`	`}`
`435`	`437`
`436`	`438`	`// At this point, we are clear to begin a new compaction process.`
`@@ -473,7 +475,7 @@ impl<D: AppData, R: AppDataResponse, N: RaftNetwork<D>, S: RaftStorage<D, R>> Ra`
`473`	`475`	`self.last_applied = last_applied;`
`474`	`476`	`}`
`475`	`477`	`self.report_metrics(Update::Ignore);`
`476`		`- self.trigger_log_compaction_if_needed();`
	`478`	`+ self.trigger_log_compaction_if_needed(false);`
`477`	`479`	`Ok(())`
`478`	`480`	`}`
`479`	`481`