preserve optimistic_slot in blockstore (#25311)

2022-05-24 12:03:28 -07:00 · 2022-05-24 12:03:28 -07:00 · 61c5a471e8
parent 0376ab41a9
commit 61c5a471e8
6 changed files with 210 additions and 118 deletions
--- a/Cargo.lock
+++ b/Cargo.lock
@ -5172,6 +5172,7 @@ dependencies = [
 "assert_cmd",
 "bs58",
 "bytecount",
+ "chrono",
 "clap 2.33.3",
 "crossbeam-channel",
 "csv",
--- a/core/src/cluster_info_vote_listener.rs
+++ b/core/src/cluster_info_vote_listener.rs
@ -483,7 +483,7 @@ impl ClusterInfoVoteListener {
            match confirmed_slots {
                Ok(confirmed_slots) => {
                    confirmation_verifier
-                        .add_new_optimistic_confirmed_slots(confirmed_slots.clone());
+                        .add_new_optimistic_confirmed_slots(confirmed_slots.clone(), &blockstore);
                }
                Err(e) => match e {
                    Error::RecvTimeout(RecvTimeoutError::Disconnected) => {
--- a/core/src/optimistic_confirmation_verifier.rs
+++ b/core/src/optimistic_confirmation_verifier.rs
@ -2,7 +2,7 @@ use {
    crate::cluster_info_vote_listener::VoteTracker,
    solana_ledger::blockstore::Blockstore,
    solana_runtime::bank::Bank,
-    solana_sdk::{clock::Slot, hash::Hash},
+    solana_sdk::{clock::Slot, hash::Hash, timing::timestamp},
    std::{collections::BTreeSet, time::Instant},
 };

@ -52,7 +52,11 @@ impl OptimisticConfirmationVerifier {
            .collect()
    }

-    pub fn add_new_optimistic_confirmed_slots(&mut self, new_optimistic_slots: Vec<(Slot, Hash)>) {
+    pub fn add_new_optimistic_confirmed_slots(
+        &mut self,
+        new_optimistic_slots: Vec<(Slot, Hash)>,
+        blockstore: &Blockstore,
+    ) {
        if new_optimistic_slots.is_empty() {
            return;
        }
@ -70,6 +74,16 @@ impl OptimisticConfirmationVerifier {
        // so ignore those slots
        for (new_optimistic_slot, hash) in new_optimistic_slots {
            if new_optimistic_slot > self.snapshot_start_slot {
+                if let Err(e) = blockstore.insert_optimistic_slot(
+                    new_optimistic_slot,
+                    &hash,
+                    timestamp().try_into().unwrap(),
+                ) {
+                    error!(
+                        "failed to record optimistic slot in blockstore: slot={}: {:?}",
+                        new_optimistic_slot, &e
+                    );
+                }
                datapoint_info!("optimistic_slot", ("slot", new_optimistic_slot, i64),);
                self.unchecked_slots.insert((new_optimistic_slot, hash));
            }
@ -142,9 +156,9 @@ impl OptimisticConfirmationVerifier {
 #[cfg(test)]
 mod test {
    use {
-        super::*, crate::vote_simulator::VoteSimulator, solana_ledger::get_tmp_ledger_path,
-        solana_runtime::bank::Bank, solana_sdk::pubkey::Pubkey, std::collections::HashMap,
-        trees::tr,
+        super::*, crate::vote_simulator::VoteSimulator,
+        solana_ledger::get_tmp_ledger_path_auto_delete, solana_runtime::bank::Bank,
+        solana_sdk::pubkey::Pubkey, std::collections::HashMap, trees::tr,
    };

    #[test]
@ -153,12 +167,23 @@ mod test {
        let bank_hash = Hash::default();
        let mut optimistic_confirmation_verifier =
            OptimisticConfirmationVerifier::new(snapshot_start_slot);
-        optimistic_confirmation_verifier
-            .add_new_optimistic_confirmed_slots(vec![(snapshot_start_slot - 1, bank_hash)]);
-        optimistic_confirmation_verifier
-            .add_new_optimistic_confirmed_slots(vec![(snapshot_start_slot, bank_hash)]);
-        optimistic_confirmation_verifier
-            .add_new_optimistic_confirmed_slots(vec![(snapshot_start_slot + 1, bank_hash)]);
+        let blockstore_path = get_tmp_ledger_path_auto_delete!();
+        let blockstore = Blockstore::open(blockstore_path.path()).unwrap();
+        optimistic_confirmation_verifier.add_new_optimistic_confirmed_slots(
+            vec![(snapshot_start_slot - 1, bank_hash)],
+            &blockstore,
+        );
+        assert_eq!(blockstore.get_latest_optimistic_slots(10).unwrap().len(), 0);
+        optimistic_confirmation_verifier.add_new_optimistic_confirmed_slots(
+            vec![(snapshot_start_slot, bank_hash)],
+            &blockstore,
+        );
+        assert_eq!(blockstore.get_latest_optimistic_slots(10).unwrap().len(), 0);
+        optimistic_confirmation_verifier.add_new_optimistic_confirmed_slots(
+            vec![(snapshot_start_slot + 1, bank_hash)],
+            &blockstore,
+        );
+        assert_eq!(blockstore.get_latest_optimistic_slots(10).unwrap().len(), 1);
        assert_eq!(optimistic_confirmation_verifier.unchecked_slots.len(), 1);
        assert!(optimistic_confirmation_verifier
            .unchecked_slots
@ -171,24 +196,23 @@ mod test {
        let mut optimistic_confirmation_verifier =
            OptimisticConfirmationVerifier::new(snapshot_start_slot);
        let bad_bank_hash = Hash::new(&[42u8; 32]);
-        let blockstore_path = get_tmp_ledger_path!();
-        {
-            let blockstore = Blockstore::open(&blockstore_path).unwrap();
-            let optimistic_slots = vec![(1, bad_bank_hash), (3, Hash::default())];
-            optimistic_confirmation_verifier.add_new_optimistic_confirmed_slots(optimistic_slots);
-            let vote_simulator = setup_forks();
-            let bank1 = vote_simulator.bank_forks.read().unwrap().get(1).unwrap();
-            assert_eq!(
-                optimistic_confirmation_verifier
-                    .verify_for_unrooted_optimistic_slots(&bank1, &blockstore),
-                vec![(1, bad_bank_hash)]
-            );
-            assert_eq!(optimistic_confirmation_verifier.unchecked_slots.len(), 1);
-            assert!(optimistic_confirmation_verifier
-                .unchecked_slots
-                .contains(&(3, Hash::default())));
-        }
-        Blockstore::destroy(&blockstore_path).expect("Expected successful database destruction");
+        let blockstore_path = get_tmp_ledger_path_auto_delete!();
+        let blockstore = Blockstore::open(blockstore_path.path()).unwrap();
+        let optimistic_slots = vec![(1, bad_bank_hash), (3, Hash::default())];
+        optimistic_confirmation_verifier
+            .add_new_optimistic_confirmed_slots(optimistic_slots, &blockstore);
+        assert_eq!(blockstore.get_latest_optimistic_slots(10).unwrap().len(), 2);
+        let vote_simulator = setup_forks();
+        let bank1 = vote_simulator.bank_forks.read().unwrap().get(1).unwrap();
+        assert_eq!(
+            optimistic_confirmation_verifier
+                .verify_for_unrooted_optimistic_slots(&bank1, &blockstore),
+            vec![(1, bad_bank_hash)]
+        );
+        assert_eq!(optimistic_confirmation_verifier.unchecked_slots.len(), 1);
+        assert!(optimistic_confirmation_verifier
+            .unchecked_slots
+            .contains(&(3, Hash::default())));
    }

    #[test]
@ -196,98 +220,98 @@ mod test {
        let snapshot_start_slot = 0;
        let mut optimistic_confirmation_verifier =
            OptimisticConfirmationVerifier::new(snapshot_start_slot);
-        let blockstore_path = get_tmp_ledger_path!();
-        {
-            let blockstore = Blockstore::open(&blockstore_path).unwrap();
-            let mut vote_simulator = setup_forks();
-            let optimistic_slots: Vec<_> = vec![1, 3, 5]
-                .into_iter()
-                .map(|s| {
-                    (
-                        s,
-                        vote_simulator
-                            .bank_forks
-                            .read()
-                            .unwrap()
-                            .get(s)
-                            .unwrap()
-                            .hash(),
-                    )
-                })
-                .collect();
+        let blockstore_path = get_tmp_ledger_path_auto_delete!();
+        let blockstore = Blockstore::open(blockstore_path.path()).unwrap();
+        let mut vote_simulator = setup_forks();
+        let optimistic_slots: Vec<_> = vec![1, 3, 5]
+            .into_iter()
+            .map(|s| {
+                (
+                    s,
+                    vote_simulator
+                        .bank_forks
+                        .read()
+                        .unwrap()
+                        .get(s)
+                        .unwrap()
+                        .hash(),
+                )
+            })
+            .collect();

-            // If root is on same fork, nothing should be returned
+        // If root is on same fork, nothing should be returned
+        optimistic_confirmation_verifier
+            .add_new_optimistic_confirmed_slots(optimistic_slots.clone(), &blockstore);
+        assert_eq!(blockstore.get_latest_optimistic_slots(10).unwrap().len(), 3);
+        let bank5 = vote_simulator.bank_forks.read().unwrap().get(5).unwrap();
+        assert!(optimistic_confirmation_verifier
+            .verify_for_unrooted_optimistic_slots(&bank5, &blockstore)
+            .is_empty());
+        // 5 is >= than all the unchecked slots, so should clear everything
+        assert!(optimistic_confirmation_verifier.unchecked_slots.is_empty());
+
+        // If root is on same fork, nothing should be returned
+        optimistic_confirmation_verifier
+            .add_new_optimistic_confirmed_slots(optimistic_slots.clone(), &blockstore);
+        let bank3 = vote_simulator.bank_forks.read().unwrap().get(3).unwrap();
+        assert!(optimistic_confirmation_verifier
+            .verify_for_unrooted_optimistic_slots(&bank3, &blockstore)
+            .is_empty());
+        // 3 is bigger than only slot 1, so slot 5 should be left over
+        assert_eq!(optimistic_confirmation_verifier.unchecked_slots.len(), 1);
+        assert!(optimistic_confirmation_verifier
+            .unchecked_slots
+            .contains(&optimistic_slots[2]));
+
+        // If root is on different fork, the slots < root on different fork should
+        // be returned
+        optimistic_confirmation_verifier
+            .add_new_optimistic_confirmed_slots(optimistic_slots.clone(), &blockstore);
+        let bank4 = vote_simulator.bank_forks.read().unwrap().get(4).unwrap();
+        assert_eq!(
            optimistic_confirmation_verifier
-                .add_new_optimistic_confirmed_slots(optimistic_slots.clone());
-            let bank5 = vote_simulator.bank_forks.read().unwrap().get(5).unwrap();
-            assert!(optimistic_confirmation_verifier
-                .verify_for_unrooted_optimistic_slots(&bank5, &blockstore)
-                .is_empty());
-            // 5 is >= than all the unchecked slots, so should clear everything
-            assert!(optimistic_confirmation_verifier.unchecked_slots.is_empty());
+                .verify_for_unrooted_optimistic_slots(&bank4, &blockstore),
+            vec![optimistic_slots[1]]
+        );
+        // 4 is bigger than only slots 1 and 3, so slot 5 should be left over
+        assert_eq!(optimistic_confirmation_verifier.unchecked_slots.len(), 1);
+        assert!(optimistic_confirmation_verifier
+            .unchecked_slots
+            .contains(&optimistic_slots[2]));

-            // If root is on same fork, nothing should be returned
+        // Now set a root at slot 5, purging BankForks of slots < 5
+        vote_simulator.set_root(5);
+
+        // Add a new bank 7 that descends from 6
+        let bank6 = vote_simulator.bank_forks.read().unwrap().get(6).unwrap();
+        vote_simulator
+            .bank_forks
+            .write()
+            .unwrap()
+            .insert(Bank::new_from_parent(&bank6, &Pubkey::default(), 7));
+        let bank7 = vote_simulator.bank_forks.read().unwrap().get(7).unwrap();
+        assert!(!bank7.ancestors.contains_key(&3));
+
+        // Should return slots 1, 3 as part of the rooted fork because there's no
+        // ancestry information
+        optimistic_confirmation_verifier
+            .add_new_optimistic_confirmed_slots(optimistic_slots.clone(), &blockstore);
+        assert_eq!(
            optimistic_confirmation_verifier
-                .add_new_optimistic_confirmed_slots(optimistic_slots.clone());
-            let bank3 = vote_simulator.bank_forks.read().unwrap().get(3).unwrap();
-            assert!(optimistic_confirmation_verifier
-                .verify_for_unrooted_optimistic_slots(&bank3, &blockstore)
-                .is_empty());
-            // 3 is bigger than only slot 1, so slot 5 should be left over
-            assert_eq!(optimistic_confirmation_verifier.unchecked_slots.len(), 1);
-            assert!(optimistic_confirmation_verifier
-                .unchecked_slots
-                .contains(&optimistic_slots[2]));
+                .verify_for_unrooted_optimistic_slots(&bank7, &blockstore),
+            optimistic_slots[0..=1].to_vec()
+        );
+        assert!(optimistic_confirmation_verifier.unchecked_slots.is_empty());

-            // If root is on different fork, the slots < root on different fork should
-            // be returned
-            optimistic_confirmation_verifier
-                .add_new_optimistic_confirmed_slots(optimistic_slots.clone());
-            let bank4 = vote_simulator.bank_forks.read().unwrap().get(4).unwrap();
-            assert_eq!(
-                optimistic_confirmation_verifier
-                    .verify_for_unrooted_optimistic_slots(&bank4, &blockstore),
-                vec![optimistic_slots[1]]
-            );
-            // 4 is bigger than only slots 1 and 3, so slot 5 should be left over
-            assert_eq!(optimistic_confirmation_verifier.unchecked_slots.len(), 1);
-            assert!(optimistic_confirmation_verifier
-                .unchecked_slots
-                .contains(&optimistic_slots[2]));
-
-            // Now set a root at slot 5, purging BankForks of slots < 5
-            vote_simulator.set_root(5);
-
-            // Add a new bank 7 that descends from 6
-            let bank6 = vote_simulator.bank_forks.read().unwrap().get(6).unwrap();
-            vote_simulator
-                .bank_forks
-                .write()
-                .unwrap()
-                .insert(Bank::new_from_parent(&bank6, &Pubkey::default(), 7));
-            let bank7 = vote_simulator.bank_forks.read().unwrap().get(7).unwrap();
-            assert!(!bank7.ancestors.contains_key(&3));
-
-            // Should return slots 1, 3 as part of the rooted fork because there's no
-            // ancestry information
-            optimistic_confirmation_verifier
-                .add_new_optimistic_confirmed_slots(optimistic_slots.clone());
-            assert_eq!(
-                optimistic_confirmation_verifier
-                    .verify_for_unrooted_optimistic_slots(&bank7, &blockstore),
-                optimistic_slots[0..=1].to_vec()
-            );
-            assert!(optimistic_confirmation_verifier.unchecked_slots.is_empty());
-
-            // If we know set the root in blockstore, should return nothing
-            blockstore.set_roots(vec![1, 3].iter()).unwrap();
-            optimistic_confirmation_verifier.add_new_optimistic_confirmed_slots(optimistic_slots);
-            assert!(optimistic_confirmation_verifier
-                .verify_for_unrooted_optimistic_slots(&bank7, &blockstore)
-                .is_empty());
-            assert!(optimistic_confirmation_verifier.unchecked_slots.is_empty());
-        }
-        Blockstore::destroy(&blockstore_path).expect("Expected successful database destruction");
+        // If we know set the root in blockstore, should return nothing
+        blockstore.set_roots(vec![1, 3].iter()).unwrap();
+        optimistic_confirmation_verifier
+            .add_new_optimistic_confirmed_slots(optimistic_slots, &blockstore);
+        assert!(optimistic_confirmation_verifier
+            .verify_for_unrooted_optimistic_slots(&bank7, &blockstore)
+            .is_empty());
+        assert!(optimistic_confirmation_verifier.unchecked_slots.is_empty());
+        assert_eq!(blockstore.get_latest_optimistic_slots(10).unwrap().len(), 3);
    }

    fn setup_forks() -> VoteSimulator {
--- a/ledger-tool/Cargo.toml
+++ b/ledger-tool/Cargo.toml
@ -12,6 +12,7 @@ documentation = "https://docs.rs/solana-ledger-tool"
 [dependencies]
 bs58 = "0.4.0"
 clap = "2.33.1"
+chrono = "0.4.11"
 crossbeam-channel = "0.5"
 csv = "1.1.6"
 dashmap = "4.0.2"
--- a/ledger-tool/src/main.rs
+++ b/ledger-tool/src/main.rs
@ -1,6 +1,7 @@
 #![allow(clippy::integer_arithmetic)]
 use {
    crate::{bigtable::*, ledger_path::*},
+    chrono::{DateTime, Utc},
    clap::{
        crate_description, crate_name, value_t, value_t_or_exit, values_t_or_exit, App,
        AppSettings, Arg, ArgMatches, SubCommand,
@ -83,6 +84,7 @@ use {
            atomic::{AtomicBool, Ordering},
            Arc, RwLock,
        },
+        time::{Duration, UNIX_EPOCH},
    },
 };

@ -714,6 +716,7 @@ fn analyze_storage(database: &Database) {
    analyze_column::<PerfSamples>(database, "PerfSamples");
    analyze_column::<BlockHeight>(database, "BlockHeight");
    analyze_column::<ProgramCosts>(database, "ProgramCosts");
+    analyze_column::<OptimisticSlots>(database, "OptimisticSlots");
 }

 fn open_blockstore(
@ -950,6 +953,7 @@ fn main() {
    }

    const DEFAULT_ROOT_COUNT: &str = "1";
+    const DEFAULT_LATEST_OPTIMISTIC_SLOTS_COUNT: &str = "1";
    const DEFAULT_MAX_SLOTS_ROOT_REPAIR: &str = "2000";
    solana_logger::setup_with_default("solana=info");

@ -1745,6 +1749,20 @@ fn main() {
                    .help("Number of roots in the output"),
            )
        )
+        .subcommand(
+            SubCommand::with_name("latest-optimistic-slots")
+                .about("Output up to the most recent <num-slots> optimistic \
+                        slots with their hashes and timestamps.")
+                .arg(
+                    Arg::with_name("num_slots")
+                        .long("num-slots")
+                        .value_name("NUM")
+                        .takes_value(true)
+                        .default_value(DEFAULT_LATEST_OPTIMISTIC_SLOTS_COUNT)
+                        .required(false)
+                        .help("Number of slots in the output"),
+                )
+        )
        .subcommand(
            SubCommand::with_name("repair-roots")
                .about("Traverses the AncestorIterator backward from a last known root \
@ -3439,6 +3457,26 @@ fn main() {
                        }
                    });
            }
+            ("latest-optimistic-slots", Some(arg_matches)) => {
+                let blockstore =
+                    open_blockstore(&ledger_path, AccessType::Secondary, wal_recovery_mode);
+                let num_slots = value_t_or_exit!(arg_matches, "num_slots", usize);
+                let slots = blockstore
+                    .get_latest_optimistic_slots(num_slots)
+                    .expect("Failed to get latest optimistic slots");
+                println!("{:>20} {:>44} {:>32}", "Slot", "Hash", "Timestamp");
+                for (slot, hash, timestamp) in slots.iter() {
+                    let time_str = {
+                        let secs: u64 = (timestamp / 1_000) as u64;
+                        let nanos: u32 = ((timestamp % 1_000) * 1_000_000) as u32;
+                        let t = UNIX_EPOCH + Duration::new(secs, nanos);
+                        let datetime: DateTime<Utc> = t.into();
+                        datetime.to_rfc3339()
+                    };
+                    let hash_str = format!("{}", hash);
+                    println!("{:>20} {:>44} {:>32}", slot, &hash_str, &time_str);
+                }
+            }
            ("repair-roots", Some(arg_matches)) => {
                let blockstore =
                    open_blockstore(&ledger_path, AccessType::Primary, wal_recovery_mode);
--- a/ledger/src/blockstore.rs
+++ b/ledger/src/blockstore.rs
@ -67,7 +67,10 @@ use {
 };
 pub mod blockstore_purge;
 pub use {
-    crate::{blockstore_db::BlockstoreError, blockstore_meta::SlotMeta},
+    crate::{
+        blockstore_db::BlockstoreError,
+        blockstore_meta::{OptimisticSlotMetaVersioned, SlotMeta},
+    },
    blockstore_purge::PurgeType,
    rocksdb::properties as RocksProperties,
 };
@ -3018,6 +3021,31 @@ impl Blockstore {
            .unwrap_or(false)
    }

+    pub fn insert_optimistic_slot(
+        &self,
+        slot: Slot,
+        hash: &Hash,
+        timestamp: UnixTimestamp,
+    ) -> Result<()> {
+        let slot_data = OptimisticSlotMetaVersioned::new(*hash, timestamp);
+        self.optimistic_slots_cf.put(slot, &slot_data)
+    }
+
+    pub fn get_latest_optimistic_slots(
+        &self,
+        num: usize,
+    ) -> Result<Vec<(Slot, Hash, UnixTimestamp)>> {
+        Ok(self
+            .db
+            .iter::<cf::OptimisticSlots>(IteratorMode::End)?
+            .take(num)
+            .map(|(slot, data)| {
+                let meta: OptimisticSlotMetaVersioned = deserialize(&data).unwrap();
+                (slot, meta.hash(), meta.timestamp())
+            })
+            .collect())
+    }
+
    pub fn set_duplicate_confirmed_slots_and_hashes(
        &self,
        duplicate_confirmed_slot_hashes: impl Iterator<Item = (Slot, Hash)>,