Send votes to next leader's TPU instead of our TPU (#16663)

(cherry picked from commit c8b474cd0b)

Co-authored-by: Michael Vines <mvines@gmail.com>

core/src/banking_stage.rs

@@ -567,15 +567,8 @@ impl BankingStage {
             }
             return;
         }
-        let next_leader = match poh_recorder
-            .lock()
-            .unwrap()
-            .leader_after_n_slots(FORWARD_TRANSACTIONS_TO_LEADER_AT_SLOT_OFFSET)
-        {
-            Some(pubkey) => pubkey,
-            None => return,
-        };
-        let addr = match cluster_info.lookup_contact_info(&next_leader, |ci| ci.tpu_forwards) {
+
+        let addr = match next_leader_tpu_forwards(cluster_info, poh_recorder) {
             Some(addr) => addr,
             None => return,
         };
@@ -1369,6 +1362,36 @@ impl BankingStage {
     }
 }
 
+pub(crate) fn next_leader_tpu(
+    cluster_info: &ClusterInfo,
+    poh_recorder: &Arc<Mutex<PohRecorder>>,
+) -> Option<std::net::SocketAddr> {
+    if let Some(leader_pubkey) = poh_recorder
+        .lock()
+        .unwrap()
+        .leader_after_n_slots(FORWARD_TRANSACTIONS_TO_LEADER_AT_SLOT_OFFSET)
+    {
+        cluster_info.lookup_contact_info(&leader_pubkey, |leader| leader.tpu)
+    } else {
+        None
+    }
+}
+
+fn next_leader_tpu_forwards(
+    cluster_info: &ClusterInfo,
+    poh_recorder: &Arc<Mutex<PohRecorder>>,
+) -> Option<std::net::SocketAddr> {
+    if let Some(leader_pubkey) = poh_recorder
+        .lock()
+        .unwrap()
+        .leader_after_n_slots(FORWARD_TRANSACTIONS_TO_LEADER_AT_SLOT_OFFSET)
+    {
+        cluster_info.lookup_contact_info(&leader_pubkey, |leader| leader.tpu_forwards)
+    } else {
+        None
+    }
+}
+
 pub fn create_test_recorder(
     bank: &Arc<Bank>,
     blockstore: &Arc<Blockstore>,

core/src/cluster_info.rs

@@ -1183,8 +1183,8 @@ impl ClusterInfo {
             .process_push_message(&self_pubkey, vec![vote], now);
     }
 
-    pub fn send_vote(&self, vote: &Transaction) -> Result<()> {
-        let tpu = self.my_contact_info().tpu;
+    pub fn send_vote(&self, vote: &Transaction, tpu: Option<SocketAddr>) -> Result<()> {
+        let tpu = tpu.unwrap_or_else(|| self.my_contact_info().tpu);
         let buf = serialize(vote)?;
         self.socket.send_to(&buf, &tpu)?;
         Ok(())

core/src/replay_stage.rs

@@ -538,6 +538,7 @@ impl ReplayStage {
 
                         Self::handle_votable_bank(
                             &vote_bank,
+                            &poh_recorder,
                             switch_fork_decision,
                             &bank_forks,
                             &mut tower,
@@ -1280,6 +1281,7 @@ impl ReplayStage {
     #[allow(clippy::too_many_arguments)]
     fn handle_votable_bank(
         bank: &Arc<Bank>,
+        poh_recorder: &Arc<Mutex<PohRecorder>>,
         switch_fork_decision: &SwitchForkDecision,
         bank_forks: &Arc<RwLock<BankForks>>,
         tower: &mut Tower,
@@ -1380,6 +1382,7 @@ impl ReplayStage {
         Self::push_vote(
             cluster_info,
             bank,
+            poh_recorder,
             vote_account_pubkey,
             authorized_voter_keypairs,
             last_vote,
@@ -1390,9 +1393,11 @@ impl ReplayStage {
         );
     }
 
+    #[allow(clippy::too_many_arguments)]
     fn push_vote(
         cluster_info: &ClusterInfo,
         bank: &Arc<Bank>,
+        poh_recorder: &Arc<Mutex<PohRecorder>>,
         vote_account_pubkey: &Pubkey,
         authorized_voter_keypairs: &[Arc<Keypair>],
         vote: Vote,
@@ -1482,7 +1487,10 @@ impl ReplayStage {
             vote_signatures.clear();
         }
 
-        let _ = cluster_info.send_vote(&vote_tx);
+        let _ = cluster_info.send_vote(
+            &vote_tx,
+            crate::banking_stage::next_leader_tpu(cluster_info, poh_recorder),
+        );
         cluster_info.push_vote(tower, vote_tx);
     }