wip

Author: akhi3030

Cargo.lock

@@ -31,8 +31,11 @@ use {
         bank_forks::BankForks,
     },
     solana_version::version,
-    solana_votor::{common::block_timeout, event::LeaderWindowInfo, votor::LeaderWindowNotifier},
-    solana_votor_messages::migration::MigrationStatus,
+    solana_votor::{
+        common::block_timeout, consensus_rewards::ConsensusRewards, event::LeaderWindowInfo,
+        votor::LeaderWindowNotifier,
+    },
+    solana_votor_messages::{migration::MigrationStatus, rewards_certificate::RewardsCertificates},
     stats::{BlockCreationLoopMetrics, SlotMetrics},
     std::{
         sync::{
@@ -104,6 +107,7 @@ struct LeaderContext {
     slot_status_notifier: Option<SlotStatusNotifier>,
     banking_tracer: Arc<BankingTracer>,
     replay_highest_frozen: Arc<ReplayHighestFrozen>,
+    consensus_rewards: ConsensusRewards,
 
     // Metrics
     metrics: BlockCreationLoopMetrics,
@@ -134,9 +138,14 @@ enum StartLeaderError {
     ),
 }
 
-fn produce_block_footer(block_producer_time_nanos: u64) -> VersionedBlockMarker {
+fn produce_block_footer(
+    block_producer_time_nanos: u64,
+    rewards_certs: RewardsCertificates,
+) -> VersionedBlockMarker {
     let footer = BlockFooterV1 {
         block_producer_time_nanos,
+        skip_rewards_certificate: rewards_certs.skip,
+        notar_rewards_certificate: rewards_certs.notar,
         block_user_agent: format!("agave/{}", version!()).into_bytes(),
     };
 
@@ -181,14 +190,15 @@ fn start_loop(config: BlockCreationLoopConfig) {
         blockstore,
         poh_recorder: poh_recorder.clone(),
         record_receiver,
-        leader_schedule_cache,
+        leader_schedule_cache: leader_schedule_cache.clone(),
         bank_forks,
         rpc_subscriptions,
         slot_status_notifier,
         banking_tracer,
         replay_highest_frozen,
         metrics: BlockCreationLoopMetrics::default(),
         slot_metrics: SlotMetrics::default(),
+        consensus_rewards: ConsensusRewards::new(cluster_info.clone(), leader_schedule_cache),
     };
 
     info!("{my_pubkey}: Block creation loop initialized");
@@ -301,6 +311,7 @@ fn produce_window(
         if let Err(e) = record_and_complete_block(
             ctx.poh_recorder.as_ref(),
             &mut ctx.record_receiver,
+            &mut ctx.consensus_rewards,
             skip_timer,
             timeout,
         ) {
@@ -342,6 +353,7 @@ fn produce_window(
 fn record_and_complete_block(
     poh_recorder: &RwLock<PohRecorder>,
     record_receiver: &mut RecordReceiver,
+    consensus_rewards: &mut ConsensusRewards,
     block_timer: Instant,
     block_timeout: Duration,
 ) -> Result<(), PohRecorderError> {
@@ -378,7 +390,9 @@ fn record_and_complete_block(
     // Construct and send the block footer
     let mut w_poh_recorder = poh_recorder.write().unwrap();
     let block_producer_time_nanos = w_poh_recorder.working_bank_block_producer_time_nanos();
-    let footer = produce_block_footer(block_producer_time_nanos);
+    // XXX: how to look up the slot.
+    let rewards_certs = consensus_rewards.build_rewards_certs(u64::MAX);
+    let footer = produce_block_footer(block_producer_time_nanos, rewards_certs);
     w_poh_recorder.send_marker(footer)?;
 
     // Alpentick and clear bank

core/src/block_creation_loop.rs

@@ -23,7 +23,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,
@@ -92,6 +95,7 @@ pub struct BLSSigVerifier {
     consensus_metrics_sender: ConsensusMetricsEventSender,
     last_checked_root_slot: Slot,
     alpenglow_last_voted: Arc<AlpenglowLastVoted>,
+    consensus_rewards: ConsensusRewards,
 }
 
 impl BLSSigVerifier {
@@ -172,6 +176,19 @@ impl BLSSigVerifier {
                         id: *solana_pubkey,
                         vote: vote_message.vote,
                     });
+
+                    // XXX: this has to be more complicated.
+                    // Some votes that the rewards container wants will get verified normally below so we can have check if the rewards container wants it there and then send it a copy.
+                    // For other votes, we check if the rewards container wants it, then we should verify them in some low priority context and then send them to the container.
+                    if self
+                        .consensus_rewards
+                        .wants_vote(root_bank.slot(), &vote_message)
+                    {
+                        // XXX: actually verify the vote.
+                        self.consensus_rewards
+                            .add_vote_message(root_bank.slot(), vote_message.clone());
+                    }
+
                     // Only need votes newer than root slot
                     if vote_message.vote.slot() <= root_bank.slot() {
                         self.stats.received_old.fetch_add(1, Ordering::Relaxed);
@@ -248,6 +265,7 @@ impl BLSSigVerifier {
         message_sender: Sender<ConsensusMessage>,
         consensus_metrics_sender: ConsensusMetricsEventSender,
         alpenglow_last_voted: Arc<AlpenglowLastVoted>,
+        consensus_rewards: ConsensusRewards,
     ) -> Self {
         Self {
             sharable_banks,
@@ -259,6 +277,7 @@ impl BLSSigVerifier {
             consensus_metrics_sender,
             last_checked_root_slot: 0,
             alpenglow_last_voted,
+            consensus_rewards,
         }
     }
 

core/src/bls_sigverify/bls_sigverifier.rs

@@ -60,6 +60,7 @@ use {
     },
     solana_turbine::{retransmit_stage::RetransmitStage, xdp::XdpSender},
     solana_votor::{
+        consensus_rewards::ConsensusRewards,
         event::{VotorEventReceiver, VotorEventSender},
         vote_history::VoteHistory,
         vote_history_storage::VoteHistoryStorage,
@@ -268,12 +269,15 @@ impl Tvu {
         .unwrap();
         let alpenglow_sigverify_service = {
             let sharable_banks = bank_forks.read().unwrap().sharable_banks();
+            let consensus_rewards =
+                ConsensusRewards::new(cluster_info.clone(), leader_schedule_cache.clone());
             let verifier = BLSSigVerifier::new(
                 sharable_banks,
                 verified_vote_sender.clone(),
                 consensus_message_sender.clone(),
                 consensus_metrics_sender.clone(),
                 alpenglow_last_voted.clone(),
+                consensus_rewards,
             );
             BLSSigverifyService::new(bls_packet_receiver, verifier)
         };

core/src/tvu.rs

@@ -39,6 +39,7 @@ solana-runtime-transaction = { workspace = true }
 solana-sha256-hasher = { workspace = true }
 solana-transaction = { workspace = true }
 solana-transaction-error = { workspace = true }
+solana-votor-messages = { workspace = true }
 thiserror = { workspace = true }
 
 [dev-dependencies]

entry/Cargo.toml

@@ -88,6 +88,7 @@ use {
     },
     solana_clock::Slot,
     solana_hash::Hash,
+    solana_votor_messages::rewards_certificate::RewardsCertificate,
     std::{error::Error, fmt},
 };
 
@@ -269,6 +270,8 @@ pub enum VersionedBlockFooter {
 #[derive(Clone, PartialEq, Eq, Debug)]
 pub struct BlockFooterV1 {
     pub block_producer_time_nanos: u64,
+    pub skip_rewards_certificate: RewardsCertificate,
+    pub notar_rewards_certificate: RewardsCertificate,
     pub block_user_agent: Vec<u8>,
 }
 
@@ -924,47 +927,49 @@ impl BlockFooterV1 {
 
     /// Serializes to bytes with user agent length capping.
     fn to_bytes(&self) -> Result<Vec<u8>, BlockComponentError> {
-        let mut buffer =
-            Vec::with_capacity(8 + 1 + self.block_user_agent.len().min(Self::MAX_USER_AGENT_LEN));
+        unimplemented!()
+        // let mut buffer =
+        //     Vec::with_capacity(8 + 1 + self.block_user_agent.len().min(Self::MAX_USER_AGENT_LEN));
 
-        // Serialize timestamp
-        buffer.extend_from_slice(&self.block_producer_time_nanos.to_le_bytes());
+        // // Serialize timestamp
+        // buffer.extend_from_slice(&self.block_producer_time_nanos.to_le_bytes());
 
-        // Serialize user agent with length capping
-        let capped_len = self.block_user_agent.len().min(Self::MAX_USER_AGENT_LEN);
-        buffer.push(capped_len as u8);
-        buffer.extend_from_slice(&self.block_user_agent[..capped_len]);
+        // // Serialize user agent with length capping
+        // let capped_len = self.block_user_agent.len().min(Self::MAX_USER_AGENT_LEN);
+        // buffer.push(capped_len as u8);
+        // buffer.extend_from_slice(&self.block_user_agent[..capped_len]);
 
-        Ok(buffer)
+        // Ok(buffer)
     }
 
     /// Deserializes from bytes with validation.
-    fn from_bytes(data: &[u8]) -> Result<Self, BlockComponentError> {
-        if data.len() < Self::HEADER_SIZE {
-            return Err(BlockComponentError::InsufficientData);
-        }
-
-        // Read timestamp
-        // Unwrap: HEADER_SIZE = TIMESTAMP_SIZE + USER_AGENT_LEN_SIZE > TIMESTAMP_SIZE, so this will
-        // never fail.
-        let time_bytes = data[..Self::TIMESTAMP_SIZE].try_into().unwrap();
-        let block_producer_time_nanos = u64::from_le_bytes(time_bytes);
-
-        // Read user agent length
-        let user_agent_len = data[Self::TIMESTAMP_SIZE] as usize;
-
-        // Validate remaining data size
-        if data.len() < Self::HEADER_SIZE + user_agent_len {
-            return Err(BlockComponentError::InsufficientData);
-        }
-
-        // Read user agent bytes
-        let block_user_agent = data[Self::HEADER_SIZE..Self::HEADER_SIZE + user_agent_len].to_vec();
-
-        Ok(Self {
-            block_producer_time_nanos,
-            block_user_agent,
-        })
+    fn from_bytes(_data: &[u8]) -> Result<Self, BlockComponentError> {
+        unimplemented!()
+        // if data.len() < Self::HEADER_SIZE {
+        //     return Err(BlockComponentError::InsufficientData);
+        // }
+
+        // // Read timestamp
+        // // Unwrap: HEADER_SIZE = TIMESTAMP_SIZE + USER_AGENT_LEN_SIZE > TIMESTAMP_SIZE, so this will
+        // // never fail.
+        // let time_bytes = data[..Self::TIMESTAMP_SIZE].try_into().unwrap();
+        // let block_producer_time_nanos = u64::from_le_bytes(time_bytes);
+
+        // // Read user agent length
+        // let user_agent_len = data[Self::TIMESTAMP_SIZE] as usize;
+
+        // // Validate remaining data size
+        // if data.len() < Self::HEADER_SIZE + user_agent_len {
+        //     return Err(BlockComponentError::InsufficientData);
+        // }
+
+        // // Read user agent bytes
+        // let block_user_agent = data[Self::HEADER_SIZE..Self::HEADER_SIZE + user_agent_len].to_vec();
+
+        // Ok(Self {
+        //     block_producer_time_nanos,
+        //     block_user_agent,
+        // })
     }
 
     /// Returns the serialized size in bytes without actually serializing.

entry/src/block_component.rs

@@ -4,6 +4,7 @@
 
 pub mod consensus_message;
 pub mod migration;
+pub mod rewards_certificate;
 pub mod vote;
 
 #[cfg_attr(feature = "frozen-abi", macro_use)]

votor-messages/src/lib.rs

@@ -0,0 +1,31 @@
+//! Defines aggregates used for rewards.
+
+use {solana_bls_signatures::Signature as BLSSignature, solana_clock::Slot};
+
+/// The reward certificates that will be included in the block footer.
+pub struct RewardsCertificates {
+    /// The notar rewards certificate.
+    pub notar: RewardsCertificate,
+    /// The skip rewards certificate.
+    pub skip: RewardsCertificate,
+}
+
+/// Different types of rewards aggregates.
+#[derive(Clone, PartialEq, Eq, Debug)]
+pub enum RewardsCertificateType {
+    /// validator voted skip.
+    Skip(Slot),
+    /// validator voted notar.
+    Notar(Slot),
+}
+
+/// An aggregate used to pay rewards.
+#[derive(Clone, PartialEq, Eq, Debug)]
+pub struct RewardsCertificate {
+    /// The type of the aggregate.
+    pub cert_type: RewardsCertificateType,
+    /// The signature
+    pub signature: BLSSignature,
+    /// The bitmap for validators, see solana-signer-store for encoding format
+    pub bitmap: Vec<u8>,
+}