Fix: ProgressEntry::is_log_range_inflight() checks a log range, not a…

… log entry

This bug causes replication tries to send pruged log.

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

@@ -358,7 +358,7 @@ where
         // Check if any replication task is going to use the log that are going to purge.
         let mut in_use = false;
         for (id, prog_entry) in self.leader.progress.iter() {
-            if prog_entry.is_inflight(&purge_upto) {
+            if prog_entry.is_log_range_inflight(&purge_upto) {
                 tracing::debug!("log {} is in use by {}", purge_upto, id);
                 in_use = true;
             }

openraft/src/progress/entry.rs

@@ -73,15 +73,15 @@ impl<NID: NodeId> ProgressEntry<NID> {
         self
     }
 
-    /// Return if a log id is inflight sending.
+    /// Return if a range of log id `..=log_id` is inflight sending.
     ///
     /// `prev_log_id` is never inflight.
-    pub(crate) fn is_inflight(&self, log_id: &LogId<NID>) -> bool {
+    pub(crate) fn is_log_range_inflight(&self, upto: &LogId<NID>) -> bool {
         match &self.inflight {
             Inflight::None => false,
             Inflight::Logs { log_id_range, .. } => {
-                let lid = Some(*log_id);
-                lid > log_id_range.prev_log_id && lid <= log_id_range.last_log_id
+                let lid = Some(*upto);
+                lid > log_id_range.prev_log_id
             }
             Inflight::Snapshot { last_log_id: _, .. } => false,
         }
@@ -257,19 +257,19 @@ mod tests {
         Inflight::logs(Some(log_id(prev_index)), Some(log_id(last_index)))
     }
     #[test]
-    fn test_is_inflight() -> anyhow::Result<()> {
+    fn test_is_log_range_inflight() -> anyhow::Result<()> {
         let mut pe = ProgressEntry::empty(20);
-        assert_eq!(false, pe.is_inflight(&log_id(2)));
+        assert_eq!(false, pe.is_log_range_inflight(&log_id(2)));
 
         pe.inflight = inflight_logs(2, 4);
-        assert_eq!(false, pe.is_inflight(&log_id(1)));
-        assert_eq!(false, pe.is_inflight(&log_id(2)));
-        assert_eq!(true, pe.is_inflight(&log_id(3)));
-        assert_eq!(true, pe.is_inflight(&log_id(4)));
-        assert_eq!(false, pe.is_inflight(&log_id(5)));
+        assert_eq!(false, pe.is_log_range_inflight(&log_id(1)));
+        assert_eq!(false, pe.is_log_range_inflight(&log_id(2)));
+        assert_eq!(true, pe.is_log_range_inflight(&log_id(3)));
+        assert_eq!(true, pe.is_log_range_inflight(&log_id(4)));
+        assert_eq!(true, pe.is_log_range_inflight(&log_id(5)));
 
         pe.inflight = Inflight::snapshot(Some(log_id(5)));
-        assert_eq!(false, pe.is_inflight(&log_id(5)));
+        assert_eq!(false, pe.is_log_range_inflight(&log_id(5)));
 
         Ok(())
     }

tests/tests/snapshot/t40_purge_in_snapshot_logs.rs

@@ -49,6 +49,10 @@ async fn purge_in_snapshot_logs() -> Result<()> {
             )
             .await?;
         let mut sto0 = router.get_storage_handle(&0)?;
+
+        // Wait for purge to complete.
+        sleep(Duration::from_millis(500)).await;
+
         let logs = sto0.try_get_log_entries(..).await?;
         assert_eq!(max_keep as usize, logs.len());
     }