Rpc: enable getConfirmedBlock and getConfirmedTransaction to return c…

…onfirmed (not yet finalized) data (bp #16142) (#16160)

* Rpc: enable getConfirmedBlock and getConfirmedTransaction to return confirmed (not yet finalized) data (#16142)

* Add Blockstore block and tx apis that allow unrooted responses

* Add TransactionStatusMessage, and send on bank freeze; also refactor TransactionStatusSender

* Track highest slot with tx-status writes complete

* Rename and unpub fn

* Add commitment to GetConfirmed input configs

* Support confirmed blocks in getConfirmedBlock

* Support confirmed txs in getConfirmedTransaction

* Update sigs-for-addr2 comment

* Enable confirmed block in cli

* Enable confirmed transaction in cli

* Review comments

* Rename blockstore method

(cherry picked from commit 433f1ea)

# Conflicts:
#	core/src/replay_stage.rs

* Fix conflict

Co-authored-by: Tyera Eulberg <teulberg@gmail.com>
Co-authored-by: Tyera Eulberg <tyera@solana.com>

cli/src/cli.rs

@@ -26,7 +26,10 @@ use solana_client::{
     client_error::{ClientError, ClientErrorKind, Result as ClientResult},
     nonce_utils,
     rpc_client::RpcClient,
-    rpc_config::{RpcLargestAccountsFilter, RpcSendTransactionConfig, RpcTransactionLogsFilter},
+    rpc_config::{
+        RpcConfirmedTransactionConfig, RpcLargestAccountsFilter, RpcSendTransactionConfig,
+        RpcTransactionLogsFilter,
+    },
     rpc_response::RpcKeyedAccount,
 };
 #[cfg(not(test))]
@@ -1027,9 +1030,13 @@ fn process_confirm(
                 let mut transaction = None;
                 let mut get_transaction_error = None;
                 if config.verbose {
-                    match rpc_client
-                        .get_confirmed_transaction(signature, UiTransactionEncoding::Base64)
-                    {
+                    match rpc_client.get_confirmed_transaction_with_config(
+                        signature,
+                        RpcConfirmedTransactionConfig {
+                            encoding: Some(UiTransactionEncoding::Base64),
+                            commitment: Some(CommitmentConfig::confirmed()),
+                        },
+                    ) {
                         Ok(confirmed_transaction) => {
                             let decoded_transaction = confirmed_transaction
                                 .transaction

cli/src/cluster_query.rs

@@ -24,8 +24,9 @@ use solana_client::{
     pubsub_client::PubsubClient,
     rpc_client::{GetConfirmedSignaturesForAddress2Config, RpcClient},
     rpc_config::{
-        RpcAccountInfoConfig, RpcLargestAccountsConfig, RpcLargestAccountsFilter,
-        RpcProgramAccountsConfig, RpcTransactionLogsConfig, RpcTransactionLogsFilter,
+        RpcAccountInfoConfig, RpcConfirmedBlockConfig, RpcLargestAccountsConfig,
+        RpcLargestAccountsFilter, RpcProgramAccountsConfig, RpcTransactionLogsConfig,
+        RpcTransactionLogsFilter,
     },
     rpc_filter,
     rpc_response::SlotInfo,
@@ -963,8 +964,16 @@ pub fn process_get_block(
         rpc_client.get_slot_with_commitment(CommitmentConfig::finalized())?
     };
 
-    let encoded_confirmed_block =
-        rpc_client.get_confirmed_block_with_encoding(slot, UiTransactionEncoding::Base64)?;
+    let encoded_confirmed_block = rpc_client
+        .get_confirmed_block_with_config(
+            slot,
+            RpcConfirmedBlockConfig {
+                encoding: Some(UiTransactionEncoding::Base64),
+                commitment: Some(CommitmentConfig::confirmed()),
+                ..RpcConfirmedBlockConfig::default()
+            },
+        )?
+        .into();
     let cli_block = CliBlock {
         encoded_confirmed_block,
         slot,

cli/src/stake.rs

@@ -1675,7 +1675,7 @@ pub(crate) fn fetch_epoch_rewards(
             .get(0)
             .ok_or_else(|| format!("Unable to fetch first confirmed block for epoch {}", epoch))?;
 
-        let first_confirmed_block = match rpc_client.get_configured_confirmed_block(
+        let first_confirmed_block = match rpc_client.get_confirmed_block_with_config(
             first_confirmed_block_in_epoch,
             RpcConfirmedBlockConfig::rewards_only(),
         ) {

client/src/rpc_client.rs

@@ -4,9 +4,10 @@ use crate::{
     mock_sender::{MockSender, Mocks},
     rpc_config::RpcAccountInfoConfig,
     rpc_config::{
-        RpcConfirmedBlockConfig, RpcGetConfirmedSignaturesForAddress2Config,
-        RpcLargestAccountsConfig, RpcProgramAccountsConfig, RpcSendTransactionConfig,
-        RpcSimulateTransactionConfig, RpcTokenAccountsFilter,
+        RpcConfirmedBlockConfig, RpcConfirmedTransactionConfig,
+        RpcGetConfirmedSignaturesForAddress2Config, RpcLargestAccountsConfig,
+        RpcProgramAccountsConfig, RpcSendTransactionConfig, RpcSimulateTransactionConfig,
+        RpcTokenAccountsFilter,
     },
     rpc_request::{RpcError, RpcRequest, RpcResponseErrorData, TokenAccountsFilter},
     rpc_response::*,
@@ -527,7 +528,7 @@ impl RpcClient {
         self.send(RpcRequest::GetConfirmedBlock, json!([slot, encoding]))
     }
 
-    pub fn get_configured_confirmed_block(
+    pub fn get_confirmed_block_with_config(
         &self,
         slot: Slot,
         config: RpcConfirmedBlockConfig,
@@ -619,6 +620,17 @@ impl RpcClient {
         )
     }
 
+    pub fn get_confirmed_transaction_with_config(
+        &self,
+        signature: &Signature,
+        config: RpcConfirmedTransactionConfig,
+    ) -> ClientResult<EncodedConfirmedTransaction> {
+        self.send(
+            RpcRequest::GetConfirmedTransaction,
+            json!([signature.to_string(), config]),
+        )
+    }
+
     pub fn get_block_time(&self, slot: Slot) -> ClientResult<UnixTimestamp> {
         let request = RpcRequest::GetBlockTime;
         let response = self.sender.send(request, json!([slot]));

client/src/rpc_config.rs

@@ -135,6 +135,8 @@ pub struct RpcConfirmedBlockConfig {
     pub encoding: Option<UiTransactionEncoding>,
     pub transaction_details: Option<TransactionDetails>,
     pub rewards: Option<bool>,
+    #[serde(flatten)]
+    pub commitment: Option<CommitmentConfig>,
 }
 
 impl EncodingConfig for RpcConfirmedBlockConfig {
@@ -159,12 +161,15 @@ impl RpcConfirmedBlockConfig {
 #[serde(rename_all = "camelCase")]
 pub struct RpcConfirmedTransactionConfig {
     pub encoding: Option<UiTransactionEncoding>,
+    #[serde(flatten)]
+    pub commitment: Option<CommitmentConfig>,
 }
 
 impl EncodingConfig for RpcConfirmedTransactionConfig {
     fn new_with_encoding(encoding: &Option<UiTransactionEncoding>) -> Self {
         Self {
             encoding: *encoding,
+            ..Self::default()
         }
     }
 }

core/src/banking_stage.rs

@@ -12,10 +12,8 @@ use itertools::Itertools;
 use lru::LruCache;
 use retain_mut::RetainMut;
 use solana_ledger::{
-    blockstore::Blockstore,
-    blockstore_processor::{send_transaction_status_batch, TransactionStatusSender},
-    entry::hash_transactions,
-    leader_schedule_cache::LeaderScheduleCache,
+    blockstore::Blockstore, blockstore_processor::TransactionStatusSender,
+    entry::hash_transactions, leader_schedule_cache::LeaderScheduleCache,
 };
 use solana_measure::{measure::Measure, thread_mem_usage};
 use solana_metrics::{inc_new_counter_debug, inc_new_counter_info};
@@ -765,7 +763,7 @@ impl BankingStage {
             if let Some(transaction_status_sender) = transaction_status_sender {
                 let post_balances = bank.collect_balances(batch);
                 let post_token_balances = collect_token_balances(&bank, &batch, &mut mint_decimals);
-                send_transaction_status_batch(
+                transaction_status_sender.send_transaction_status_batch(
                     bank.clone(),
                     batch.transactions(),
                     batch.iteration_order_vec(),
@@ -774,7 +772,6 @@ impl BankingStage {
                     TransactionTokenBalancesSet::new(pre_token_balances, post_token_balances),
                     inner_instructions,
                     transaction_logs,
-                    transaction_status_sender,
                 );
             }
         }
@@ -2346,6 +2343,7 @@ mod tests {
             let (transaction_status_sender, transaction_status_receiver) = unbounded();
             let transaction_status_service = TransactionStatusService::new(
                 transaction_status_receiver,
+                Arc::new(AtomicU64::default()),
                 blockstore.clone(),
                 &Arc::new(AtomicBool::new(false)),
             );
@@ -2366,7 +2364,7 @@ mod tests {
 
             transaction_status_service.join().unwrap();
 
-            let confirmed_block = blockstore.get_confirmed_block(bank.slot(), false).unwrap();
+            let confirmed_block = blockstore.get_rooted_block(bank.slot(), false).unwrap();
             assert_eq!(confirmed_block.transactions.len(), 3);
 
             for TransactionWithStatusMeta { transaction, meta } in

core/src/replay_stage.rs

@@ -1433,6 +1433,9 @@ impl ReplayStage {
                     );
                     did_complete_bank = true;
                     info!("bank frozen: {}", bank.slot());
+                    if let Some(transaction_status_sender) = transaction_status_sender.clone() {
+                        transaction_status_sender.send_transaction_status_freeze_message(&bank);
+                    }
                     bank.freeze();
                     heaviest_subtree_fork_choice
                         .add_new_leaf_slot(bank.slot(), Some(bank.parent_slot()));
@@ -2110,7 +2113,7 @@ pub(crate) mod tests {
     use std::{
         fs::remove_dir_all,
         iter,
-        sync::{Arc, RwLock},
+        sync::{atomic::AtomicU64, Arc, RwLock},
     };
     use trees::tr;
 
@@ -2764,6 +2767,7 @@ pub(crate) mod tests {
         previous_slot: Slot,
         bank: Arc<Bank>,
         blockstore: Arc<Blockstore>,
+        max_complete_transaction_status_slot: Arc<AtomicU64>,
     ) -> Vec<Signature> {
         let mint_keypair = keypairs[0];
         let keypair1 = keypairs[1];
@@ -2797,12 +2801,13 @@ pub(crate) mod tests {
         let (replay_vote_sender, _replay_vote_receiver) = unbounded();
         let transaction_status_service = TransactionStatusService::new(
             transaction_status_receiver,
+            max_complete_transaction_status_slot,
             blockstore,
             &Arc::new(AtomicBool::new(false)),
         );
 
         // Check that process_entries successfully writes can_commit transactions statuses, and
-        // that they are matched properly by get_confirmed_block
+        // that they are matched properly by get_rooted_block
         let _result = blockstore_processor::process_entries(
             &bank,
             &entries,
@@ -2849,9 +2854,10 @@ pub(crate) mod tests {
                 bank0.slot(),
                 bank1,
                 blockstore.clone(),
+                Arc::new(AtomicU64::default()),
             );
 
-            let confirmed_block = blockstore.get_confirmed_block(slot, false).unwrap();
+            let confirmed_block = blockstore.get_rooted_block(slot, false).unwrap();
             assert_eq!(confirmed_block.transactions.len(), 3);
 
             for TransactionWithStatusMeta { transaction, meta } in