Make request_snapshot more safer (#499)

close #498

Signed-off-by: Wenbo Zhang <ethercflow@gmail.com>

Author: ethercflow

harness/tests/integration_cases/test_raft.rs

@@ -4860,11 +4860,7 @@ fn test_follower_request_snapshot() {
     let prev_snapshot_idx = s.get_metadata().index;
     let request_idx = nt.peers[&1].raft_log.committed;
     assert!(prev_snapshot_idx < request_idx);
-    nt.peers
-        .get_mut(&2)
-        .unwrap()
-        .request_snapshot(request_idx)
-        .unwrap();
+    nt.peers.get_mut(&2).unwrap().request_snapshot().unwrap();
 
     // Send the request snapshot message.
     let req_snap = nt.peers.get_mut(&2).unwrap().msgs.pop().unwrap();
@@ -4909,11 +4905,7 @@ fn test_request_snapshot_unavailable() {
     let prev_snapshot_idx = s.get_metadata().index;
     let request_idx = nt.peers[&1].raft_log.committed;
     assert!(prev_snapshot_idx < request_idx);
-    nt.peers
-        .get_mut(&2)
-        .unwrap()
-        .request_snapshot(request_idx)
-        .unwrap();
+    nt.peers.get_mut(&2).unwrap().request_snapshot().unwrap();
 
     // Send the request snapshot message.
     let req_snap = nt.peers.get_mut(&2).unwrap().msgs.pop().unwrap();
@@ -4966,12 +4958,7 @@ fn test_request_snapshot_matched_change() {
     nt.peers.get_mut(&2).unwrap().raft_log.committed -= 1;
 
     // Request the latest snapshot.
-    let request_idx = nt.peers[&2].raft_log.committed;
-    nt.peers
-        .get_mut(&2)
-        .unwrap()
-        .request_snapshot(request_idx)
-        .unwrap();
+    nt.peers.get_mut(&2).unwrap().request_snapshot().unwrap();
     let req_snap = nt.peers.get_mut(&2).unwrap().msgs.pop().unwrap();
     // The request snapshot is ignored because it is considered as out of order.
     nt.peers.get_mut(&1).unwrap().step(req_snap).unwrap();
@@ -5015,12 +5002,7 @@ fn test_request_snapshot_none_replicate() {
         .state = ProgressState::Probe;
 
     // Request the latest snapshot.
-    let request_idx = nt.peers[&2].raft_log.committed;
-    nt.peers
-        .get_mut(&2)
-        .unwrap()
-        .request_snapshot(request_idx)
-        .unwrap();
+    nt.peers.get_mut(&2).unwrap().request_snapshot().unwrap();
     let req_snap = nt.peers.get_mut(&2).unwrap().msgs.pop().unwrap();
     nt.peers.get_mut(&1).unwrap().step(req_snap).unwrap();
     assert!(nt.peers[&1].prs().get(2).unwrap().pending_request_snapshot != 0);
@@ -5042,12 +5024,7 @@ fn test_request_snapshot_step_down() {
 
     // Recover and request the latest snapshot.
     nt.recover();
-    let request_idx = nt.peers[&2].raft_log.committed;
-    nt.peers
-        .get_mut(&2)
-        .unwrap()
-        .request_snapshot(request_idx)
-        .unwrap();
+    nt.peers.get_mut(&2).unwrap().request_snapshot().unwrap();
     nt.send(vec![new_message(3, 3, MessageType::MsgBeat, 0)]);
     assert!(
         nt.peers[&2].pending_request_snapshot == INVALID_INDEX,
@@ -5061,12 +5038,7 @@ fn test_request_snapshot_step_down() {
 fn test_request_snapshot_on_role_change() {
     let (mut nt, _) = prepare_request_snapshot();
 
-    let request_idx = nt.peers[&2].raft_log.committed;
-    nt.peers
-        .get_mut(&2)
-        .unwrap()
-        .request_snapshot(request_idx)
-        .unwrap();
+    nt.peers.get_mut(&2).unwrap().request_snapshot().unwrap();
 
     // Becoming follower does not reset pending_request_snapshot.
     let (term, id) = (nt.peers[&1].term, nt.peers[&1].id);
@@ -5086,6 +5058,28 @@ fn test_request_snapshot_on_role_change() {
     );
 }
 
+// Abort request snapshot if term change.
+#[test]
+fn test_request_snapshot_after_term_change() {
+    let (mut nt, _) = prepare_request_snapshot();
+
+    nt.peers.get_mut(&2).unwrap().request_snapshot().unwrap();
+
+    assert!(
+        nt.peers[&2].pending_request_snapshot != INVALID_INDEX,
+        "{}",
+        nt.peers[&2].pending_request_snapshot
+    );
+
+    let term = nt.peers[&1].term;
+    nt.peers.get_mut(&2).unwrap().reset(term + 1);
+    assert!(
+        nt.peers[&2].pending_request_snapshot == INVALID_INDEX,
+        "{}",
+        nt.peers[&2].pending_request_snapshot
+    );
+}
+
 /// Tests group commit.
 ///
 /// 1. Logs should be replicated to at least different groups before committed;

harness/tests/integration_cases/test_raft_snap.rs

@@ -160,11 +160,16 @@ fn test_request_snapshot() {
 
     // Raft can not step request snapshot if there is no leader.
     assert_eq!(
-        sm.raft
-            .as_mut()
-            .unwrap()
-            .request_snapshot(INVALID_INDEX + 1)
-            .unwrap_err(),
+        sm.raft.as_mut().unwrap().request_snapshot().unwrap_err(),
+        Error::RequestSnapshotDropped
+    );
+
+    let term = sm.term;
+    sm.become_follower(term + 1, 2);
+
+    // Raft can not step request snapshot if last raft log's term mismatch current term.
+    assert_eq!(
+        sm.raft.as_mut().unwrap().request_snapshot().unwrap_err(),
         Error::RequestSnapshotDropped
     );
 
@@ -173,11 +178,7 @@ fn test_request_snapshot() {
 
     // Raft can not step request snapshot if itself is a leader.
     assert_eq!(
-        sm.raft
-            .as_mut()
-            .unwrap()
-            .request_snapshot(INVALID_INDEX + 1)
-            .unwrap_err(),
+        sm.raft.as_mut().unwrap().request_snapshot().unwrap_err(),
         Error::RequestSnapshotDropped
     );
 

src/raft.rs

@@ -2407,7 +2407,7 @@ impl<T: Storage> Raft<T> {
     }
 
     /// Request a snapshot from a leader.
-    pub fn request_snapshot(&mut self, request_index: u64) -> Result<()> {
+    pub fn request_snapshot(&mut self) -> Result<()> {
         if self.state == StateRole::Leader {
             info!(
                 self.logger,
@@ -2416,7 +2416,7 @@ impl<T: Storage> Raft<T> {
         } else if self.leader_id == INVALID_ID {
             info!(
                 self.logger,
-                "drop request snapshot because of no leader";
+                "no leader; dropping request snapshot";
                 "term" => self.term,
             );
         } else if self.snap().is_some() {
@@ -2430,9 +2430,19 @@ impl<T: Storage> Raft<T> {
                 "there is a pending snapshot; dropping request snapshot";
             );
         } else {
-            self.pending_request_snapshot = request_index;
-            self.send_request_snapshot();
-            return Ok(());
+            let request_index = self.raft_log.last_index();
+            let request_index_term = self.raft_log.term(request_index).unwrap();
+            if self.term == request_index_term {
+                self.pending_request_snapshot = request_index;
+                self.send_request_snapshot();
+                return Ok(());
+            }
+            info! {
+                self.logger,
+                "mismatched term; dropping request snapshot";
+                "term" => self.term,
+                "last_term" => request_index_term,
+            };
         }
         Err(Error::RequestSnapshotDropped)
     }
@@ -2837,6 +2847,7 @@ impl<T: Storage> Raft<T> {
         m.reject_hint = self.raft_log.last_index();
         m.to = self.leader_id;
         m.request_snapshot = self.pending_request_snapshot;
+        m.log_term = self.raft_log.term(m.reject_hint).unwrap();
         self.r.send(m, &mut self.msgs);
     }
 

src/raft_log.rs

@@ -828,12 +828,12 @@ mod test {
         for i in 1..=unstable_index {
             storage
                 .wl()
-                .append(&[new_entry(i as u64, i as u64)])
+                .append(&[new_entry(i, i)])
                 .expect("append failed");
         }
         let mut raft_log = RaftLog::new(storage, default_logger());
         for i in unstable_index..last_index {
-            raft_log.append(&[new_entry(i as u64 + 1, i as u64 + 1)]);
+            raft_log.append(&[new_entry(i + 1, i + 1)]);
         }
         assert!(
             raft_log.maybe_commit(last_index, last_term),

src/raw_node.rs

@@ -743,9 +743,10 @@ impl<T: Storage> RawNode<T> {
     }
 
     /// Request a snapshot from a leader.
-    /// The snapshot's index must be greater or equal to the request_index.
-    pub fn request_snapshot(&mut self, request_index: u64) -> Result<()> {
-        self.raft.request_snapshot(request_index)
+    /// The snapshot's index must be greater or equal to the request_index (last_index) or
+    /// the leader's term must be greater than the request term (last_index's term).
+    pub fn request_snapshot(&mut self) -> Result<()> {
+        self.raft.request_snapshot()
     }
 
     /// TransferLeader tries to transfer leadership to the given transferee.

src/storage.rs

@@ -538,7 +538,7 @@ mod test {
     }
 
     fn size_of<T: PbMessage>(m: &T) -> u32 {
-        m.compute_size() as u32
+        m.compute_size()
     }
 
     fn new_snapshot(index: u64, term: u64, voters: Vec<u64>) -> Snapshot {