Add largest_confirmed_root to BlockCommitmentCache (#9640)

* Add largest_confirmed_root to BlockCommitmentCache

* clippy

* Add blockstore to BlockCommitmentCache to check root

* Add rooted_stake helper fn and test

* Nodes that are behind should correctly id confirmed roots

* Simplify rooted_stake collector

core/src/commitment.rs

@@ -1,4 +1,5 @@
 use crate::consensus::VOTE_THRESHOLD_SIZE;
+use solana_ledger::blockstore::Blockstore;
 use solana_measure::measure::Measure;
 use solana_metrics::inc_new_counter_info;
 use solana_runtime::bank::Bank;
@@ -45,11 +46,12 @@ impl BlockCommitment {
     }
 }
 
-#[derive(Default)]
 pub struct BlockCommitmentCache {
     block_commitment: HashMap<Slot, BlockCommitment>,
+    largest_confirmed_root: Slot,
     total_stake: u64,
     bank: Arc<Bank>,
+    blockstore: Arc<Blockstore>,
     root: Slot,
 }
 
@@ -70,22 +72,41 @@ impl std::fmt::Debug for BlockCommitmentCache {
 impl BlockCommitmentCache {
     pub fn new(
         block_commitment: HashMap<Slot, BlockCommitment>,
+        largest_confirmed_root: Slot,
         total_stake: u64,
         bank: Arc<Bank>,
+        blockstore: Arc<Blockstore>,
         root: Slot,
     ) -> Self {
         Self {
             block_commitment,
+            largest_confirmed_root,
             total_stake,
             bank,
+            blockstore,
             root,
         }
     }
 
+    pub fn default_with_blockstore(blockstore: Arc<Blockstore>) -> Self {
+        Self {
+            block_commitment: HashMap::default(),
+            largest_confirmed_root: Slot::default(),
+            total_stake: u64::default(),
+            bank: Arc::new(Bank::default()),
+            blockstore,
+            root: Slot::default(),
+        }
+    }
+
     pub fn get_block_commitment(&self, slot: Slot) -> Option<&BlockCommitment> {
         self.block_commitment.get(&slot)
     }
 
+    pub fn largest_confirmed_root(&self) -> Slot {
+        self.largest_confirmed_root
+    }
+
     pub fn total_stake(&self) -> u64 {
         self.total_stake
     }
@@ -123,22 +144,21 @@ impl BlockCommitmentCache {
     }
 
     pub fn is_confirmed_rooted(&self, slot: Slot) -> bool {
-        self.get_block_commitment(slot)
+        slot <= self.largest_confirmed_root()
-            .map(|block_commitment| {
+            && (self.blockstore.is_root(slot) || self.bank.status_cache_ancestors().contains(&slot))
-                (block_commitment.get_rooted_stake() as f64 / self.total_stake as f64)
-                    > VOTE_THRESHOLD_SIZE
-            })
-            .unwrap_or(false)
     }
 
     #[cfg(test)]
-    pub fn new_for_tests() -> Self {
+    pub fn new_for_tests_with_blockstore(blockstore: Arc<Blockstore>) -> Self {
         let mut block_commitment: HashMap<Slot, BlockCommitment> = HashMap::new();
         block_commitment.insert(0, BlockCommitment::default());
         Self {
             block_commitment,
+            blockstore,
             total_stake: 42,
-            ..Self::default()
+            largest_confirmed_root: Slot::default(),
+            bank: Arc::new(Bank::default()),
+            root: Slot::default(),
         }
     }
 }
@@ -159,6 +179,18 @@ impl CommitmentAggregationData {
     }
 }
 
+fn get_largest_confirmed_root(mut rooted_stake: Vec<(Slot, u64)>, total_stake: u64) -> Slot {
+    rooted_stake.sort_by(|a, b| a.0.cmp(&b.0).reverse());
+    let mut stake_sum = 0;
+    for (root, stake) in rooted_stake {
+        stake_sum += stake;
+        if (stake_sum as f64 / total_stake as f64) > VOTE_THRESHOLD_SIZE {
+            return root;
+        }
+    }
+    0
+}
+
 pub struct AggregateCommitmentService {
     t_commitment: JoinHandle<()>,
 }
@@ -216,12 +248,18 @@ impl AggregateCommitmentService {
             }
 
             let mut aggregate_commitment_time = Measure::start("aggregate-commitment-ms");
-            let block_commitment = Self::aggregate_commitment(&ancestors, &aggregation_data.bank);
+            let (block_commitment, rooted_stake) =
+                Self::aggregate_commitment(&ancestors, &aggregation_data.bank);
+
+            let largest_confirmed_root =
+                get_largest_confirmed_root(rooted_stake, aggregation_data.total_staked);
 
             let mut new_block_commitment = BlockCommitmentCache::new(
                 block_commitment,
+                largest_confirmed_root,
                 aggregation_data.total_staked,
                 aggregation_data.bank,
+                block_commitment_cache.read().unwrap().blockstore.clone(),
                 aggregation_data.root,
             );
 
@@ -236,7 +274,10 @@ impl AggregateCommitmentService {
         }
     }
 
-    pub fn aggregate_commitment(ancestors: &[Slot], bank: &Bank) -> HashMap<Slot, BlockCommitment> {
+    pub fn aggregate_commitment(
+        ancestors: &[Slot],
+        bank: &Bank,
+    ) -> (HashMap<Slot, BlockCommitment>, Vec<(Slot, u64)>) {
         assert!(!ancestors.is_empty());
 
         // Check ancestors is sorted
@@ -245,6 +286,7 @@ impl AggregateCommitmentService {
         }
 
         let mut commitment = HashMap::new();
+        let mut rooted_stake: Vec<(Slot, u64)> = Vec::new();
         for (_, (lamports, account)) in bank.vote_accounts().into_iter() {
             if lamports == 0 {
                 continue;
@@ -257,17 +299,19 @@ impl AggregateCommitmentService {
             let vote_state = vote_state.unwrap();
             Self::aggregate_commitment_for_vote_account(
                 &mut commitment,
+                &mut rooted_stake,
                 &vote_state,
                 ancestors,
                 lamports,
             );
         }
 
-        commitment
+        (commitment, rooted_stake)
     }
 
     fn aggregate_commitment_for_vote_account(
         commitment: &mut HashMap<Slot, BlockCommitment>,
+        rooted_stake: &mut Vec<(Slot, u64)>,
         vote_state: &VoteState,
         ancestors: &[Slot],
         lamports: u64,
@@ -286,6 +330,7 @@ impl AggregateCommitmentService {
                     break;
                 }
             }
+            rooted_stake.push((root, lamports));
         }
 
         for vote in &vote_state.votes {
@@ -311,7 +356,10 @@ impl AggregateCommitmentService {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use solana_ledger::genesis_utils::{create_genesis_config, GenesisConfigInfo};
+    use solana_ledger::{
+        genesis_utils::{create_genesis_config, GenesisConfigInfo},
+        get_tmp_ledger_path,
+    };
     use solana_sdk::pubkey::Pubkey;
     use solana_stake_program::stake_state;
     use solana_vote_program::vote_state::{self, VoteStateVersions};
@@ -329,6 +377,8 @@ mod tests {
     #[test]
     fn test_get_confirmations() {
         let bank = Arc::new(Bank::default());
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
         // Build BlockCommitmentCache with votes at depths 0 and 1 for 2 slots
         let mut cache0 = BlockCommitment::default();
         cache0.increase_confirmation_stake(1, 5);
@@ -346,7 +396,8 @@ mod tests {
         block_commitment.entry(0).or_insert(cache0.clone());
         block_commitment.entry(1).or_insert(cache1.clone());
         block_commitment.entry(2).or_insert(cache2.clone());
-        let block_commitment_cache = BlockCommitmentCache::new(block_commitment, 50, bank, 0);
+        let block_commitment_cache =
+            BlockCommitmentCache::new(block_commitment, 0, 50, bank, blockstore, 0);
 
         assert_eq!(block_commitment_cache.get_confirmation_count(0), Some(2));
         assert_eq!(block_commitment_cache.get_confirmation_count(1), Some(1));
@@ -354,17 +405,68 @@ mod tests {
         assert_eq!(block_commitment_cache.get_confirmation_count(3), None,);
     }
 
+    #[test]
+    fn test_is_confirmed_rooted() {
+        let bank = Arc::new(Bank::default());
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
+        blockstore.set_roots(&[0, 1]).unwrap();
+        // Build BlockCommitmentCache with rooted slots
+        let mut cache0 = BlockCommitment::default();
+        cache0.increase_rooted_stake(50);
+        let mut cache1 = BlockCommitment::default();
+        cache1.increase_rooted_stake(40);
+        let mut cache2 = BlockCommitment::default();
+        cache2.increase_rooted_stake(20);
+
+        let mut block_commitment = HashMap::new();
+        block_commitment.entry(1).or_insert(cache0.clone());
+        block_commitment.entry(2).or_insert(cache1.clone());
+        block_commitment.entry(3).or_insert(cache2.clone());
+        let largest_confirmed_root = 1;
+        let block_commitment_cache = BlockCommitmentCache::new(
+            block_commitment,
+            largest_confirmed_root,
+            50,
+            bank,
+            blockstore,
+            0,
+        );
+
+        assert!(block_commitment_cache.is_confirmed_rooted(0));
+        assert!(block_commitment_cache.is_confirmed_rooted(1));
+        assert!(!block_commitment_cache.is_confirmed_rooted(2));
+        assert!(!block_commitment_cache.is_confirmed_rooted(3));
+    }
+
+    #[test]
+    fn test_get_largest_confirmed_root() {
+        assert_eq!(get_largest_confirmed_root(vec![], 10), 0);
+        let mut rooted_stake = vec![];
+        rooted_stake.push((0, 5));
+        rooted_stake.push((1, 5));
+        assert_eq!(get_largest_confirmed_root(rooted_stake, 10), 0);
+        let mut rooted_stake = vec![];
+        rooted_stake.push((1, 5));
+        rooted_stake.push((0, 10));
+        rooted_stake.push((2, 5));
+        rooted_stake.push((1, 4));
+        assert_eq!(get_largest_confirmed_root(rooted_stake, 10), 1);
+    }
+
     #[test]
     fn test_aggregate_commitment_for_vote_account_1() {
         let ancestors = vec![3, 4, 5, 7, 9, 11];
         let mut commitment = HashMap::new();
+        let mut rooted_stake = vec![];
         let lamports = 5;
         let mut vote_state = VoteState::default();
 
-        let root = ancestors.last().unwrap();
+        let root = ancestors.last().unwrap().clone();
-        vote_state.root_slot = Some(*root);
+        vote_state.root_slot = Some(root);
         AggregateCommitmentService::aggregate_commitment_for_vote_account(
             &mut commitment,
+            &mut rooted_stake,
             &vote_state,
             &ancestors,
             lamports,
@@ -375,12 +477,14 @@ mod tests {
             expected.increase_rooted_stake(lamports);
             assert_eq!(*commitment.get(&a).unwrap(), expected);
         }
+        assert_eq!(rooted_stake[0], (root, lamports));
     }
 
     #[test]
     fn test_aggregate_commitment_for_vote_account_2() {
         let ancestors = vec![3, 4, 5, 7, 9, 11];
         let mut commitment = HashMap::new();
+        let mut rooted_stake = vec![];
         let lamports = 5;
         let mut vote_state = VoteState::default();
 
@@ -389,6 +493,7 @@ mod tests {
         vote_state.process_slot_vote_unchecked(*ancestors.last().unwrap());
         AggregateCommitmentService::aggregate_commitment_for_vote_account(
             &mut commitment,
+            &mut rooted_stake,
             &vote_state,
             &ancestors,
             lamports,
@@ -405,12 +510,14 @@ mod tests {
                 assert_eq!(*commitment.get(&a).unwrap(), expected);
             }
         }
+        assert_eq!(rooted_stake[0], (root, lamports));
     }
 
     #[test]
     fn test_aggregate_commitment_for_vote_account_3() {
         let ancestors = vec![3, 4, 5, 7, 9, 10, 11];
         let mut commitment = HashMap::new();
+        let mut rooted_stake = vec![];
         let lamports = 5;
         let mut vote_state = VoteState::default();
 
@@ -421,6 +528,7 @@ mod tests {
         vote_state.process_slot_vote_unchecked(ancestors[6]);
         AggregateCommitmentService::aggregate_commitment_for_vote_account(
             &mut commitment,
+            &mut rooted_stake,
             &vote_state,
             &ancestors,
             lamports,
@@ -441,6 +549,7 @@ mod tests {
                 assert_eq!(*commitment.get(&a).unwrap(), expected);
             }
         }
+        assert_eq!(rooted_stake[0], (root, lamports));
     }
 
     #[test]
@@ -450,6 +559,8 @@ mod tests {
             mut genesis_config, ..
         } = create_genesis_config(10_000);
 
+        let rooted_stake_amount = 40;
+
         let sk1 = Pubkey::new_rand();
         let pk1 = Pubkey::new_rand();
         let mut vote_account1 = vote_state::create_account(&pk1, &Pubkey::new_rand(), 0, 100);
@@ -460,12 +571,36 @@ mod tests {
         let mut vote_account2 = vote_state::create_account(&pk2, &Pubkey::new_rand(), 0, 50);
         let stake_account2 =
             stake_state::create_account(&sk2, &pk2, &vote_account2, &genesis_config.rent, 50);
+        let sk3 = Pubkey::new_rand();
+        let pk3 = Pubkey::new_rand();
+        let mut vote_account3 = vote_state::create_account(&pk3, &Pubkey::new_rand(), 0, 1);
+        let stake_account3 = stake_state::create_account(
+            &sk3,
+            &pk3,
+            &vote_account3,
+            &genesis_config.rent,
+            rooted_stake_amount,
+        );
+        let sk4 = Pubkey::new_rand();
+        let pk4 = Pubkey::new_rand();
+        let mut vote_account4 = vote_state::create_account(&pk4, &Pubkey::new_rand(), 0, 1);
+        let stake_account4 = stake_state::create_account(
+            &sk4,
+            &pk4,
+            &vote_account4,
+            &genesis_config.rent,
+            rooted_stake_amount,
+        );
 
         genesis_config.accounts.extend(vec![
             (pk1, vote_account1.clone()),
             (sk1, stake_account1),
             (pk2, vote_account2.clone()),
             (sk2, stake_account2),
+            (pk3, vote_account3.clone()),
+            (sk3, stake_account3),
+            (pk4, vote_account4.clone()),
+            (sk4, stake_account4),
         ]);
 
         // Create bank
@@ -485,7 +620,20 @@ mod tests {
         VoteState::to(&versioned, &mut vote_account2).unwrap();
         bank.store_account(&pk2, &vote_account2);
 
-        let commitment = AggregateCommitmentService::aggregate_commitment(&ancestors, &bank);
+        let mut vote_state3 = VoteState::from(&vote_account3).unwrap();
+        vote_state3.root_slot = Some(1);
+        let versioned = VoteStateVersions::Current(Box::new(vote_state3));
+        VoteState::to(&versioned, &mut vote_account3).unwrap();
+        bank.store_account(&pk3, &vote_account3);
+
+        let mut vote_state4 = VoteState::from(&vote_account4).unwrap();
+        vote_state4.root_slot = Some(2);
+        let versioned = VoteStateVersions::Current(Box::new(vote_state4));
+        VoteState::to(&versioned, &mut vote_account4).unwrap();
+        bank.store_account(&pk4, &vote_account4);
+
+        let (commitment, rooted_stake) =
+            AggregateCommitmentService::aggregate_commitment(&ancestors, &bank);
 
         for a in ancestors {
             if a <= 3 {
@@ -509,5 +657,7 @@ mod tests {
                 assert!(commitment.get(&a).is_none());
             }
         }
+        assert_eq!(rooted_stake.len(), 2);
+        assert_eq!(get_largest_confirmed_root(rooted_stake, 100), 1)
     }
 }

core/src/replay_stage.rs

@@ -88,7 +88,6 @@ struct SkippedSlotsInfo {
     last_skipped_slot: u64,
 }
 
-#[derive(Default)]
 pub struct ReplayStageConfig {
     pub my_pubkey: Pubkey,
     pub vote_account: Pubkey,
@@ -1980,7 +1979,9 @@ pub(crate) mod tests {
             let exit = Arc::new(AtomicBool::new(false));
             let subscriptions = Arc::new(RpcSubscriptions::new(
                 &exit,
-                Arc::new(RwLock::new(BlockCommitmentCache::default())),
+                Arc::new(RwLock::new(BlockCommitmentCache::default_with_blockstore(
+                    blockstore.clone(),
+                ))),
             ));
             let mut bank_forks = BankForks::new(0, bank0);
 
@@ -2368,7 +2369,11 @@ pub(crate) mod tests {
             bank.store_account(&pubkey, &leader_vote_account);
         }
 
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore.clone()),
+        ));
         let (lockouts_sender, _) = AggregateCommitmentService::new(
             &Arc::new(AtomicBool::new(false)),
             block_commitment_cache.clone(),

core/src/rpc.rs

@@ -1477,8 +1477,10 @@ pub mod tests {
             .or_insert(commitment_slot1.clone());
         let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::new(
             block_commitment,
+            0,
             10,
             bank.clone(),
+            blockstore.clone(),
             0,
         )));
 
@@ -1591,14 +1593,16 @@ pub mod tests {
         let validator_exit = create_validator_exit(&exit);
         let (bank_forks, alice, _) = new_bank_forks();
         let bank = bank_forks.read().unwrap().working_bank();
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
         let ledger_path = get_tmp_ledger_path!();
-        let blockstore = Blockstore::open(&ledger_path).unwrap();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore.clone()),
+        ));
         let request_processor = JsonRpcRequestProcessor::new(
             JsonRpcConfig::default(),
             bank_forks,
             block_commitment_cache,
-            Arc::new(blockstore),
+            blockstore,
             StorageState::default(),
             validator_exit,
         );
@@ -2070,7 +2074,7 @@ pub mod tests {
                 .expect("actual response deserialization");
         let result = result.as_ref().unwrap();
         assert_eq!(expected_res, result.status);
-        assert_eq!(Some(2), result.confirmations);
+        assert_eq!(None, result.confirmations);
 
         // Test getSignatureStatus request on unprocessed tx
         let tx = system_transaction::transfer(&alice, &bob_pubkey, 10, blockhash);
@@ -2234,9 +2238,11 @@ pub mod tests {
     fn test_rpc_send_bad_tx() {
         let exit = Arc::new(AtomicBool::new(false));
         let validator_exit = create_validator_exit(&exit);
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
         let ledger_path = get_tmp_ledger_path!();
-        let blockstore = Blockstore::open(&ledger_path).unwrap();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore.clone()),
+        ));
 
         let mut io = MetaIoHandler::default();
         let rpc = RpcSolImpl;
@@ -2247,7 +2253,7 @@ pub mod tests {
                     JsonRpcConfig::default(),
                     new_bank_forks().0,
                     block_commitment_cache,
-                    Arc::new(blockstore),
+                    blockstore,
                     StorageState::default(),
                     validator_exit,
                 );
@@ -2335,14 +2341,16 @@ pub mod tests {
     fn test_rpc_request_processor_config_default_trait_validator_exit_fails() {
         let exit = Arc::new(AtomicBool::new(false));
         let validator_exit = create_validator_exit(&exit);
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
         let ledger_path = get_tmp_ledger_path!();
-        let blockstore = Blockstore::open(&ledger_path).unwrap();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore.clone()),
+        ));
         let request_processor = JsonRpcRequestProcessor::new(
             JsonRpcConfig::default(),
             new_bank_forks().0,
             block_commitment_cache,
-            Arc::new(blockstore),
+            blockstore,
             StorageState::default(),
             validator_exit,
         );
@@ -2354,16 +2362,18 @@ pub mod tests {
     fn test_rpc_request_processor_allow_validator_exit_config() {
         let exit = Arc::new(AtomicBool::new(false));
         let validator_exit = create_validator_exit(&exit);
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
         let ledger_path = get_tmp_ledger_path!();
-        let blockstore = Blockstore::open(&ledger_path).unwrap();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore.clone()),
+        ));
         let mut config = JsonRpcConfig::default();
         config.enable_validator_exit = true;
         let request_processor = JsonRpcRequestProcessor::new(
             config,
             new_bank_forks().0,
             block_commitment_cache,
-            Arc::new(blockstore),
+            blockstore,
             StorageState::default(),
             validator_exit,
         );
@@ -2418,6 +2428,8 @@ pub mod tests {
         let exit = Arc::new(AtomicBool::new(false));
         let validator_exit = create_validator_exit(&exit);
         let bank_forks = new_bank_forks().0;
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
 
         let commitment_slot0 = BlockCommitment::new([8; MAX_LOCKOUT_HISTORY + 1]);
         let commitment_slot1 = BlockCommitment::new([9; MAX_LOCKOUT_HISTORY + 1]);
@@ -2430,12 +2442,12 @@ pub mod tests {
             .or_insert(commitment_slot1.clone());
         let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::new(
             block_commitment,
+            0,
             42,
             bank_forks.read().unwrap().working_bank(),
+            blockstore.clone(),
             0,
         )));
-        let ledger_path = get_tmp_ledger_path!();
-        let blockstore = Blockstore::open(&ledger_path).unwrap();
 
         let mut config = JsonRpcConfig::default();
         config.enable_validator_exit = true;
@@ -2443,7 +2455,7 @@ pub mod tests {
             config,
             bank_forks,
             block_commitment_cache,
-            Arc::new(blockstore),
+            blockstore,
             StorageState::default(),
             validator_exit,
         );

core/src/rpc_pubsub.rs

@@ -7,8 +7,13 @@ use jsonrpc_pubsub::{typed::Subscriber, Session, SubscriptionId};
 use solana_client::rpc_response::{
     Response as RpcResponse, RpcAccount, RpcKeyedAccount, RpcSignatureResult,
 };
+#[cfg(test)]
+use solana_ledger::blockstore::Blockstore;
 use solana_sdk::{clock::Slot, pubkey::Pubkey, signature::Signature};
-use std::sync::{atomic, Arc};
+use std::{
+    str::FromStr,
+    sync::{atomic, Arc},
+};
 
 // Suppress needless_return due to
 //   https://github.com/paritytech/jsonrpc/blob/2d38e6424d8461cdf72e78425ce67d51af9c6586/derive/src/lib.rs#L204
@@ -118,7 +123,6 @@ pub trait RpcSolPubSub {
     fn root_unsubscribe(&self, meta: Option<Self::Metadata>, id: SubscriptionId) -> Result<bool>;
 }
 
-#[derive(Default)]
 pub struct RpcSolPubSubImpl {
     uid: Arc<atomic::AtomicUsize>,
     subscriptions: Arc<RpcSubscriptions>,
@@ -129,9 +133,14 @@ impl RpcSolPubSubImpl {
         let uid = Arc::new(atomic::AtomicUsize::default());
         Self { uid, subscriptions }
     }
-}
 
-use std::str::FromStr;
+    #[cfg(test)]
+    fn default_with_blockstore(blockstore: Arc<Blockstore>) -> Self {
+        let uid = Arc::new(atomic::AtomicUsize::default());
+        let subscriptions = Arc::new(RpcSubscriptions::default_with_blockstore(blockstore));
+        Self { uid, subscriptions }
+    }
+}
 
 fn param<T: FromStr>(param_str: &str, thing: &str) -> Result<T> {
     param_str.parse::<T>().map_err(|_e| Error {
@@ -322,8 +331,11 @@ mod tests {
     use jsonrpc_pubsub::{PubSubHandler, Session};
     use serial_test_derive::serial;
     use solana_budget_program::{self, budget_instruction};
-    use solana_ledger::bank_forks::BankForks;
+    use solana_ledger::{
-    use solana_ledger::genesis_utils::{create_genesis_config, GenesisConfigInfo};
+        bank_forks::BankForks,
+        genesis_utils::{create_genesis_config, GenesisConfigInfo},
+        get_tmp_ledger_path,
+    };
     use solana_runtime::bank::Bank;
     use solana_sdk::{
         pubkey::Pubkey,
@@ -370,12 +382,16 @@ mod tests {
         let bank = Bank::new(&genesis_config);
         let blockhash = bank.last_blockhash();
         let bank_forks = Arc::new(RwLock::new(BankForks::new(0, bank)));
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
         let rpc = RpcSolPubSubImpl {
             subscriptions: Arc::new(RpcSubscriptions::new(
                 &Arc::new(AtomicBool::new(false)),
-                Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
+                Arc::new(RwLock::new(
+                    BlockCommitmentCache::new_for_tests_with_blockstore(blockstore),
                 )),
-            ..RpcSolPubSubImpl::default()
+            )),
+            uid: Arc::new(atomic::AtomicUsize::default()),
         };
 
         // Test signature subscriptions
@@ -416,11 +432,13 @@ mod tests {
         let bank = Bank::new(&genesis_config);
         let arc_bank = Arc::new(bank);
         let blockhash = arc_bank.last_blockhash();
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
 
         let session = create_session();
 
         let mut io = PubSubHandler::default();
-        let rpc = RpcSolPubSubImpl::default();
+        let rpc = RpcSolPubSubImpl::default_with_blockstore(blockstore);
         io.extend_with(rpc.to_delegate());
 
         let tx = system_transaction::transfer(&alice, &bob_pubkey, 20, blockhash);
@@ -475,13 +493,17 @@ mod tests {
         let bank = Bank::new(&genesis_config);
         let blockhash = bank.last_blockhash();
         let bank_forks = Arc::new(RwLock::new(BankForks::new(0, bank)));
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
 
         let rpc = RpcSolPubSubImpl {
             subscriptions: Arc::new(RpcSubscriptions::new(
                 &Arc::new(AtomicBool::new(false)),
-                Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
+                Arc::new(RwLock::new(
+                    BlockCommitmentCache::new_for_tests_with_blockstore(blockstore),
                 )),
-            ..RpcSolPubSubImpl::default()
+            )),
+            uid: Arc::new(atomic::AtomicUsize::default()),
         };
         let session = create_session();
         let (subscriber, _id_receiver, receiver) = Subscriber::new_test("accountNotification");
@@ -569,9 +591,11 @@ mod tests {
     fn test_account_unsubscribe() {
         let bob_pubkey = Pubkey::new_rand();
         let session = create_session();
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
 
         let mut io = PubSubHandler::default();
-        let rpc = RpcSolPubSubImpl::default();
+        let rpc = RpcSolPubSubImpl::default_with_blockstore(blockstore);
 
         io.extend_with(rpc.to_delegate());
 
@@ -615,13 +639,17 @@ mod tests {
         let bank = Bank::new(&genesis_config);
         let blockhash = bank.last_blockhash();
         let bank_forks = Arc::new(RwLock::new(BankForks::new(0, bank)));
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
         let bob = Keypair::new();
 
-        let mut rpc = RpcSolPubSubImpl::default();
+        let mut rpc = RpcSolPubSubImpl::default_with_blockstore(blockstore.clone());
         let exit = Arc::new(AtomicBool::new(false));
         let subscriptions = RpcSubscriptions::new(
             &exit,
-            Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
+            Arc::new(RwLock::new(
+                BlockCommitmentCache::new_for_tests_with_blockstore(blockstore),
+            )),
         );
         rpc.subscriptions = Arc::new(subscriptions);
         let session = create_session();
@@ -652,11 +680,15 @@ mod tests {
         let bank = Bank::new(&genesis_config);
         let blockhash = bank.last_blockhash();
         let bank_forks = Arc::new(RwLock::new(BankForks::new(0, bank)));
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
         let bob = Keypair::new();
 
-        let mut rpc = RpcSolPubSubImpl::default();
+        let mut rpc = RpcSolPubSubImpl::default_with_blockstore(blockstore.clone());
         let exit = Arc::new(AtomicBool::new(false));
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests()));
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::new_for_tests_with_blockstore(blockstore.clone()),
+        ));
 
         let subscriptions = RpcSubscriptions::new(&exit, block_commitment_cache.clone());
         rpc.subscriptions = Arc::new(subscriptions);
@@ -683,8 +715,14 @@ mod tests {
         cache0.increase_confirmation_stake(1, 10);
         let mut block_commitment = HashMap::new();
         block_commitment.entry(0).or_insert(cache0.clone());
-        let mut new_block_commitment =
+        let mut new_block_commitment = BlockCommitmentCache::new(
-            BlockCommitmentCache::new(block_commitment, 10, bank1.clone(), 0);
+            block_commitment,
+            0,
+            10,
+            bank1.clone(),
+            blockstore.clone(),
+            0,
+        );
         let mut w_block_commitment_cache = block_commitment_cache.write().unwrap();
         std::mem::swap(&mut *w_block_commitment_cache, &mut new_block_commitment);
         drop(w_block_commitment_cache);
@@ -698,7 +736,8 @@ mod tests {
         cache0.increase_confirmation_stake(2, 10);
         let mut block_commitment = HashMap::new();
         block_commitment.entry(0).or_insert(cache0.clone());
-        let mut new_block_commitment = BlockCommitmentCache::new(block_commitment, 10, bank2, 0);
+        let mut new_block_commitment =
+            BlockCommitmentCache::new(block_commitment, 0, 10, bank2, blockstore.clone(), 0);
         let mut w_block_commitment_cache = block_commitment_cache.write().unwrap();
         std::mem::swap(&mut *w_block_commitment_cache, &mut new_block_commitment);
         drop(w_block_commitment_cache);
@@ -728,7 +767,9 @@ mod tests {
     #[test]
     #[serial]
     fn test_slot_subscribe() {
-        let rpc = RpcSolPubSubImpl::default();
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
+        let rpc = RpcSolPubSubImpl::default_with_blockstore(blockstore);
         let session = create_session();
         let (subscriber, _id_receiver, receiver) = Subscriber::new_test("slotNotification");
         rpc.slot_subscribe(session, subscriber);
@@ -753,7 +794,9 @@ mod tests {
     #[test]
     #[serial]
     fn test_slot_unsubscribe() {
-        let rpc = RpcSolPubSubImpl::default();
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
+        let rpc = RpcSolPubSubImpl::default_with_blockstore(blockstore);
         let session = create_session();
         let (subscriber, _id_receiver, receiver) = Subscriber::new_test("slotNotification");
         rpc.slot_subscribe(session, subscriber);

core/src/rpc_pubsub_service.rs

@@ -73,6 +73,7 @@ impl PubSubService {
 mod tests {
     use super::*;
     use crate::commitment::BlockCommitmentCache;
+    use solana_ledger::{blockstore::Blockstore, get_tmp_ledger_path};
     use std::{
         net::{IpAddr, Ipv4Addr},
         sync::RwLock,
@@ -82,9 +83,13 @@ mod tests {
     fn test_pubsub_new() {
         let pubsub_addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(0, 0, 0, 0)), 0);
         let exit = Arc::new(AtomicBool::new(false));
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
         let subscriptions = Arc::new(RpcSubscriptions::new(
             &exit,
-            Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
+            Arc::new(RwLock::new(
+                BlockCommitmentCache::new_for_tests_with_blockstore(blockstore),
+            )),
         ));
         let pubsub_service = PubSubService::new(&subscriptions, pubsub_addr, &exit);
         let thread = pubsub_service.thread_hdl.thread();

core/src/rpc_service.rs

@@ -379,16 +379,18 @@ mod tests {
             solana_net_utils::find_available_port_in_range(ip_addr, (10000, 65535)).unwrap(),
         );
         let bank_forks = Arc::new(RwLock::new(BankForks::new(bank.slot(), bank)));
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
         let ledger_path = get_tmp_ledger_path!();
-        let blockstore = Blockstore::open(&ledger_path).unwrap();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore.clone()),
+        ));
         let mut rpc_service = JsonRpcService::new(
             rpc_addr,
             JsonRpcConfig::default(),
             None,
             bank_forks,
             block_commitment_cache,
-            Arc::new(blockstore),
+            blockstore,
             cluster_info,
             Hash::default(),
             &PathBuf::from("farf"),

core/src/rpc_subscriptions.rs

@@ -11,7 +11,7 @@ use serde::Serialize;
 use solana_client::rpc_response::{
     Response, RpcAccount, RpcKeyedAccount, RpcResponseContext, RpcSignatureResult,
 };
-use solana_ledger::bank_forks::BankForks;
+use solana_ledger::{bank_forks::BankForks, blockstore::Blockstore};
 use solana_runtime::bank::Bank;
 use solana_sdk::{
     account::Account, clock::Slot, pubkey::Pubkey, signature::Signature, transaction,
@@ -246,14 +246,14 @@ pub struct RpcSubscriptions {
     exit: Arc<AtomicBool>,
 }
 
-impl Default for RpcSubscriptions {
+// impl Default for RpcSubscriptions {
-    fn default() -> Self {
+//     fn default() -> Self {
-        Self::new(
+//         Self::new(
-            &Arc::new(AtomicBool::new(false)),
+//             &Arc::new(AtomicBool::new(false)),
-            Arc::new(RwLock::new(BlockCommitmentCache::default())),
+//             Arc::new(RwLock::new(BlockCommitmentCache::default())),
-        )
+//         )
-    }
+//     }
-}
+// }
 
 impl Drop for RpcSubscriptions {
     fn drop(&mut self) {
@@ -324,6 +324,15 @@ impl RpcSubscriptions {
         }
     }
 
+    pub fn default_with_blockstore(blockstore: Arc<Blockstore>) -> Self {
+        Self::new(
+            &Arc::new(AtomicBool::new(false)),
+            Arc::new(RwLock::new(BlockCommitmentCache::default_with_blockstore(
+                blockstore,
+            ))),
+        )
+    }
+
     fn check_account(
         pubkey: &Pubkey,
         bank_forks: &Arc<RwLock<BankForks>>,
@@ -621,7 +630,11 @@ pub(crate) mod tests {
     use jsonrpc_pubsub::typed::Subscriber;
     use serial_test_derive::serial;
     use solana_budget_program;
-    use solana_ledger::genesis_utils::{create_genesis_config, GenesisConfigInfo};
+    use solana_ledger::{
+        blockstore::Blockstore,
+        genesis_utils::{create_genesis_config, GenesisConfigInfo},
+        get_tmp_ledger_path,
+    };
     use solana_sdk::{
         signature::{Keypair, Signer},
         system_transaction,
@@ -662,6 +675,8 @@ pub(crate) mod tests {
             mint_keypair,
             ..
         } = create_genesis_config(100);
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
         let bank = Bank::new(&genesis_config);
         let blockhash = bank.last_blockhash();
         let bank_forks = Arc::new(RwLock::new(BankForks::new(0, bank)));
@@ -688,7 +703,9 @@ pub(crate) mod tests {
         let exit = Arc::new(AtomicBool::new(false));
         let subscriptions = RpcSubscriptions::new(
             &exit,
-            Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
+            Arc::new(RwLock::new(
+                BlockCommitmentCache::new_for_tests_with_blockstore(blockstore),
+            )),
         );
         subscriptions.add_account_subscription(alice.pubkey(), None, sub_id.clone(), subscriber);
 
@@ -735,6 +752,8 @@ pub(crate) mod tests {
             mint_keypair,
             ..
         } = create_genesis_config(100);
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
         let bank = Bank::new(&genesis_config);
         let blockhash = bank.last_blockhash();
         let bank_forks = Arc::new(RwLock::new(BankForks::new(0, bank)));
@@ -761,7 +780,9 @@ pub(crate) mod tests {
         let exit = Arc::new(AtomicBool::new(false));
         let subscriptions = RpcSubscriptions::new(
             &exit,
-            Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
+            Arc::new(RwLock::new(
+                BlockCommitmentCache::new_for_tests_with_blockstore(blockstore),
+            )),
         );
         subscriptions.add_program_subscription(
             solana_budget_program::id(),
@@ -816,6 +837,8 @@ pub(crate) mod tests {
             mint_keypair,
             ..
         } = create_genesis_config(100);
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
         let bank = Bank::new(&genesis_config);
         let blockhash = bank.last_blockhash();
         let mut bank_forks = BankForks::new(0, bank);
@@ -854,7 +877,8 @@ pub(crate) mod tests {
         let mut block_commitment = HashMap::new();
         block_commitment.entry(0).or_insert(cache0.clone());
         block_commitment.entry(1).or_insert(cache1.clone());
-        let block_commitment_cache = BlockCommitmentCache::new(block_commitment, 10, bank1, 0);
+        let block_commitment_cache =
+            BlockCommitmentCache::new(block_commitment, 0, 10, bank1, blockstore, 0);
 
         let exit = Arc::new(AtomicBool::new(false));
         let subscriptions =
@@ -957,9 +981,13 @@ pub(crate) mod tests {
             Subscriber::new_test("slotNotification");
         let sub_id = SubscriptionId::Number(0 as u64);
         let exit = Arc::new(AtomicBool::new(false));
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
         let subscriptions = RpcSubscriptions::new(
             &exit,
-            Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
+            Arc::new(RwLock::new(
+                BlockCommitmentCache::new_for_tests_with_blockstore(blockstore),
+            )),
         );
         subscriptions.add_slot_subscription(sub_id.clone(), subscriber);
 
@@ -999,9 +1027,13 @@ pub(crate) mod tests {
             Subscriber::new_test("rootNotification");
         let sub_id = SubscriptionId::Number(0 as u64);
         let exit = Arc::new(AtomicBool::new(false));
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
         let subscriptions = RpcSubscriptions::new(
             &exit,
-            Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
+            Arc::new(RwLock::new(
+                BlockCommitmentCache::new_for_tests_with_blockstore(blockstore),
+            )),
         );
         subscriptions.add_root_subscription(sub_id.clone(), subscriber);
 

core/src/storage_stage.rs

@@ -662,7 +662,10 @@ pub fn test_cluster_info(id: &Pubkey) -> Arc<ClusterInfo> {
 mod tests {
     use super::*;
     use rayon::prelude::*;
-    use solana_ledger::genesis_utils::{create_genesis_config, GenesisConfigInfo};
+    use solana_ledger::{
+        genesis_utils::{create_genesis_config, GenesisConfigInfo},
+        get_tmp_ledger_path,
+    };
     use solana_runtime::bank::Bank;
     use solana_sdk::{
         hash::Hasher,
@@ -690,7 +693,11 @@ mod tests {
             &[bank.clone()],
             vec![0],
         )));
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
+        let ledger_path = get_tmp_ledger_path!();
+        let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore),
+        ));
         let (_slot_sender, slot_receiver) = channel();
         let storage_state = StorageState::new(
             &bank.last_blockhash(),

core/src/tvu.rs

@@ -298,7 +298,9 @@ pub mod tests {
         let vote_keypair = Keypair::new();
         let storage_keypair = Arc::new(Keypair::new());
         let leader_schedule_cache = Arc::new(LeaderScheduleCache::new_from_bank(&bank));
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore.clone()),
+        ));
         let (retransmit_slots_sender, _retransmit_slots_receiver) = unbounded();
         let tvu = Tvu::new(
             &vote_keypair.pubkey(),
@@ -317,10 +319,7 @@ pub mod tests {
             blockstore,
             &StorageState::default(),
             l_receiver,
-            &Arc::new(RpcSubscriptions::new(
+            &Arc::new(RpcSubscriptions::new(&exit, block_commitment_cache.clone())),
-                &exit,
-                Arc::new(RwLock::new(BlockCommitmentCache::default())),
-            )),
             &poh_recorder,
             &leader_schedule_cache,
             &exit,

core/src/validator.rs

@@ -211,7 +211,6 @@ impl Validator {
         }
 
         let bank_forks = Arc::new(RwLock::new(bank_forks));
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
 
         let mut validator_exit = ValidatorExit::default();
         let exit_ = exit.clone();
@@ -244,6 +243,9 @@ impl Validator {
         );
 
         let blockstore = Arc::new(blockstore);
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore.clone()),
+        ));
 
         let subscriptions = Arc::new(RpcSubscriptions::new(&exit, block_commitment_cache.clone()));
 

core/tests/client.rs

@@ -6,6 +6,7 @@ use solana_core::{
     commitment::BlockCommitmentCache, rpc_pubsub_service::PubSubService,
     rpc_subscriptions::RpcSubscriptions, validator::TestValidator,
 };
+use solana_ledger::{blockstore::Blockstore, get_tmp_ledger_path};
 use solana_sdk::{
     commitment_config::CommitmentConfig, pubkey::Pubkey, rpc_port, signature::Signer,
     system_transaction,
@@ -85,9 +86,13 @@ fn test_slot_subscription() {
         rpc_port::DEFAULT_RPC_PUBSUB_PORT,
     );
     let exit = Arc::new(AtomicBool::new(false));
+    let ledger_path = get_tmp_ledger_path!();
+    let blockstore = Arc::new(Blockstore::open(&ledger_path).unwrap());
     let subscriptions = Arc::new(RpcSubscriptions::new(
         &exit,
-        Arc::new(RwLock::new(BlockCommitmentCache::default())),
+        Arc::new(RwLock::new(BlockCommitmentCache::default_with_blockstore(
+            blockstore,
+        ))),
     ));
     let pubsub_service = PubSubService::new(&subscriptions, pubsub_addr, &exit);
     std::thread::sleep(Duration::from_millis(400));

core/tests/storage_stage.rs

@@ -60,7 +60,9 @@ mod tests {
             &[bank.clone()],
             vec![0],
         )));
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore.clone()),
+        ));
         let cluster_info = test_cluster_info(&keypair.pubkey());
 
         let (bank_sender, bank_receiver) = channel();
@@ -181,7 +183,9 @@ mod tests {
             &[bank.clone()],
             vec![0],
         )));
-        let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
+        let block_commitment_cache = Arc::new(RwLock::new(
+            BlockCommitmentCache::default_with_blockstore(blockstore.clone()),
+        ));
 
         let cluster_info = test_cluster_info(&keypair.pubkey());
         let (bank_sender, bank_receiver) = channel();