migration: validate that migrationary blocks are VoM (#566)

Split from #502 

#### Problem
Blocks in the migration period should not contain user transactions.
#565 handles this on the block packing side, but we need verification on
the replay side.

#### Summary of Changes
Mark migration period blocks as vote only. During replay mark vote only
blocks which contain user transactions as dead.

Author: AshwinSekar

core/src/replay_stage.rs

@@ -809,6 +809,7 @@ impl ReplayStage {
                     &slot_status_notifier,
                     &mut progress,
                     &mut replay_timing,
+                    migration_status.as_ref(),
                 );
                 generate_new_bank_forks_time.stop();
 
@@ -4359,6 +4360,7 @@ impl ReplayStage {
         slot_status_notifier: &Option<SlotStatusNotifier>,
         progress: &mut ProgressMap,
         replay_timing: &mut ReplayLoopTiming,
+        migration_status: &MigrationStatus,
     ) {
         // Find the next slot that chains to the old slot
         let mut generate_new_bank_forks_read_lock =
@@ -4406,14 +4408,21 @@ impl ReplayStage {
                     parent_slot,
                     forks.root()
                 );
+                // Migration period banks are VoM
+                let options = NewBankOptions {
+                    vote_only_bank: migration_status.should_bank_be_vote_only(child_slot),
+                };
+                if options.vote_only_bank {
+                    info!("Replaying block in slot {child_slot} in VoM");
+                }
                 let child_bank = Self::new_bank_from_parent_with_notify(
                     parent_bank.clone(),
                     child_slot,
                     forks.root(),
                     &leader,
                     rpc_subscriptions,
                     slot_status_notifier,
-                    NewBankOptions::default(),
+                    options,
                 );
                 blockstore_processor::set_alpenglow_ticks(&child_bank);
                 let empty: Vec<Pubkey> = vec![];
@@ -4433,7 +4442,6 @@ impl ReplayStage {
         let mut generate_new_bank_forks_write_lock =
             Measure::start("generate_new_bank_forks_write_lock");
 
-        // TODO(ksn): should we have this if-statement check?
         if !new_banks.is_empty() {
             let mut forks = bank_forks.write().unwrap();
             let root = forks.root();
@@ -4805,6 +4813,7 @@ pub(crate) mod tests {
             &None,
             &mut progress,
             &mut replay_timing,
+            &MigrationStatus::default(),
         );
         assert!(bank_forks
             .read()
@@ -4829,6 +4838,7 @@ pub(crate) mod tests {
             &None,
             &mut progress,
             &mut replay_timing,
+            &MigrationStatus::default(),
         );
         assert!(bank_forks
             .read()
@@ -6732,6 +6742,7 @@ pub(crate) mod tests {
             &None,
             &mut progress,
             &mut replay_timing,
+            &MigrationStatus::default(),
         );
         assert_eq!(bank_forks.read().unwrap().active_bank_slots(), vec![3]);
 
@@ -6762,6 +6773,7 @@ pub(crate) mod tests {
             &None,
             &mut progress,
             &mut replay_timing,
+            &MigrationStatus::default(),
         );
         assert_eq!(bank_forks.read().unwrap().active_bank_slots(), vec![5]);
 
@@ -6793,6 +6805,7 @@ pub(crate) mod tests {
             &None,
             &mut progress,
             &mut replay_timing,
+            &MigrationStatus::default(),
         );
         assert_eq!(bank_forks.read().unwrap().active_bank_slots(), vec![6]);
 
@@ -6823,6 +6836,7 @@ pub(crate) mod tests {
             &None,
             &mut progress,
             &mut replay_timing,
+            &MigrationStatus::default(),
         );
         assert_eq!(bank_forks.read().unwrap().active_bank_slots(), vec![7]);
     }

ledger/src/blockstore_processor.rs

@@ -833,6 +833,9 @@ pub enum BlockstoreProcessorError {
 
     #[error("non consecutive leader slot for bank {0} parent {1}")]
     NonConsecutiveLeaderSlot(Slot, Slot),
+
+    #[error("user transactions found in vote only mode bank at slot {0}")]
+    UserTransactionsInVoteOnlyBank(Slot),
 }
 
 /// Callback for accessing bank state after each slot is confirmed while
@@ -1673,14 +1676,30 @@ fn confirm_slot_entries(
         .expect("Transaction verification generates entries");
 
     let mut replay_timer = Measure::start("replay_elapsed");
+    let is_vote_only_bank = bank.vote_only_bank();
     let replay_entries: Vec<_> = entries
         .into_iter()
         .zip(entry_tx_starting_indexes)
-        .map(|(entry, tx_starting_index)| ReplayEntry {
-            entry,
-            starting_index: tx_starting_index,
+        .map(|(entry, tx_starting_index)| {
+            // If bank is in vote-only mode, validate that entries contain only vote transactions
+            if let EntryType::Transactions(ref transactions) = entry {
+                if is_vote_only_bank
+                    && transactions
+                        .iter()
+                        .any(|tx| !tx.is_simple_vote_transaction())
+                {
+                    return Err(BlockstoreProcessorError::UserTransactionsInVoteOnlyBank(
+                        bank.slot(),
+                    ));
+                }
+            }
+            Ok(ReplayEntry {
+                entry,
+                starting_index: tx_starting_index,
+            })
         })
-        .collect();
+        .collect::<result::Result<Vec<_>, _>>()?;
+
     let process_result = process_entries(
         bank,
         replay_tx_thread_pool,