diff --git a/core/src/commitment.rs b/core/src/commitment.rs index 8f10bdfd6e..3c940208cb 100644 --- a/core/src/commitment.rs +++ b/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, + largest_confirmed_root: Slot, total_stake: u64, bank: Arc, + blockstore: Arc, root: Slot, } @@ -70,22 +72,41 @@ impl std::fmt::Debug for BlockCommitmentCache { impl BlockCommitmentCache { pub fn new( block_commitment: HashMap, + largest_confirmed_root: Slot, total_stake: u64, bank: Arc, + blockstore: Arc, root: Slot, ) -> Self { Self { block_commitment, + largest_confirmed_root, total_stake, bank, + blockstore, root, } } + pub fn default_with_blockstore(blockstore: Arc) -> 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) - .map(|block_commitment| { - (block_commitment.get_rooted_stake() as f64 / self.total_stake as f64) - > VOTE_THRESHOLD_SIZE - }) - .unwrap_or(false) + slot <= self.largest_confirmed_root() + && (self.blockstore.is_root(slot) || self.bank.status_cache_ancestors().contains(&slot)) } #[cfg(test)] - pub fn new_for_tests() -> Self { + pub fn new_for_tests_with_blockstore(blockstore: Arc) -> Self { let mut block_commitment: HashMap = 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 { + pub fn aggregate_commitment( + ancestors: &[Slot], + bank: &Bank, + ) -> (HashMap, 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, + 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(); - vote_state.root_slot = Some(*root); + let root = ancestors.last().unwrap().clone(); + 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) } } diff --git a/core/src/replay_stage.rs b/core/src/replay_stage.rs index 02acc98582..6c931bc7fe 100644 --- a/core/src/replay_stage.rs +++ b/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(), diff --git a/core/src/rpc.rs b/core/src/rpc.rs index d51d2888ed..9fff3fa1a4 100644 --- a/core/src/rpc.rs +++ b/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, ); diff --git a/core/src/rpc_pubsub.rs b/core/src/rpc_pubsub.rs index 1d37cdfd02..6c45c2fad4 100644 --- a/core/src/rpc_pubsub.rs +++ b/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, id: SubscriptionId) -> Result; } -#[derive(Default)] pub struct RpcSolPubSubImpl { uid: Arc, subscriptions: Arc, @@ -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) -> Self { + let uid = Arc::new(atomic::AtomicUsize::default()); + let subscriptions = Arc::new(RpcSubscriptions::default_with_blockstore(blockstore)); + Self { uid, subscriptions } + } +} fn param(param_str: &str, thing: &str) -> Result { param_str.parse::().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::genesis_utils::{create_genesis_config, GenesisConfigInfo}; + use solana_ledger::{ + 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 = - BlockCommitmentCache::new(block_commitment, 10, bank1.clone(), 0); + let mut new_block_commitment = BlockCommitmentCache::new( + 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); diff --git a/core/src/rpc_pubsub_service.rs b/core/src/rpc_pubsub_service.rs index e58bdc775c..47586642ce 100644 --- a/core/src/rpc_pubsub_service.rs +++ b/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(); diff --git a/core/src/rpc_service.rs b/core/src/rpc_service.rs index 7e0197ac4e..b361847be8 100644 --- a/core/src/rpc_service.rs +++ b/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"), diff --git a/core/src/rpc_subscriptions.rs b/core/src/rpc_subscriptions.rs index b890ac3ffd..04e91312be 100644 --- a/core/src/rpc_subscriptions.rs +++ b/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, } -impl Default for RpcSubscriptions { - fn default() -> Self { - Self::new( - &Arc::new(AtomicBool::new(false)), - Arc::new(RwLock::new(BlockCommitmentCache::default())), - ) - } -} +// impl Default for RpcSubscriptions { +// fn default() -> Self { +// Self::new( +// &Arc::new(AtomicBool::new(false)), +// 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) -> 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>, @@ -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); diff --git a/core/src/storage_stage.rs b/core/src/storage_stage.rs index 38ab4d53a4..865e84ac6f 100644 --- a/core/src/storage_stage.rs +++ b/core/src/storage_stage.rs @@ -662,7 +662,10 @@ pub fn test_cluster_info(id: &Pubkey) -> Arc { 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(), diff --git a/core/src/tvu.rs b/core/src/tvu.rs index bfca12168d..ec889e14f0 100644 --- a/core/src/tvu.rs +++ b/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( - &exit, - Arc::new(RwLock::new(BlockCommitmentCache::default())), - )), + &Arc::new(RpcSubscriptions::new(&exit, block_commitment_cache.clone())), &poh_recorder, &leader_schedule_cache, &exit, diff --git a/core/src/validator.rs b/core/src/validator.rs index aaf22c9430..7cb44462b8 100644 --- a/core/src/validator.rs +++ b/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())); diff --git a/core/tests/client.rs b/core/tests/client.rs index 7b1d9645df..7e16276a14 100644 --- a/core/tests/client.rs +++ b/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)); diff --git a/core/tests/storage_stage.rs b/core/tests/storage_stage.rs index b6c936fc10..51c62e7158 100644 --- a/core/tests/storage_stage.rs +++ b/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();