feat: perform all BlockComponent checks in blockstore processor

core/src/replay_stage.rs

@@ -3690,34 +3690,6 @@ impl ReplayStage {
                     let _ = cluster_slots_update_sender.send(vec![bank_slot]);
                 }
 
-                // Verify and process block components (e.g., 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_processor
-                        .read()
-                        .unwrap()
-                        .finish(migration_status)
-                    {
-                        warn!("Block component processing failed for slot {bank_slot}: {err:?}",);
-                        let root = bank_forks.read().unwrap().root();
-                        Self::mark_dead_slot(
-                            blockstore,
-                            bank,
-                            root,
-                            &BlockstoreProcessorError::BlockComponentProcessor(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",

ledger/src/blockstore_processor.rs

@@ -1526,8 +1526,8 @@ pub fn confirm_slot(
     //     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.
+    //     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
@@ -1541,11 +1541,22 @@ pub fn confirm_slot(
         .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 && ix == last_entry_batch_index.unwrap();
 
+                // 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,
@@ -1563,9 +1574,18 @@ pub fn confirm_slot(
                 )?;
             }
             BlockComponent::BlockMarker(marker) => {
-                // Skip verification for the genesis block
                 if let Some(parent_bank) = bank.parent() {
-                    processor.on_marker(bank.clone_without_scheduler(), parent_bank, &marker)?;
+                    processor
+                        .on_marker(
+                            bank.clone_without_scheduler(),
+                            parent_bank,
+                            &marker,
+                            migration_status,
+                            is_final,
+                        )
+                        .inspect_err(|err| {
+                            warn!("Block component processing failed for slot {slot}: {err:?}",);
+                        })?;
                 }
                 progress.num_shreds += num_shreds as u64;
             }