fix: reset progress when heartbeat response indicates conflict

With `allow_log_reversion` enabled:

Before this fix, `RaftCore` ignored `conflict` messages from `Heartbeat` RPCs,
preventing leaders from discovering follower state changes. When a follower's
state reverted and responded with a conflict message, the leader wouldn't
retransmit necessary data to the follower.

This commit ensures `conflict` responses are always processed properly and
progress is reset to trigger data retransmission to the follower.

In this commit, a `Heartbeat` message uses `committed` log id as the
`prev_log_id` to detect Follower state reversion.

Author: drmingdrmer

openraft/src/core/heartbeat/errors/mod.rs

@@ -0,0 +1,5 @@
+mod raft_core_closed;
+mod stopped;
+
+pub(crate) use raft_core_closed::RaftCoreClosed;
+pub(crate) use stopped::Stopped;

openraft/src/core/heartbeat/errors/raft_core_closed.rs

@@ -0,0 +1,3 @@
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+#[error("RaftCore closed receiver")]
+pub(crate) struct RaftCoreClosed;

openraft/src/core/heartbeat/errors/stopped.rs

@@ -0,0 +1,10 @@
+use crate::core::heartbeat::errors::raft_core_closed::RaftCoreClosed;
+
+#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
+pub enum Stopped {
+    #[error("HeartbeatWorkerStopped: {0}")]
+    RaftCoreClosed(#[from] RaftCoreClosed),
+
+    #[error("HeartbeatWorkerStopped: received shutdown signal")]
+    ReceivedShutdown,
+}

openraft/src/core/heartbeat/mod.rs

@@ -1,3 +1,4 @@
+pub(crate) mod errors;
 pub(crate) mod event;
 pub(crate) mod handle;
 pub(crate) mod worker;

openraft/src/core/heartbeat/worker.rs

@@ -6,11 +6,16 @@ use futures::FutureExt;
 
 use crate::async_runtime::watch::WatchReceiver;
 use crate::async_runtime::MpscUnboundedSender;
+use crate::core::heartbeat::errors::RaftCoreClosed;
+use crate::core::heartbeat::errors::Stopped;
 use crate::core::heartbeat::event::HeartbeatEvent;
 use crate::core::notification::Notification;
 use crate::network::v2::RaftNetworkV2;
 use crate::network::RPCOption;
 use crate::raft::AppendEntriesRequest;
+use crate::raft::AppendEntriesResponse;
+use crate::replication::response::ReplicationResult;
+use crate::replication::Progress;
 use crate::type_config::alias::MpscUnboundedSenderOf;
 use crate::type_config::alias::OneshotReceiverOf;
 use crate::type_config::alias::WatchReceiverOf;
@@ -59,14 +64,19 @@ where
     C: RaftTypeConfig,
     N: RaftNetworkV2<C>,
 {
-    pub(crate) async fn run(mut self, mut rx_shutdown: OneshotReceiverOf<C, ()>) {
+    pub(crate) async fn run(self, rx_shutdown: OneshotReceiverOf<C, ()>) {
+        let res = self.do_run(rx_shutdown).await;
+        tracing::info!("HeartbeatWorker finished with result: {:?}", res);
+    }
+
+    pub(crate) async fn do_run(mut self, mut rx_shutdown: OneshotReceiverOf<C, ()>) -> Result<(), Stopped> {
         loop {
             tracing::debug!("{} is waiting for a new heartbeat event.", self);
 
             futures::select! {
                 _ = (&mut rx_shutdown).fuse() => {
                     tracing::info!("{} is shutdown.", self);
-                    return;
+                    return Err(Stopped::ReceivedShutdown);
                 },
                 _ = self.rx.changed().fuse() => {},
             }
@@ -83,7 +93,9 @@ where
 
             let payload = AppendEntriesRequest {
                 vote: heartbeat.session_id.leader_vote.clone().into_vote(),
-                prev_log_id: None,
+                // Use committed log id as prev_log_id to detect follower state reversion.
+                // prev_log_id == None does not conflict.
+                prev_log_id: heartbeat.committed.clone(),
                 leader_commit: heartbeat.committed.clone(),
                 entries: vec![],
             };
@@ -92,22 +104,65 @@ where
             tracing::debug!("{} sent a heartbeat: {}, result: {:?}", self, heartbeat, res);
 
             match res {
-                Ok(Ok(_)) => {
-                    let res = self.tx_notification.send(Notification::HeartbeatProgress {
+                Ok(Ok(x)) => {
+                    let response: AppendEntriesResponse<C> = x;
+
+                    match response {
+                        AppendEntriesResponse::Success => {}
+                        AppendEntriesResponse::PartialSuccess(_matching) => {}
+                        AppendEntriesResponse::HigherVote(vote) => {
+                            tracing::debug!(
+                                "seen a higher vote({vote}) from {}; when:(sending heartbeat)",
+                                self.target
+                            );
+
+                            let noti = Notification::HigherVote {
+                                target: self.target.clone(),
+                                higher: vote,
+                                leader_vote: heartbeat.session_id.committed_vote(),
+                            };
+
+                            self.send_notification(noti, "Seeing higher Vote")?;
+                        }
+                        AppendEntriesResponse::Conflict => {
+                            let conflict = heartbeat.committed.unwrap();
+
+                            let noti = Notification::ReplicationProgress {
+                                has_payload: false,
+                                progress: Progress {
+                                    session_id: heartbeat.session_id.clone(),
+                                    target: self.target.clone(),
+                                    result: Ok(ReplicationResult(Err(conflict))),
+                                },
+                            };
+
+                            self.send_notification(noti, "Seeing conflict")?;
+                        }
+                    }
+
+                    let noti = Notification::HeartbeatProgress {
                         session_id: heartbeat.session_id.clone(),
                         sending_time: heartbeat.time,
                         target: self.target.clone(),
-                    });
+                    };
 
-                    if res.is_err() {
-                        tracing::error!("{} failed to send a heartbeat progress to RaftCore. quit", self);
-                        return;
-                    }
+                    self.send_notification(noti, "send HeartbeatProgress")?;
                 }
                 _ => {
                     tracing::warn!("{} failed to send a heartbeat: {:?}", self, res);
                 }
             }
         }
     }
+
+    fn send_notification(&self, notification: Notification<C>, when: impl fmt::Display) -> Result<(), RaftCoreClosed> {
+        let res = self.tx_notification.send(notification);
+
+        if let Err(e) = res {
+            let notification = e.0;
+            tracing::error!("{self} failed to send {notification} to RaftCore; when:({when})");
+            return Err(RaftCoreClosed);
+        }
+        Ok(())
+    }
 }

openraft/src/core/notification.rs

@@ -50,7 +50,13 @@ where C: RaftTypeConfig
     LocalIO { io_id: IOId<C> },
 
     /// Result of executing a command sent from network worker.
-    ReplicationProgress { progress: replication::Progress<C> },
+    ReplicationProgress {
+        /// If this progress from RPC with payload.
+        ///
+        /// `has_payload`: contain payload and should reset `inflight` state if conflict.
+        has_payload: bool,
+        progress: replication::Progress<C>,
+    },
 
     HeartbeatProgress {
         session_id: ReplicationSessionId<C>,
@@ -105,8 +111,9 @@ where C: RaftTypeConfig
             }
             Self::StorageError { error } => write!(f, "StorageError: {}", error),
             Self::LocalIO { io_id } => write!(f, "IOFlushed: {}", io_id),
-            Self::ReplicationProgress { progress } => {
-                write!(f, "{}", progress)
+            Self::ReplicationProgress { has_payload, progress } => {
+                let payload = if *has_payload { "no-payload" } else { "has-payload" };
+                write!(f, "{payload}: {}", progress)
             }
             Self::HeartbeatProgress {
                 session_id: leader_vote,

openraft/src/core/raft_core.rs

@@ -1399,15 +1399,15 @@ where
                 }
             }
 
-            Notification::ReplicationProgress { progress } => {
+            Notification::ReplicationProgress { has_payload, progress } => {
                 // 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") {
                     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);
+                    self.engine.replication_handler().update_progress(progress.target, progress.result, has_payload);
                 }
             }
 

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

@@ -213,15 +213,18 @@ where C: RaftTypeConfig
 
     /// Update progress when replicated data(logs or snapshot) does not match follower/learner state
     /// and is rejected.
+    ///
+    /// If `has_payload` is true, the `inflight` state is reset because AppendEntries RPC
+    /// manages the inflight state.
     #[tracing::instrument(level = "debug", skip_all)]
-    pub(crate) fn update_conflicting(&mut self, target: C::NodeId, conflict: LogIdOf<C>) {
+    pub(crate) fn update_conflicting(&mut self, target: C::NodeId, conflict: LogIdOf<C>, has_payload: bool) {
         // TODO(2): test it?
 
         let prog_entry = self.leader.progress.get_mut(&target).unwrap();
 
         let mut updater = progress::entry::update::Updater::new(self.config, prog_entry);
 
-        updater.update_conflicting(conflict.index());
+        updater.update_conflicting(conflict.index(), has_payload);
     }
 
     /// Enable one-time replication reset for a specific node upon log reversion detection.
@@ -254,13 +257,17 @@ where C: RaftTypeConfig
 
     /// Update replication progress when a response is received.
     #[tracing::instrument(level = "debug", skip_all)]
-    pub(crate) fn update_progress(&mut self, target: C::NodeId, repl_res: Result<ReplicationResult<C>, String>) {
+    pub(crate) fn update_progress(
+        &mut self,
+        target: C::NodeId,
+        repl_res: Result<ReplicationResult<C>, String>,
+        has_payload: bool,
+    ) {
         tracing::debug!(
-            target = display(&target),
-            result = display(repl_res.display()),
-            progress = display(&self.leader.progress),
-            "{}",
-            func_name!()
+            "{}: target={target}, result={}, has_payload={has_payload}, current progresses={}",
+            func_name!(),
+            repl_res.display(),
+            self.leader.progress
         );
 
         match repl_res {
@@ -269,7 +276,7 @@ where C: RaftTypeConfig
                     self.update_matching(target, matching);
                 }
                 Err(conflict) => {
-                    self.update_conflicting(target, conflict);
+                    self.update_conflicting(target, conflict, has_payload);
                 }
             },
             Err(err_str) => {

openraft/src/progress/entry/tests.rs

@@ -79,7 +79,7 @@ fn test_update_conflicting() -> anyhow::Result<()> {
     pe.inflight = inflight_logs(5, 10);
 
     let engine_config = EngineConfig::new_default(1);
-    pe.new_updater(&engine_config).update_conflicting(5);
+    pe.new_updater(&engine_config).update_conflicting(5, true);
 
     assert_eq!(Inflight::None, pe.inflight);
     assert_eq!(&Some(log_id(3)), pe.borrow());

openraft/src/progress/entry/update.rs

@@ -20,28 +20,42 @@ where C: RaftTypeConfig
         Self { engine_config, entry }
     }
 
-    /// Update conflicting log index.
+    /// Update the conflicting log index for this follower.
     ///
-    /// Conflicting log index is the last found log index on a follower that is not matching the
-    /// leader log.
+    /// The conflicting log index is the last log index found on a follower that does not match
+    /// the leader's log at that position.
     ///
-    /// Usually if follower's data is not lost, `conflict` is always greater than or equal
-    /// `matching`. But for testing purpose, a follower is allowed to clean its data and wait
-    /// for leader to replicate all data to it.
+    /// If `has_payload` is true, the `inflight` state is reset because AppendEntries RPC
+    /// manages the inflight state.
     ///
-    /// To allow a follower to clean its data, set the config [`Config::allow_log_reversion`] .
+    /// Normally, the `conflict` index should be greater than or equal to the `matching` index
+    /// when follower data is intact. However, for testing purposes, a follower may clean its
+    /// data and require the leader to replicate all data from the beginning.
+    ///
+    /// To allow follower log reversion, enable [`Config::allow_log_reversion`].
     ///
     /// [`Config::allow_log_reversion`]: `crate::config::Config::allow_log_reversion`
-    pub(crate) fn update_conflicting(&mut self, conflict: u64) {
+    pub(crate) fn update_conflicting(&mut self, conflict: u64, has_payload: bool) {
         tracing::debug!(
             "update_conflict: current progress_entry: {}; conflict: {}",
             self.entry,
             conflict
         );
 
-        self.entry.inflight.conflict(conflict);
+        // The inflight may be None if the conflict is caused by a heartbeat response.
+        if has_payload {
+            self.entry.inflight.conflict(conflict);
+        }
+
+        if conflict >= self.entry.searching_end {
+            tracing::debug!(
+                "conflict {} >= searching_end {}; no need to update",
+                conflict,
+                self.entry.searching_end
+            );
+            return;
+        }
 
-        debug_assert!(conflict < self.entry.searching_end);
         self.entry.searching_end = conflict;
 
         // An already matching log id is found lost: