gaspersic/solana
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

RPC Block Subscription (#21787)

Browse Source 
* add stuff

* compiling

* add notify block

* wip

* feat: add blockSubscribe pubsub method

* address PR comments

Co-authored-by: Lucas B <buffalu@jito.network>
Co-authored-by: Zano <segfaultdoctor@protonmail.com>
This commit is contained in:
segfaultdoctor
2021-12-17 18:03:09 -05:00
committed by GitHub
parent 5f054cd51b
commit 76098dd42a
20 changed files with 1022 additions and 157 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
rpc/src/optimistically_confirmed_bank_tracker.rs
View File

@ -321,6 +321,7 @@ mod tests {
accounts_background_service::AbsRequestSender, commitment::BlockCommitmentCache,
},
solana_sdk::pubkey::Pubkey,
std::sync::atomic::AtomicU64,
};
#[test]
@ -343,8 +344,10 @@ mod tests {
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::default()));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks.clone(),
block_commitment_cache,
optimistically_confirmed_bank.clone(),

5
rpc/src/rpc.rs
View File

@ -2101,7 +2101,7 @@ fn verify_and_parse_signatures_for_address_params(
Ok((address, before, until, limit))
}
fn check_is_at_least_confirmed(commitment: CommitmentConfig) -> Result<()> {
pub(crate) fn check_is_at_least_confirmed(commitment: CommitmentConfig) -> Result<()> {
if !commitment.is_at_least_confirmed() {
return Err(Error::invalid_params(
"Method does not support commitment below `confirmed`",
@ -7807,9 +7807,10 @@ pub mod tests {
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let mut pending_optimistically_confirmed_banks = HashSet::new();
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks.clone(),
block_commitment_cache.clone(),
optimistically_confirmed_bank.clone(),

138
rpc/src/rpc_pubsub.rs
View File

@ -2,11 +2,13 @@
use {
crate::{
rpc::check_is_at_least_confirmed,
rpc_pubsub_service::PubSubConfig,
rpc_subscription_tracker::{
AccountSubscriptionParams, LogsSubscriptionKind, LogsSubscriptionParams,
ProgramSubscriptionParams, SignatureSubscriptionParams, SubscriptionControl,
SubscriptionId, SubscriptionParams, SubscriptionToken,
AccountSubscriptionParams, BlockSubscriptionKind, BlockSubscriptionParams,
LogsSubscriptionKind, LogsSubscriptionParams, ProgramSubscriptionParams,
SignatureSubscriptionParams, SubscriptionControl, SubscriptionId, SubscriptionParams,
SubscriptionToken,
},
},
dashmap::DashMap,
@ -16,15 +18,17 @@ use {
solana_account_decoder::{UiAccount, UiAccountEncoding},
solana_client::{
rpc_config::{
RpcAccountInfoConfig, RpcProgramAccountsConfig, RpcSignatureSubscribeConfig,
RpcTransactionLogsConfig, RpcTransactionLogsFilter,
RpcAccountInfoConfig, RpcBlockSubscribeConfig, RpcBlockSubscribeFilter,
RpcProgramAccountsConfig, RpcSignatureSubscribeConfig, RpcTransactionLogsConfig,
RpcTransactionLogsFilter,
},
rpc_response::{
Response as RpcResponse, RpcKeyedAccount, RpcLogsResponse, RpcSignatureResult, RpcVote,
SlotInfo, SlotUpdate,
Response as RpcResponse, RpcBlockUpdate, RpcKeyedAccount, RpcLogsResponse,
RpcSignatureResult, RpcVote, SlotInfo, SlotUpdate,
},
},
solana_sdk::{clock::Slot, pubkey::Pubkey, signature::Signature},
solana_transaction_status::UiTransactionEncoding,
std::{str::FromStr, sync::Arc},
};
@ -187,6 +191,28 @@ pub trait RpcSolPubSub {
id: PubSubSubscriptionId,
) -> Result<bool>;
// Subscribe to block data and content
#[pubsub(subscription = "blockNotification", subscribe, name = "blockSubscribe")]
fn block_subscribe(
&self,
meta: Self::Metadata,
subscriber: Subscriber<Arc<RpcBlockUpdate>>,
filter: RpcBlockSubscribeFilter,
config: Option<RpcBlockSubscribeConfig>,
);
// Unsubscribe from block notification subscription.
#[pubsub(
subscription = "blockNotification",
unsubscribe,
name = "blockUnsubscribe"
)]
fn block_unsubscribe(
&self,
meta: Option<Self::Metadata>,
id: PubSubSubscriptionId,
) -> Result<bool>;
// Get notification when vote is encountered
#[pubsub(subscription = "voteNotification", subscribe, name = "voteSubscribe")]
fn vote_subscribe(&self, meta: Self::Metadata, subscriber: Subscriber<RpcVote>);
@ -295,6 +321,18 @@ mod internal {
#[rpc(name = "slotsUpdatesUnsubscribe")]
fn slots_updates_unsubscribe(&self, id: SubscriptionId) -> Result<bool>;
// Subscribe to block data and content
#[rpc(name = "blockSubscribe")]
fn block_subscribe(
&self,
filter: RpcBlockSubscribeFilter,
config: Option<RpcBlockSubscribeConfig>,
) -> Result<SubscriptionId>;
// Unsubscribe from block notification subscription.
#[rpc(name = "blockUnsubscribe")]
fn block_unsubscribe(&self, id: SubscriptionId) -> Result<bool>;
// Get notification when vote is encountered
#[rpc(name = "voteSubscribe")]
fn vote_subscribe(&self) -> Result<SubscriptionId>;
@ -475,6 +513,42 @@ impl RpcSolPubSubInternal for RpcSolPubSubImpl {
self.unsubscribe(id)
}
fn block_subscribe(
&self,
filter: RpcBlockSubscribeFilter,
config: Option<RpcBlockSubscribeConfig>,
) -> Result<SubscriptionId> {
if !self.config.enable_block_subscription {
return Err(Error::new(jsonrpc_core::ErrorCode::MethodNotFound));
}
let config = config.unwrap_or_default();
let commitment = config.commitment.unwrap_or_default();
check_is_at_least_confirmed(commitment)?;
let params = BlockSubscriptionParams {
commitment: config.commitment.unwrap_or_default(),
encoding: config.encoding.unwrap_or(UiTransactionEncoding::Base64),
kind: match filter {
RpcBlockSubscribeFilter::All => BlockSubscriptionKind::All,
RpcBlockSubscribeFilter::MentionsAccountOrProgram(key) => {
BlockSubscriptionKind::MentionsAccountOrProgram(param::<Pubkey>(
&key,
"mentions_account_or_program",
)?)
}
},
transaction_details: config.transaction_details.unwrap_or_default(),
show_rewards: config.show_rewards.unwrap_or_default(),
};
self.subscribe(SubscriptionParams::Block(params))
}
fn block_unsubscribe(&self, id: SubscriptionId) -> Result<bool> {
if !self.config.enable_block_subscription {
return Err(Error::new(jsonrpc_core::ErrorCode::MethodNotFound));
}
self.unsubscribe(id)
}
fn vote_subscribe(&self) -> Result<SubscriptionId> {
if !self.config.enable_vote_subscription {
return Err(Error::new(jsonrpc_core::ErrorCode::MethodNotFound));
@ -539,7 +613,10 @@ mod tests {
solana_stake_program::stake_state,
solana_vote_program::vote_state::Vote,
std::{
sync::{atomic::AtomicBool, RwLock},
sync::{
atomic::{AtomicBool, AtomicU64},
RwLock,
},
thread::sleep,
time::Duration,
},
@ -578,8 +655,10 @@ mod tests {
let bank = Bank::new_for_tests(&genesis_config);
let blockhash = bank.last_blockhash();
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let rpc_subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&Arc::new(AtomicBool::new(false)),
max_complete_transaction_status_slot,
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks),
@ -705,7 +784,11 @@ mod tests {
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let mut io = IoHandler::<()>::default();
let subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(bank_forks));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(
max_complete_transaction_status_slot,
bank_forks,
));
let (rpc, _receiver) = rpc_pubsub_service::test_connection(&subscriptions);
io.extend_with(rpc.to_delegate());
@ -756,9 +839,10 @@ mod tests {
let bank0 = bank_forks.read().unwrap().get(0).unwrap().clone();
let bank1 = Bank::new_from_parent(&bank0, &Pubkey::default(), 1);
bank_forks.write().unwrap().insert(bank1);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let rpc_subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&Arc::new(AtomicBool::new(false)),
max_complete_transaction_status_slot,
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests_with_slots(
1, 1,
@ -873,9 +957,10 @@ mod tests {
let bank0 = bank_forks.read().unwrap().get(0).unwrap().clone();
let bank1 = Bank::new_from_parent(&bank0, &Pubkey::default(), 1);
bank_forks.write().unwrap().insert(bank1);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let rpc_subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&Arc::new(AtomicBool::new(false)),
max_complete_transaction_status_slot,
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests_with_slots(
1, 1,
@ -963,7 +1048,11 @@ mod tests {
))));
let mut io = IoHandler::<()>::default();
let subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(bank_forks));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(
max_complete_transaction_status_slot,
bank_forks,
));
let (rpc, _receiver) = rpc_pubsub_service::test_connection(&subscriptions);
io.extend_with(rpc.to_delegate());
@ -1007,8 +1096,10 @@ mod tests {
let bob = Keypair::new();
let exit = Arc::new(AtomicBool::new(false));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let rpc_subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks),
@ -1058,9 +1149,10 @@ mod tests {
let exit = Arc::new(AtomicBool::new(false));
let block_commitment_cache = Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests()));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks.clone(),
block_commitment_cache,
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks),
@ -1128,7 +1220,11 @@ mod tests {
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(10_000);
let bank = Bank::new_for_tests(&genesis_config);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let rpc_subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(bank_forks));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let rpc_subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(
max_complete_transaction_status_slot,
bank_forks,
));
let (rpc, mut receiver) = rpc_pubsub_service::test_connection(&rpc_subscriptions);
rpc.slot_subscribe().unwrap();
@ -1156,7 +1252,11 @@ mod tests {
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(10_000);
let bank = Bank::new_for_tests(&genesis_config);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let rpc_subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(bank_forks));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let rpc_subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(
max_complete_transaction_status_slot,
bank_forks,
));
let (rpc, mut receiver) = rpc_pubsub_service::test_connection(&rpc_subscriptions);
let sub_id = rpc.slot_subscribe().unwrap();
@ -1198,8 +1298,10 @@ mod tests {
// Setup Subscriptions
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks,
block_commitment_cache,
optimistically_confirmed_bank,
@ -1228,7 +1330,11 @@ mod tests {
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(10_000);
let bank = Bank::new_for_tests(&genesis_config);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let rpc_subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(bank_forks));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let rpc_subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(
max_complete_transaction_status_slot,
bank_forks,
));
let (rpc, _receiver) = rpc_pubsub_service::test_connection(&rpc_subscriptions);
let sub_id = rpc.vote_subscribe().unwrap();

44
rpc/src/rpc_pubsub_service.rs
View File

@ -33,6 +33,7 @@ pub const DEFAULT_WORKER_THREADS: usize = 1;
#[derive(Debug, Clone)]
pub struct PubSubConfig {
pub enable_block_subscription: bool,
pub enable_vote_subscription: bool,
pub max_active_subscriptions: usize,
pub queue_capacity_items: usize,
@ -44,6 +45,7 @@ pub struct PubSubConfig {
impl Default for PubSubConfig {
fn default() -> Self {
Self {
enable_block_subscription: false,
enable_vote_subscription: false,
max_active_subscriptions: MAX_ACTIVE_SUBSCRIPTIONS,
queue_capacity_items: DEFAULT_QUEUE_CAPACITY_ITEMS,
@ -57,6 +59,7 @@ impl Default for PubSubConfig {
impl PubSubConfig {
pub fn default_for_tests() -> Self {
Self {
enable_block_subscription: false,
enable_vote_subscription: false,
max_active_subscriptions: MAX_ACTIVE_SUBSCRIPTIONS,
queue_capacity_items: DEFAULT_TEST_QUEUE_CAPACITY_ITEMS,
@ -142,6 +145,9 @@ fn count_final(params: &SubscriptionParams) {
SubscriptionParams::Vote => {
inc_new_counter_info!("rpc-pubsub-final-votes", 1);
}
SubscriptionParams::Block(_) => {
inc_new_counter_info!("rpc-pubsub-final-slot-txs", 1);
}
}
}
@ -187,16 +193,17 @@ pub struct TestBroadcastReceiver {
#[cfg(test)]
impl TestBroadcastReceiver {
pub fn recv(&mut self) -> String {
use {
std::{
thread::sleep,
time::{Duration, Instant},
},
tokio::sync::broadcast::error::TryRecvError,
return match self.recv_timeout(std::time::Duration::from_secs(5)) {
Err(err) => panic!("broadcast receiver error: {}", err),
Ok(str) => str,
};
}
let timeout = Duration::from_secs(5);
let started = Instant::now();
pub fn recv_timeout(&mut self, timeout: std::time::Duration) -> Result<String, String> {
use std::thread::sleep;
use tokio::sync::broadcast::error::TryRecvError;
let started = std::time::Instant::now();
loop {
match self.inner.try_recv() {
@ -206,17 +213,16 @@ impl TestBroadcastReceiver {
started.elapsed().as_millis()
);
if let Some(json) = self.handler.handle(notification).expect("handler failed") {
return json.to_string();
return Ok(json.to_string());
}
}
Err(TryRecvError::Empty) => {
assert!(
started.elapsed() <= timeout,
"TestBroadcastReceiver: no data, timeout reached"
);
sleep(Duration::from_millis(50));
if started.elapsed() > timeout {
return Err("TestBroadcastReceiver: no data, timeout reached".into());
}
sleep(std::time::Duration::from_millis(50));
}
Err(err) => panic!("broadcast receiver error: {}", err),
Err(e) => return Err(e.to_string()),
}
}
}
@ -230,6 +236,7 @@ pub fn test_connection(
let rpc_impl = RpcSolPubSubImpl::new(
PubSubConfig {
enable_block_subscription: true,
enable_vote_subscription: true,
queue_capacity_items: 100,
..PubSubConfig::default()
@ -383,7 +390,10 @@ mod tests {
},
std::{
net::{IpAddr, Ipv4Addr},
sync::{atomic::AtomicBool, RwLock},
sync::{
atomic::{AtomicBool, AtomicU64},
RwLock,
},
},
};
@ -391,6 +401,7 @@ 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 max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(10_000);
let bank = Bank::new_for_tests(&genesis_config);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
@ -398,6 +409,7 @@ mod tests {
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks,
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
optimistically_confirmed_bank,

32
rpc/src/rpc_subscription_tracker.rs
View File

@ -11,6 +11,7 @@ use {
solana_sdk::{
clock::Slot, commitment_config::CommitmentConfig, pubkey::Pubkey, signature::Signature,
},
solana_transaction_status::{TransactionDetails, UiTransactionEncoding},
std::{
collections::{
hash_map::{Entry, HashMap},
@ -44,6 +45,7 @@ impl From<SubscriptionId> for u64 {
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum SubscriptionParams {
Account(AccountSubscriptionParams),
Block(BlockSubscriptionParams),
Logs(LogsSubscriptionParams),
Program(ProgramSubscriptionParams),
Signature(SignatureSubscriptionParams),
@ -62,6 +64,7 @@ impl SubscriptionParams {
SubscriptionParams::Signature(_) => "signatureNotification",
SubscriptionParams::Slot => "slotNotification",
SubscriptionParams::SlotsUpdates => "slotsUpdatesNotification",
SubscriptionParams::Block(_) => "blockNotification",
SubscriptionParams::Root => "rootNotification",
SubscriptionParams::Vote => "voteNotification",
}
@ -73,6 +76,7 @@ impl SubscriptionParams {
SubscriptionParams::Logs(params) => Some(params.commitment),
SubscriptionParams::Program(params) => Some(params.commitment),
SubscriptionParams::Signature(params) => Some(params.commitment),
SubscriptionParams::Block(params) => Some(params.commitment),
SubscriptionParams::Slot
| SubscriptionParams::SlotsUpdates
| SubscriptionParams::Root
@ -83,12 +87,13 @@ impl SubscriptionParams {
fn is_commitment_watcher(&self) -> bool {
let commitment = match self {
SubscriptionParams::Account(params) => &params.commitment,
SubscriptionParams::Block(params) => &params.commitment,
SubscriptionParams::Logs(params) => &params.commitment,
SubscriptionParams::Program(params) => &params.commitment,
SubscriptionParams::Signature(params) => &params.commitment,
SubscriptionParams::Slot
SubscriptionParams::Root
| SubscriptionParams::Slot
| SubscriptionParams::SlotsUpdates
| SubscriptionParams::Root
| SubscriptionParams::Vote => return false,
};
!commitment.is_confirmed()
@ -97,12 +102,13 @@ impl SubscriptionParams {
fn is_gossip_watcher(&self) -> bool {
let commitment = match self {
SubscriptionParams::Account(params) => &params.commitment,
SubscriptionParams::Block(params) => &params.commitment,
SubscriptionParams::Logs(params) => &params.commitment,
SubscriptionParams::Program(params) => &params.commitment,
SubscriptionParams::Signature(params) => &params.commitment,
SubscriptionParams::Slot
SubscriptionParams::Root
| SubscriptionParams::Slot
| SubscriptionParams::SlotsUpdates
| SubscriptionParams::Root
| SubscriptionParams::Vote => return false,
};
commitment.is_confirmed()
@ -127,6 +133,21 @@ pub struct AccountSubscriptionParams {
pub commitment: CommitmentConfig,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct BlockSubscriptionParams {
pub commitment: CommitmentConfig,
pub encoding: UiTransactionEncoding,
pub kind: BlockSubscriptionKind,
pub transaction_details: TransactionDetails,
pub show_rewards: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub enum BlockSubscriptionKind {
All,
MentionsAccountOrProgram(Pubkey),
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct LogsSubscriptionParams {
pub kind: LogsSubscriptionKind,
@ -473,12 +494,15 @@ impl SubscriptionsTracker {
) -> &HashMap<Signature, HashMap<SubscriptionId, Arc<SubscriptionInfo>>> {
&self.by_signature
}
pub fn commitment_watchers(&self) -> &HashMap<SubscriptionId, Arc<SubscriptionInfo>> {
&self.commitment_watchers
}
pub fn gossip_watchers(&self) -> &HashMap<SubscriptionId, Arc<SubscriptionInfo>> {
&self.gossip_watchers
}
pub fn node_progress_watchers(&self) -> &HashMap<SubscriptionParams, Arc<SubscriptionInfo>> {
&self.node_progress_watchers
}

693
rpc/src/rpc_subscriptions.rs
View File

@ -1,14 +1,14 @@
//! The `pubsub` module implements a threaded subscription service on client RPC request
use {
crate::{
optimistically_confirmed_bank_tracker::OptimisticallyConfirmedBank,
parsed_token_accounts::{get_parsed_token_account, get_parsed_token_accounts},
rpc_pubsub_service::PubSubConfig,
rpc_subscription_tracker::{
AccountSubscriptionParams, LogsSubscriptionKind, LogsSubscriptionParams,
ProgramSubscriptionParams, SignatureSubscriptionParams, SubscriptionControl,
SubscriptionId, SubscriptionInfo, SubscriptionParams, SubscriptionsTracker,
AccountSubscriptionParams, BlockSubscriptionKind, BlockSubscriptionParams,
LogsSubscriptionKind, LogsSubscriptionParams, ProgramSubscriptionParams,
SignatureSubscriptionParams, SubscriptionControl, SubscriptionId, SubscriptionInfo,
SubscriptionParams, SubscriptionsTracker,
},
},
crossbeam_channel::{Receiver, RecvTimeoutError, SendError, Sender},
@ -18,10 +18,12 @@ use {
solana_client::{
rpc_filter::RpcFilterType,
rpc_response::{
ProcessedSignatureResult, ReceivedSignatureResult, Response, RpcKeyedAccount,
RpcLogsResponse, RpcResponseContext, RpcSignatureResult, RpcVote, SlotInfo, SlotUpdate,
ProcessedSignatureResult, ReceivedSignatureResult, Response, RpcBlockUpdate,
RpcBlockUpdateError, RpcKeyedAccount, RpcLogsResponse, RpcResponseContext,
RpcSignatureResult, RpcVote, SlotInfo, SlotUpdate,
},
},
solana_ledger::{blockstore::Blockstore, get_tmp_ledger_path},
solana_measure::measure::Measure,
solana_rayon_threadlimit::get_thread_count,
solana_runtime::{
@ -37,6 +39,7 @@ use {
timing::timestamp,
transaction,
},
solana_transaction_status::ConfirmedBlock,
solana_vote_program::vote_state::VoteTransaction,
std::{
cell::RefCell,
@ -44,7 +47,7 @@ use {
io::Cursor,
iter, str,
sync::{
atomic::{AtomicBool, AtomicUsize, Ordering},
atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
Arc, Mutex, RwLock, Weak,
},
thread::{Builder, JoinHandle},
@ -130,7 +133,7 @@ fn check_commitment_and_notify<P, S, B, F, X>(
params: &P,
subscription: &SubscriptionInfo,
bank_forks: &Arc<RwLock<BankForks>>,
commitment_slots: &CommitmentSlots,
slot: Slot,
bank_method: B,
filter_results: F,
notifier: &RpcNotifier,
@ -142,20 +145,6 @@ where
F: Fn(X, &P, Slot, Arc<Bank>) -> (Box<dyn Iterator<Item = S>>, Slot),
X: Clone + Default,
{
let commitment = if let Some(commitment) = subscription.commitment() {
commitment
} else {
error!("missing commitment in check_commitment_and_notify");
return false;
};
let slot = if commitment.is_finalized() {
commitment_slots.highest_confirmed_root
} else if commitment.is_confirmed() {
commitment_slots.highest_confirmed_slot
} else {
commitment_slots.slot
};
let mut notified = false;
if let Some(bank) = bank_forks.read().unwrap().get(slot).cloned() {
let results = bank_method(&bank, params);
@ -175,6 +164,7 @@ where
notified = true;
}
}
notified
}
@ -287,6 +277,46 @@ impl RpcNotifier {
}
}
fn filter_block_result_txs(
block: ConfirmedBlock,
last_modified_slot: Slot,
params: &BlockSubscriptionParams,
) -> Option<RpcBlockUpdate> {
let transactions = match params.kind {
BlockSubscriptionKind::All => block.transactions,
BlockSubscriptionKind::MentionsAccountOrProgram(pk) => block
.transactions
.into_iter()
.filter(|tx| tx.transaction.message.account_keys.contains(&pk))
.collect(),
};
if transactions.is_empty() {
if let BlockSubscriptionKind::MentionsAccountOrProgram(_) = params.kind {
return None;
}
}
let block = ConfirmedBlock {
transactions,
..block
}
.configure(
params.encoding,
params.transaction_details,
params.show_rewards,
);
// If last_modified_slot < last_notified_slot, then the last notif was for a fork.
// That's the risk clients take when subscribing to non-finalized commitments.
// This code lets the logic for dealing with forks live on the client side.
Some(RpcBlockUpdate {
slot: last_modified_slot,
block: Some(block),
err: None,
})
}
fn filter_account_result(
result: Option<(AccountSharedData, Slot)>,
params: &AccountSubscriptionParams,
@ -416,14 +446,7 @@ fn initial_last_notified_slot(
0
}
}
// last_notified_slot is not utilized for these subscriptions
SubscriptionParams::Logs(_)
| SubscriptionParams::Program(_)
| SubscriptionParams::Signature(_)
| SubscriptionParams::Slot
| SubscriptionParams::SlotsUpdates
| SubscriptionParams::Root
| SubscriptionParams::Vote => 0,
_ => 0,
}
}
@ -480,12 +503,16 @@ impl Drop for RpcSubscriptions {
impl RpcSubscriptions {
pub fn new(
exit: &Arc<AtomicBool>,
max_complete_transaction_status_slot: Arc<AtomicU64>,
blockstore: Arc<Blockstore>,
bank_forks: Arc<RwLock<BankForks>>,
block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
optimistically_confirmed_bank: Arc<RwLock<OptimisticallyConfirmedBank>>,
) -> Self {
Self::new_with_config(
exit,
max_complete_transaction_status_slot,
blockstore,
bank_forks,
block_commitment_cache,
optimistically_confirmed_bank,
@ -495,12 +522,38 @@ impl RpcSubscriptions {
pub fn new_for_tests(
exit: &Arc<AtomicBool>,
max_complete_transaction_status_slot: Arc<AtomicU64>,
bank_forks: Arc<RwLock<BankForks>>,
block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
optimistically_confirmed_bank: Arc<RwLock<OptimisticallyConfirmedBank>>,
) -> Self {
let ledger_path = get_tmp_ledger_path!();
let blockstore = Blockstore::open(&ledger_path).unwrap();
let blockstore = Arc::new(blockstore);
Self::new_with_config(
exit,
max_complete_transaction_status_slot,
blockstore,
bank_forks,
block_commitment_cache,
optimistically_confirmed_bank,
&PubSubConfig::default_for_tests(),
)
}
pub fn new_for_tests_with_blockstore(
exit: &Arc<AtomicBool>,
max_complete_transaction_status_slot: Arc<AtomicU64>,
blockstore: Arc<Blockstore>,
bank_forks: Arc<RwLock<BankForks>>,
block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
optimistically_confirmed_bank: Arc<RwLock<OptimisticallyConfirmedBank>>,
) -> Self {
Self::new_with_config(
exit,
max_complete_transaction_status_slot,
blockstore,
bank_forks,
block_commitment_cache,
optimistically_confirmed_bank,
@ -510,6 +563,8 @@ impl RpcSubscriptions {
pub fn new_with_config(
exit: &Arc<AtomicBool>,
max_complete_transaction_status_slot: Arc<AtomicU64>,
blockstore: Arc<Blockstore>,
bank_forks: Arc<RwLock<BankForks>>,
block_commitment_cache: Arc<RwLock<BlockCommitmentCache>>,
optimistically_confirmed_bank: Arc<RwLock<OptimisticallyConfirmedBank>>,
@ -541,6 +596,8 @@ impl RpcSubscriptions {
pool.install(|| {
Self::process_notifications(
exit_clone,
max_complete_transaction_status_slot,
blockstore,
notifier,
notification_receiver,
subscriptions,
@ -568,11 +625,19 @@ impl RpcSubscriptions {
}
// For tests only...
pub fn default_with_bank_forks(bank_forks: Arc<RwLock<BankForks>>) -> Self {
pub fn default_with_bank_forks(
max_complete_transaction_status_slot: Arc<AtomicU64>,
bank_forks: Arc<RwLock<BankForks>>,
) -> Self {
let ledger_path = get_tmp_ledger_path!();
let blockstore = Blockstore::open(&ledger_path).unwrap();
let blockstore = Arc::new(blockstore);
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
Self::new(
&Arc::new(AtomicBool::new(false)),
max_complete_transaction_status_slot,
blockstore,
bank_forks,
Arc::new(RwLock::new(BlockCommitmentCache::default())),
optimistically_confirmed_bank,
@ -641,6 +706,8 @@ impl RpcSubscriptions {
fn process_notifications(
exit: Arc<AtomicBool>,
max_complete_transaction_status_slot: Arc<AtomicU64>,
blockstore: Arc<Blockstore>,
notifier: RpcNotifier,
notification_receiver: Receiver<TimestampedNotificationEntry>,
mut subscriptions: SubscriptionsTracker,
@ -719,27 +786,32 @@ impl RpcSubscriptions {
}
}
NotificationEntry::Bank(commitment_slots) => {
RpcSubscriptions::notify_accounts_logs_programs_signatures(
const SOURCE: &str = "bank";
RpcSubscriptions::notify_watchers(
max_complete_transaction_status_slot.clone(),
subscriptions.commitment_watchers(),
&bank_forks,
&blockstore,
&commitment_slots,
&notifier,
"bank",
)
SOURCE,
);
}
NotificationEntry::Gossip(slot) => {
let commitment_slots = CommitmentSlots {
highest_confirmed_slot: slot,
..CommitmentSlots::default()
};
RpcSubscriptions::notify_accounts_logs_programs_signatures(
const SOURCE: &str = "gossip";
RpcSubscriptions::notify_watchers(
max_complete_transaction_status_slot.clone(),
subscriptions.gossip_watchers(),
&bank_forks,
&blockstore,
&commitment_slots,
&notifier,
"gossip",
)
SOURCE,
);
}
NotificationEntry::SignaturesReceived((slot, slot_signatures)) => {
for slot_signature in &slot_signatures {
@ -785,100 +857,205 @@ impl RpcSubscriptions {
}
}
fn notify_accounts_logs_programs_signatures(
fn notify_watchers(
max_complete_transaction_status_slot: Arc<AtomicU64>,
subscriptions: &HashMap<SubscriptionId, Arc<SubscriptionInfo>>,
bank_forks: &Arc<RwLock<BankForks>>,
blockstore: &Blockstore,
commitment_slots: &CommitmentSlots,
notifier: &RpcNotifier,
source: &'static str,
) {
let mut total_time = Measure::start("notify_accounts_logs_programs_signatures");
let mut total_time = Measure::start("notify_watchers");
let num_accounts_found = AtomicUsize::new(0);
let num_accounts_notified = AtomicUsize::new(0);
let num_blocks_found = AtomicUsize::new(0);
let num_blocks_notified = AtomicUsize::new(0);
let num_logs_found = AtomicUsize::new(0);
let num_logs_notified = AtomicUsize::new(0);
let num_signatures_found = AtomicUsize::new(0);
let num_signatures_notified = AtomicUsize::new(0);
let num_programs_found = AtomicUsize::new(0);
let num_programs_notified = AtomicUsize::new(0);
let num_signatures_found = AtomicUsize::new(0);
let num_signatures_notified = AtomicUsize::new(0);
let subscriptions = subscriptions.into_par_iter();
subscriptions.for_each(|(_id, subscription)| {
let slot = if let Some(commitment) = subscription.commitment() {
if commitment.is_finalized() {
Some(commitment_slots.highest_confirmed_root)
} else if commitment.is_confirmed() {
Some(commitment_slots.highest_confirmed_slot)
} else {
Some(commitment_slots.slot)
}
} else {
error!("missing commitment in notify_watchers");
None
};
match subscription.params() {
SubscriptionParams::Account(params) => {
let notified = check_commitment_and_notify(
params,
subscription,
bank_forks,
commitment_slots,
|bank, params| bank.get_account_modified_slot(&params.pubkey),
filter_account_result,
notifier,
false,
);
num_accounts_found.fetch_add(1, Ordering::Relaxed);
if let Some(slot) = slot {
let notified = check_commitment_and_notify(
params,
subscription,
bank_forks,
slot,
|bank, params| bank.get_account_modified_slot(&params.pubkey),
filter_account_result,
notifier,
false,
);
if notified {
num_accounts_notified.fetch_add(1, Ordering::Relaxed);
if notified {
num_accounts_notified.fetch_add(1, Ordering::Relaxed);
}
}
}
SubscriptionParams::Block(params) => {
num_blocks_found.fetch_add(1, Ordering::Relaxed);
if let Some(slot) = slot {
if let Some(bank) = bank_forks.read().unwrap().get(slot) {
// We're calling it unnotified in this context
// because, logically, it gets set to `last_notified_slot + 1`
// on the final iteration of the loop down below.
// This is used to notify blocks for slots that were
// potentially missed due to upstream transient errors
// that led to this notification not being triggered for
// a slot.
//
// e.g.
// notify_watchers is triggered for Slot 1
// some time passes
// notify_watchers is triggered for Slot 4
// this will try to fetch blocks for slots 2, 3, and 4
// as long as they are ancestors of `slot`
let mut w_last_unnotified_slot =
subscription.last_notified_slot.write().unwrap();
// would mean it's the first notification for this subscription connection
if *w_last_unnotified_slot == 0 {
*w_last_unnotified_slot = slot;
}
let mut slots_to_notify: Vec<_> =
(*w_last_unnotified_slot..slot).collect();
let ancestors = bank.proper_ancestors_set();
slots_to_notify = slots_to_notify
.into_iter()
.filter(|slot| ancestors.contains(slot))
.collect();
slots_to_notify.push(slot);
for s in slots_to_notify {
// To avoid skipping a slot that fails this condition,
// caused by non-deterministic concurrency accesses, we
// break out of the loop. Besides if the current `s` is
// greater, then any `s + K` is also greater.
if s > max_complete_transaction_status_slot.load(Ordering::SeqCst) {
break;
}
match blockstore.get_complete_block(s, false) {
Ok(block) => {
if let Some(res) = filter_block_result_txs(block, s, params)
{
notifier.notify(
Response {
context: RpcResponseContext { slot: s },
value: res,
},
subscription,
false,
);
num_blocks_notified.fetch_add(1, Ordering::Relaxed);
// the next time this subscription is notified it will
// try to fetch all slots between (s + 1) to `slot`, inclusively
*w_last_unnotified_slot = s + 1;
}
}
Err(e) => {
// we don't advance `w_last_unnotified_slot` so that
// it'll retry on the next notification trigger
error!("get_complete_block error: {}", e);
notifier.notify(
Response {
context: RpcResponseContext { slot: s },
value: RpcBlockUpdate {
slot,
block: None,
err: Some(RpcBlockUpdateError::BlockStoreError),
},
},
subscription,
false,
);
}
}
}
}
}
}
SubscriptionParams::Logs(params) => {
let notified = check_commitment_and_notify(
params,
subscription,
bank_forks,
commitment_slots,
get_transaction_logs,
filter_logs_results,
notifier,
false,
);
num_logs_found.fetch_add(1, Ordering::Relaxed);
if let Some(slot) = slot {
let notified = check_commitment_and_notify(
params,
subscription,
bank_forks,
slot,
get_transaction_logs,
filter_logs_results,
notifier,
false,
);
if notified {
num_logs_notified.fetch_add(1, Ordering::Relaxed);
if notified {
num_logs_notified.fetch_add(1, Ordering::Relaxed);
}
}
}
SubscriptionParams::Program(params) => {
let notified = check_commitment_and_notify(
params,
subscription,
bank_forks,
commitment_slots,
|bank, params| {
bank.get_program_accounts_modified_since_parent(&params.pubkey)
},
filter_program_results,
notifier,
false,
);
num_programs_found.fetch_add(1, Ordering::Relaxed);
if let Some(slot) = slot {
let notified = check_commitment_and_notify(
params,
subscription,
bank_forks,
slot,
|bank, params| {
bank.get_program_accounts_modified_since_parent(&params.pubkey)
},
filter_program_results,
notifier,
false,
);
if notified {
num_programs_notified.fetch_add(1, Ordering::Relaxed);
if notified {
num_programs_notified.fetch_add(1, Ordering::Relaxed);
}
}
}
SubscriptionParams::Signature(params) => {
let notified = check_commitment_and_notify(
params,
subscription,
bank_forks,
commitment_slots,
|bank, params| {
bank.get_signature_status_processed_since_parent(&params.signature)
},
filter_signature_result,
notifier,
true, // Unsubscribe.
);
num_signatures_found.fetch_add(1, Ordering::Relaxed);
if let Some(slot) = slot {
let notified = check_commitment_and_notify(
params,
subscription,
bank_forks,
slot,
|bank, params| {
bank.get_signature_status_processed_since_parent(&params.signature)
},
filter_signature_result,
notifier,
true, // Unsubscribe.
);
if notified {
num_signatures_notified.fetch_add(1, Ordering::Relaxed);
if notified {
num_signatures_notified.fetch_add(1, Ordering::Relaxed);
}
}
}
_ => error!("wrong subscription type in alps map"),
@ -997,13 +1174,14 @@ pub(crate) mod tests {
optimistically_confirmed_bank_tracker::{
BankNotification, OptimisticallyConfirmedBank, OptimisticallyConfirmedBankTracker,
},
rpc::create_test_transactions_and_populate_blockstore,
rpc_pubsub::RpcSolPubSubInternal,
rpc_pubsub_service,
},
serial_test::serial,
solana_client::rpc_config::{
RpcAccountInfoConfig, RpcProgramAccountsConfig, RpcSignatureSubscribeConfig,
RpcTransactionLogsFilter,
RpcTransactionLogsFilter, {RpcBlockSubscribeConfig, RpcBlockSubscribeFilter},
},
solana_runtime::{
commitment::BlockCommitment,
@ -1016,7 +1194,11 @@ pub(crate) mod tests {
stake, system_instruction, system_program, system_transaction,
transaction::Transaction,
},
std::{collections::HashSet, sync::atomic::Ordering::Relaxed},
solana_transaction_status::{TransactionDetails, UiTransactionEncoding},
std::{
collections::HashSet,
sync::atomic::{AtomicU64, Ordering::Relaxed},
},
};
fn make_account_result(lamports: u64, subscription: u64, data: &str) -> serde_json::Value {
@ -1056,8 +1238,10 @@ pub(crate) mod tests {
let alice = Keypair::new();
let exit = Arc::new(AtomicBool::new(false));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests_with_slots(
1, 1,
@ -1153,6 +1337,294 @@ pub(crate) mod tests {
}
}
#[test]
#[serial]
fn test_check_confirmed_block_subscribe() {
let exit = Arc::new(AtomicBool::new(false));
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(10_000);
let bank = Bank::new_for_tests(&genesis_config);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let ledger_path = get_tmp_ledger_path!();
let blockstore = Blockstore::open(&ledger_path).unwrap();
let blockstore = Arc::new(blockstore);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests_with_blockstore(
&exit,
max_complete_transaction_status_slot,
blockstore.clone(),
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
optimistically_confirmed_bank,
));
let (rpc, mut receiver) = rpc_pubsub_service::test_connection(&subscriptions);
let filter = RpcBlockSubscribeFilter::All;
let config = RpcBlockSubscribeConfig {
commitment: Some(CommitmentConfig::confirmed()),
encoding: Some(UiTransactionEncoding::Json),
transaction_details: Some(TransactionDetails::Signatures),
show_rewards: None,
};
let params = BlockSubscriptionParams {
kind: BlockSubscriptionKind::All,
commitment: config.commitment.unwrap(),
encoding: config.encoding.unwrap(),
transaction_details: config.transaction_details.unwrap(),
show_rewards: config.show_rewards.unwrap_or_default(),
};
let sub_id = rpc.block_subscribe(filter, Some(config)).unwrap();
subscriptions
.control
.assert_subscribed(&SubscriptionParams::Block(params.clone()));
let bank = bank_forks.read().unwrap().working_bank();
let keypair1 = Keypair::new();
let keypair2 = Keypair::new();
let keypair3 = Keypair::new();
let keypair4 = Keypair::new();
let max_complete_transaction_status_slot = Arc::new(AtomicU64::new(blockstore.max_root()));
let _confirmed_block_signatures = create_test_transactions_and_populate_blockstore(
vec![&keypair1, &keypair2, &keypair3, &keypair4],
0,
bank,
blockstore.clone(),
max_complete_transaction_status_slot,
);
let slot = 0;
subscriptions.notify_gossip_subscribers(slot);
let actual_resp = receiver.recv();
let actual_resp = serde_json::from_str::<serde_json::Value>(&actual_resp).unwrap();
let block = blockstore.get_complete_block(slot, false).unwrap();
let block = block.configure(params.encoding, params.transaction_details, false);
let expected_resp = RpcBlockUpdate {
slot,
block: Some(block),
err: None,
};
let expected_resp = json!({
"jsonrpc": "2.0",
"method": "blockNotification",
"params": {
"result": {
"context": { "slot": slot },
"value": expected_resp,
},
"subscription": 0,
}
});
assert_eq!(expected_resp, actual_resp);
// should not trigger since commitment NOT set to finalized
subscriptions.notify_subscribers(CommitmentSlots {
slot,
root: slot,
highest_confirmed_slot: slot,
highest_confirmed_root: slot,
});
let should_err = receiver.recv_timeout(Duration::from_millis(300));
assert!(should_err.is_err());
rpc.slot_unsubscribe(sub_id).unwrap();
subscriptions
.control
.assert_unsubscribed(&SubscriptionParams::Block(params));
}
#[test]
#[serial]
fn test_check_confirmed_block_subscribe_with_mentions() {
let exit = Arc::new(AtomicBool::new(false));
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(10_000);
let bank = Bank::new_for_tests(&genesis_config);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let ledger_path = get_tmp_ledger_path!();
let blockstore = Blockstore::open(&ledger_path).unwrap();
let blockstore = Arc::new(blockstore);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests_with_blockstore(
&exit,
max_complete_transaction_status_slot,
blockstore.clone(),
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
optimistically_confirmed_bank,
));
let (rpc, mut receiver) = rpc_pubsub_service::test_connection(&subscriptions);
let keypair1 = Keypair::new();
let filter =
RpcBlockSubscribeFilter::MentionsAccountOrProgram(keypair1.pubkey().to_string());
let config = RpcBlockSubscribeConfig {
commitment: Some(CommitmentConfig::confirmed()),
encoding: Some(UiTransactionEncoding::Json),
transaction_details: Some(TransactionDetails::Signatures),
show_rewards: None,
};
let params = BlockSubscriptionParams {
kind: BlockSubscriptionKind::MentionsAccountOrProgram(keypair1.pubkey()),
commitment: config.commitment.unwrap(),
encoding: config.encoding.unwrap(),
transaction_details: config.transaction_details.unwrap(),
show_rewards: config.show_rewards.unwrap_or_default(),
};
let sub_id = rpc.block_subscribe(filter, Some(config)).unwrap();
subscriptions
.control
.assert_subscribed(&SubscriptionParams::Block(params.clone()));
let bank = bank_forks.read().unwrap().working_bank();
let keypair2 = Keypair::new();
let keypair3 = Keypair::new();
let keypair4 = Keypair::new();
let max_complete_transaction_status_slot = Arc::new(AtomicU64::new(blockstore.max_root()));
let _confirmed_block_signatures = create_test_transactions_and_populate_blockstore(
vec![&keypair1, &keypair2, &keypair3, &keypair4],
0,
bank,
blockstore.clone(),
max_complete_transaction_status_slot,
);
let slot = 0;
subscriptions.notify_gossip_subscribers(slot);
let actual_resp = receiver.recv();
let actual_resp = serde_json::from_str::<serde_json::Value>(&actual_resp).unwrap();
// make sure it filtered out the other keypairs
let mut block = blockstore.get_complete_block(slot, false).unwrap();
block.transactions.retain(|tx| {
tx.transaction
.message
.account_keys
.contains(&keypair1.pubkey())
});
let block = block.configure(params.encoding, params.transaction_details, false);
let expected_resp = RpcBlockUpdate {
slot,
block: Some(block),
err: None,
};
let expected_resp = json!({
"jsonrpc": "2.0",
"method": "blockNotification",
"params": {
"result": {
"context": { "slot": slot },
"value": expected_resp,
},
"subscription": 0,
}
});
assert_eq!(expected_resp, actual_resp);
rpc.slot_unsubscribe(sub_id).unwrap();
subscriptions
.control
.assert_unsubscribed(&SubscriptionParams::Block(params));
}
#[test]
#[serial]
fn test_check_finalized_block_subscribe() {
let exit = Arc::new(AtomicBool::new(false));
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(10_000);
let bank = Bank::new_for_tests(&genesis_config);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let ledger_path = get_tmp_ledger_path!();
let blockstore = Blockstore::open(&ledger_path).unwrap();
let blockstore = Arc::new(blockstore);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests_with_blockstore(
&exit,
max_complete_transaction_status_slot,
blockstore.clone(),
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
optimistically_confirmed_bank,
));
let (rpc, mut receiver) = rpc_pubsub_service::test_connection(&subscriptions);
let filter = RpcBlockSubscribeFilter::All;
let config = RpcBlockSubscribeConfig {
commitment: Some(CommitmentConfig::finalized()),
encoding: Some(UiTransactionEncoding::Json),
transaction_details: Some(TransactionDetails::Signatures),
show_rewards: None,
};
let params = BlockSubscriptionParams {
kind: BlockSubscriptionKind::All,
commitment: config.commitment.unwrap(),
encoding: config.encoding.unwrap(),
transaction_details: config.transaction_details.unwrap(),
show_rewards: config.show_rewards.unwrap_or_default(),
};
let sub_id = rpc.block_subscribe(filter, Some(config)).unwrap();
subscriptions
.control
.assert_subscribed(&SubscriptionParams::Block(params.clone()));
let bank = bank_forks.read().unwrap().working_bank();
let keypair1 = Keypair::new();
let keypair2 = Keypair::new();
let keypair3 = Keypair::new();
let keypair4 = Keypair::new();
let max_complete_transaction_status_slot = Arc::new(AtomicU64::new(blockstore.max_root()));
let _confirmed_block_signatures = create_test_transactions_and_populate_blockstore(
vec![&keypair1, &keypair2, &keypair3, &keypair4],
0,
bank,
blockstore.clone(),
max_complete_transaction_status_slot,
);
let slot = 0;
subscriptions.notify_subscribers(CommitmentSlots {
slot,
root: slot,
highest_confirmed_slot: slot,
highest_confirmed_root: slot,
});
let actual_resp = receiver.recv();
let actual_resp = serde_json::from_str::<serde_json::Value>(&actual_resp).unwrap();
let block = blockstore.get_complete_block(slot, false).unwrap();
let block = block.configure(params.encoding, params.transaction_details, false);
let expected_resp = RpcBlockUpdate {
slot,
block: Some(block),
err: None,
};
let expected_resp = json!({
"jsonrpc": "2.0",
"method": "blockNotification",
"params": {
"result": {
"context": { "slot": slot },
"value": expected_resp,
},
"subscription": 0,
}
});
assert_eq!(expected_resp, actual_resp);
// should not trigger since commitment set to finalized
subscriptions.notify_gossip_subscribers(slot);
let should_err = receiver.recv_timeout(Duration::from_millis(300));
assert!(should_err.is_err());
rpc.slot_unsubscribe(sub_id).unwrap();
subscriptions
.control
.assert_unsubscribed(&SubscriptionParams::Block(params));
}
#[test]
#[serial]
fn test_check_program_subscribe() {
@ -1184,8 +1656,10 @@ pub(crate) mod tests {
let exit = Arc::new(AtomicBool::new(false));
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks,
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
optimistically_confirmed_bank,
@ -1329,9 +1803,10 @@ pub(crate) mod tests {
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let mut pending_optimistically_confirmed_banks = HashSet::new();
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests_with_slots(
1, 1,
@ -1498,9 +1973,10 @@ pub(crate) mod tests {
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let mut pending_optimistically_confirmed_banks = HashSet::new();
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests_with_slots(
1, 1,
@ -1608,9 +2084,10 @@ pub(crate) mod tests {
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let mut pending_optimistically_confirmed_banks = HashSet::new();
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests_with_slots(
1, 1,
@ -1794,8 +2271,10 @@ pub(crate) mod tests {
let exit = Arc::new(AtomicBool::new(false));
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks,
Arc::new(RwLock::new(block_commitment_cache)),
optimistically_confirmed_bank,
@ -1966,8 +2445,10 @@ pub(crate) mod tests {
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks,
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
optimistically_confirmed_bank,
@ -2010,8 +2491,10 @@ pub(crate) mod tests {
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks,
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests())),
optimistically_confirmed_bank,
@ -2066,8 +2549,10 @@ pub(crate) mod tests {
let mut pending_optimistically_confirmed_banks = HashSet::new();
let exit = Arc::new(AtomicBool::new(false));
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let subscriptions = Arc::new(RpcSubscriptions::new_for_tests(
&exit,
max_complete_transaction_status_slot,
bank_forks.clone(),
Arc::new(RwLock::new(BlockCommitmentCache::new_for_tests_with_slots(
1, 1,
@ -2218,8 +2703,12 @@ pub(crate) mod tests {
fn test_total_subscriptions() {
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(100);
let bank = Bank::new_for_tests(&genesis_config);
let max_complete_transaction_status_slot = Arc::new(AtomicU64::default());
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(bank_forks));
let subscriptions = Arc::new(RpcSubscriptions::default_with_bank_forks(
max_complete_transaction_status_slot,
bank_forks,
));
let (rpc1, _receiver1) = rpc_pubsub_service::test_connection(&subscriptions);
let sub_id1 = rpc1