Refactor: Remove request_id from replication

openraft/src/core/raft_core.rs

@@ -37,7 +37,6 @@ use crate::core::sm;
 use crate::core::ServerState;
 use crate::display_ext::DisplayInstantExt;
 use crate::display_ext::DisplayOptionExt;
-use crate::display_ext::DisplayResultExt;
 use crate::display_ext::DisplaySlice;
 use crate::display_ext::DisplaySliceExt;
 use crate::engine::Command;
@@ -68,7 +67,6 @@ use crate::network::RPCOption;
 use crate::network::RPCTypes;
 use crate::network::RaftNetworkFactory;
 use crate::progress::entry::ProgressEntry;
-use crate::progress::Inflight;
 use crate::progress::Progress;
 use crate::quorum::QuorumSet;
 use crate::raft::message::TransferLeaderRequest;
@@ -81,8 +79,6 @@ use crate::raft::VoteResponse;
 use crate::raft_state::io_state::io_id::IOId;
 use crate::raft_state::LogStateReader;
 use crate::replication::request::Replicate;
-use crate::replication::request_id::RequestId;
-use crate::replication::response::ReplicationResult;
 use crate::replication::ReplicationCore;
 use crate::replication::ReplicationHandle;
 use crate::replication::ReplicationSessionId;
@@ -1426,7 +1422,11 @@ where
  // If vote or membership changes, ignore the message.
  // There is chance delayed message reports a wrong state.
  if self.does_replication_session_match(&progress.session_id, "ReplicationProgress") {
- self.handle_replication_progress(progress.target, progress.request_id, progress.result);
+ tracing::debug!(progress = display(&progress), "recv Notification::ReplicationProgress");
+
+ // replication_handler() won't panic because:
+ // The leader is still valid because progress.session_id.leader_vote does not change.
+ self.engine.replication_handler().update_progress(progress.target, progress.result);
  }
  }
 
@@ -1442,9 +1442,9 @@ where
  sending_time = display(sending_time.display()),
  "HeartbeatProgress"
  );
- if self.engine.leader.is_some() {
-  self.engine.replication_handler().update_leader_clock(target, sending_time);
- }
+ // replication_handler() won't panic because:
+ // The leader is still valid because progress.session_id.leader_vote does not change.
+ self.engine.replication_handler().update_leader_clock(target, sending_time);
  }
  }
 
@@ -1551,33 +1551,6 @@ where
  self.engine.elect();
  }
 
- #[tracing::instrument(level = "debug", skip_all)]
- fn handle_replication_progress(
- &mut self,
- target: C::NodeId,
- request_id: u64,
- result: Result<ReplicationResult<C>, String>,
- ) {
- tracing::debug!(
- target = display(target),
- request_id = display(request_id),
- result = display(result.display()),
- "handle_replication_progress"
- );
-
- #[allow(clippy::collapsible_if)]
- if tracing::enabled!(Level::DEBUG) {
- if !self.replications.contains_key(&target) {
- tracing::warn!("leader has removed target: {}", target);
- };
- }
-
- // A leader may have stepped down.
- if self.engine.leader.is_some() {
- self.engine.replication_handler().update_progress(target, request_id, result);
- }
- }
-
  /// If a message is sent by a previous server state but is received by current server state,
  /// it is a stale message and should be just ignored.
  fn does_vote_match(&self, sender_vote: &Vote<C::NodeId>, msg: impl fmt::Display) -> bool {
@@ -1790,21 +1763,7 @@ where
  }
  Command::Replicate { req, target } => {
  let node = self.replications.get(&target).expect("replication to target node exists");
-
- match req {
- Inflight::None => {
- let _ = node.tx_repl.send(Replicate::Heartbeat);
- }
- Inflight::Logs { id, log_id_range } => {
- let _ = node.tx_repl.send(Replicate::logs(RequestId::new_append_entries(id), log_id_range));
- }
- Inflight::Snapshot { id, last_log_id } => {
- // unwrap: The replication channel must not be dropped or it is a bug.
- node.tx_repl.send(Replicate::snapshot(RequestId::new_snapshot(id), last_log_id)).map_err(
- |_e| StorageError::read_snapshot(None, AnyError::error("replication channel closed")),
- )?;
- }
- }
+ let _ = node.tx_repl.send(req);
  }
  Command::BroadcastTransferLeader { req } => self.broadcast_transfer_leader(req).await,
 

openraft/src/docs/protocol/snapshot_replication.md

@@ -112,7 +112,7 @@ Installing snapshot includes two steps:
 The final step is to purge logs up to [`snapshot_meta.last_log_id`].
 This step is necessary because:
 
-- 1) A local log that is <= [`snapshot_meta.last_log_id`] may conflict with the leader, and can not be used anymore.
+- 1) A local log that is `<=` [`snapshot_meta.last_log_id`] may conflict with the leader, and can not be used anymore.
 
 - 2) There may be a hole in the logs, if `snapshot_last_log_id > local_last_log_id`:
 

openraft/src/engine/command.rs

@@ -11,14 +11,14 @@ use crate::engine::CommandKind;
 use crate::error::Infallible;
 use crate::error::InitializeError;
 use crate::error::InstallSnapshotError;
-use crate::progress::Inflight;
 use crate::raft::message::TransferLeaderRequest;
 use crate::raft::AppendEntriesResponse;
 use crate::raft::InstallSnapshotResponse;
 use crate::raft::SnapshotResponse;
 use crate::raft::VoteRequest;
 use crate::raft::VoteResponse;
 use crate::raft_state::IOId;
+use crate::replication::request::Replicate;
 use crate::replication::ReplicationSessionId;
 use crate::type_config::alias::OneshotSenderOf;
 use crate::vote::CommittedVote;
@@ -96,7 +96,7 @@ where C: RaftTypeConfig
  },
 
  /// Replicate log entries or snapshot to a target.
- Replicate { target: C::NodeId, req: Inflight<C> },
+ Replicate { target: C::NodeId, req: Replicate<C> },
 
  /// Broadcast transfer Leader message to all other nodes.
  BroadcastTransferLeader { req: TransferLeaderRequest<C> },

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

@@ -14,10 +14,11 @@ use crate::engine::Command;
 use crate::engine::Engine;
 use crate::engine::ReplicationProgress;
 use crate::entry::RaftEntry;
+use crate::log_id_range::LogIdRange;
 use crate::progress::entry::ProgressEntry;
-use crate::progress::Inflight;
 use crate::raft_state::IOId;
 use crate::raft_state::LogStateReader;
+use crate::replication::request::Replicate;
 use crate::testing::blank_ent;
 use crate::testing::log_id;
 use crate::type_config::TypeConfigExt;
@@ -153,11 +154,11 @@ fn test_leader_append_entries_normal() -> anyhow::Result<()> {
  },
  Command::Replicate {
  target: 2,
- req: Inflight::logs(None, Some(log_id(3, 1, 6))).with_id(1),
+ req: Replicate::logs(LogIdRange::new(None, Some(log_id(3, 1, 6)))),
  },
  Command::Replicate {
  target: 3,
- req: Inflight::logs(None, Some(log_id(3, 1, 6))).with_id(1),
+ req: Replicate::logs(LogIdRange::new(None, Some(log_id(3, 1, 6)))),
  },
  ],
  eng.output.take_commands()
@@ -285,7 +286,7 @@ fn test_leader_append_entries_with_membership_log() -> anyhow::Result<()> {
  },
  Command::Replicate {
  target: 2,
- req: Inflight::logs(None, Some(log_id(3, 1, 6))).with_id(1),
+ req: Replicate::logs(LogIdRange::new(None, Some(log_id(3, 1, 6))))
  },
  ],
  eng.output.take_commands()

openraft/src/engine/handler/replication_handler/append_membership_test.rs

@@ -155,32 +155,27 @@ fn test_leader_append_membership_update_learner_process() -> anyhow::Result<()>
  if let Some(l) = &mut eng.leader.as_mut() {
  assert_eq!(
  &ProgressEntry::new(Some(log_id(1, 1, 4)))
- .with_inflight(Inflight::logs(Some(log_id(1, 1, 4)), Some(log_id(5, 1, 10))).with_id(1))
- .with_curr_inflight_id(1),
+ .with_inflight(Inflight::logs(Some(log_id(1, 1, 4)), Some(log_id(5, 1, 10)))),
  l.progress.get(&4),
  "learner-4 progress should be transferred to voter progress"
  );
 
  assert_eq!(
  &ProgressEntry::new(Some(log_id(1, 1, 3)))
- .with_inflight(Inflight::logs(Some(log_id(1, 1, 3)), Some(log_id(5, 1, 10))).with_id(1))
- .with_curr_inflight_id(1),
+ .with_inflight(Inflight::logs(Some(log_id(1, 1, 3)), Some(log_id(5, 1, 10)))),
  l.progress.get(&3),
  "voter-3 progress should be transferred to learner progress"
  );
 
  assert_eq!(
  &ProgressEntry::new(Some(log_id(1, 1, 5)))
- .with_inflight(Inflight::logs(Some(log_id(1, 1, 5)), Some(log_id(5, 1, 10))).with_id(1))
- .with_curr_inflight_id(1),
+ .with_inflight(Inflight::logs(Some(log_id(1, 1, 5)), Some(log_id(5, 1, 10)))),
  l.progress.get(&5),
  "learner-5 has previous value"
  );
 
  assert_eq!(
- &ProgressEntry::empty(11)
- .with_inflight(Inflight::logs(None, Some(log_id(5, 1, 10))).with_id(1))
- .with_curr_inflight_id(1),
+ &ProgressEntry::empty(11).with_inflight(Inflight::logs(None, Some(log_id(5, 1, 10)))),
  l.progress.get(&6)
  );
  } else {

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

@@ -13,6 +13,7 @@ use crate::progress::Progress;
 use crate::proposer::Leader;
 use crate::proposer::LeaderQuorumSet;
 use crate::raft_state::LogStateReader;
+use crate::replication::request::Replicate;
 use crate::replication::response::ReplicationResult;
 use crate::type_config::alias::InstantOf;
 use crate::EffectiveMembership;
@@ -145,10 +146,9 @@ where C: RaftTypeConfig
  /// Update progress when replicated data(logs or snapshot) matches on follower/learner and is
  /// accepted.
  #[tracing::instrument(level = "debug", skip_all)]
- pub(crate) fn update_matching(&mut self, node_id: C::NodeId, inflight_id: u64, log_id: Option<LogId<C::NodeId>>) {
+ pub(crate) fn update_matching(&mut self, node_id: C::NodeId, log_id: Option<LogId<C::NodeId>>) {
  tracing::debug!(
  node_id = display(node_id),
- inflight_id = display(inflight_id),
  log_id = display(log_id.display()),
  "{}",
  func_name!()
@@ -161,13 +161,7 @@ where C: RaftTypeConfig
  let quorum_accepted = *self
  .leader
  .progress
- .update_with(&node_id, |prog_entry| {
- let res = prog_entry.update_matching(inflight_id, log_id);
- if let Err(e) = &res {
- tracing::error!(error = display(e), "update_matching");
- panic!("update_matching error: {}", e);
- }
- })
+ .update_with(&node_id, |prog_entry| prog_entry.update_matching(log_id))
  .expect("it should always update existing progress");
 
  tracing::debug!(
@@ -213,33 +207,19 @@ where C: RaftTypeConfig
  /// Update progress when replicated data(logs or snapshot) does not match follower/learner state
  /// and is rejected.
  #[tracing::instrument(level = "debug", skip_all)]
- pub(crate) fn update_conflicting(&mut self, target: C::NodeId, inflight_id: u64, conflict: LogId<C::NodeId>) {
+ pub(crate) fn update_conflicting(&mut self, target: C::NodeId, conflict: LogId<C::NodeId>) {
  // TODO(2): test it?
 
  let prog_entry = self.leader.progress.get_mut(&target).unwrap();
 
- debug_assert_eq!(
- prog_entry.inflight.get_id(),
- Some(inflight_id),
- "inflight({:?}) id should match: {}",
- prog_entry.inflight,
- inflight_id
- );
-
- prog_entry.update_conflicting(inflight_id, conflict.index).unwrap();
+ prog_entry.update_conflicting(conflict.index);
  }
 
  /// Update replication progress when a response is received.
  #[tracing::instrument(level = "debug", skip_all)]
- pub(crate) fn update_progress(
- &mut self,
- target: C::NodeId,
- request_id: u64,
- repl_res: Result<ReplicationResult<C>, String>,
- ) {
+ pub(crate) fn update_progress(&mut self, target: C::NodeId, repl_res: Result<ReplicationResult<C>, String>) {
  tracing::debug!(
  target = display(target),
- request_id = display(request_id),
  result = display(repl_res.display()),
  progress = display(&self.leader.progress),
  "{}",
@@ -249,29 +229,17 @@ where C: RaftTypeConfig
  match repl_res {
  Ok(p) => match p.0 {
  Ok(matching) => {
- self.update_matching(target, request_id, matching);
+ self.update_matching(target, matching);
  }
  Err(conflict) => {
- self.update_conflicting(target, request_id, conflict);
+ self.update_conflicting(target, conflict);
  }
  },
  Err(err_str) => {
- tracing::warn!(
- request_id = display(request_id),
- result = display(&err_str),
- "update progress error"
- );
+ tracing::warn!(result = display(&err_str), "update progress error");
 
  // Reset inflight state and it will retry.
  let p = self.leader.progress.get_mut(&target).unwrap();
-
- debug_assert!(
- p.inflight.is_my_id(request_id),
- "inflight({:?}) id should match: {}",
- p.inflight,
- request_id
- );
-
  p.inflight = Inflight::None;
  }
  };
@@ -328,10 +296,12 @@ where C: RaftTypeConfig
 
  #[tracing::instrument(level = "debug", skip_all)]
  pub(crate) fn send_to_target(output: &mut EngineOutput<C>, target: &C::NodeId, inflight: &Inflight<C>) {
- output.push_command(Command::Replicate {
- target: *target,
- req: *inflight,
- });
+ let req = match inflight {
+ Inflight::None => unreachable!("no data to send"),
+ Inflight::Logs { log_id_range } => Replicate::logs(*log_id_range),
+ Inflight::Snapshot { last_log_id } => Replicate::snapshot(*last_log_id),
+ };
+ output.push_command(Command::Replicate { target: *target, req });
  }
 
  /// Try to run a pending purge job, if no tasks are using the logs to be purged.
@@ -403,8 +373,7 @@ where C: RaftTypeConfig
  // TODO: It should be self.state.last_log_id() but None is ok.
  prog_entry.inflight = Inflight::logs(None, upto);
 
- let inflight_id = prog_entry.inflight.get_id().unwrap();
- self.update_matching(id, inflight_id, upto);
+ self.update_matching(id, upto);
  }
  }