refactor

lolraft/src/process/command_log/mod.rs

@@ -106,7 +106,7 @@ impl Inner {
 
         // If the same snapshot already exists, we can skip the insertion.
         if new_snapshot_index == cur_snapshot_index {
-            return Ok(())
+            return Ok(());
         }
 
         self.storage.insert_entry(new_snapshot_index, e).await?;

lolraft/src/process/peer_svc/replication.rs

@@ -31,7 +31,7 @@ impl PeerSvc {
         })
     }
 
-    pub async fn advance_replication(&self, follower_id: NodeId) -> Result<bool> {
+    pub async fn advance_replication(&self, follower_id: NodeId) -> Result<()> {
         let peer_context = self
             .peer_contexts
             .read()
@@ -43,10 +43,7 @@ impl PeerSvc {
         let cur_last_log_index = self.command_log.get_log_last_index().await?;
 
         // More entries to send?
-        let should_send = old_progress.next_index <= cur_last_log_index;
-        if !should_send {
-            return Ok(false);
-        }
+        ensure!(old_progress.next_index <= cur_last_log_index);
 
         // The entries to be sent may be deleted due to a previous compaction.
         // In this case, replication will reset from the current snapshot index.
@@ -62,7 +59,7 @@ impl PeerSvc {
                 .get_mut(&follower_id)
                 .context(Error::PeerNotFound(follower_id.clone()))?
                 .progress = new_progress;
-            return Ok(true);
+            return Ok(());
         }
 
         let n_max_possible = cur_last_log_index - old_progress.next_index + 1;
@@ -107,6 +104,6 @@ impl PeerSvc {
             .context(Error::PeerNotFound(follower_id.clone()))?
             .progress = new_progress;
 
-        Ok(true)
+        Ok(())
     }
 }

lolraft/src/process/thread/replication.rs

@@ -13,12 +13,11 @@ impl Thread {
             return Ok(false);
         }
 
-        let cont = self
-            .peers
+        self.peers
             .advance_replication(self.follower_id.clone())
             .await?;
 
-        Ok(cont)
+        Ok(true)
     }
 
     fn do_loop(self) -> ThreadHandle {

tests/env/tests/tests.rs

@@ -61,4 +61,4 @@ async fn drop_env() -> Result<()> {
     }
 
     Ok(())
-}
+}

tests/lol-tests/tests/multi_raft.rs

@@ -30,11 +30,10 @@ async fn N3_L100_K3_multi_raft_cluster() -> Result<()> {
         futs.push(fut);
     }
     futures::future::try_join_all(futs).await?;
-    
+
     Ok(())
 }
 
-
 #[serial]
 #[tokio::test(flavor = "multi_thread")]
 async fn N1_L100_K3_multi_raft_io() -> Result<()> {

tests/testapp/src/lib.rs

@@ -76,9 +76,7 @@ impl Client {
         use tokio_retry::Retry;
 
         // 200ms, 400, 800, 1600, 3200
-        let strategy = ExponentialBackoff::from_millis(2)
-            .factor(100)
-            .take(8);
+        let strategy = ExponentialBackoff::from_millis(2).factor(100).take(8);
 
         let fut = Retry::spawn(strategy, || {
             let mut cli = self.cli.clone();