Don't store votes unless we are leader soon (#17803)

(cherry picked from commit 0feac57)

Author: sakridge

core/src/cluster_info_vote_listener.rs

@@ -33,7 +33,7 @@ use solana_runtime::{
     vote_sender_types::{ReplayVoteReceiver, ReplayedVote},
 };
 use solana_sdk::{
-    clock::{Epoch, Slot, DEFAULT_MS_PER_SLOT},
+    clock::{Epoch, Slot, DEFAULT_MS_PER_SLOT, DEFAULT_TICKS_PER_SLOT},
     epoch_schedule::EpochSchedule,
     hash::Hash,
     pubkey::Pubkey,
@@ -384,9 +384,14 @@ impl ClusterInfoVoteListener {
                 return Ok(());
             }
 
+            let would_be_leader = poh_recorder
+                .lock()
+                .unwrap()
+                .would_be_leader(20 * DEFAULT_TICKS_PER_SLOT);
             if let Err(e) = verified_vote_packets.receive_and_process_vote_packets(
                 &verified_vote_label_packets_receiver,
                 &mut update_version,
+                would_be_leader,
             ) {
                 match e {
                     Error::CrossbeamRecvTimeoutError(RecvTimeoutError::Disconnected) => {

core/src/verified_vote_packets.rs

@@ -15,17 +15,25 @@ impl VerifiedVotePackets {
         &mut self,
         vote_packets_receiver: &VerifiedLabelVotePacketsReceiver,
         last_update_version: &mut u64,
+        would_be_leader: bool,
     ) -> Result<()> {
         let timer = Duration::from_millis(200);
         let vote_packets = vote_packets_receiver.recv_timeout(timer)?;
         *last_update_version += 1;
-        for (label, slot, packet) in vote_packets {
-            self.0.insert(label, (*last_update_version, slot, packet));
-        }
-        while let Ok(vote_packets) = vote_packets_receiver.try_recv() {
+        if would_be_leader {
             for (label, slot, packet) in vote_packets {
                 self.0.insert(label, (*last_update_version, slot, packet));
             }
+        } else {
+            self.0.clear();
+            self.0.shrink_to_fit();
+        }
+        while let Ok(vote_packets) = vote_packets_receiver.try_recv() {
+            if would_be_leader {
+                for (label, slot, packet) in vote_packets {
+                    self.0.insert(label, (*last_update_version, slot, packet));
+                }
+            }
         }
         Ok(())
     }
@@ -137,7 +145,7 @@ mod tests {
         s.send(vec![(label1.clone(), 42, later_packets)]).unwrap();
         let mut verified_vote_packets = VerifiedVotePackets(HashMap::new());
         verified_vote_packets
-            .receive_and_process_vote_packets(&r, &mut update_version)
+            .receive_and_process_vote_packets(&r, &mut update_version, true)
             .unwrap();
 
         // Test timestamps for same batch are the same
@@ -171,15 +179,15 @@ mod tests {
         s.send(vec![(label2.clone(), 51, Packets::default())])
             .unwrap();
         verified_vote_packets
-            .receive_and_process_vote_packets(&r, &mut update_version)
+            .receive_and_process_vote_packets(&r, &mut update_version, true)
             .unwrap();
         let update_version2 = verified_vote_packets.get_vote_packets(&label2).unwrap().0;
         assert!(update_version2 > update_version1);
 
         // Test empty doesn't bump the version
         let before = update_version;
         assert_matches!(
-            verified_vote_packets.receive_and_process_vote_packets(&r, &mut update_version),
+            verified_vote_packets.receive_and_process_vote_packets(&r, &mut update_version, true),
             Err(Error::CrossbeamRecvTimeoutError(RecvTimeoutError::Timeout))
         );
         assert_eq!(before, update_version);