raftstore: update applied_index_term after snapshot (#10476)

* add test case

Signed-off-by: Jay Lee <BusyJayLee@gmail.com>

* raftstore: update applied_term after snapshot

After applying snapshot applied_term should also be updated, otherwise
it can produce snapshot with wrong term and cause panic in follower.

Close #10225

Signed-off-by: Jay Lee <BusyJayLee@gmail.com>

Co-authored-by: Ti Chi Robot <ti-community-prow-bot@tidb.io>

components/raftstore/src/store/peer_storage.rs

@@ -1615,6 +1615,16 @@ where
         self.schedule_applying_snapshot();
         let prev_region = self.region().clone();
         self.set_region(snap_region);
+        if self.truncated_index() == self.applied_index() {
+            self.applied_index_term = self.truncated_term();
+        } else {
+            panic!(
+                "{} applied index should be equal to truncated index after snapshot: {} != {}",
+                self.tag,
+                self.applied_index(),
+                self.truncated_index()
+            );
+        }
 
         Some(ApplySnapResult {
             prev_region,

tests/integrations/raftstore/test_snap.rs

@@ -658,3 +658,57 @@ fn random_long_vec(length: usize) -> Vec<u8> {
     (0..length).for_each(|_| value.push(rng.gen::<u8>()));
     value
 }
+
+/// Snapshot is generated using apply term from apply thread, which should be set
+/// correctly otherwise lead to unconsistency.
+#[test]
+fn test_correct_snapshot_term() {
+    // Use five replicas so leader can send a snapshot to a new peer without
+    // committing extra logs.
+    let mut cluster = new_server_cluster(0, 5);
+    let pd_client = cluster.pd_client.clone();
+    pd_client.disable_default_operator();
+
+    // Use run conf change to make new node be initialized with term 0.
+    let r = cluster.run_conf_change();
+
+    // 5 will catch up logs with just a snapshot.
+    cluster.add_send_filter(IsolationFilterFactory::new(5));
+    // 4 will catch up logs using snapshot from 5.
+    cluster.add_send_filter(IsolationFilterFactory::new(4));
+
+    pd_client.must_add_peer(r, new_peer(2, 2));
+    pd_client.must_add_peer(r, new_peer(3, 3));
+    pd_client.must_add_peer(r, new_peer(4, 4));
+    pd_client.must_add_peer(r, new_peer(5, 5));
+    cluster.must_put(b"k0", b"v0");
+
+    cluster.clear_send_filters();
+    // So peer 4 will not apply snapshot from current leader.
+    cluster.add_send_filter(CloneFilterFactory(
+        RegionPacketFilter::new(1, 4)
+            .msg_type(MessageType::MsgSnapshot)
+            .direction(Direction::Recv),
+    ));
+    // So no new log will be applied. When peer 5 becomes leader, it won't have
+    // chance to update apply index in apply worker.
+    for i in 1..=3 {
+        cluster.add_send_filter(CloneFilterFactory(
+            RegionPacketFilter::new(1, i)
+                .msg_type(MessageType::MsgAppendResponse)
+                .direction(Direction::Send),
+        ));
+    }
+    cluster.must_transfer_leader(1, new_peer(5, 5));
+    // Clears send filters so peer 4 can accept snapshot from peer 5. If peer 5
+    // didn't set apply index correctly using snapshot in apply worker, the snapshot
+    // will be generated as term 0. Raft consider term of missing index as 0, so
+    // peer 4 will accept the snapshot and think it has already applied it, hence fast
+    // forward it then panic.
+    cluster.clear_send_filters();
+    must_get_equal(&cluster.get_engine(4), b"k0", b"v0");
+    cluster.clear_send_filters();
+    cluster.must_put(b"k1", b"v1");
+    // If peer 4 panicks, it won't be able to apply new writes.
+    must_get_equal(&cluster.get_engine(4), b"k1", b"v1");
+}