Fix pre-check of blockstore slots during load_bank_forks (#25632)

Fix pre-check of blockstore slts during load_bank_forks. Now iterates from starting_slot to halt_slot via slot_meta.next_slots to confirm they are connected.

ledger-tool/src/main.rs

@@ -774,7 +774,6 @@ fn load_bank_forks(
             snapshot_archive_path.unwrap_or_else(|| blockstore.ledger_path().to_path_buf());
         let incremental_snapshot_archives_dir =
             incremental_snapshot_archive_path.unwrap_or_else(|| full_snapshot_archives_dir.clone());
-
         if let Some(full_snapshot_slot) =
             snapshot_utils::get_highest_full_snapshot_archive_slot(&full_snapshot_archives_dir)
         {
@@ -798,20 +797,12 @@ fn load_bank_forks(
     };
 
     if let Some(halt_slot) = process_options.halt_at_slot {
-        for slot in starting_slot..=halt_slot {
-            if let Ok(Some(slot_meta)) = blockstore.meta(slot) {
-                if !slot_meta.is_full() {
-                    eprintln!("Unable to process from slot {} to {} due to blockstore slot {} not being full",
-                        starting_slot, halt_slot, slot);
-                    exit(1);
-                }
-            } else {
-                eprintln!(
-                    "Unable to process from slot {} to {} due to blockstore missing slot {}",
-                    starting_slot, halt_slot, slot
-                );
-                exit(1);
-            }
+        if !blockstore.slots_connected(starting_slot, halt_slot) {
+            eprintln!(
+                "Unable to load bank forks at slot {} due to disconnected blocks.",
+                halt_slot,
+            );
+            exit(1);
         }
     }
 

ledger/src/blockstore.rs

@@ -52,7 +52,7 @@ use {
         borrow::Cow,
         cell::RefCell,
         cmp,
-        collections::{hash_map::Entry as HashMapEntry, BTreeSet, HashMap, HashSet},
+        collections::{hash_map::Entry as HashMapEntry, BTreeSet, HashMap, HashSet, VecDeque},
         convert::TryInto,
         fmt::Write,
         fs,
@@ -527,6 +527,24 @@ impl Blockstore {
         Ok(slot_iterator.map(move |(rooted_slot, _)| rooted_slot))
     }
 
+    /// 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();
+        while let Some(slot) = next_slots.pop_front() {
+            if let Ok(Some(slot_meta)) = self.meta(slot) {
+                if slot_meta.is_full() {
+                    match slot.cmp(&ending_slot) {
+                        cmp::Ordering::Less => next_slots.extend(slot_meta.next_slots),
+                        cmp::Ordering::Equal => return true,
+                        cmp::Ordering::Greater => {} // slot is greater than the ending slot, so all its children would be as well
+                    }
+                }
+            }
+        }
+
+        false
+    }
+
     fn get_recovery_data_shreds<'a>(
         index: &'a Index,
         slot: Slot,
@@ -5380,6 +5398,49 @@ pub mod tests {
 
         }
     */
+    #[test]
+    pub fn test_slots_connected_chain() {
+        let ledger_path = get_tmp_ledger_path_auto_delete!();
+        let blockstore = Blockstore::open(ledger_path.path()).unwrap();
+
+        let num_slots = 3;
+        for slot in 1..=num_slots {
+            let (shreds, _) = make_slot_entries(slot, slot.saturating_sub(1), 100);
+            blockstore.insert_shreds(shreds, None, true).unwrap();
+
+            let meta = blockstore.meta(slot).unwrap().unwrap();
+            assert_eq!(slot, meta.slot);
+            assert!(meta.is_full());
+            assert!(meta.next_slots.is_empty());
+        }
+
+        assert!(blockstore.slots_connected(1, 3));
+        assert!(!blockstore.slots_connected(1, 4)); // slot 4 does not exist
+    }
+
+    #[test]
+    pub fn test_slots_connected_disconnected() {
+        let ledger_path = get_tmp_ledger_path_auto_delete!();
+        let blockstore = Blockstore::open(ledger_path.path()).unwrap();
+
+        fn make_and_insert_slot(blockstore: &Blockstore, slot: Slot, parent_slot: Slot) {
+            let (shreds, _) = make_slot_entries(slot, parent_slot, 100);
+            blockstore.insert_shreds(shreds, None, true).unwrap();
+
+            let meta = blockstore.meta(slot).unwrap().unwrap();
+            assert_eq!(slot, meta.slot);
+            assert!(meta.is_full());
+            assert!(meta.next_slots.is_empty());
+        }
+
+        make_and_insert_slot(&blockstore, 1, 0);
+        make_and_insert_slot(&blockstore, 2, 1);
+        make_and_insert_slot(&blockstore, 4, 2);
+
+        assert!(!blockstore.slots_connected(1, 3)); // Slot 3 does not exit
+        assert!(blockstore.slots_connected(1, 4));
+    }
+
     #[test]
     pub fn test_get_slots_since() {
         let ledger_path = get_tmp_ledger_path_auto_delete!();