squash commits

Signed-off-by: Fullstop000 <[email protected]>

harness/src/network.rs

@@ -161,13 +161,21 @@ impl Network {
     /// Read out all messages generated by peers in the `Network`.
     ///
     /// Note: messages are not filtered by any configured filters.
+    #[inline]
     pub fn read_messages(&mut self) -> Vec<Message> {
         self.peers
             .iter_mut()
             .flat_map(|(_peer, progress)| progress.read_messages())
             .collect()
     }
 
+    #[inline]
+    pub fn read_peer_messages(&mut self, id: u64) -> Vec<Message> {
+        self.peers
+            .get_mut(&id)
+            .map_or(vec![], |node| node.read_messages())
+    }
+
     /// Instruct the cluster to `step` through the given messages.
     ///
     /// NOTE: the given `msgs` won't be filtered by its filters.
@@ -177,7 +185,11 @@ impl Network {
             let mut new_msgs = vec![];
             for m in msgs.drain(..) {
                 let resp = {
-                    let p = self.peers.get_mut(&m.to).unwrap();
+                    let p = if m.to_proxy {
+                        self.peers.get_mut(&m.get_proxy()).unwrap()
+                    } else {
+                        self.peers.get_mut(&m.to).unwrap()
+                    };
                     let _ = p.step(m);
                     p.read_messages()
                 };
@@ -197,7 +209,11 @@ impl Network {
     /// Unlike `send` this does not gather and send any responses. It also does not ignore errors.
     pub fn dispatch(&mut self, messages: impl IntoIterator<Item = Message>) -> Result<()> {
         for message in self.filter(messages.into_iter().map(Into::into)) {
-            let to = message.to;
+            let to = if message.to_proxy {
+                message.proxy
+            } else {
+                message.to
+            };
             let peer = self.peers.get_mut(&to).unwrap();
             peer.step(message)?;
         }

harness/tests/integration_cases/mod.rs

@@ -14,6 +14,7 @@
 mod test_membership_changes;
 mod test_raft;
 mod test_raft_flow_control;
+mod test_raft_follower_replication;
 mod test_raft_paper;
 mod test_raft_snap;
 mod test_raw_node;

harness/tests/integration_cases/test_raft.rs

@@ -138,10 +138,7 @@ fn next_ents(r: &mut Raft<MemStorage>, s: &MemStorage) -> Vec<Entry> {
 
 fn do_send_append(raft: &mut Raft<MemStorage>, to: u64) {
     let mut prs = raft.take_prs();
-    {
-        let pr = prs.get_mut(to).unwrap();
-        raft.send_append(to, pr);
-    }
+    raft.send_append(to, &mut prs, false);
     raft.set_prs(prs);
 }