raftstore: retry pending read index requests (tikv#6348)

Signed-off-by: qupeng <qupeng@pingcap.com>

src/raftstore/store/fsm/peer.rs

@@ -761,6 +761,8 @@ impl<'a, T: Transport, C: PdClient> PeerFsmDelegate<'a, T, C> {
             return;
         }
 
+        self.fsm.peer.retry_pending_reads(&self.ctx.cfg);
+
         let mut res = None;
         if self.ctx.cfg.hibernate_regions {
             if self.fsm.group_state == GroupState::Idle {

tests/integrations/raftstore/test_replica_read.rs

@@ -19,8 +19,8 @@ fn test_replica_read_not_applied() {
     let mut cluster = new_node_cluster(0, 3);
 
     // Increase the election tick to make this test case running reliably.
-    configure_for_lease_read(&mut cluster, Some(50), Some(10_000));
-    let max_lease = Duration::from_secs(2);
+    configure_for_lease_read(&mut cluster, Some(50), Some(30));
+    let max_lease = Duration::from_secs(1);
     cluster.cfg.raft_store.raft_store_max_leader_lease = ReadableDuration(max_lease);
     // After the leader has committed to its term, pending reads on followers can be responsed.
     // However followers can receive `ReadIndexResp` after become candidate if the leader has
@@ -70,21 +70,18 @@ fn test_replica_read_not_applied() {
         router.send_raft_message(raft_msg).unwrap();
     }
 
-    // However, the old read index request could be blocked in raftstore forever, need more fix.
-    // TODO: this needs to be fixed.
+    // The old read index request won't be blocked forever as it's retried internally.
     cluster.sim.wl().clear_send_filters(1);
     cluster.sim.wl().clear_recv_filters(2);
-    assert!(resp1_ch.recv_timeout(Duration::from_secs(3)).is_err());
+    let resp1 = resp1_ch.recv_timeout(Duration::from_secs(6)).unwrap();
+    let exp_value = resp1.get_responses()[0].get_get().get_value();
+    assert_eq!(exp_value, b"v2");
 
     // New read index requests can be resolved quickly.
     let resp2_ch = async_read_on_peer(&mut cluster, new_peer(3, 3), r1, b"k1", true, true);
     let resp2 = resp2_ch.recv_timeout(Duration::from_secs(3)).unwrap();
     let exp_value = resp2.get_responses()[0].get_get().get_value();
     assert_eq!(exp_value, b"v2");
-
-    let resp1 = resp1_ch.recv_timeout(Duration::from_secs(3)).unwrap();
-    let exp_value = resp1.get_responses()[0].get_get().get_value();
-    assert_eq!(exp_value, b"v2");
 }
 
 #[test]