feat: perform all BlockComponent checks in blockstore processor (anza-xyz#594)

Right now, we need to complete replay to verify that `BlockComponent`s
in a block satisfy a number of invariants (e.g., exactly one header,
exactly one footer, etc.).

@AshwinSekar found a clever way to avoid this:
anza-xyz#575 (comment).

Implement the above.

Author: ksn6

ledger/src/blockstore_processor.rs

@@ -21,7 +21,10 @@ use {
     },
     solana_clock::{MAX_PROCESSING_AGE, Slot},
     solana_cost_model::{cost_model::CostModel, transaction_cost::TransactionCost},
-    solana_entry::entry::{self, Entry, EntrySlice, EntryType, create_ticks},
+    solana_entry::{
+        block_component::BlockComponent,
+        entry::{self, Entry, EntrySlice, EntryType, create_ticks},
+    },
     solana_genesis_config::GenesisConfig,
     solana_hash::Hash,
     solana_keypair::Keypair,
@@ -32,6 +35,7 @@ use {
         bank::{Bank, PreCommitResult, TransactionBalancesSet},
         bank_forks::BankForks,
         bank_utils,
+        block_component_processor::BlockComponentProcessorError,
         commitment::VOTE_THRESHOLD_SIZE,
         dependency_tracker::DependencyTracker,
         installed_scheduler_pool::BankWithScheduler,
@@ -824,6 +828,9 @@ pub enum BlockstoreProcessorError {
     #[error("user transactions found in vote only mode bank at slot {0}")]
     UserTransactionsInVoteOnlyBank(Slot),
 
+    #[error("block component processor error: {0}")]
+    BlockComponentProcessor(#[from] BlockComponentProcessorError),
+
     #[error("invalid parent -> child chained merkle root at slot {0} parent {1}")]
     ChainedBlockIdFailure(Slot, Slot),
 }
@@ -1494,10 +1501,10 @@ pub fn confirm_slot(
 ) -> result::Result<(), BlockstoreProcessorError> {
     let slot = bank.slot();
 
-    let slot_entries_load_result = {
+    let (slot_components, completed_ranges, 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() {
@@ -1508,20 +1515,79 @@ pub fn confirm_slot(
         load_result
     }?;
 
-    confirm_slot_entries(
-        bank,
-        replay_tx_thread_pool,
-        slot_entries_load_result,
-        timing,
-        progress,
-        skip_verification,
-        transaction_status_sender,
-        entry_notification_sender,
-        replay_vote_sender,
-        log_messages_bytes_limit,
-        prioritization_fee_cache,
-        migration_status,
-    )
+    // Process block components for Alpenglow slots. Note that we don't need to run migration checks
+    // for BlockMarkers here, despite BlockMarkers only being active post-Alpenglow. 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 processing a
+    //     BlockMarker, resulting in the slot being marked as dead.
+    let mut processor = bank.block_component_processor.write().unwrap();
+
+    // Find the index of the last EntryBatch in slot_components
+    let last_entry_batch_index = slot_components
+        .iter()
+        .rposition(|bc| matches!(bc, BlockComponent::EntryBatch(_)));
+
+    for (ix, (completed_range, component)) in completed_ranges
+        .iter()
+        .zip(slot_components.into_iter())
+        .enumerate()
+    {
+        let num_shreds = completed_range.end - completed_range.start;
+        let is_final = slot_full && ix == completed_ranges.len() - 1;
+
+        match component {
+            BlockComponent::EntryBatch(entries) => {
+                let slot_full = slot_full
+                    && last_entry_batch_index.is_some_and(|last_entry_batch_ix| {
+                        ix == last_entry_batch_ix
+                    });
+
+                // Skip block component validation for genesis block. Slot 0 is handled specially,
+                // since it won't have the required block markers (e.g., the header and the footer).
+                if bank.slot() != 0 {
+                    processor
+                        .on_entry_batch(migration_status, is_final)
+                        .inspect_err(|err| {
+                            warn!("Block component processing failed for slot {slot}: {err:?}",);
+                        })?;
+                }
+
+                confirm_slot_entries(
+                    bank,
+                    replay_tx_thread_pool,
+                    (entries, num_shreds as u64, slot_full),
+                    timing,
+                    progress,
+                    skip_verification,
+                    transaction_status_sender,
+                    entry_notification_sender,
+                    replay_vote_sender,
+                    log_messages_bytes_limit,
+                    prioritization_fee_cache,
+                    migration_status,
+                )?;
+            }
+            BlockComponent::BlockMarker(marker) => {
+                processor
+                    .on_marker(&marker, migration_status, is_final)
+                    .inspect_err(|err| {
+                        warn!("Block component processing failed for slot {slot}: {err:?}",);
+                    })?;
+                progress.num_shreds += num_shreds as u64;
+            }
+        }
+    }
+
+    Ok(())
 }
 
 #[allow(clippy::too_many_arguments)]

runtime/src/bank.rs

@@ -43,6 +43,7 @@ use {
             },
         },
         bank_forks::BankForks,
+        block_component_processor::BlockComponentProcessor,
         epoch_stakes::{
             BLSPubkeyToRankMap, DeserializableVersionedEpochStakes, NodeVoteAccounts,
             VersionedEpochStakes,
@@ -600,6 +601,7 @@ impl PartialEq for Bank {
             stats_for_accounts_lt_hash: _,
             block_id,
             expected_bank_hash: _,
+            block_component_processor: _,
             bank_hash_stats: _,
             epoch_rewards_calculation_cache: _,
             // Ignore new fields explicitly if they do not impact PartialEq.
@@ -946,6 +948,9 @@ pub struct Bank {
     /// later when the bank is frozen.
     expected_bank_hash: RwLock<Option<Hash>>,
 
+    /// Processor for validating the per-slot block component stream.
+    pub block_component_processor: RwLock<BlockComponentProcessor>,
+
     /// Accounts stats for computing the bank hash
     bank_hash_stats: AtomicBankHashStats,
 
@@ -1160,6 +1165,7 @@ impl Bank {
             stats_for_accounts_lt_hash: AccountsLtHashStats::default(),
             block_id: RwLock::new(None),
             expected_bank_hash: RwLock::new(None),
+            block_component_processor: RwLock::new(BlockComponentProcessor::default()),
             bank_hash_stats: AtomicBankHashStats::default(),
             epoch_rewards_calculation_cache: Arc::new(Mutex::new(HashMap::default())),
         };
@@ -1408,6 +1414,7 @@ impl Bank {
             stats_for_accounts_lt_hash: AccountsLtHashStats::default(),
             block_id: RwLock::new(None),
             expected_bank_hash: RwLock::new(None),
+            block_component_processor: RwLock::new(BlockComponentProcessor::default()),
             bank_hash_stats: AtomicBankHashStats::default(),
             epoch_rewards_calculation_cache: parent.epoch_rewards_calculation_cache.clone(),
         };
@@ -1970,6 +1977,7 @@ impl Bank {
             cache_for_accounts_lt_hash: DashMap::default(),
             stats_for_accounts_lt_hash: AccountsLtHashStats::default(),
             block_id: RwLock::new(None),
+            block_component_processor: RwLock::new(BlockComponentProcessor::default()),
             bank_hash_stats: AtomicBankHashStats::new(&fields.bank_hash_stats),
             epoch_rewards_calculation_cache: Arc::new(Mutex::new(HashMap::default())),
             expected_bank_hash: RwLock::new(None),

runtime/src/block_component_processor.rs

@@ -27,19 +27,8 @@ pub struct BlockComponentProcessor {
 }
 
 impl BlockComponentProcessor {
-    pub fn finish(
-        &self,
-        migration_status: &MigrationStatus,
-    ) -> Result<(), BlockComponentProcessorError> {
-        // Pre-migration: blocks with block components should be marked as dead
-        if !migration_status.is_alpenglow_enabled() {
-            match self.has_footer || self.has_header {
-                false => return Ok(()),
-                true => return Err(BlockComponentProcessorError::BlockComponentPreMigration),
-            }
-        }
-
-        // Post-migration: both header and footer are required
+    fn on_final(&self) -> Result<(), BlockComponentProcessorError> {
+        // Post-migration: both header and footer are required.
         if !self.has_footer {
             return Err(BlockComponentProcessorError::MissingBlockFooter);
         }
@@ -51,25 +40,64 @@ impl BlockComponentProcessor {
         Ok(())
     }
 
+    pub fn on_entry_batch(
+        &mut self,
+        migration_status: &MigrationStatus,
+        is_final: bool,
+    ) -> Result<(), BlockComponentProcessorError> {
+        if !migration_status.is_alpenglow_enabled() {
+            return Ok(());
+        }
+
+        // The block header must be the first component of each block.
+        if !self.has_header {
+            return Err(BlockComponentProcessorError::MissingBlockHeader);
+        }
+
+        if is_final {
+            self.on_final()
+        } else {
+            Ok(())
+        }
+    }
+
     pub fn on_marker(
         &mut self,
         marker: &VersionedBlockMarker,
+        migration_status: &MigrationStatus,
+        is_final: bool,
     ) -> Result<(), BlockComponentProcessorError> {
+        // Pre-migration: blocks with block components should be marked as dead.
+        if !migration_status.is_alpenglow_enabled() {
+            return Err(BlockComponentProcessorError::BlockComponentPreMigration);
+        }
+
         let VersionedBlockMarker::V1(marker) = marker;
 
         match marker {
             BlockMarkerV1::BlockFooter(footer) => self.on_footer(footer.inner()),
             BlockMarkerV1::BlockHeader(header) => self.on_header(header.inner()),
-            // We process UpdateParent messages on shred ingest, so no callback needed here
+            // We process UpdateParent messages on shred ingest, so no callback needed here.
             BlockMarkerV1::UpdateParent(_) => Ok(()),
             BlockMarkerV1::GenesisCertificate(_) => Ok(()),
+        }?;
+
+        if is_final {
+            self.on_final()
+        } else {
+            Ok(())
         }
     }
 
     fn on_footer(
         &mut self,
         _footer: &VersionedBlockFooter,
     ) -> Result<(), BlockComponentProcessorError> {
+        // The block header must be the first component of each block.
+        if !self.has_header {
+            return Err(BlockComponentProcessorError::MissingBlockHeader);
+        }
+
         if self.has_footer {
             return Err(BlockComponentProcessorError::MultipleBlockFooters);
         }