wip: pay rewards

core/Cargo.toml

@@ -70,6 +70,7 @@ lru = { workspace = true }
 min-max-heap = { workspace = true }
 num_cpus = { workspace = true }
 num_enum = { workspace = true }
+parking_lot = { workspace = true }
 prio-graph = { workspace = true }
 qualifier_attr = { workspace = true }
 quinn = { workspace = true }

core/src/block_creation_loop.rs

@@ -10,6 +10,7 @@ use {
         replay_stage::{Finalizer, ReplayStage},
     },
     crossbeam_channel::Receiver,
+    parking_lot::RwLock as PLRwLock,
     solana_clock::Slot,
     solana_entry::block_component::BlockFooterV1,
     solana_gossip::cluster_info::ClusterInfo,
@@ -30,9 +31,13 @@ use {
         bank::{Bank, NewBankOptions},
         bank_forks::BankForks,
         block_component_processor::BlockComponentProcessor,
+        validated_reward_certificate::ValidatedRewardCertificate,
     },
     solana_version::version,
-    solana_votor::{common::block_timeout, event::LeaderWindowInfo},
+    solana_votor::{
+        common::block_timeout, consensus_rewards::ConsensusRewards, event::LeaderWindowInfo,
+    },
+    solana_votor_messages::rewards_certificate::{NotarRewardCertificate, SkipRewardCertificate},
     stats::{BlockCreationLoopMetrics, SlotMetrics},
     std::{
         sync::{
@@ -78,6 +83,7 @@ pub struct BlockCreationLoopConfig {
     pub poh_recorder: Arc<RwLock<PohRecorder>>,
     pub leader_schedule_cache: Arc<LeaderScheduleCache>,
     pub rpc_subscriptions: Option<Arc<RpcSubscriptions>>,
+    pub consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
 
     // Notifiers
     pub banking_tracer: Arc<BankingTracer>,
@@ -108,6 +114,7 @@ struct LeaderContext {
     slot_status_notifier: Option<SlotStatusNotifier>,
     banking_tracer: Arc<BankingTracer>,
     replay_highest_frozen: Arc<ReplayHighestFrozen>,
+    consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
 
     // Metrics
     metrics: BlockCreationLoopMetrics,
@@ -159,6 +166,7 @@ fn start_loop(config: BlockCreationLoopConfig) {
         replay_highest_frozen,
         highest_parent_ready,
         optimistic_parent_receiver,
+        consensus_rewards,
     } = config;
 
     // Similar to Votor, if this loop dies kill the validator
@@ -196,6 +204,7 @@ fn start_loop(config: BlockCreationLoopConfig) {
         replay_highest_frozen,
         metrics: BlockCreationLoopMetrics::default(),
         slot_metrics: SlotMetrics::default(),
+        consensus_rewards,
     };
 
     // Setup poh
@@ -316,6 +325,7 @@ fn produce_window(
         if let Err(e) = record_and_complete_block(
             ctx.poh_recorder.as_ref(),
             &mut ctx.record_receiver,
+            ctx.consensus_rewards.clone(),
             skip_timer,
             timeout,
         ) {
@@ -369,7 +379,11 @@ fn skew_block_producer_time_nanos(
 
 /// Produces a block footer with the current timestamp and version information.
 /// The bank_hash field is left as default and will be filled in after the bank freezes.
-fn produce_block_footer(bank: Arc<Bank>) -> BlockFooterV1 {
+fn produce_block_footer(
+    bank: Arc<Bank>,
+    skip_reward_certificate: Option<SkipRewardCertificate>,
+    notar_reward_certificate: Option<NotarRewardCertificate>,
+) -> BlockFooterV1 {
     let mut block_producer_time_nanos = SystemTime::now()
         .duration_since(UNIX_EPOCH)
         .expect("Misconfigured system clock; couldn't measure block producer time.")
@@ -396,6 +410,8 @@ fn produce_block_footer(bank: Arc<Bank>) -> BlockFooterV1 {
         bank_hash: Hash::default(),
         block_producer_time_nanos: block_producer_time_nanos as u64,
         block_user_agent: format!("agave/{}", version!()).into_bytes(),
+        skip_reward_certificate,
+        notar_reward_certificate,
     }
 }
 
@@ -409,9 +425,17 @@ fn produce_block_footer(bank: Arc<Bank>) -> BlockFooterV1 {
 fn record_and_complete_block(
     poh_recorder: &RwLock<PohRecorder>,
     record_receiver: &mut RecordReceiver,
+    consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
     block_timer: Instant,
     block_timeout: Duration,
 ) -> Result<(), PohRecorderError> {
+    // Taking a read lock on consensus_rewards can contend with the write lock in bls_sigverifier.
+    // We are ready to produce the block footer now, while we gather other bits of data, we can block on the consensus_rewards lock in a separate thread to minimise contention.
+    let handle = std::thread::spawn(move || {
+        // XXX: how to look up the slot.
+        let slot = u64::MAX;
+        consensus_rewards.read().build_rewards_certs(slot)
+    });
     loop {
         let remaining_slot_time = block_timeout.saturating_sub(block_timer.elapsed());
         if remaining_slot_time.is_zero() {
@@ -462,11 +486,25 @@ fn record_and_complete_block(
 
     // Produce the footer with the current timestamp
     let working_bank = w_poh_recorder.working_bank().unwrap();
-    let footer = produce_block_footer(working_bank.bank.clone_without_scheduler());
+    let (skip_reward_certificate, notar_reward_certificate, validators) = handle.join().unwrap();
+    let footer = produce_block_footer(
+        working_bank.bank.clone_without_scheduler(),
+        skip_reward_certificate,
+        notar_reward_certificate,
+    );
+    let validated_reward_cert = ValidatedRewardCertificate::new_for_block_producer(validators);
 
+    let epoch_stakes = working_bank
+        .bank
+        .epoch_stakes(working_bank.bank.epoch())
+        .unwrap();
+    let epoch_schedule = working_bank.bank.epoch_schedule();
     BlockComponentProcessor::update_bank_with_footer(
         working_bank.bank.clone_without_scheduler(),
         &footer,
+        validated_reward_cert,
+        epoch_stakes,
+        epoch_schedule,
     );
 
     drop(bank);

core/src/bls_sigverify/bls_sigverifier.rs

@@ -9,12 +9,14 @@ use {
     },
     bitvec::prelude::{BitVec, Lsb0},
     crossbeam_channel::{Sender, TrySendError},
+    parking_lot::RwLock as PLRwLock,
     rayon::iter::{
         IndexedParallelIterator, IntoParallelIterator, IntoParallelRefIterator, ParallelIterator,
     },
     solana_bls_signatures::{
         pubkey::{Pubkey as BlsPubkey, PubkeyProjective, VerifiablePubkey},
         signature::SignatureProjective,
+        Signature as BLSSignature,
     },
     solana_clock::Slot,
     solana_measure::measure::Measure,
@@ -23,7 +25,10 @@ use {
     solana_runtime::{bank::Bank, bank_forks::SharableBanks, epoch_stakes::BLSPubkeyToRankMap},
     solana_signer_store::{decode, DecodeError},
     solana_streamer::packet::PacketBatch,
-    solana_votor::consensus_metrics::{ConsensusMetricsEvent, ConsensusMetricsEventSender},
+    solana_votor::{
+        consensus_metrics::{ConsensusMetricsEvent, ConsensusMetricsEventSender},
+        consensus_rewards::ConsensusRewards,
+    },
     solana_votor_messages::{
         consensus_message::{Certificate, CertificateType, ConsensusMessage, VoteMessage},
         vote::Vote,
@@ -61,8 +66,29 @@ fn aggregate_keys_from_bitmap(
     PubkeyProjective::par_aggregate(pubkeys.par_iter()).ok()
 }
 
+pub fn verify_base2_certificate(
+    vote: Vote,
+    signature: &BLSSignature,
+    bit_vec: &BitVec<u8, Lsb0>,
+    key_to_rank_map: &Arc<BLSPubkeyToRankMap>,
+) -> Result<(), CertVerifyError> {
+    let Ok(signed_payload) = bincode::serialize(&vote) else {
+        return Err(CertVerifyError::SerializationFailed);
+    };
+
+    let Some(aggregate_bls_pubkey) = aggregate_keys_from_bitmap(bit_vec, key_to_rank_map) else {
+        return Err(CertVerifyError::KeyAggregationFailed);
+    };
+
+    if let Ok(true) = aggregate_bls_pubkey.verify_signature(signature, &signed_payload) {
+        Ok(())
+    } else {
+        Err(CertVerifyError::SignatureVerificationFailed)
+    }
+}
+
 #[derive(Debug, Error, PartialEq)]
-enum CertVerifyError {
+pub enum CertVerifyError {
     #[error("Failed to find key to rank map for slot {0}")]
     KeyToRankMapNotFound(Slot),
 
@@ -92,6 +118,7 @@ pub struct BLSSigVerifier {
     consensus_metrics_sender: ConsensusMetricsEventSender,
     last_checked_root_slot: Slot,
     alpenglow_last_voted: Arc<AlpenglowLastVoted>,
+    consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
 }
 
 impl BLSSigVerifier {
@@ -172,8 +199,17 @@ impl BLSSigVerifier {
                         id: *solana_pubkey,
                         vote: vote_message.vote,
                     });
-                    // Only need votes newer than root slot
+
+                    // consensus pool does not need votes for slots other than root slot however the rewards container may still need them.
                     if vote_message.vote.slot() <= root_bank.slot() {
+                        // the only module that takes a write lock on consensus_rewards is this one and it does not take the write lock while it is verifying votes so this should never block.
+                        if self
+                            .consensus_rewards
+                            .read()
+                            .wants_vote(root_bank.slot(), &vote_message)
+                        {
+                            // XXX: actually verify and send the votes.  The verification and sending should happen off the critical path.
+                        }
                         self.stats.received_old.fetch_add(1, Ordering::Relaxed);
                         packet.meta_mut().set_discard(true);
                         continue;
@@ -219,8 +255,8 @@ impl BLSSigVerifier {
             || self.verify_and_send_certificates(certs_to_verify, &root_bank),
         );
 
-        votes_result?;
-        certs_result?;
+        let rewards_votes = votes_result?;
+        let () = certs_result?;
 
         // Send to RPC service for last voted tracking
         self.alpenglow_last_voted
@@ -235,6 +271,19 @@ impl BLSSigVerifier {
             warn!("could not send consensus metrics, receive side of channel is closed");
         }
 
+        {
+            // This should be the only place that is taking a write lock on consensus_rewards.
+            // This lock should not contend with any other operations in this module.
+            // It can contend with the read lock in the block creation loop though as such we should hold it for as little time as possible.
+            let mut guard = self.consensus_rewards.write();
+            let root_slot = root_bank.slot();
+            for v in rewards_votes {
+                if let Some(rank_map) = get_key_to_rank_map(&root_bank, v.vote.slot()) {
+                    guard.add_vote_message(root_slot, rank_map.clone(), v);
+                }
+            }
+        }
+
         self.stats.maybe_report_stats();
 
         Ok(())
@@ -248,6 +297,7 @@ impl BLSSigVerifier {
         message_sender: Sender<ConsensusMessage>,
         consensus_metrics_sender: ConsensusMetricsEventSender,
         alpenglow_last_voted: Arc<AlpenglowLastVoted>,
+        consensus_rewards: Arc<PLRwLock<ConsensusRewards>>,
     ) -> Self {
         Self {
             sharable_banks,
@@ -259,14 +309,32 @@ impl BLSSigVerifier {
             consensus_metrics_sender,
             last_checked_root_slot: 0,
             alpenglow_last_voted,
+            consensus_rewards,
         }
     }
 
+    /// Verifies votes and sends verified votes to the consensus pool.
+    /// Also returns a copy of the verified votes that the rewards container is interested is so that the caller can send them to it.
     fn verify_and_send_votes(
         &self,
         votes_to_verify: Vec<VoteToVerify>,
-    ) -> Result<(), BLSSigVerifyServiceError<ConsensusMessage>> {
+    ) -> Result<Vec<VoteMessage>, BLSSigVerifyServiceError<ConsensusMessage>> {
         let verified_votes = self.verify_votes(votes_to_verify);
+
+        let rewards_votes = {
+            // the only module that takes a write lock on consensus_rewards is this one and it does not take the write lock while it is verifying votes so this should never block.
+            let guard = self.consensus_rewards.read();
+            let root_slot = self.sharable_banks.root().slot();
+            verified_votes
+                .iter()
+                .filter_map(|vote| {
+                    guard
+                        .wants_vote(root_slot, &vote.vote_message)
+                        .then_some(vote.vote_message.clone())
+                })
+                .collect()
+        };
+
         self.stats
             .total_valid_packets
             .fetch_add(verified_votes.len() as u64, Ordering::Relaxed);
@@ -285,7 +353,7 @@ impl BLSSigVerifier {
                 }
             }
 
-            // Send the BLS vote messaage to certificate pool
+            // Send the votes to the consensus pool
             match self
                 .message_sender
                 .try_send(ConsensusMessage::Vote(vote.vote_message))
@@ -319,7 +387,7 @@ impl BLSSigVerifier {
             }
         }
 
-        Ok(())
+        Ok(rewards_votes)
     }
 
     fn verify_votes(&self, votes_to_verify: Vec<VoteToVerify>) -> Vec<VoteToVerify> {
@@ -523,9 +591,12 @@ impl BLSSigVerifier {
         };
 
         match decoded_bitmap {
-            solana_signer_store::Decoded::Base2(bit_vec) => {
-                self.verify_base2_certificate(cert_to_verify, &bit_vec, key_to_rank_map)?
-            }
+            solana_signer_store::Decoded::Base2(bit_vec) => verify_base2_certificate(
+                cert_to_verify.cert_type.to_source_vote(),
+                &cert_to_verify.signature,
+                &bit_vec,
+                key_to_rank_map,
+            )?,
             solana_signer_store::Decoded::Base3(bit_vec1, bit_vec2) => self
                 .verify_base3_certificate(cert_to_verify, &bit_vec1, &bit_vec2, key_to_rank_map)?,
         }
@@ -538,32 +609,6 @@ impl BLSSigVerifier {
         Ok(())
     }
 
-    fn verify_base2_certificate(
-        &self,
-        cert_to_verify: &Certificate,
-        bit_vec: &BitVec<u8, Lsb0>,
-        key_to_rank_map: &Arc<BLSPubkeyToRankMap>,
-    ) -> Result<(), CertVerifyError> {
-        let original_vote = cert_to_verify.cert_type.to_source_vote();
-
-        let Ok(signed_payload) = bincode::serialize(&original_vote) else {
-            return Err(CertVerifyError::SerializationFailed);
-        };
-
-        let Some(aggregate_bls_pubkey) = aggregate_keys_from_bitmap(bit_vec, key_to_rank_map)
-        else {
-            return Err(CertVerifyError::KeyAggregationFailed);
-        };
-
-        if let Ok(true) =
-            aggregate_bls_pubkey.verify_signature(&cert_to_verify.signature, &signed_payload)
-        {
-            Ok(())
-        } else {
-            Err(CertVerifyError::SignatureVerificationFailed)
-        }
-    }
-
     fn verify_base3_certificate(
         &self,
         cert_to_verify: &Certificate,