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

v1.9: Enforce tx metadata upload to bigtable (#23212)

Browse Source 
* Enforce tx metadata upload with static types (#23028)

* resolve conflicts

* fix test
This commit is contained in:
Justin Starry
2022-02-23 12:03:17 +08:00
committed by GitHub
parent 8413700a2f
commit 450404f800
16 changed files with 367 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files

109
rpc/src/rpc.rs
View File

@ -76,9 +76,9 @@ use {
solana_storage_bigtable::Error as StorageError,
solana_streamer::socket::SocketAddrSpace,
solana_transaction_status::{
ConfirmedBlock, ConfirmedTransactionStatusWithSignature, EncodedConfirmedTransaction,
Reward, RewardType, TransactionConfirmationStatus, TransactionStatus, UiConfirmedBlock,
UiTransactionEncoding,
ConfirmedBlockWithOptionalMetadata, ConfirmedTransactionStatusWithSignature,
ConfirmedTransactionWithOptionalMetadata, EncodedConfirmedTransaction, Reward, RewardType,
TransactionConfirmationStatus, TransactionStatus, UiConfirmedBlock, UiTransactionEncoding,
},
solana_vote_program::vote_state::{VoteState, MAX_LOCKOUT_HISTORY},
spl_token::{
@ -914,12 +914,8 @@ impl JsonRpcRequestProcessor {
&self,
result: &std::result::Result<T, BlockstoreError>,
slot: Slot,
) -> Result<()>
where
T: std::fmt::Debug,
{
if result.is_err() {
let err = result.as_ref().unwrap_err();
) -> Result<()> {
if let Err(err) = result {
debug!(
"check_blockstore_root, slot: {:?}, max root: {:?}, err: {:?}",
slot,
@ -940,21 +936,16 @@ impl JsonRpcRequestProcessor {
&self,
result: &std::result::Result<T, BlockstoreError>,
slot: Slot,
) -> Result<()>
where
T: std::fmt::Debug,
{
if result.is_err() {
if let BlockstoreError::SlotCleanedUp = result.as_ref().unwrap_err() {
return Err(RpcCustomError::BlockCleanedUp {
slot,
first_available_block: self
.blockstore
.get_first_available_block()
.unwrap_or_default(),
}
.into());
) -> Result<()> {
if let Err(BlockstoreError::SlotCleanedUp) = result {
return Err(RpcCustomError::BlockCleanedUp {
slot,
first_available_block: self
.blockstore
.get_first_available_block()
.unwrap_or_default(),
}
.into());
}
Ok(())
}
@ -962,15 +953,9 @@ impl JsonRpcRequestProcessor {
fn check_bigtable_result<T>(
&self,
result: &std::result::Result<T, solana_storage_bigtable::Error>,
) -> Result<()>
where
T: std::fmt::Debug,
{
if result.is_err() {
let err = result.as_ref().unwrap_err();
if let solana_storage_bigtable::Error::BlockNotFound(slot) = err {
return Err(RpcCustomError::LongTermStorageSlotSkipped { slot: *slot }.into());
}
) -> Result<()> {
if let Err(solana_storage_bigtable::Error::BlockNotFound(slot)) = result {
return Err(RpcCustomError::LongTermStorageSlotSkipped { slot: *slot }.into());
}
Ok(())
}
@ -1013,7 +998,7 @@ impl JsonRpcRequestProcessor {
self.check_status_is_complete(slot)?;
let result = self.blockstore.get_rooted_block(slot, true);
self.check_blockstore_root(&result, slot)?;
let configure_block = |confirmed_block: ConfirmedBlock| {
let configure_block = |confirmed_block: ConfirmedBlockWithOptionalMetadata| {
let mut confirmed_block =
confirmed_block.configure(encoding, transaction_details, show_rewards);
if slot == 0 {
@ -1031,7 +1016,10 @@ impl JsonRpcRequestProcessor {
}
}
self.check_slot_cleaned_up(&result, slot)?;
return Ok(result.ok().map(configure_block));
return Ok(result
.ok()
.map(ConfirmedBlockWithOptionalMetadata::from)
.map(configure_block));
} else if commitment.is_confirmed() {
// Check if block is confirmed
let confirmed_bank = self.bank(Some(CommitmentConfig::confirmed()));
@ -1053,7 +1041,11 @@ impl JsonRpcRequestProcessor {
}
}
}
confirmed_block.configure(encoding, transaction_details, show_rewards)
ConfirmedBlockWithOptionalMetadata::from(confirmed_block).configure(
encoding,
transaction_details,
show_rewards,
)
}));
}
}
@ -1399,7 +1391,10 @@ impl JsonRpcRequestProcessor {
.get(confirmed_transaction.slot)
.map(|bank| bank.clock().unix_timestamp);
}
return Ok(Some(confirmed_transaction.encode(encoding)));
return Ok(Some(
ConfirmedTransactionWithOptionalMetadata::from(confirmed_transaction)
.encode(encoding),
));
}
if confirmed_transaction.slot
<= self
@ -1408,7 +1403,10 @@ impl JsonRpcRequestProcessor {
.unwrap()
.highest_confirmed_root()
{
return Ok(Some(confirmed_transaction.encode(encoding)));
return Ok(Some(
ConfirmedTransactionWithOptionalMetadata::from(confirmed_transaction)
.encode(encoding),
));
}
}
None => {
@ -4315,15 +4313,7 @@ pub fn create_test_transactions_and_populate_blockstore(
);
let ix_error_signature = ix_error_tx.signatures[0];
let entry_2 = solana_entry::entry::next_entry(&entry_1.hash, 1, vec![ix_error_tx]);
// Failed transaction
let fail_tx = solana_sdk::system_transaction::transfer(
mint_keypair,
&keypair2.pubkey(),
rent_exempt_amount,
Hash::default(),
);
let entry_3 = solana_entry::entry::next_entry(&entry_2.hash, 1, vec![fail_tx]);
let entries = vec![entry_1, entry_2, entry_3];
let entries = vec![entry_1, entry_2];
let shreds = solana_ledger::blockstore::entries_to_test_shreds(
entries.clone(),
@ -4349,17 +4339,20 @@ pub fn create_test_transactions_and_populate_blockstore(
// Check that process_entries successfully writes can_commit transactions statuses, and
// that they are matched properly by get_rooted_block
let _result = solana_ledger::blockstore_processor::process_entries_for_tests(
&bank,
entries,
true,
Some(
&solana_ledger::blockstore_processor::TransactionStatusSender {
sender: transaction_status_sender,
enable_cpi_and_log_storage: false,
},
assert_eq!(
solana_ledger::blockstore_processor::process_entries_for_tests(
&bank,
entries,
true,
Some(
&solana_ledger::blockstore_processor::TransactionStatusSender {
sender: transaction_status_sender,
enable_cpi_and_log_storage: false,
},
),
Some(&replay_vote_sender),
),
Some(&replay_vote_sender),
Ok(())
);
transaction_status_service.join().unwrap();
@ -6605,7 +6598,7 @@ pub mod tests {
let confirmed_block: Option<EncodedConfirmedBlock> =
serde_json::from_value(result["result"].clone()).unwrap();
let confirmed_block = confirmed_block.unwrap();
assert_eq!(confirmed_block.transactions.len(), 3);
assert_eq!(confirmed_block.transactions.len(), 2);
assert_eq!(confirmed_block.rewards, vec![]);
for EncodedTransactionWithStatusMeta { transaction, meta } in
@ -6650,7 +6643,7 @@ pub mod tests {
let confirmed_block: Option<EncodedConfirmedBlock> =
serde_json::from_value(result["result"].clone()).unwrap();
let confirmed_block = confirmed_block.unwrap();
assert_eq!(confirmed_block.transactions.len(), 3);
assert_eq!(confirmed_block.transactions.len(), 2);
assert_eq!(confirmed_block.rewards, vec![]);
for EncodedTransactionWithStatusMeta { transaction, meta } in

75
rpc/src/rpc_subscriptions.rs
View File

@ -39,7 +39,7 @@ use {
timing::timestamp,
transaction,
},
solana_transaction_status::ConfirmedBlock,
solana_transaction_status::{ConfirmedBlock, ConfirmedBlockWithOptionalMetadata},
solana_vote_program::vote_state::Vote,
std::{
cell::RefCell,
@ -278,30 +278,26 @@ impl RpcNotifier {
}
fn filter_block_result_txs(
block: ConfirmedBlock,
mut block: ConfirmedBlock,
last_modified_slot: Slot,
params: &BlockSubscriptionParams,
) -> Option<RpcBlockUpdate> {
let transactions = match params.kind {
block.transactions = match params.kind {
BlockSubscriptionKind::All => block.transactions,
BlockSubscriptionKind::MentionsAccountOrProgram(pk) => block
.transactions
.into_iter()
.filter(|tx| tx.transaction.message.account_keys.contains(&pk))
.filter(|tx_with_meta| tx_with_meta.transaction.message.account_keys.contains(&pk))
.collect(),
};
if transactions.is_empty() {
if block.transactions.is_empty() {
if let BlockSubscriptionKind::MentionsAccountOrProgram(_) = params.kind {
return None;
}
}
let block = ConfirmedBlock {
transactions,
..block
}
.configure(
let block = ConfirmedBlockWithOptionalMetadata::from(block).configure(
params.encoding,
params.transaction_details,
params.show_rewards,
@ -964,12 +960,13 @@ impl RpcSubscriptions {
}
match blockstore.get_complete_block(s, false) {
Ok(block) => {
if let Some(res) = filter_block_result_txs(block, s, params)
if let Some(block_update) =
filter_block_result_txs(block, s, params)
{
notifier.notify(
Response {
context: RpcResponseContext { slot: s },
value: res,
value: block_update,
},
subscription,
false,
@ -1346,8 +1343,13 @@ pub(crate) mod tests {
#[serial]
fn test_check_confirmed_block_subscribe() {
let exit = Arc::new(AtomicBool::new(false));
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(10_000);
let GenesisConfigInfo {
genesis_config,
mint_keypair,
..
} = create_genesis_config(10_000);
let bank = Bank::new_for_tests(&genesis_config);
let rent_exempt_amount = bank.get_minimum_balance_for_rent_exemption(0);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
@ -1388,10 +1390,11 @@ pub(crate) mod tests {
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()));
bank.transfer(rent_exempt_amount, &mint_keypair, &keypair2.pubkey())
.unwrap();
let _confirmed_block_signatures = create_test_transactions_and_populate_blockstore(
vec![&keypair1, &keypair2, &keypair3, &keypair4],
vec![&mint_keypair, &keypair1, &keypair2, &keypair3],
0,
bank,
blockstore.clone(),
@ -1404,7 +1407,11 @@ pub(crate) mod tests {
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 block = ConfirmedBlockWithOptionalMetadata::from(block).configure(
params.encoding,
params.transaction_details,
false,
);
let expected_resp = RpcBlockUpdate {
slot,
block: Some(block),
@ -1443,8 +1450,13 @@ pub(crate) mod tests {
#[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 GenesisConfigInfo {
genesis_config,
mint_keypair,
..
} = create_genesis_config(10_000);
let bank = Bank::new_for_tests(&genesis_config);
let rent_exempt_amount = bank.get_minimum_balance_for_rent_exemption(0);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
@ -1486,10 +1498,11 @@ pub(crate) mod tests {
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()));
bank.transfer(rent_exempt_amount, &mint_keypair, &keypair2.pubkey())
.unwrap();
let _confirmed_block_signatures = create_test_transactions_and_populate_blockstore(
vec![&keypair1, &keypair2, &keypair3, &keypair4],
vec![&mint_keypair, &keypair1, &keypair2, &keypair3],
0,
bank,
blockstore.clone(),
@ -1509,7 +1522,11 @@ pub(crate) mod tests {
.account_keys
.contains(&keypair1.pubkey())
});
let block = block.configure(params.encoding, params.transaction_details, false);
let block = ConfirmedBlockWithOptionalMetadata::from(block).configure(
params.encoding,
params.transaction_details,
false,
);
let expected_resp = RpcBlockUpdate {
slot,
block: Some(block),
@ -1538,8 +1555,13 @@ pub(crate) mod tests {
#[serial]
fn test_check_finalized_block_subscribe() {
let exit = Arc::new(AtomicBool::new(false));
let GenesisConfigInfo { genesis_config, .. } = create_genesis_config(10_000);
let GenesisConfigInfo {
genesis_config,
mint_keypair,
..
} = create_genesis_config(10_000);
let bank = Bank::new_for_tests(&genesis_config);
let rent_exempt_amount = bank.get_minimum_balance_for_rent_exemption(0);
let bank_forks = Arc::new(RwLock::new(BankForks::new(bank)));
let optimistically_confirmed_bank =
OptimisticallyConfirmedBank::locked_from_bank_forks_root(&bank_forks);
@ -1579,10 +1601,11 @@ pub(crate) mod tests {
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()));
bank.transfer(rent_exempt_amount, &mint_keypair, &keypair2.pubkey())
.unwrap();
let _confirmed_block_signatures = create_test_transactions_and_populate_blockstore(
vec![&keypair1, &keypair2, &keypair3, &keypair4],
vec![&mint_keypair, &keypair1, &keypair2, &keypair3],
0,
bank,
blockstore.clone(),
@ -1600,7 +1623,11 @@ pub(crate) mod tests {
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 block = ConfirmedBlockWithOptionalMetadata::from(block).configure(
params.encoding,
params.transaction_details,
false,
);
let expected_resp = RpcBlockUpdate {
slot,
block: Some(block),