Use estimated current slot in tpu client (solana-labs#28997)

tpu-client/src/nonblocking/tpu_client.rs

@@ -145,10 +145,18 @@ impl LeaderTpuCache {
     }
 
     // Get the TPU sockets for the current leader and upcoming leaders according to fanout size
-    fn get_leader_sockets(&self, fanout_slots: u64) -> Vec<SocketAddr> {
+    fn get_leader_sockets(
+        &self,
+        estimated_current_slot: Slot,
+        fanout_slots: u64,
+    ) -> Vec<SocketAddr> {
         let mut leader_set = HashSet::new();
         let mut leader_sockets = Vec::new();
-        for leader_slot in self.first_slot..self.first_slot + fanout_slots {
+        // `first_slot` might have been advanced since caller last read the `estimated_current_slot`
+        // value. Take the greater of the two values to ensure we are reading from the latest
+        // leader schedule.
+        let current_slot = std::cmp::max(estimated_current_slot, self.first_slot);
+        for leader_slot in current_slot..current_slot + fanout_slots {
             if let Some(leader) = self.get_slot_leader(leader_slot) {
                 if let Some(tpu_socket) = self.leader_tpu_map.get(leader) {
                     if leader_set.insert(*leader) {
@@ -628,10 +636,11 @@ impl LeaderTpuService {
     }
 
     pub fn leader_tpu_sockets(&self, fanout_slots: u64) -> Vec<SocketAddr> {
+        let current_slot = self.recent_slots.estimated_current_slot();
         self.leader_tpu_cache
             .read()
             .unwrap()
-            .get_leader_sockets(fanout_slots)
+            .get_leader_sockets(current_slot, fanout_slots)
     }
 
     async fn run(