Refactor: Use CommittedLeaderId in place of LeaderId for LogIOId

In this commit, `LeaderId` has been replaced with `CommittedLeaderId`
for identifying log IO requests. This change is made because
`CommittedLeaderId` is shorter and sufficiently identifies the necessary
attributes since only an established leader (a committed leader) can
write or replicate logs. Thus, using `CommittedLeaderId` streamlines the
identification process for log IO requests.

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

@@ -136,9 +136,6 @@ where C: RaftTypeConfig
  self.state.extend_log_ids(&entries);
  self.append_membership(entries.iter());
 
- // TODO: with asynchronous IO in future,
- // do not write log until vote being committed,
- // or consistency is broken.
  self.output.push_command(Command::AppendInputEntries {
  // A follower should always use the node's vote.
  vote: *self.state.vote_ref(),

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

@@ -65,9 +65,25 @@ where C: RaftTypeConfig
  }
  }
 
- // TODO: with asynchronous IO in future,
- // do not write log until vote being committed,
- // or consistency is broken.
+ // TODO: In future implementations with asynchronous IO,
+ // ensure logs are not written until the vote is committed
+ // to maintain consistency.
+ // ---
+ // Currently, IO requests to `RaftLogStorage` are executed
+ // within the `RaftCore` task. This means an `AppendLog` request
+ // won't be submitted to `RaftLogStorage` until `save_vote()` completes,
+ // which ensures consistency.
+ // ---
+ // However, when `RaftLogStorage` is moved to a separate task,
+ // `RaftCore` will communicate with `RaftLogStorage` via a channel.
+ // This change could result in `AppendLog` requests being submitted
+ // before the previous `save_vote()` request is finished.
+ // ---
+ // This scenario creates a risk where a log entry becomes visible and
+ // is replicated by `ReplicationCore` to other nodes before the vote
+ // is flushed to disk. If the vote isn't flushed and the server restarts,
+ // the vote could revert to a previous state. This could allow a new leader
+ // to be elected with a smaller vote (term), breaking consistency.
  self.output.push_command(Command::AppendInputEntries {
  // A leader should always use the leader's vote.
  // It is allowed to be different from local vote.

openraft/src/raft_state/io_state/log_io_id.rs

@@ -1,7 +1,7 @@
 use std::fmt;
 
 use crate::display_ext::DisplayOptionExt;
-use crate::LeaderId;
+use crate::CommittedLeaderId;
 use crate::LogId;
 use crate::NodeId;
 
@@ -18,22 +18,22 @@ use crate::NodeId;
 #[derive(PartialEq, Eq)]
 pub(crate) struct LogIOId<NID: NodeId> {
  /// The id of the leader that performs the log io operation.
- pub(crate) leader_id: LeaderId<NID>,
+ pub(crate) committed_leader_id: CommittedLeaderId<NID>,
 
  /// The last log id that has been flushed to storage.
  pub(crate) log_id: Option<LogId<NID>>,
 }
 
 impl<NID: NodeId> fmt::Display for LogIOId<NID> {
  fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
- write!(f, "by_leader({}):{}", self.leader_id, self.log_id.display())
+ write!(f, "by_leader({}):{}", self.committed_leader_id, self.log_id.display())
  }
 }
 
 impl<NID: NodeId> LogIOId<NID> {
- pub(crate) fn new(leader_id: impl Into<LeaderId<NID>>, log_id: Option<LogId<NID>>) -> Self {
+ pub(crate) fn new(committed_leader_id: impl Into<CommittedLeaderId<NID>>, log_id: Option<LogId<NID>>) -> Self {
  Self {
- leader_id: leader_id.into(),
+ committed_leader_id: committed_leader_id.into(),
  log_id,
  }
  }

openraft/src/vote/leader_id/impl_into_leader_id.rs

@@ -1,9 +1,9 @@
 use crate::node::NodeId;
-use crate::vote::leader_id::LeaderId;
+use crate::vote::leader_id::CommittedLeaderId;
 use crate::vote::Vote;
 
-impl<NID: NodeId> From<Vote<NID>> for LeaderId<NID> {
+impl<NID: NodeId> From<Vote<NID>> for CommittedLeaderId<NID> {
  fn from(vote: Vote<NID>) -> Self {
- vote.leader_id
+ vote.leader_id.to_committed()
  }
 }