Ingest replayed Alpenglow votes into cert pool

core/src/banking_stage/committer.rs

@@ -107,10 +107,14 @@ impl Committer {
             .collect();
 
         let ((), find_and_send_votes_us) = measure_us!({
+            // We are going to send Alpenglow votes received through TPU port directly to cert pool,
+            // whether or not the transaction is committed. So we don't need to set alpenglow_vote_sender
+            // here.
             bank_utils::find_and_send_votes(
                 batch.sanitized_transactions(),
                 &commit_results,
                 Some(&self.replay_vote_sender),
+                None,
             );
             self.collect_balances_and_send_status_batch(
                 commit_results,

core/src/replay_stage.rs

@@ -78,7 +78,7 @@ use {
         commitment::BlockCommitmentCache,
         installed_scheduler_pool::BankWithScheduler,
         prioritization_fee_cache::PrioritizationFeeCache,
-        vote_sender_types::{AlpenglowVoteReceiver, ReplayVoteSender},
+        vote_sender_types::{AlpenglowVoteReceiver, AlpenglowVoteSender, ReplayVoteSender},
     },
     solana_sdk::{
         clock::{BankId, Slot, NUM_CONSECUTIVE_LEADER_SLOTS},
@@ -308,6 +308,7 @@ pub struct ReplaySenders {
     pub drop_bank_sender: Sender<Vec<BankWithScheduler>>,
     pub block_metadata_notifier: Option<BlockMetadataNotifierArc>,
     pub dumped_slots_sender: Sender<Vec<(u64, Hash)>>,
+    pub alpenglow_vote_sender: AlpenglowVoteSender,
 }
 
 pub struct ReplayReceivers {
@@ -605,6 +606,7 @@ impl ReplayStage {
             drop_bank_sender,
             block_metadata_notifier,
             dumped_slots_sender,
+            alpenglow_vote_sender,
         } = senders;
 
         let ReplayReceivers {
@@ -788,6 +790,7 @@ impl ReplayStage {
                     &verify_recyclers,
                     &mut heaviest_subtree_fork_choice,
                     &replay_vote_sender,
+                    &alpenglow_vote_sender,
                     &bank_notification_sender,
                     &rpc_subscriptions,
                     &slot_status_notifier,
@@ -2696,6 +2699,7 @@ impl ReplayStage {
         transaction_status_sender: Option<&TransactionStatusSender>,
         entry_notification_sender: Option<&EntryNotifierSender>,
         replay_vote_sender: &ReplayVoteSender,
+        alpenglow_vote_sender: &AlpenglowVoteSender,
         verify_recyclers: &VerifyRecyclers,
         log_messages_bytes_limit: Option<usize>,
         prioritization_fee_cache: &PrioritizationFeeCache,
@@ -2716,6 +2720,7 @@ impl ReplayStage {
             transaction_status_sender,
             entry_notification_sender,
             Some(replay_vote_sender),
+            Some(alpenglow_vote_sender),
             verify_recyclers,
             false,
             log_messages_bytes_limit,
@@ -3674,6 +3679,7 @@ impl ReplayStage {
         entry_notification_sender: Option<&EntryNotifierSender>,
         verify_recyclers: &VerifyRecyclers,
         replay_vote_sender: &ReplayVoteSender,
+        alpenglow_vote_sender: &AlpenglowVoteSender,
         replay_timing: &mut ReplayLoopTiming,
         log_messages_bytes_limit: Option<usize>,
         active_bank_slots: &[Slot],
@@ -3757,6 +3763,7 @@ impl ReplayStage {
                             transaction_status_sender,
                             entry_notification_sender,
                             &replay_vote_sender.clone(),
+                            &alpenglow_vote_sender.clone(),
                             &verify_recyclers.clone(),
                             log_messages_bytes_limit,
                             prioritization_fee_cache,
@@ -3789,6 +3796,7 @@ impl ReplayStage {
         entry_notification_sender: Option<&EntryNotifierSender>,
         verify_recyclers: &VerifyRecyclers,
         replay_vote_sender: &ReplayVoteSender,
+        alpenglow_vote_sender: &AlpenglowVoteSender,
         replay_timing: &mut ReplayLoopTiming,
         log_messages_bytes_limit: Option<usize>,
         bank_slot: Slot,
@@ -3846,6 +3854,7 @@ impl ReplayStage {
                     transaction_status_sender,
                     entry_notification_sender,
                     &replay_vote_sender.clone(),
+                    &alpenglow_vote_sender.clone(),
                     &verify_recyclers.clone(),
                     log_messages_bytes_limit,
                     prioritization_fee_cache,
@@ -4313,6 +4322,7 @@ impl ReplayStage {
         verify_recyclers: &VerifyRecyclers,
         heaviest_subtree_fork_choice: &mut HeaviestSubtreeForkChoice,
         replay_vote_sender: &ReplayVoteSender,
+        alpenglow_vote_sender: &AlpenglowVoteSender,
         bank_notification_sender: &Option<BankNotificationSenderConfig>,
         rpc_subscriptions: &Arc<RpcSubscriptions>,
         slot_status_notifier: &Option<SlotStatusNotifier>,
@@ -4362,6 +4372,7 @@ impl ReplayStage {
                     entry_notification_sender,
                     verify_recyclers,
                     replay_vote_sender,
+                    alpenglow_vote_sender,
                     replay_timing,
                     log_messages_bytes_limit,
                     &active_bank_slots,
@@ -4382,6 +4393,7 @@ impl ReplayStage {
                         entry_notification_sender,
                         verify_recyclers,
                         replay_vote_sender,
+                        alpenglow_vote_sender,
                         replay_timing,
                         log_messages_bytes_limit,
                         *bank_slot,
@@ -6076,6 +6088,7 @@ pub(crate) mod tests {
     {
         let ledger_path = get_tmp_ledger_path!();
         let (replay_vote_sender, _replay_vote_receiver) = unbounded();
+        let (alpenglow_vote_sender, _alpenglow_vote_receiver) = unbounded();
         let res = {
             let ReplayBlockstoreComponents {
                 blockstore,
@@ -6120,6 +6133,7 @@ pub(crate) mod tests {
                 None,
                 None,
                 &replay_vote_sender,
+                &alpenglow_vote_sender,
                 &VerifyRecyclers::default(),
                 None,
                 &PrioritizationFeeCache::new(0u64),
@@ -10289,6 +10303,7 @@ pub(crate) mod tests {
             None,
             None,
             None,
+            None,
             &mut ExecuteTimings::default(),
         )
         .unwrap();

core/src/tvu.rs

@@ -45,7 +45,7 @@ use {
         bank_forks::BankForks,
         commitment::BlockCommitmentCache,
         prioritization_fee_cache::PrioritizationFeeCache,
-        vote_sender_types::{AlpenglowVoteReceiver, ReplayVoteSender},
+        vote_sender_types::{AlpenglowVoteReceiver, AlpenglowVoteSender, ReplayVoteSender},
     },
     solana_sdk::{clock::Slot, pubkey::Pubkey, signature::Keypair},
     solana_turbine::retransmit_stage::RetransmitStage,
@@ -145,6 +145,7 @@ impl Tvu {
         completed_data_sets_sender: Option<CompletedDataSetsSender>,
         bank_notification_sender: Option<BankNotificationSenderConfig>,
         duplicate_confirmed_slots_receiver: DuplicateConfirmedSlotsReceiver,
+        alpenglow_vote_sender: AlpenglowVoteSender,
         alpenglow_vote_receiver: AlpenglowVoteReceiver,
         tvu_config: TvuConfig,
         max_slots: &Arc<MaxSlots>,
@@ -301,6 +302,7 @@ impl Tvu {
             drop_bank_sender,
             block_metadata_notifier,
             dumped_slots_sender,
+            alpenglow_vote_sender,
         };
 
         let replay_receivers = ReplayReceivers {
@@ -514,6 +516,7 @@ pub mod tests {
         let (_gossip_verified_vote_hash_sender, gossip_verified_vote_hash_receiver) = unbounded();
         let (_verified_vote_sender, verified_vote_receiver) = unbounded();
         let (replay_vote_sender, _replay_vote_receiver) = unbounded();
+        let (alpenglow_vote_sender, _alpenglow_vote_receiver) = unbounded();
         let (_, gossip_confirmed_slots_receiver) = unbounded();
         let (_, alpenglow_vote_receiver) = unbounded();
         let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
@@ -580,6 +583,7 @@ pub mod tests {
             /*completed_data_sets_sender:*/ None,
             None,
             gossip_confirmed_slots_receiver,
+            alpenglow_vote_sender,
             alpenglow_vote_receiver,
             TvuConfig::default(),
             &Arc::new(MaxSlots::default()),

core/src/validator.rs

@@ -946,6 +946,7 @@ impl Validator {
         );
 
         let (replay_vote_sender, replay_vote_receiver) = unbounded();
+        let (alpenglow_vote_sender, alpenglow_vote_receiver) = unbounded();
 
         // block min prioritization fee cache should be readable by RPC, and writable by validator
         // (by both replay stage and banking stage)
@@ -1010,6 +1011,7 @@ impl Validator {
                     config.runtime_config.log_messages_bytes_limit,
                     transaction_status_sender.clone(),
                     Some(replay_vote_sender.clone()),
+                    Some(alpenglow_vote_sender.clone()),
                     prioritization_fee_cache.clone(),
                 );
                 bank_forks
@@ -1354,7 +1356,6 @@ impl Validator {
         let (verified_vote_sender, verified_vote_receiver) = unbounded();
         let (gossip_verified_vote_hash_sender, gossip_verified_vote_hash_receiver) = unbounded();
         let (duplicate_confirmed_slot_sender, duplicate_confirmed_slots_receiver) = unbounded();
-        let (alpenglow_vote_sender, alpenglow_vote_receiver) = unbounded();
         let entry_notification_sender = entry_notifier_service
             .as_ref()
             .map(|service| service.sender_cloned());
@@ -1509,6 +1510,7 @@ impl Validator {
             completed_data_sets_sender,
             bank_notification_sender.clone(),
             duplicate_confirmed_slots_receiver,
+            alpenglow_vote_sender.clone(),
             alpenglow_vote_receiver,
             TvuConfig {
                 max_ledger_shreds: config.max_ledger_shreds,

core/tests/unified_scheduler.rs

@@ -92,6 +92,7 @@ fn test_scheduler_waited_by_drop_bank_service() {
         None,
         None,
         None,
+        None,
         ignored_prioritization_fee_cache,
     );
     let pool = pool_raw.clone();
@@ -229,7 +230,14 @@ fn test_scheduler_producing_blocks() {
         None,
         Some(leader_schedule_cache),
     );
-    let pool = DefaultSchedulerPool::new(None, None, None, None, ignored_prioritization_fee_cache);
+    let pool = DefaultSchedulerPool::new(
+        None,
+        None,
+        None,
+        None,
+        None,
+        ignored_prioritization_fee_cache,
+    );
     let channels = {
         let banking_tracer = BankingTracer::new_disabled();
         banking_tracer.create_channels(true)

ledger-tool/src/ledger_utils.rs

@@ -378,6 +378,7 @@ pub fn load_and_process_ledger(
         }
         BlockVerificationMethod::UnifiedScheduler => {
             let no_replay_vote_sender = None;
+            let no_alpenglow_vote_sender = None;
             let ignored_prioritization_fee_cache = Arc::new(PrioritizationFeeCache::new(0u64));
             bank_forks
                 .write()
@@ -387,6 +388,7 @@ pub fn load_and_process_ledger(
                     process_options.runtime_config.log_messages_bytes_limit,
                     transaction_status_sender.clone(),
                     no_replay_vote_sender,
+                    no_alpenglow_vote_sender,
                     ignored_prioritization_fee_cache,
                 ));
         }

ledger/benches/blockstore_processor.rs

@@ -159,6 +159,7 @@ fn bench_execute_batch(
                 &bank,
                 None,
                 None,
+                None,
                 &mut timing,
                 None,
                 &prioritization_fee_cache,