Purge next slots to avoid a blockstore_processor panic on restart (#10281) (#10285)

(cherry picked from commit 5ac2ae1178)

Co-authored-by: Michael Vines <mvines@gmail.com>

ledger-tool/src/main.rs

@@ -1187,6 +1187,7 @@ fn main() {
             let end_slot = value_t_or_exit!(arg_matches, "end_slot", Slot);
             let blockstore = open_blockstore(&ledger_path);
             blockstore.purge_slots(start_slot, end_slot);
+            blockstore.purge_from_next_slots(start_slot, end_slot);
         }
         ("list-roots", Some(arg_matches)) => {
             let blockstore = open_blockstore(&ledger_path);

ledger/src/blockstore.rs

@@ -350,6 +350,33 @@ impl Blockstore {
         self.purge_slots_with_delay(from_slot, to_slot, None)
     }
 
+    /// Ensures that the SlotMeta::next_slots vector for all slots contain no references in the
+    /// [from_slot,to_slot] range
+    ///
+    /// Dangerous; Use with care
+    pub fn purge_from_next_slots(&self, from_slot: Slot, to_slot: Slot) {
+        for (slot, mut meta) in self
+            .slot_meta_iterator(0)
+            .expect("unable to iterate over meta")
+        {
+            if slot > to_slot {
+                break;
+            }
+
+            let original_len = meta.next_slots.len();
+            meta.next_slots
+                .retain(|slot| *slot < from_slot || *slot > to_slot);
+            if meta.next_slots.len() != original_len {
+                info!("purge_from_next_slots: adjusted meta for slot {}", slot);
+                self.put_meta_bytes(
+                    slot,
+                    &bincode::serialize(&meta).expect("couldn't update meta"),
+                )
+                .expect("couldn't update meta");
+            }
+        }
+    }
+
     // Returns whether or not all columns successfully purged the slot range
     fn run_purge(&self, from_slot: Slot, to_slot: Slot) -> Result<bool> {
         let mut write_batch = self