Make sure to root local slots even with hard fork

Author: ryoqun

core/src/consensus.rs

@@ -1072,7 +1072,7 @@ impl Tower {
         if let Some(last_voted_slot) = self.last_voted_slot() {
             if tower_root <= replayed_root {
                 // Normally, we goes into this clause with possible help of
-                // reconcile_blockstore_roots_with_tower()
+                // reconcile_blockstore_roots_with_external_source()
                 if slot_history.check(last_voted_slot) == Check::TooOld {
                     // We could try hard to anchor with other older votes, but opt to simplify the
                     // following logic
@@ -1320,45 +1320,63 @@ impl TowerError {
     }
 }
 
+#[derive(Debug)]
+pub enum ExternalRootSource {
+    Tower(Slot),
+    HardFork(Slot),
+}
+
+impl ExternalRootSource {
+    fn root(&self) -> Slot {
+        match self {
+            ExternalRootSource::Tower(slot) => *slot,
+            ExternalRootSource::HardFork(slot) => *slot,
+        }
+    }
+}
+
 // Given an untimely crash, tower may have roots that are not reflected in blockstore,
 // or the reverse of this.
 // That's because we don't impose any ordering guarantee or any kind of write barriers
 // between tower (plain old POSIX fs calls) and blockstore (through RocksDB), when
 // `ReplayState::handle_votable_bank()` saves tower before setting blockstore roots.
-pub fn reconcile_blockstore_roots_with_tower(
-    tower: &Tower,
+pub fn reconcile_blockstore_roots_with_external_source(
+    external_source: ExternalRootSource,
     blockstore: &Blockstore,
+    last_blockstore_root: &mut Slot,
 ) -> blockstore_db::Result<()> {
-    let tower_root = tower.root();
-    let last_blockstore_root = blockstore.last_root();
-    if last_blockstore_root < tower_root {
-        // Ensure tower_root itself to exist and be marked as rooted in the blockstore
+    let external_root = external_source.root();
+    if *last_blockstore_root < external_root {
+        // Ensure external_root itself to exist and be marked as rooted in the blockstore
         // in addition to its ancestors.
-        let new_roots: Vec<_> = AncestorIterator::new_inclusive(tower_root, blockstore)
-            .take_while(|current| match current.cmp(&last_blockstore_root) {
+        let new_roots: Vec<_> = AncestorIterator::new_inclusive(external_root, blockstore)
+            .take_while(|current| match current.cmp(last_blockstore_root) {
                 Ordering::Greater => true,
                 Ordering::Equal => false,
                 Ordering::Less => panic!(
-                    "couldn't find a last_blockstore_root upwards from: {}!?",
-                    tower_root
+                    "last_blockstore_root({}) is skipped while traversing blockstore (currently at {}) from external root ({})!?",
+                    last_blockstore_root,
+                    current,
+                    external_root
                 ),
             })
             .collect();
         if !new_roots.is_empty() {
             info!(
-                "Reconciling slots as root based on tower root: {:?} ({}..{}) ",
-                new_roots, tower_root, last_blockstore_root
+                "Reconciling slots as root based on external root: {:?} (external: {:?}, blockstore: {})",
+                new_roots, external_source, last_blockstore_root
             );
             blockstore.set_roots(new_roots.iter())?;
+            *last_blockstore_root = blockstore.last_root();
         } else {
             // This indicates we're in bad state; but still don't panic here.
             // That's because we might have a chance of recovering properly with
             // newer snapshot.
             warn!(
-                "Couldn't find any ancestor slots from tower root ({}) \
+                "Couldn't find any ancestor slots from external source ({:?}) \
                  towards blockstore root ({}); blockstore pruned or only \
-                 tower moved into new ledger?",
-                tower_root, last_blockstore_root,
+                 tower moved into new ledger or just hard fork?",
+                external_source, last_blockstore_root,
             );
         }
     }
@@ -2814,7 +2832,12 @@ pub mod test {
 
             let mut tower = Tower::default();
             tower.vote_state.root_slot = Some(4);
-            reconcile_blockstore_roots_with_tower(&tower, &blockstore).unwrap();
+            reconcile_blockstore_roots_with_external_source(
+                ExternalRootSource::Tower(tower.root()),
+                &blockstore,
+                &mut blockstore.last_root(),
+            )
+            .unwrap();
 
             assert!(!blockstore.is_root(0));
             assert!(blockstore.is_root(1));
@@ -2825,7 +2848,9 @@ pub mod test {
     }
 
     #[test]
-    #[should_panic(expected = "couldn't find a last_blockstore_root upwards from: 4!?")]
+    #[should_panic(
+        expected = "last_blockstore_root(3) is skipped while traversing blockstore (currently at 1) from external root (4)!?"
+    )]
     fn test_reconcile_blockstore_roots_with_tower_panic_no_common_root() {
         solana_logger::setup();
         let blockstore_path = get_tmp_ledger_path!();
@@ -2846,7 +2871,12 @@ pub mod test {
 
             let mut tower = Tower::default();
             tower.vote_state.root_slot = Some(4);
-            reconcile_blockstore_roots_with_tower(&tower, &blockstore).unwrap();
+            reconcile_blockstore_roots_with_external_source(
+                ExternalRootSource::Tower(tower.root()),
+                &blockstore,
+                &mut blockstore.last_root(),
+            )
+            .unwrap();
         }
         Blockstore::destroy(&blockstore_path).expect("Expected successful database destruction");
     }
@@ -2869,7 +2899,12 @@ pub mod test {
             let mut tower = Tower::default();
             tower.vote_state.root_slot = Some(4);
             assert_eq!(blockstore.last_root(), 0);
-            reconcile_blockstore_roots_with_tower(&tower, &blockstore).unwrap();
+            reconcile_blockstore_roots_with_external_source(
+                ExternalRootSource::Tower(tower.root()),
+                &blockstore,
+                &mut blockstore.last_root(),
+            )
+            .unwrap();
             assert_eq!(blockstore.last_root(), 0);
         }
         Blockstore::destroy(&blockstore_path).expect("Expected successful database destruction");

core/src/validator.rs

@@ -8,7 +8,7 @@ use {
         cache_block_meta_service::{CacheBlockMetaSender, CacheBlockMetaService},
         cluster_info_vote_listener::VoteTracker,
         completed_data_sets_service::CompletedDataSetsService,
-        consensus::{reconcile_blockstore_roots_with_tower, Tower},
+        consensus::{reconcile_blockstore_roots_with_external_source, ExternalRootSource, Tower},
         ledger_metric_report_service::LedgerMetricReportService,
         poh_timing_report_service::PohTimingReportService,
         rewards_recorder_service::{RewardsRecorderSender, RewardsRecorderService},
@@ -512,6 +512,7 @@ impl Validator {
             genesis_config,
             bank_forks,
             blockstore,
+            original_blockstore_root,
             ledger_signal_receiver,
             completed_slots_receiver,
             leader_schedule_cache,
@@ -667,6 +668,7 @@ impl Validator {
             vote_account,
             &start_progress,
             &blockstore,
+            original_blockstore_root,
             &bank_forks,
             &leader_schedule_cache,
             &blockstore_process_options,
@@ -1224,6 +1226,17 @@ fn check_poh_speed(genesis_config: &GenesisConfig, maybe_hash_samples: Option<u6
     }
 }
 
+fn maybe_cluster_restart_with_hard_fork(config: &ValidatorConfig, root_slot: Slot) -> Option<Slot> {
+    // detect cluster restart (hard fork) indirectly via wait_for_supermajority...
+    if let Some(wait_slot_for_supermajority) = config.wait_for_supermajority {
+        if wait_slot_for_supermajority == root_slot {
+            return Some(wait_slot_for_supermajority);
+        }
+    }
+
+    None
+}
+
 fn post_process_restored_tower(
     restored_tower: crate::consensus::Result<Tower>,
     validator_identity: &Pubkey,
@@ -1237,29 +1250,28 @@ fn post_process_restored_tower(
         .and_then(|tower| {
             let root_bank = bank_forks.root_bank();
             let slot_history = root_bank.get_slot_history();
+            // make sure tower isn't corrupted first before the following hard fork check
             let tower = tower.adjust_lockouts_after_replay(root_bank.slot(), &slot_history);
 
-            if let Some(wait_slot_for_supermajority) = config.wait_for_supermajority {
-                if root_bank.slot() == wait_slot_for_supermajority {
-                    // intentionally fail to restore tower; we're supposedly in a new hard fork; past
-                    // out-of-chain vote state doesn't make sense at all
-                    // what if --wait-for-supermajority again if the validator restarted?
-                    let message = format!("Hardfork is detected; discarding tower restoration result: {:?}", tower);
-                    datapoint_error!(
-                        "tower_error",
-                        (
-                            "error",
-                            message,
-                            String
-                        ),
-                    );
-                    error!("{}", message);
+            if let Some(hard_fork_restart_slot) = maybe_cluster_restart_with_hard_fork(config, root_bank.slot()) {
+                // intentionally fail to restore tower; we're supposedly in a new hard fork; past
+                // out-of-chain vote state doesn't make sense at all
+                // what if --wait-for-supermajority again if the validator restarted?
+                let message = format!("Hard fork is detected; discarding tower restoration result: {:?}", tower);
+                datapoint_error!(
+                    "tower_error",
+                    (
+                        "error",
+                        message,
+                        String
+                    ),
+                );
+                error!("{}", message);
 
-                    // unconditionally relax tower requirement so that we can always restore tower
-                    // from root bank.
-                    should_require_tower = false;
-                    return Err(crate::consensus::TowerError::HardFork(wait_slot_for_supermajority));
-                }
+                // unconditionally relax tower requirement so that we can always restore tower
+                // from root bank.
+                should_require_tower = false;
+                return Err(crate::consensus::TowerError::HardFork(hard_fork_restart_slot));
             }
 
             if let Some(warp_slot) = config.warp_slot {
@@ -1326,6 +1338,7 @@ fn load_blockstore(
     GenesisConfig,
     Arc<RwLock<BankForks>>,
     Arc<Blockstore>,
+    Slot,
     Receiver<bool>,
     CompletedSlotsReceiver,
     LeaderScheduleCache,
@@ -1378,6 +1391,9 @@ fn load_blockstore(
     .expect("Failed to open ledger database");
     blockstore.set_no_compaction(config.no_rocksdb_compaction);
     blockstore.shred_timing_point_sender = poh_timing_point_sender;
+    // following boot sequence (esp BankForks) could set root. so stash the original value
+    // of blockstore root away here as soon as possible.
+    let original_blockstore_root = blockstore.last_root();
 
     let blockstore = Arc::new(blockstore);
     let blockstore_root_scan = BlockstoreRootScan::new(config, &blockstore, exit);
@@ -1482,6 +1498,7 @@ fn load_blockstore(
         genesis_config,
         bank_forks,
         blockstore,
+        original_blockstore_root,
         ledger_signal_receiver,
         completed_slots_receiver,
         leader_schedule_cache,
@@ -1521,6 +1538,7 @@ struct ProcessBlockStore<'a> {
     vote_account: &'a Pubkey,
     start_progress: &'a Arc<RwLock<ValidatorStartProgress>>,
     blockstore: &'a Blockstore,
+    original_blockstore_root: Slot,
     bank_forks: &'a Arc<RwLock<BankForks>>,
     leader_schedule_cache: &'a LeaderScheduleCache,
     process_options: &'a blockstore_processor::ProcessOptions,
@@ -1539,6 +1557,7 @@ impl<'a> ProcessBlockStore<'a> {
         vote_account: &'a Pubkey,
         start_progress: &'a Arc<RwLock<ValidatorStartProgress>>,
         blockstore: &'a Blockstore,
+        original_blockstore_root: Slot,
         bank_forks: &'a Arc<RwLock<BankForks>>,
         leader_schedule_cache: &'a LeaderScheduleCache,
         process_options: &'a blockstore_processor::ProcessOptions,
@@ -1553,6 +1572,7 @@ impl<'a> ProcessBlockStore<'a> {
             vote_account,
             start_progress,
             blockstore,
+            original_blockstore_root,
             bank_forks,
             leader_schedule_cache,
             process_options,
@@ -1622,12 +1642,30 @@ impl<'a> ProcessBlockStore<'a> {
             self.tower = Some({
                 let restored_tower = Tower::restore(self.config.tower_storage.as_ref(), self.id);
                 if let Ok(tower) = &restored_tower {
-                    reconcile_blockstore_roots_with_tower(tower, self.blockstore).unwrap_or_else(
-                        |err| {
-                            error!("Failed to reconcile blockstore with tower: {:?}", err);
-                            abort()
-                        },
-                    );
+                    reconcile_blockstore_roots_with_external_source(
+                        ExternalRootSource::Tower(tower.root()),
+                        self.blockstore,
+                        &mut self.original_blockstore_root,
+                    )
+                    .unwrap_or_else(|err| {
+                        error!("Failed to reconcile blockstore with tower: {:?}", err);
+                        abort()
+                    });
+                }
+
+                if let Some(hard_fork_restart_slot) = maybe_cluster_restart_with_hard_fork(
+                    self.config,
+                    self.bank_forks.read().unwrap().root_bank().slot(),
+                ) {
+                    reconcile_blockstore_roots_with_external_source(
+                        ExternalRootSource::HardFork(hard_fork_restart_slot),
+                        self.blockstore,
+                        &mut self.original_blockstore_root,
+                    )
+                    .unwrap_or_else(|err| {
+                        error!("Failed to reconcile blockstore with hard fork: {:?}", err);
+                        abort()
+                    });
                 }
 
                 post_process_restored_tower(

ledger/src/blockstore.rs

@@ -523,13 +523,28 @@ impl Blockstore {
         Ok(slot_iterator.take_while(move |((shred_slot, _), _)| *shred_slot == slot))
     }
 
-    pub fn rooted_slot_iterator(&self, slot: Slot) -> Result<impl Iterator<Item = u64> + '_> {
+    fn prepare_rooted_slot_iterator(
+        &self,
+        slot: Slot,
+        direction: IteratorDirection,
+    ) -> Result<impl Iterator<Item = Slot> + '_> {
         let slot_iterator = self
             .db
-            .iter::<cf::Root>(IteratorMode::From(slot, IteratorDirection::Forward))?;
+            .iter::<cf::Root>(IteratorMode::From(slot, direction))?;
         Ok(slot_iterator.map(move |(rooted_slot, _)| rooted_slot))
     }
 
+    pub fn rooted_slot_iterator(&self, slot: Slot) -> Result<impl Iterator<Item = Slot> + '_> {
+        self.prepare_rooted_slot_iterator(slot, IteratorDirection::Forward)
+    }
+
+    pub fn reversed_rooted_slot_iterator(
+        &self,
+        slot: Slot,
+    ) -> Result<impl Iterator<Item = Slot> + '_> {
+        self.prepare_rooted_slot_iterator(slot, IteratorDirection::Reverse)
+    }
+
     /// Determines if starting_slot and ending_slot are connected
     pub fn slots_connected(&self, starting_slot: Slot, ending_slot: Slot) -> bool {
         let mut next_slots: VecDeque<_> = vec![starting_slot].into();
@@ -1725,6 +1740,13 @@ impl Blockstore {
         self.meta_cf.put_bytes(slot, bytes)
     }
 
+    /// Manually update the meta for a slot.
+    /// Can interfere with automatic meta update and potentially break chaining.
+    /// Dangerous. Use with care.
+    pub fn put_meta(&self, slot: Slot, meta: &SlotMeta) -> Result<()> {
+        self.put_meta_bytes(slot, &bincode::serialize(meta)?)
+    }
+
     // Given a start and end entry index, find all the missing
     // indexes in the ledger in the range [start_index, end_index)
     // for the slot with the specified slot

ledger/src/blockstore_meta.rs

@@ -214,6 +214,10 @@ impl SlotMeta {
         Some(self.consumed) == self.last_index.map(|ix| ix + 1)
     }
 
+    pub fn unset_parent(&mut self) {
+        self.parent_slot = None;
+    }
+
     pub fn clear_unconfirmed_slot(&mut self) {
         let mut new_self = SlotMeta::new_orphan(self.slot);
         std::mem::swap(&mut new_self.next_slots, &mut self.next_slots);