Feature: Add new methods to openraft::metrics::Wait

Add `log_index()`, `log_index_at_least()`, `applied_index()` and
`applied_index_at_least()` to `openraft::metrics::Wait` to replace
`Wait::log()` and `Wait::log_at_least()`.

These two methods are deprecated since `0.9.0` because the names does
not imply that they will be blocked by `last_applied` log index.

Thanks to @tvsfx

cluster_benchmark/tests/benchmark/bench_cluster.rs

@@ -122,7 +122,7 @@ async fn do_bench(bench_config: &BenchConfig) -> anyhow::Result<()> {
         handles.push(h)
     }
 
-    leader.wait(timeout()).log_at_least(Some(total), "commit all written logs").await?;
+    leader.wait(timeout()).applied_index_at_least(Some(total), "commit all written logs").await?;
 
     let elapsed = now.elapsed();
 

cluster_benchmark/tests/benchmark/network.rs

@@ -73,7 +73,7 @@ impl Router {
 
         for (id, s) in rafts.iter_mut() {
             tracing::info!(log_index, "--- wait init log: {}, index: {}", id, log_index);
-            s.wait(timeout()).log(Some(log_index), "init").await?;
+            s.wait(timeout()).applied_index(Some(log_index), "init").await?;
         }
 
         Ok(())

macros/src/lib.rs

@@ -1,6 +1,6 @@
 use proc_macro::TokenStream;
 
-/// This macro either emits `#[async_trait::async_trait]` or `#[asnc_trait::async_trait(?Send)]`
+/// This macro either emits `#[async_trait::async_trait]` or `#[async_trait::async_trait(?Send)]`
 /// based on the activated feature set.
 ///
 /// This assumes that the `[async_trait](https://crates.io/crates/async-trait)` crate is imported

openraft/src/metrics/wait.rs

@@ -128,6 +128,7 @@ where
     }
 
     /// Wait until applied exactly `want_log`(inclusive) logs or timeout.
+    #[deprecated(note = "use `log_index()` and `applied_index()` instead, deprecated since 0.9.0")]
     #[tracing::instrument(level = "trace", skip(self), fields(msg=msg.to_string().as_str()))]
     pub async fn log(&self, want_log_index: Option<u64>, msg: impl ToString) -> Result<RaftMetrics<NID, N>, WaitError> {
         self.metrics(
@@ -144,6 +145,7 @@ where
     }
 
     /// Wait until applied at least `want_log`(inclusive) logs or timeout.
+    #[deprecated(note = "use `log_index_at_least()` and `applied_index_at_least()` instead, deprecated since 0.9.0")]
     #[tracing::instrument(level = "trace", skip(self), fields(msg=msg.to_string().as_str()))]
     pub async fn log_at_least(
         &self,
@@ -163,6 +165,59 @@ where
         .await
     }
 
+    /// Block until the last log index becomes exactly `index`(inclusive) or timeout.
+    #[tracing::instrument(level = "trace", skip(self), fields(msg=msg.to_string().as_str()))]
+    pub async fn log_index(&self, index: Option<u64>, msg: impl ToString) -> Result<RaftMetrics<NID, N>, WaitError> {
+        self.metrics(
+            |x| x.last_log_index == index,
+            &format!("{} .last_log_index == {:?}", msg.to_string(), index),
+        )
+        .await
+    }
+
+    /// Block until the last log index becomes at least `index`(inclusive) or timeout.
+    #[tracing::instrument(level = "trace", skip(self), fields(msg=msg.to_string().as_str()))]
+    pub async fn log_index_at_least(
+        &self,
+        index: Option<u64>,
+        msg: impl ToString,
+    ) -> Result<RaftMetrics<NID, N>, WaitError> {
+        self.metrics(
+            |x| x.last_log_index >= index,
+            &format!("{} .last_log_index >= {:?}", msg.to_string(), index),
+        )
+        .await
+    }
+
+    /// Block until the applied index becomes exactly `index`(inclusive) or timeout.
+    #[tracing::instrument(level = "trace", skip(self), fields(msg=msg.to_string().as_str()))]
+    pub async fn applied_index(
+        &self,
+        index: Option<u64>,
+        msg: impl ToString,
+    ) -> Result<RaftMetrics<NID, N>, WaitError> {
+        self.metrics(
+            |x| x.last_applied.index() == index,
+            &format!("{} .last_applied.index == {:?}", msg.to_string(), index),
+        )
+        .await
+    }
+
+    /// Block until the last applied log index become at least `index`(inclusive) or timeout.
+    /// Note that this also implies `last_log_id >= index`.
+    #[tracing::instrument(level = "trace", skip(self), fields(msg=msg.to_string().as_str()))]
+    pub async fn applied_index_at_least(
+        &self,
+        index: Option<u64>,
+        msg: impl ToString,
+    ) -> Result<RaftMetrics<NID, N>, WaitError> {
+        self.metrics(
+            |m| m.last_log_index >= index && m.last_applied.index() >= index,
+            &format!("{} .last_applied.index >= {:?}", msg.to_string(), index),
+        )
+        .await
+    }
+
     /// Wait for `state` to become `want_state` or timeout.
     #[tracing::instrument(level = "trace", skip(self), fields(msg=msg.to_string().as_str()))]
     pub async fn state(&self, want_state: ServerState, msg: impl ToString) -> Result<RaftMetrics<NID, N>, WaitError> {

openraft/src/metrics/wait_test.rs

@@ -41,7 +41,7 @@ async fn test_wait() -> anyhow::Result<()> {
     }
 
     {
-        // wait for log
+        // wait for applied log
         let (init, w, tx) = init_wait_test::<u64, ()>();
 
         let h = tokio::spawn(async move {
@@ -52,10 +52,10 @@ async fn test_wait() -> anyhow::Result<()> {
             let rst = tx.send(update);
             assert!(rst.is_ok());
         });
-        let got = w.log(Some(3), "log").await?;
-        let got_least2 = w.log_at_least(Some(2), "log").await?;
-        let got_least3 = w.log_at_least(Some(3), "log").await?;
-        let got_least4 = w.log_at_least(Some(4), "log").await;
+        let got = w.applied_index(Some(3), "log").await?;
+        let got_least2 = w.applied_index_at_least(Some(2), "log").await?;
+        let got_least3 = w.applied_index_at_least(Some(3), "log").await?;
+        let got_least4 = w.applied_index_at_least(Some(4), "log").await;
         h.await?;
 
         assert_eq!(Some(3), got.last_log_index);
@@ -173,6 +173,34 @@ async fn test_wait() -> anyhow::Result<()> {
     Ok(())
 }
 
+#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
+async fn test_wait_log_index() -> anyhow::Result<()> {
+    // wait for applied log
+    let (init, w, tx) = init_wait_test::<u64, ()>();
+
+    let h = tokio::spawn(async move {
+        sleep(Duration::from_millis(10)).await;
+        let mut update = init.clone();
+        update.last_log_index = Some(3);
+        let rst = tx.send(update);
+        assert!(rst.is_ok());
+    });
+
+    let got = w.log_index(Some(3), "log").await?;
+    let got_least2 = w.log_index_at_least(Some(2), "log").await?;
+    let got_least3 = w.log_index_at_least(Some(3), "log").await?;
+    let got_least4 = w.log_index_at_least(Some(4), "log").await;
+    h.await?;
+
+    assert_eq!(Some(3), got.last_log_index);
+    assert_eq!(Some(3), got_least2.last_log_index);
+    assert_eq!(Some(3), got_least3.last_log_index);
+
+    assert!(got_least4.is_err());
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 8)]
 async fn test_wait_vote() -> anyhow::Result<()> {
     let (init, w, tx) = init_wait_test::<u64, ()>();

tests/tests/append_entries/t11_append_inconsistent_log.rs

@@ -109,7 +109,7 @@ async fn append_inconsistent_log() -> Result<()> {
     router
         .wait(&0, Some(Duration::from_millis(2000)))
         // leader appends at least one blank log. There may be more than one transient leaders
-        .log_at_least(Some(log_index), "sync log to node 0")
+        .applied_index_at_least(Some(log_index), "sync log to node 0")
         .await?;
 
     let logs = sto0.get_log_entries(60..=60).await?;

tests/tests/append_entries/t60_enable_heartbeat.rs

@@ -37,7 +37,7 @@ async fn enable_heartbeat() -> Result<()> {
             // no new log will be sent, .
             router
                 .wait(&node_id, timeout())
-                .log_at_least(Some(log_index), format!("node {} emit heartbeat log", node_id))
+                .applied_index_at_least(Some(log_index), format!("node {} emit heartbeat log", node_id))
                 .await?;
 
             // leader lease is extended.

tests/tests/append_entries/t61_heartbeat_reject_vote.rs

@@ -69,15 +69,15 @@ async fn heartbeat_reject_vote() -> Result<()> {
     {
         // TODO: this part can be removed when blank-log heartbeat is removed.
         sleep(Duration::from_millis(1500)).await;
-        router.wait(&1, timeout()).log(Some(log_index), "no log is written").await?;
+        router.wait(&1, timeout()).applied_index(Some(log_index), "no log is written").await?;
     }
 
     tracing::info!(log_index, "--- disable heartbeat, vote request will be granted");
     {
         node0.runtime_config().heartbeat(false);
         sleep(Duration::from_millis(1500)).await;
 
-        router.wait(&1, timeout()).log(Some(log_index), "no log is written").await?;
+        router.wait(&1, timeout()).applied_index(Some(log_index), "no log is written").await?;
 
         let res = node1.vote(VoteRequest::new(Vote::new(10, 2), Some(log_id(10, 1, 10)))).await?;
         assert!(res.vote_granted, "vote is granted after leader lease expired");

tests/tests/append_entries/t61_large_heartbeat.rs

@@ -30,7 +30,7 @@ async fn large_heartbeat() -> Result<()> {
     router.client_request_many(0, "foo", 10).await?;
     log_index += 10;
 
-    router.wait(&1, Some(Duration::from_millis(3_000))).log(Some(log_index), "").await?;
+    router.wait(&1, Some(Duration::from_millis(3_000))).applied_index(Some(log_index), "").await?;
 
     Ok(())
 }

tests/tests/append_entries/t90_issue_216_stale_last_log_id.rs

@@ -57,10 +57,10 @@ async fn stale_last_log_id() -> Result<()> {
         log_index += n_ops as u64;
     }
 
-    router.wait(&1, Some(Duration::from_millis(1000))).log(Some(log_index), "").await?;
-    router.wait(&2, Some(Duration::from_millis(1000))).log(Some(log_index), "").await?;
-    router.wait(&3, Some(Duration::from_millis(1000))).log(Some(log_index), "").await?;
-    router.wait(&4, Some(Duration::from_millis(1000))).log(Some(log_index), "").await?;
+    router.wait(&1, Some(Duration::from_millis(1000))).applied_index(Some(log_index), "").await?;
+    router.wait(&2, Some(Duration::from_millis(1000))).applied_index(Some(log_index), "").await?;
+    router.wait(&3, Some(Duration::from_millis(1000))).applied_index(Some(log_index), "").await?;
+    router.wait(&4, Some(Duration::from_millis(1000))).applied_index(Some(log_index), "").await?;
 
     Ok(())
 }

tests/tests/client_api/t12_trigger_purge_log.rs

@@ -34,7 +34,10 @@ async fn trigger_purge_log() -> anyhow::Result<()> {
         log_index += router.client_request_many(0, "0", 10).await?;
 
         for id in [0, 1, 2] {
-            router.wait(&id, timeout()).log(Some(log_index), format_args!("node-{} write logs", id)).await?;
+            router
+                .wait(&id, timeout())
+                .applied_index(Some(log_index), format_args!("node-{} write logs", id))
+                .await?;
         }
     }
 
@@ -53,7 +56,10 @@ async fn trigger_purge_log() -> anyhow::Result<()> {
         log_index += router.client_request_many(0, "0", 10).await?;
 
         for id in [0, 1, 2] {
-            router.wait(&id, timeout()).log(Some(log_index), format_args!("node-{} write logs", id)).await?;
+            router
+                .wait(&id, timeout())
+                .applied_index(Some(log_index), format_args!("node-{} write logs", id))
+                .await?;
         }
     }
 

tests/tests/client_api/t13_trigger_snapshot.rs

@@ -41,8 +41,8 @@ async fn trigger_snapshot() -> anyhow::Result<()> {
         router.client_request_many(0, "0", 10).await?;
         log_index += 10;
 
-        router.wait(&0, timeout()).log(Some(log_index), "node-0 write logs").await?;
-        router.wait(&1, timeout()).log(Some(log_index), "node-1 write logs").await?;
+        router.wait(&0, timeout()).applied_index(Some(log_index), "node-0 write logs").await?;
+        router.wait(&1, timeout()).applied_index(Some(log_index), "node-1 write logs").await?;
     }
 
     tracing::info!(log_index, "--- trigger snapshot for node-0");

tests/tests/fixtures/mod.rs

@@ -641,7 +641,7 @@ impl TypedRaftRouter {
         msg: &str,
     ) -> anyhow::Result<()> {
         for i in node_ids.iter() {
-            self.wait(i, timeout).log(want_log, msg).await?;
+            self.wait(i, timeout).applied_index(want_log, msg).await?;
         }
         Ok(())
     }

tests/tests/life_cycle/t10_initialization.rs

@@ -81,7 +81,7 @@ async fn initialization() -> anyhow::Result<()> {
         log_index += 1;
 
         for node_id in [0, 1, 2] {
-            router.wait(&node_id, timeout()).log(Some(log_index), "init").await?;
+            router.wait(&node_id, timeout()).applied_index(Some(log_index), "init").await?;
         }
     }
 

tests/tests/life_cycle/t50_single_follower_restart.rs

@@ -63,7 +63,7 @@ async fn single_follower_restart() -> anyhow::Result<()> {
         router.client_request_many(0, "foo", 1).await?;
         log_index += 1;
 
-        router.wait(&0, timeout()).log(Some(log_index), "node-0 works").await?;
+        router.wait(&0, timeout()).applied_index(Some(log_index), "node-0 works").await?;
     }
 
     Ok(())

tests/tests/life_cycle/t50_single_leader_restart_re_apply_logs.rs

@@ -49,7 +49,7 @@ async fn single_leader_restart_re_apply_logs() -> anyhow::Result<()> {
 
     tracing::info!(log_index, "--- a single leader should re-apply all logs");
     {
-        router.wait(&0, timeout()).log(Some(log_index), "node-0 works").await?;
+        router.wait(&0, timeout()).applied_index(Some(log_index), "node-0 works").await?;
     }
 
     Ok(())

tests/tests/life_cycle/t90_issue_607_single_restart.rs

@@ -48,7 +48,7 @@ async fn single_restart() -> anyhow::Result<()> {
         router.client_request_many(0, "foo", 1).await?;
         log_index += 1;
 
-        router.wait(&0, timeout()).log(Some(log_index), "node-0 works").await?;
+        router.wait(&0, timeout()).applied_index(Some(log_index), "node-0 works").await?;
     }
 
     Ok(())

tests/tests/log_store/t10_save_committed.rs

@@ -29,7 +29,7 @@ async fn write_committed_log_id_to_log_store() -> Result<()> {
     log_index += router.client_request_many(0, "0", 10).await?;
 
     for i in [0, 1, 2] {
-        router.wait(&i, timeout()).log(Some(log_index), "write logs").await?;
+        router.wait(&i, timeout()).applied_index(Some(log_index), "write logs").await?;
     }
 
     for id in [0, 1, 2] {

tests/tests/membership/t10_learner_restart.rs

@@ -49,7 +49,7 @@ async fn learner_restart() -> Result<()> {
 
     // restart node-1, assert the state as expected.
     let restarted = Raft::new(1, config.clone(), router.clone(), sto1, sm1).await?;
-    restarted.wait(timeout()).log(Some(log_index), "log after restart").await?;
+    restarted.wait(timeout()).applied_index(Some(log_index), "log after restart").await?;
     restarted.wait(timeout()).state(ServerState::Learner, "server state after restart").await?;
 
     Ok(())

tests/tests/membership/t11_add_learner.rs

@@ -46,7 +46,7 @@ async fn add_learner_basic() -> Result<()> {
         log_index += 1;
 
         assert_eq!(log_index, res.log_id.index);
-        router.wait(&0, timeout()).log(Some(log_index), "commit re-adding leader log").await?;
+        router.wait(&0, timeout()).applied_index(Some(log_index), "commit re-adding leader log").await?;
     }
 
     tracing::info!(log_index, "--- add new node node-1");
@@ -86,7 +86,7 @@ async fn add_learner_basic() -> Result<()> {
         log_index += 1;
 
         assert_eq!(log_index, res.log_id.index);
-        router.wait(&0, timeout()).log(Some(log_index), "commit re-adding node-1 log").await?;
+        router.wait(&0, timeout()).applied_index(Some(log_index), "commit re-adding node-1 log").await?;
 
         let metrics = router.get_raft_handle(&0)?.metrics().borrow().clone();
         let node_ids = metrics.membership_config.membership().nodes().map(|x| *x.0).collect::<Vec<_>>();
@@ -122,7 +122,7 @@ async fn add_learner_non_blocking() -> Result<()> {
         router.client_request_many(0, "learner_add", 100 - log_index as usize).await?;
         log_index = 100;
 
-        router.wait(&0, timeout()).log(Some(log_index), "received 100 logs").await?;
+        router.wait(&0, timeout()).applied_index(Some(log_index), "received 100 logs").await?;
 
         router.new_raft_node(1).await;
 
@@ -279,7 +279,7 @@ async fn check_learner_after_leader_transferred() -> Result<()> {
     {
         router
             .wait(&orig_leader_id, timeout())
-            .log(Some(log_index), "old leader commits 2 membership log")
+            .applied_index(Some(log_index), "old leader commits 2 membership log")
             .await?;
     }
 
@@ -291,7 +291,7 @@ async fn check_learner_after_leader_transferred() -> Result<()> {
         for id in [1, 3, 4] {
             router
                 .wait(&id, timeout())
-                .log_at_least(
+                .applied_index_at_least(
                     Some(log_index),
                     "node in new cluster finally commit at least one blank leader-initialize log",
                 )
@@ -325,7 +325,7 @@ async fn check_learner_after_leader_transferred() -> Result<()> {
         log_index += 1;
 
         for i in [1, 2, 3, 4] {
-            router.wait(&i, timeout()).log_at_least(Some(log_index), "learner recv new log").await?;
+            router.wait(&i, timeout()).applied_index_at_least(Some(log_index), "learner recv new log").await?;
         }
     }