feat: parse BlockComponents during replay

Author: ksn6

Cargo.lock

@@ -3534,6 +3534,34 @@ impl ReplayStage {
                     let _ = cluster_slots_update_sender.send(vec![bank_slot]);
                 }
 
+                // Verify block components (header, footer) before freezing
+                // Only verify blocks that were replayed from blockstore (not leader blocks)
+                if !is_leader_block {
+                    if let Err(err) = bank
+                        .block_component_verifier
+                        .read()
+                        .unwrap()
+                        .finish(migration_status)
+                    {
+                        warn!("Block component verification failed for slot {bank_slot}: {err:?}",);
+                        let root = bank_forks.read().unwrap().root();
+                        Self::mark_dead_slot(
+                            blockstore,
+                            bank,
+                            root,
+                            &BlockstoreProcessorError::BlockComponentVerifier(err),
+                            rpc_subscriptions,
+                            slot_status_notifier,
+                            progress,
+                            duplicate_slots_to_repair,
+                            ancestor_hashes_replay_update_sender,
+                            purge_repair_slot_counter,
+                            &mut tbft_structs,
+                        );
+                        continue;
+                    }
+                }
+
                 bank.freeze();
                 datapoint_info!(
                     "bank_frozen",

core/src/replay_stage.rs

@@ -3928,14 +3928,13 @@ impl Blockstore {
             .map(|x| x.0)
     }
 
-    /// Returns the entry vector for the slot starting with `shred_start_index`, the number of
-    /// shreds that comprise the entry vector, and whether the slot is full (consumed all shreds).
-    pub fn get_slot_entries_with_shred_info(
+    /// Helper function that contains the common logic for getting slot data with shred info
+    fn get_slot_data_with_shred_info_common(
         &self,
         slot: Slot,
         start_index: u64,
         allow_dead_slots: bool,
-    ) -> Result<(Vec<Entry>, u64, bool)> {
+    ) -> Result<(CompletedRanges, SlotMeta, u64)> {
         let (completed_ranges, slot_meta) = self.get_completed_ranges(slot, start_index)?;
 
         // Check if the slot is dead *after* fetching completed ranges to avoid a race
@@ -3945,7 +3944,7 @@ impl Blockstore {
         if self.is_dead(slot) && !allow_dead_slots {
             return Err(BlockstoreError::DeadSlot);
         } else if completed_ranges.is_empty() {
-            return Ok((vec![], 0, false));
+            return Err(BlockstoreError::SlotUnavailable);
         }
 
         let slot_meta = slot_meta.unwrap();
@@ -3954,10 +3953,49 @@ impl Blockstore {
             .map(|&Range { end, .. }| u64::from(end) - start_index)
             .unwrap_or(0);
 
+        Ok((completed_ranges, slot_meta, num_shreds))
+    }
+
+    /// Returns the entry vector for the slot starting with `shred_start_index`, the number of
+    /// shreds that comprise the entry vector, and whether the slot is full (consumed all shreds).
+    pub fn get_slot_entries_with_shred_info(
+        &self,
+        slot: Slot,
+        start_index: u64,
+        allow_dead_slots: bool,
+    ) -> Result<(Vec<Entry>, u64, bool)> {
+        let (completed_ranges, slot_meta, num_shreds) =
+            match self.get_slot_data_with_shred_info_common(slot, start_index, allow_dead_slots) {
+                Ok(data) => data,
+                Err(BlockstoreError::SlotUnavailable) => return Ok((vec![], 0, false)),
+                Err(e) => return Err(e),
+            };
+
         let entries = self.get_slot_entries_in_block(slot, completed_ranges, Some(&slot_meta))?;
         Ok((entries, num_shreds, slot_meta.is_full()))
     }
 
+    /// Returns the components vector for the slot starting with `shred_start_index`, the number of
+    /// shreds that comprise the components vector, and whether the slot is full (consumed all
+    /// shreds).
+    pub fn get_slot_components_with_shred_info(
+        &self,
+        slot: Slot,
+        start_index: u64,
+        allow_dead_slots: bool,
+    ) -> Result<(Vec<BlockComponent>, u64, bool)> {
+        let (completed_ranges, slot_meta, num_shreds) =
+            match self.get_slot_data_with_shred_info_common(slot, start_index, allow_dead_slots) {
+                Ok(data) => data,
+                Err(BlockstoreError::SlotUnavailable) => return Ok((vec![], 0, false)),
+                Err(e) => return Err(e),
+            };
+
+        let components =
+            self.get_slot_components_in_block(slot, completed_ranges, Some(&slot_meta))?;
+        Ok((components, num_shreds, slot_meta.is_full()))
+    }
+
     /// Gets accounts used in transactions in the slot range [starting_slot, ending_slot].
     /// Additionally returns a bool indicating if the set may be incomplete.
     /// Used by ledger-tool to create a minimized snapshot
@@ -4061,18 +4099,24 @@ impl Blockstore {
             .collect()
     }
 
-    /// Fetch the entries corresponding to all of the shred indices in `completed_ranges`
+    /// Helper function to process shreds in `completed_ranges` and apply a transformation
+    /// to the resulting block components.
     /// This function takes advantage of the fact that `completed_ranges` are both
     /// contiguous and in sorted order. To clarify, suppose completed_ranges is as follows:
     ///   completed_ranges = [..., (s_i..e_i), (s_i+1..e_i+1), ...]
     /// Then, the following statements are true:
     ///   s_i < e_i == s_i+1 < e_i+1
-    fn get_slot_entries_in_block(
+    fn process_slot_data_in_block<T, I, F>(
         &self,
         slot: Slot,
         completed_ranges: CompletedRanges,
         slot_meta: Option<&SlotMeta>,
-    ) -> Result<Vec<Entry>> {
+        transform: F,
+    ) -> Result<Vec<T>>
+    where
+        I: IntoIterator<Item = T>,
+        F: Fn(Vec<BlockComponent>) -> Result<I>,
+    {
         debug_assert!(completed_ranges
             .iter()
             .tuple_windows()
@@ -4118,26 +4162,60 @@ impl Blockstore {
                     .and_then(|payload| {
                         // TODO(karthik): if Alpenglow flag is disabled, return an error on special
                         // EntryBatches.
-                        BlockComponent::from_bytes_multiple(&payload)
-                            .map(|cs| {
-                                cs.into_iter()
-                                    .filter_map(|bc| bc.as_entry_batch_owned())
-                                    .flatten()
-                                    .collect_vec()
-                            })
-                            .map_err(|e| {
+                        let components =
+                            BlockComponent::from_bytes_multiple(&payload).map_err(|e| {
                                 BlockstoreError::InvalidShredData(Box::new(
                                     bincode::ErrorKind::Custom(format!(
-                                        "could not reconstruct entries: {e:?}"
+                                        "could not reconstruct block components: {e:?}"
                                     )),
                                 ))
-                            })
+                            })?;
+                        transform(components).map_err(|e| {
+                            BlockstoreError::InvalidShredData(Box::new(bincode::ErrorKind::Custom(
+                                format!("could not transform block components: {e:?}"),
+                            )))
+                        })
                     })
             })
             .flatten_ok()
             .collect()
     }
 
+    /// Fetch the components corresponding to all of the shred indices in `completed_ranges`
+    /// This function takes advantage of the fact that `completed_ranges` are both
+    /// contiguous and in sorted order. To clarify, suppose completed_ranges is as follows:
+    ///   completed_ranges = [..., (s_i..e_i), (s_i+1..e_i+1), ...]
+    /// Then, the following statements are true:
+    ///   s_i < e_i == s_i+1 < e_i+1
+    fn get_slot_components_in_block(
+        &self,
+        slot: Slot,
+        completed_ranges: CompletedRanges,
+        slot_meta: Option<&SlotMeta>,
+    ) -> Result<Vec<BlockComponent>> {
+        self.process_slot_data_in_block(slot, completed_ranges, slot_meta, |cs| Ok(cs.into_iter()))
+    }
+
+    /// Fetch the entries corresponding to all of the shred indices in `completed_ranges`
+    /// This function takes advantage of the fact that `completed_ranges` are both
+    /// contiguous and in sorted order. To clarify, suppose completed_ranges is as follows:
+    ///   completed_ranges = [..., (s_i..e_i), (s_i+1..e_i+1), ...]
+    /// Then, the following statements are true:
+    ///   s_i < e_i == s_i+1 < e_i+1
+    fn get_slot_entries_in_block(
+        &self,
+        slot: Slot,
+        completed_ranges: CompletedRanges,
+        slot_meta: Option<&SlotMeta>,
+    ) -> Result<Vec<Entry>> {
+        self.process_slot_data_in_block(slot, completed_ranges, slot_meta, |cs| {
+            Ok(cs
+                .into_iter()
+                .filter_map(|bc| bc.as_entry_batch_owned())
+                .flatten())
+        })
+    }
+
     pub fn get_entries_in_data_block(
         &self,
         slot: Slot,

ledger/src/blockstore.rs

@@ -32,6 +32,7 @@ use {
         bank::{Bank, PreCommitResult, TransactionBalancesSet},
         bank_forks::{BankForks, SetRootError},
         bank_utils,
+        block_component_verifier::BlockComponentVerifierError,
         commitment::VOTE_THRESHOLD_SIZE,
         dependency_tracker::DependencyTracker,
         installed_scheduler_pool::BankWithScheduler,
@@ -836,6 +837,9 @@ pub enum BlockstoreProcessorError {
 
     #[error("user transactions found in vote only mode bank at slot {0}")]
     UserTransactionsInVoteOnlyBank(Slot),
+
+    #[error("block component verifier error: {0}")]
+    BlockComponentVerifier(#[from] BlockComponentVerifierError),
 }
 
 /// Callback for accessing bank state after each slot is confirmed while
@@ -1511,10 +1515,10 @@ pub fn confirm_slot(
 ) -> result::Result<(), BlockstoreProcessorError> {
     let slot = bank.slot();
 
-    let slot_entries_load_result = {
+    let (slot_components, num_shreds, slot_full) = {
         let mut load_elapsed = Measure::start("load_elapsed");
         let load_result = blockstore
-            .get_slot_entries_with_shred_info(slot, progress.num_shreds, allow_dead_slots)
+            .get_slot_components_with_shred_info(slot, progress.num_shreds, allow_dead_slots)
             .map_err(BlockstoreProcessorError::FailedToLoadEntries);
         load_elapsed.stop();
         if load_result.is_err() {
@@ -1525,10 +1529,37 @@ pub fn confirm_slot(
         load_result
     }?;
 
+    // Process block component markers for Alpenglow slots. Note that we don't need to run migration
+    // checks here. Here's why:
+    //
+    // Post-Alpenglow migration - validators that have Alpenglow enabled can parse BlockComponents.
+    // Things just work.
+    //
+    // Pre-Alpenglow migration, suppose a validator receives a BlockMarker:
+    //
+    // (1) validators *incapable* of processing BlockMarkers will mark the slot as dead on shred
+    //     ingest in blockstore.
+    //
+    // (2) validators *capable* of processing BlockMarkers will store the BlockMarkers in shred
+    //     ingest, run through this verifying code here, and then error out when finish() is invoked
+    //     during replay, resulting in the slot being marked as dead.
+    if let Some(parent_bank) = bank.parent() {
+        let mut verifier = bank.block_component_verifier.write().unwrap();
+        for marker in slot_components.iter().filter_map(|bc| bc.as_marker()) {
+            verifier.on_marker(bank.clone_without_scheduler(), parent_bank.clone(), marker)?;
+        }
+    }
+
+    let slot_entries = slot_components
+        .into_iter()
+        .filter_map(|c| c.as_entry_batch_owned())
+        .flatten()
+        .collect_vec();
+
     confirm_slot_entries(
         bank,
         replay_tx_thread_pool,
-        slot_entries_load_result,
+        (slot_entries, num_shreds, slot_full),
         timing,
         progress,
         skip_verification,

ledger/src/blockstore_processor.rs

@@ -107,6 +107,7 @@ solana-compute-budget-interface = { workspace = true }
 solana-cost-model = { workspace = true }
 solana-cpi = { workspace = true }
 solana-ed25519-program = { workspace = true }
+solana-entry = { workspace = true }
 solana-epoch-info = { workspace = true }
 solana-epoch-rewards-hasher = { workspace = true }
 solana-epoch-schedule = { workspace = true }
@@ -200,6 +201,7 @@ solana-instruction-error = { workspace = true }
 solana-logger = { workspace = true }
 solana-program = { workspace = true }
 solana-program-binaries = { workspace = true }
+solana-votor-messages = { workspace = true, features = ["dev-context-only-utils"] }
 # See order-crates-for-publishing.py for using this unusual `path = "."`
 solana-runtime = { path = ".", features = ["dev-context-only-utils"] }
 solana-runtime-transaction = { workspace = true, features = [

runtime/Cargo.toml

@@ -41,6 +41,7 @@ use {
             partitioned_epoch_rewards::{EpochRewardStatus, VoteRewardsAccounts},
         },
         bank_forks::BankForks,
+        block_component_verifier::BlockComponentVerifier,
         epoch_stakes::{NodeVoteAccounts, VersionedEpochStakes},
         inflation_rewards::points::InflationPointCalculationEvent,
         installed_scheduler_pool::{BankWithScheduler, InstalledSchedulerRwLock},
@@ -556,6 +557,7 @@ impl PartialEq for Bank {
             block_id,
             bank_hash_stats: _,
             epoch_rewards_calculation_cache: _,
+            block_component_verifier: _,
             // Ignore new fields explicitly if they do not impact PartialEq.
             // Adding ".." will remove compile-time checks that if a new field
             // is added to the struct, this PartialEq is accordingly updated.
@@ -903,6 +905,9 @@ pub struct Bank {
     /// This is used to avoid recalculating the same epoch rewards at epoch boundary.
     /// The hashmap is keyed by parent_hash.
     epoch_rewards_calculation_cache: Arc<Mutex<HashMap<Hash, Arc<PartitionedRewardsCalculation>>>>,
+
+    /// Block component verifier for validating block headers/footers and clock bounds
+    pub block_component_verifier: RwLock<BlockComponentVerifier>,
 }
 
 #[derive(Debug)]
@@ -1099,6 +1104,7 @@ impl Bank {
             block_id: RwLock::new(None),
             bank_hash_stats: AtomicBankHashStats::default(),
             epoch_rewards_calculation_cache: Arc::new(Mutex::new(HashMap::default())),
+            block_component_verifier: RwLock::new(BlockComponentVerifier::default()),
         };
 
         bank.transaction_processor =
@@ -1347,6 +1353,7 @@ impl Bank {
             block_id: RwLock::new(None),
             bank_hash_stats: AtomicBankHashStats::default(),
             epoch_rewards_calculation_cache: parent.epoch_rewards_calculation_cache.clone(),
+            block_component_verifier: RwLock::new(BlockComponentVerifier::default()),
         };
 
         let (_, ancestors_time_us) = measure_us!({
@@ -1815,6 +1822,7 @@ impl Bank {
             block_id: RwLock::new(None),
             bank_hash_stats: AtomicBankHashStats::new(&fields.bank_hash_stats),
             epoch_rewards_calculation_cache: Arc::new(Mutex::new(HashMap::default())),
+            block_component_verifier: RwLock::new(BlockComponentVerifier::default()),
         };
 
         // Sanity assertions between bank snapshot and genesis config