refactor: replace has_payload with inflight_id presence check

Simplify replication progress tracking by removing the redundant
`has_payload` boolean. The presence of `inflight_id` (Some vs None)
now indicates whether a response requires inflight state update.

`InflightId` identifies whether an AppendEntries RPC carries actual log
data that needs acknowledgment. Not all RPCs have a corresponding
inflight record on the leader. For example, when synchronizing the
committed index, the leader sends an AppendEntries RPC without log
payload - such RPCs have no `InflightId`. However, RPCs that replicate
actual log entries have a corresponding inflight record on the leader,
and these RPCs carry an `InflightId`. This distinction determines
whether a response should update the inflight state.

Changes:
- Remove `has_payload` field from `Notification::ReplicationProgress`
- Change `Inflight::ack()` and `Inflight::conflict()` to require matching `InflightId`
- Restructure `Data::Logs` to include `inflight_id` field
- Move `inflight_id` tracking into `ReplicationCore`
- Update `update_conflicting()` and `update_progress()` signatures

Author: drmingdrmer

openraft/src/core/heartbeat/worker.rs

@@ -135,7 +135,6 @@ where
                             let conflict_log_id = heartbeat.matching.clone().unwrap();
 
                             let noti = Notification::ReplicationProgress {
-                                has_payload: false,
                                 progress: Progress {
                                     session_id: heartbeat.session_id.clone(),
                                     target: self.target.clone(),

openraft/src/core/notification.rs

@@ -53,11 +53,17 @@ where C: RaftTypeConfig
 
     /// Result of executing a command sent from network worker.
     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>,
+
+        /// The `InflightId` of the replication request that produced this response.
+        ///
+        /// - `Some(id)`: This response corresponds to a replication request that carries log
+        ///   payload. The `id` is used to match the response to the correct inflight state,
+        ///   allowing the leader to update `matching` or handle conflicts properly.
+        ///
+        /// - `None`: This response is from an RPC without log payload (e.g., a heartbeat to
+        ///   synchronize commit index). Such RPCs don't have corresponding inflight records, so no
+        ///   inflight state update is needed.
         inflight_id: Option<InflightId>,
     },
 
@@ -114,13 +120,8 @@ where C: RaftTypeConfig
             }
             Self::StorageError { error } => write!(f, "StorageError: {}", error),
             Self::LocalIO { io_id } => write!(f, "IOFlushed: {}", io_id),
-            Self::ReplicationProgress {
-                has_payload,
-                progress,
-                inflight_id,
-            } => {
-                let payload = if *has_payload { "no-payload" } else { "has-payload" };
-                write!(f, "{payload}: {}, inflight_id: {}", progress, inflight_id.display())
+            Self::ReplicationProgress { progress, inflight_id } => {
+                write!(f, "{}, inflight_id: {}", progress, inflight_id.display())
             }
             Self::HeartbeatProgress {
                 session_id: leader_vote,

openraft/src/core/raft_core.rs

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

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

@@ -151,11 +151,11 @@ fn test_leader_append_entries_normal() -> anyhow::Result<()> {
             },
             Command::Replicate {
                 target: 2,
-                req: Replicate::logs(LogIdRange::new(None, Some(log_id(3, 1, 6))), Some(InflightId::new(1))),
+                req: Replicate::logs(LogIdRange::new(None, Some(log_id(3, 1, 6))), InflightId::new(1)),
             },
             Command::Replicate {
                 target: 3,
-                req: Replicate::logs(LogIdRange::new(None, Some(log_id(3, 1, 6))), Some(InflightId::new(2))),
+                req: Replicate::logs(LogIdRange::new(None, Some(log_id(3, 1, 6))), InflightId::new(2)),
             },
         ],
         eng.output.take_commands()
@@ -280,7 +280,7 @@ fn test_leader_append_entries_with_membership_log() -> anyhow::Result<()> {
             },
             Command::Replicate {
                 target: 2,
-                req: Replicate::logs(LogIdRange::new(None, Some(log_id(3, 1, 6))), Some(InflightId::new(1)))
+                req: Replicate::logs(LogIdRange::new(None, Some(log_id(3, 1, 6))), InflightId::new(1))
             },
         ],
         eng.output.take_commands()

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

@@ -217,7 +217,6 @@ where C: RaftTypeConfig
         &mut self,
         target: C::NodeId,
         conflict: LogIdOf<C>,
-        has_payload: bool,
         inflight_id: Option<InflightId>,
     ) {
         // TODO(2): test it?
@@ -226,7 +225,7 @@ where C: RaftTypeConfig
 
         let mut updater = progress::entry::update::Updater::new(self.config, prog_entry);
 
-        updater.update_conflicting(conflict.index(), has_payload, inflight_id);
+        updater.update_conflicting(conflict.index(), inflight_id);
     }
 
     /// Enable one-time replication reset for a specific node upon log reversion detection.
@@ -263,13 +262,13 @@ where C: RaftTypeConfig
         &mut self,
         target: C::NodeId,
         repl_res: Result<ReplicationResult<C>, String>,
-        has_payload: bool,
         inflight_id: Option<InflightId>,
     ) {
         tracing::debug!(
-            "{}: target={target}, result={}, has_payload={has_payload}, current progresses={}",
+            "{}: target={target}, result={}, inflight_id={}, current progresses={}",
             func_name!(),
             repl_res.display(),
+            inflight_id.display(),
             self.leader.progress
         );
 
@@ -279,7 +278,7 @@ where C: RaftTypeConfig
                     self.update_matching(target, matching, inflight_id);
                 }
                 Err(conflict) => {
-                    self.update_conflicting(target, conflict, has_payload, inflight_id);
+                    self.update_conflicting(target, conflict, inflight_id);
                 }
             },
             Err(err_str) => {
@@ -349,7 +348,7 @@ where C: RaftTypeConfig
                 log_id_range,
                 inflight_id,
             } => {
-                let req = Replicate::logs(log_id_range.clone(), Some(*inflight_id));
+                let req = Replicate::logs(log_id_range.clone(), *inflight_id);
                 output.push_command(Command::Replicate {
                     target: target.clone(),
                     req,
@@ -433,7 +432,7 @@ where C: RaftTypeConfig
             // TODO: It should be self.state.last_log_id() but None is ok.
             prog_entry.inflight = Inflight::logs(None, upto.clone(), InflightId::new(0));
 
-            self.update_matching(id, upto, None);
+            self.update_matching(id, upto, Some(InflightId::new(0)));
         }
     }
 

openraft/src/engine/handler/vote_handler/become_leader_test.rs

@@ -73,7 +73,7 @@ fn test_become_leader() -> anyhow::Result<()> {
         },
         Command::Replicate {
             target: 0,
-            req: Replicate::logs(LogIdRange::new(None, Some(log_id(2, 1, 0))), Some(InflightId::new(1)))
+            req: Replicate::logs(LogIdRange::new(None, Some(log_id(2, 1, 0))), InflightId::new(1))
         }
     ]);
 

openraft/src/engine/tests/handle_vote_resp_test.rs

@@ -219,7 +219,7 @@ fn test_handle_vote_resp_equal_vote() -> anyhow::Result<()> {
                 },
                 Command::Replicate {
                     target: 2,
-                    req: Replicate::logs(LogIdRange::new(None, Some(log_id(2, 1, 1))), Some(InflightId::new(1)))
+                    req: Replicate::logs(LogIdRange::new(None, Some(log_id(2, 1, 1))), InflightId::new(1))
                 },
             ],
             eng.output.take_commands()

openraft/src/engine/tests/startup_test.rs

@@ -89,7 +89,7 @@ fn test_startup_as_leader_without_logs() -> anyhow::Result<()> {
             },
             Command::Replicate {
                 target: 3,
-                req: Replicate::logs(LogIdRange::new(None, Some(log_id(2, 2, 4))), Some(InflightId::new(1))),
+                req: Replicate::logs(LogIdRange::new(None, Some(log_id(2, 2, 4))), InflightId::new(1)),
             }
         ],
         eng.output.take_commands()
@@ -137,7 +137,7 @@ fn test_startup_as_leader_with_proposed_logs() -> anyhow::Result<()> {
             },
             Command::Replicate {
                 target: 3,
-                req: Replicate::logs(LogIdRange::new(None, Some(log_id(1, 2, 6))), Some(InflightId::new(1)))
+                req: Replicate::logs(LogIdRange::new(None, Some(log_id(1, 2, 6))), InflightId::new(1))
             }
         ],
         eng.output.take_commands()

openraft/src/progress/entry/tests.rs

@@ -82,7 +82,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, true, Some(InflightId::new(0)));
+    pe.new_updater(&engine_config).update_conflicting(5, Some(InflightId::new(0)));
 
     assert_eq!(Inflight::None, pe.inflight);
     assert_eq!(&Some(log_id(3)), pe.borrow());

openraft/src/progress/entry/update.rs

@@ -26,8 +26,9 @@ where C: RaftTypeConfig
     /// The conflicting log index is the last log index found on a follower that does not match
     /// the leader's log at that position.
     ///
-    /// If `has_payload` is true, the `inflight` state is reset because AppendEntries RPC
-    /// manages the inflight state.
+    /// If `inflight_id` is `Some`, the inflight state is reset because the response corresponds
+    /// to a replication request with log payload. If `None`, the response is from an RPC without
+    /// payload (e.g., heartbeat), and inflight state is not modified.
     ///
     /// 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
@@ -36,15 +37,15 @@ where C: RaftTypeConfig
     /// 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, has_payload: bool, inflight_id: Option<InflightId>) {
+    pub(crate) fn update_conflicting(&mut self, conflict: u64, inflight_id: Option<InflightId>) {
         tracing::debug!(
             "update_conflict: current progress_entry: {}; conflict: {}",
             self.entry,
             conflict
         );
 
         // The inflight may be None if the conflict is caused by a heartbeat response.
-        if has_payload {
+        if let Some(inflight_id) = inflight_id {
             self.entry.inflight.conflict(conflict, inflight_id);
         }
 
@@ -91,14 +92,21 @@ where C: RaftTypeConfig
         }
     }
 
+    /// Update the matching log id for this follower when replication succeeds.
+    ///
+    /// If `inflight_id` is `Some`, the inflight state is acknowledged because the response
+    /// corresponds to a replication request with log payload. If `None`, the response is from
+    /// an RPC without payload (e.g., heartbeat), and inflight state is not modified.
     pub(crate) fn update_matching(&mut self, matching: Option<LogIdOf<C>>, inflight_id: Option<InflightId>) {
         tracing::debug!(
             "update_matching: current progress_entry: {}; matching: {}",
             self.entry,
             matching.display()
         );
 
-        self.entry.inflight.ack(matching.clone(), inflight_id);
+        if let Some(inflight_id) = inflight_id {
+            self.entry.inflight.ack(matching.clone(), inflight_id);
+        }
 
         debug_assert!(matching.as_ref() >= self.entry.matching());
         self.entry.matching = matching;