refactor: rename and reorganize log I/O state handling

- Replaced `accepted_io` with `accepted_log_io` in `RaftState` for
  clearer semantics.

- Introduced `log_progress_mut` and `apply_progress_mut` methods to
  streamline access to log and apply progress.

- Improved naming conventions for log I/O progress tracking, e.g.,
  renamed methods like `update_committed` to `accept` for consistency.

- Updated `IOProgress` to include a `name` field for better debugging
  and logging context.

- Adjusted test cases and internal logic to align with the new naming
  and structure.

Author: drmingdrmer

openraft/src/core/raft_core.rs

@@ -543,7 +543,7 @@ where
 
             // --- data ---
             current_term: st.vote_ref().term(),
-            vote: st.io_state().log_progress.flushed().map(|io_id| io_id.to_vote()).unwrap_or_default(),
+            vote: st.log_progress().flushed().map(|io_id| io_id.to_vote()).unwrap_or_default(),
             last_log_index: st.last_log_id().index(),
             last_applied: st.io_applied().cloned(),
             snapshot: st.io_snapshot_last_log_id().cloned(),
@@ -575,7 +575,7 @@ where
 
         let server_metrics = RaftServerMetrics {
             id: self.id.clone(),
-            vote: st.io_state().log_progress.flushed().map(|io_id| io_id.to_vote()).unwrap_or_default(),
+            vote: st.log_progress().flushed().map(|io_id| io_id.to_vote()).unwrap_or_default(),
             state: st.server_state,
             current_leader,
             membership_config,
@@ -861,7 +861,7 @@ where
             tracing::debug!(
                 "RAFT_stats id={:<2} log_io: {}",
                 self.id,
-                self.engine.state.io_state.log_progress
+                self.engine.state.log_progress()
             );
 
             // In each loop, it does not have to check rx_shutdown and flush metrics for every RaftMsg
@@ -1368,7 +1368,7 @@ where
             }
 
             Notification::LocalIO { io_id } => {
-                self.engine.state.io_state.log_progress.flush(io_id.clone());
+                self.engine.state.log_progress_mut().flush(io_id.clone());
 
                 match io_id {
                     IOId::Log(log_io_id) => {
@@ -1448,16 +1448,18 @@ where
                             func_name!()
                         );
 
-                        self.engine.state.io_state_mut().log_progress.flush(io_id);
+                        self.engine.state.log_progress_mut().flush(io_id);
 
                         if let Some(meta) = meta {
+                            if let Some(last) = &meta.last_log_id {
+                                self.engine.state.apply_progress_mut().flush(last.clone());
+                            }
                             let st = self.engine.state.io_state_mut();
-                            st.update_applied(meta.last_log_id.clone());
                             st.update_snapshot(meta.last_log_id);
                         }
                     }
                     sm::Response::Apply(res) => {
-                        self.engine.state.io_state_mut().update_applied(Some(res.last_applied.clone()));
+                        self.engine.state.apply_progress_mut().flush(res.last_applied);
                     }
                 }
             }
@@ -1613,7 +1615,7 @@ where
         if let Some(condition) = condition {
             match condition {
                 Condition::IOFlushed { io_id } => {
-                    let curr = self.engine.state.io_state().log_progress.flushed();
+                    let curr = self.engine.state.log_progress().flushed();
                     if curr < Some(&io_id) {
                         tracing::debug!(
                             "io_id: {} has not yet flushed, currently flushed: {} postpone cmd: {}",
@@ -1625,7 +1627,7 @@ where
                     }
                 }
                 Condition::LogFlushed { log_id } => {
-                    let curr = self.engine.state.io_state().log_progress.flushed();
+                    let curr = self.engine.state.log_progress().flushed();
                     let curr = curr.and_then(|x| x.last_log_id());
                     if curr < log_id.as_ref() {
                         tracing::debug!(
@@ -1664,7 +1666,7 @@ where
 
         match cmd {
             Command::UpdateIOProgress { io_id, .. } => {
-                self.engine.state.io_state.log_progress.submit(io_id.clone());
+                self.engine.state.log_progress_mut().submit(io_id.clone());
 
                 let notify = Notification::LocalIO { io_id: io_id.clone() };
 
@@ -1689,13 +1691,13 @@ where
                 //
                 // The `submit` state must be updated before calling `append()`,
                 // because `append()` may call the callback before returning.
-                self.engine.state.io_state.log_progress.submit(io_id);
+                self.engine.state.log_progress_mut().submit(io_id);
 
                 // Submit IO request, do not wait for the response.
                 self.log_store.append(entries, callback).await?;
             }
             Command::SaveVote { vote } => {
-                self.engine.state.io_state_mut().log_progress.submit(IOId::new(&vote));
+                self.engine.state.log_progress_mut().submit(IOId::new(&vote));
                 self.log_store.save_vote(&vote).await?;
 
                 let _ = self.tx_notification.send(Notification::LocalIO {
@@ -1758,7 +1760,7 @@ where
                 already_committed,
                 upto,
             } => {
-                self.engine.state.io_state.apply_progress.submit(upto.clone());
+                self.engine.state.apply_progress_mut().submit(upto.clone());
                 let first = self.engine.state.get_log_id(already_committed.next_index()).unwrap();
                 self.apply_to_state_machine(first, upto).await?;
             }
@@ -1786,15 +1788,15 @@ where
                 self.heartbeat_handle.spawn_workers(&mut self.network_factory, &self.tx_notification, nodes).await;
             }
             Command::StateMachine { command } => {
-                let io_id = command.get_submit_io();
+                let io_id = command.get_log_progress();
 
                 if let Some(io_id) = io_id {
-                    self.engine.state.io_state.log_progress.submit(io_id);
+                    self.engine.state.log_progress_mut().submit(io_id);
                 }
 
                 // If this command update the last-applied log id, mark it as submitted(to state machine).
                 if let Some(log_id) = command.get_apply_progress() {
-                    self.engine.state.io_state.apply_progress.submit(log_id);
+                    self.engine.state.apply_progress_mut().submit(log_id);
                 }
 
                 // Just forward a state machine command to the worker.

openraft/src/core/sm/command.rs

@@ -91,8 +91,11 @@ where C: RaftTypeConfig
         }
     }
 
-    /// Return the IOId if this command submit any IO.
-    pub(crate) fn get_submit_io(&self) -> Option<IOId<C>> {
+    /// Return the [`IOId`] of the log-related I/O progress to submit, if this command submits any
+    /// log I/O.
+    ///
+    /// Log-related I/O progress includes both Vote and AppendEntries operations.
+    pub(crate) fn get_log_progress(&self) -> Option<IOId<C>> {
         match self {
             Command::BuildSnapshot => None,
             Command::GetSnapshot { .. } => None,

openraft/src/engine/engine_impl.rs

@@ -403,7 +403,7 @@ where C: RaftTypeConfig
 
             let condition = if is_ok {
                 Some(Condition::IOFlushed {
-                    io_id: self.state.accepted_io().unwrap().clone(),
+                    io_id: self.state.accepted_log_io().unwrap().clone(),
                 })
             } else {
                 None
@@ -439,7 +439,7 @@ where C: RaftTypeConfig
     pub(crate) fn handle_commit_entries(&mut self, leader_committed: Option<LogIdOf<C>>) {
         tracing::debug!(
             leader_committed = display(leader_committed.display()),
-            my_accepted = display(self.state.accepted_io().display()),
+            my_accepted = display(self.state.accepted_log_io().display()),
             my_committed = display(self.state.committed().display()),
             "{}",
             func_name!()
@@ -649,7 +649,7 @@ where C: RaftTypeConfig
         let _res = self.vote_handler().update_vote(&vote.clone().into_vote());
         debug_assert!(_res.is_ok(), "commit vote can not fail but: {:?}", _res);
 
-        self.state.accept_io(IOId::new_log_io(vote, last_log_id));
+        self.state.accept_log_io(IOId::new_log_io(vote, last_log_id));
 
         // No need to submit UpdateIOProgress command,
         // IO progress is updated by the new blank log

openraft/src/engine/handler/following_handler/append_entries_test.rs

@@ -73,7 +73,7 @@ fn test_follower_append_entries_update_accepted() -> anyhow::Result<()> {
             Vote::new(2, 1).into_committed(),
             Some(log_id(3, 1, 5))
         )),
-        eng.state.accepted_io()
+        eng.state.accepted_log_io()
     );
     assert_eq!(eng.output.take_commands(), vec![
         //
@@ -101,7 +101,7 @@ fn test_follower_append_entries_update_accepted() -> anyhow::Result<()> {
                 Vote::new(3, 1).into_committed(),
                 Some(log_id(3, 1, 4))
             )),
-            eng.state.accepted_io()
+            eng.state.accepted_log_io()
         );
         assert_eq!(eng.output.take_commands(), vec![
             //
@@ -123,7 +123,7 @@ fn test_follower_append_entries_update_accepted() -> anyhow::Result<()> {
                 Vote::new(3, 1).into_committed(),
                 Some(log_id(3, 1, 4))
             )),
-            eng.state.accepted_io()
+            eng.state.accepted_log_io()
         );
 
         assert_eq!(eng.output.take_commands(), vec![

openraft/src/engine/handler/following_handler/commit_entries_test.rs

@@ -33,7 +33,7 @@ fn eng() -> Engine<UTConfig> {
         Duration::from_millis(500),
         Vote::new_committed(2, 1),
     );
-    eng.state.io_state.update_committed(log_id(1, 1, 1));
+    eng.state.apply_progress_mut().accept(log_id(1, 1, 1));
     eng.state.membership_state = MembershipState::new(
         Arc::new(EffectiveMembership::new(Some(log_id(1, 1, 1)), m01())),
         Arc::new(EffectiveMembership::new(Some(log_id(2, 1, 3)), m23())),
@@ -64,7 +64,7 @@ fn test_following_handler_commit_entries_empty() -> anyhow::Result<()> {
 fn test_following_handler_commit_entries_ge_accepted() -> anyhow::Result<()> {
     let mut eng = eng();
     let committed_vote = eng.state.vote_ref().into_committed();
-    eng.state.io_state.log_progress.accept(IOId::new_log_io(committed_vote, Some(log_id(1, 1, 2))));
+    eng.state.log_progress_mut().accept(IOId::new_log_io(committed_vote, Some(log_id(1, 1, 2))));
 
     eng.following_handler().commit_entries(Some(log_id(2, 1, 3)));
 
@@ -96,7 +96,7 @@ fn test_following_handler_commit_entries_ge_accepted() -> anyhow::Result<()> {
 fn test_following_handler_commit_entries_le_accepted() -> anyhow::Result<()> {
     let mut eng = eng();
     let committed_vote = eng.state.vote_ref().into_committed();
-    eng.state.io_state.log_progress.accept(IOId::new_log_io(committed_vote, Some(log_id(3, 1, 4))));
+    eng.state.log_progress_mut().accept(IOId::new_log_io(committed_vote, Some(log_id(3, 1, 4))));
 
     eng.following_handler().commit_entries(Some(log_id(2, 1, 3)));
 

openraft/src/engine/handler/following_handler/install_snapshot_test.rs

@@ -38,7 +38,7 @@ fn eng() -> Engine<UTConfig> {
     let now = UTConfig::<()>::now();
     let vote = VoteOf::<UTConfig>::new_committed(2, 1);
     eng.state.vote.update(now, Duration::from_millis(500), vote);
-    eng.state.io_state_mut().update_committed(log_id(4, 1, 5));
+    eng.state.apply_progress_mut().accept(log_id(4, 1, 5));
     eng.state.log_ids = LogIdList::new(vec![
         //
         log_id(2, 1, 2),
@@ -188,7 +188,7 @@ fn test_install_snapshot_conflict() -> anyhow::Result<()> {
             Duration::from_millis(500),
             Vote::new_committed(2, 1),
         );
-        eng.state.io_state_mut().update_committed(log_id(2, 1, 3));
+        eng.state.apply_progress_mut().accept(log_id(2, 1, 3));
         eng.state.log_ids = LogIdList::new(vec![
             //
             log_id(2, 1, 2),
@@ -349,7 +349,7 @@ fn test_install_snapshot_update_accepted() -> anyhow::Result<()> {
             Vote::new(2, 1).into_committed(),
             Some(log_id(100, 1, 100))
         )),
-        eng.state.accepted_io()
+        eng.state.accepted_log_io()
     );
 
     Ok(())

openraft/src/engine/handler/following_handler/mod.rs

@@ -76,7 +76,7 @@ where C: RaftTypeConfig
         let last_log_id = entries.last().map(|ent| ent.log_id());
         let last_log_id = std::cmp::max(prev_log_id, last_log_id);
 
-        let prev_accepted = self.state.accept_io(IOId::new_log_io(self.leader_vote.clone(), last_log_id.clone()));
+        let prev_accepted = self.state.accept_log_io(IOId::new_log_io(self.leader_vote.clone(), last_log_id.clone()));
 
         let l = entries.len();
         let since = self.state.first_conflicting_index(&entries);
@@ -160,7 +160,7 @@ where C: RaftTypeConfig
     /// Commit entries that are already committed by the leader.
     #[tracing::instrument(level = "debug", skip_all)]
     pub(crate) fn commit_entries(&mut self, leader_committed: Option<LogIdOf<C>>) {
-        let accepted = self.state.accepted_io().cloned();
+        let accepted = self.state.accepted_log_io().cloned();
         let accepted = accepted.and_then(|x| x.last_log_id().cloned());
         let committed = std::cmp::min(accepted.clone(), leader_committed.clone());
 
@@ -314,8 +314,10 @@ where C: RaftTypeConfig
         }
 
         let io_id = IOId::new_log_io(self.leader_vote.clone(), Some(snap_last_log_id.clone()));
-        self.state.accept_io(io_id.clone());
-        self.state.io_state.update_committed(snap_last_log_id.clone());
+        self.state.accept_log_io(io_id.clone());
+
+        self.state.apply_progress_mut().accept(snap_last_log_id.clone());
+
         self.update_committed_membership(EffectiveMembership::new_from_stored_membership(
             meta.last_membership.clone(),
         ));

openraft/src/engine/handler/leader_handler/append_entries_test.rs

@@ -52,7 +52,7 @@ fn eng() -> Engine<UTConfig> {
     eng.state.enable_validation(false); // Disable validation for incomplete state
 
     eng.config.id = 1;
-    eng.state.io_state_mut().update_committed(log_id(0, 1, 0));
+    eng.state.apply_progress_mut().accept(log_id(0, 1, 0));
     eng.state.vote = Leased::new(
         UTConfig::<()>::now(),
         Duration::from_millis(500),
@@ -79,7 +79,7 @@ fn test_leader_append_entries_empty() -> anyhow::Result<()> {
 
     assert_eq!(
         None,
-        eng.state.accepted_io(),
+        eng.state.accepted_log_io(),
         "no accepted log updated for empty entries"
     );
     assert_eq!(
@@ -119,7 +119,7 @@ fn test_leader_append_entries_normal() -> anyhow::Result<()> {
             Vote::new(3, 1).into_committed(),
             Some(log_id(3, 1, 6))
         )),
-        eng.state.accepted_io()
+        eng.state.accepted_log_io()
     );
     assert_eq!(
         &[
@@ -185,7 +185,7 @@ fn test_leader_append_entries_single_node_leader() -> anyhow::Result<()> {
             Vote::new(3, 1).into_committed(),
             Some(log_id(3, 1, 6))
         )),
-        eng.state.accepted_io()
+        eng.state.accepted_log_io()
     );
     assert_eq!(
         &[
@@ -243,7 +243,7 @@ fn test_leader_append_entries_with_membership_log() -> anyhow::Result<()> {
             Vote::new(3, 1).into_committed(),
             Some(log_id(3, 1, 6))
         )),
-        eng.state.accepted_io()
+        eng.state.accepted_log_io()
     );
     assert_eq!(
         &[

openraft/src/engine/handler/leader_handler/get_read_log_id_test.rs

@@ -32,7 +32,7 @@ fn eng() -> Engine<UTConfig> {
     eng.state.enable_validation(false); // Disable validation for incomplete state
 
     eng.config.id = 1;
-    eng.state.io_state_mut().update_committed(log_id(0, 1, 0));
+    eng.state.apply_progress_mut().accept(log_id(0, 1, 0));
     eng.state.vote = Leased::new(
         UTConfig::<()>::now(),
         Duration::from_millis(500),
@@ -54,13 +54,13 @@ fn eng() -> Engine<UTConfig> {
 fn test_get_read_log_id() -> anyhow::Result<()> {
     let mut eng = eng();
 
-    eng.state.io_state_mut().update_committed(log_id(0, 1, 0));
+    eng.state.apply_progress_mut().accept(log_id(0, 1, 0));
     eng.leader.as_mut().unwrap().noop_log_id = Some(log_id(1, 1, 2));
 
     let got = eng.leader_handler()?.get_read_log_id();
     assert_eq!(Some(log_id(1, 1, 2)), got);
 
-    eng.state.io_state_mut().update_committed(log_id(2, 1, 3));
+    eng.state.apply_progress_mut().accept(log_id(2, 1, 3));
     let got = eng.leader_handler()?.get_read_log_id();
     assert_eq!(Some(log_id(2, 1, 3)), got);
 

openraft/src/engine/handler/leader_handler/mod.rs

@@ -71,7 +71,7 @@ where C: RaftTypeConfig
             }
         }
 
-        self.state.accept_io(IOId::new_log_io(
+        self.state.accept_log_io(IOId::new_log_io(
             self.leader.committed_vote.clone(),
             self.leader.last_log_id().cloned(),
         ));