ledger-tool: Notify geyser of transactions (#29933)

ledger-tool: Notify geyser of transactions

Cargo.lock

@@ -5784,6 +5784,7 @@ dependencies = [
  "solana-ledger",
  "solana-logger 1.16.0",
  "solana-measure",
+ "solana-rpc",
  "solana-runtime",
  "solana-sdk 1.16.0",
  "solana-stake-program",

ledger-tool/Cargo.toml

@@ -33,6 +33,7 @@ solana-geyser-plugin-manager = { path = "../geyser-plugin-manager", version = "=
 solana-ledger = { path = "../ledger", version = "=1.16.0" }
 solana-logger = { path = "../logger", version = "=1.16.0" }
 solana-measure = { path = "../measure", version = "=1.16.0" }
+solana-rpc = { path = "../rpc", version = "=1.16.0" }
 solana-runtime = { path = "../runtime", version = "=1.16.0" }
 solana-sdk = { path = "../sdk", version = "=1.16.0" }
 solana-stake-program = { path = "../programs/stake", version = "=1.16.0" }

ledger-tool/src/main.rs

@@ -36,10 +36,16 @@ use {
             AccessType, BlockstoreOptions, BlockstoreRecoveryMode, LedgerColumnOptions,
             ShredStorageType, BLOCKSTORE_DIRECTORY_ROCKS_FIFO,
         },
-        blockstore_processor::{self, BlockstoreProcessorError, ProcessOptions},
+        blockstore_processor::{
+            self, BlockstoreProcessorError, ProcessOptions, TransactionStatusSender,
+        },
         shred::Shred,
     },
     solana_measure::{measure, measure::Measure},
+    solana_rpc::{
+        transaction_notifier_interface::TransactionNotifierLock,
+        transaction_status_service::TransactionStatusService,
+    },
     solana_runtime::{
         accounts::Accounts,
         accounts_background_service::{
@@ -1078,7 +1084,7 @@ fn get_accounts_db_config(ledger_path: &Path, arg_matches: &ArgMatches<'_>) -> A
 fn load_bank_forks(
     arg_matches: &ArgMatches,
     genesis_config: &GenesisConfig,
-    blockstore: &Blockstore,
+    blockstore: Arc<Blockstore>,
     process_options: ProcessOptions,
     snapshot_archive_path: Option<PathBuf>,
     incremental_snapshot_archive_path: Option<PathBuf>,
@@ -1188,6 +1194,7 @@ fn load_bank_forks(
     info!("done. {}", measure);
 
     let mut accounts_update_notifier = Option::<AccountsUpdateNotifier>::default();
+    let mut transaction_notifier = Option::<TransactionNotifierLock>::default();
     if arg_matches.is_present("geyser_plugin_config") {
         let geyser_config_files = values_t_or_exit!(arg_matches, "geyser_plugin_config", String)
             .into_iter()
@@ -1204,12 +1211,13 @@ fn load_bank_forks(
                 },
             );
         accounts_update_notifier = geyser_service.get_accounts_update_notifier();
+        transaction_notifier = geyser_service.get_transaction_notifier();
     }
 
     let (bank_forks, leader_schedule_cache, starting_snapshot_hashes, ..) =
         bank_forks_utils::load_bank_forks(
             genesis_config,
-            blockstore,
+            blockstore.as_ref(),
             account_paths,
             None,
             snapshot_config.as_ref(),
@@ -1246,12 +1254,34 @@ fn load_bank_forks(
         None,
     );
 
+    let (transaction_status_sender, transaction_status_service) = if transaction_notifier.is_some()
+    {
+        let (transaction_status_sender, transaction_status_receiver) = unbounded();
+        let transaction_status_service = TransactionStatusService::new(
+            transaction_status_receiver,
+            Arc::default(),
+            false,
+            transaction_notifier,
+            blockstore.clone(),
+            false,
+            &exit,
+        );
+        (
+            Some(TransactionStatusSender {
+                sender: transaction_status_sender,
+            }),
+            Some(transaction_status_service),
+        )
+    } else {
+        (None, None)
+    };
+
     let result = blockstore_processor::process_blockstore_from_root(
-        blockstore,
+        blockstore.as_ref(),
         &bank_forks,
         &leader_schedule_cache,
         &process_options,
-        None,
+        transaction_status_sender.as_ref(),
         None,
         &accounts_background_request_sender,
     )
@@ -1259,6 +1289,9 @@ fn load_bank_forks(
 
     exit.store(true, Ordering::Relaxed);
     accounts_background_service.join().unwrap();
+    if let Some(service) = transaction_status_service {
+        service.join().unwrap();
+    }
 
     result
 }
@@ -2475,7 +2508,7 @@ fn main() {
                 match load_bank_forks(
                     arg_matches,
                     &genesis_config,
-                    &blockstore,
+                    Arc::new(blockstore),
                     process_options,
                     snapshot_archive_path,
                     incremental_snapshot_archive_path,
@@ -2566,7 +2599,7 @@ fn main() {
                 match load_bank_forks(
                     arg_matches,
                     &genesis_config,
-                    &blockstore,
+                    Arc::new(blockstore),
                     process_options,
                     snapshot_archive_path,
                     incremental_snapshot_archive_path,
@@ -2801,7 +2834,7 @@ fn main() {
                 let (bank_forks, ..) = load_bank_forks(
                     arg_matches,
                     &genesis_config,
-                    &blockstore,
+                    Arc::new(blockstore),
                     process_options,
                     snapshot_archive_path,
                     incremental_snapshot_archive_path,
@@ -2844,7 +2877,7 @@ fn main() {
                 match load_bank_forks(
                     arg_matches,
                     &open_genesis_config_by(&ledger_path, arg_matches),
-                    &blockstore,
+                    Arc::new(blockstore),
                     process_options,
                     snapshot_archive_path,
                     incremental_snapshot_archive_path,
@@ -2948,12 +2981,12 @@ fn main() {
                     usize
                 );
                 let genesis_config = open_genesis_config_by(&ledger_path, arg_matches);
-                let blockstore = open_blockstore(
+                let blockstore = Arc::new(open_blockstore(
                     &ledger_path,
                     AccessType::Secondary,
                     wal_recovery_mode,
                     force_update_to_open,
-                );
+                ));
 
                 let snapshot_slot = if Some("ROOT") == arg_matches.value_of("snapshot_slot") {
                     blockstore
@@ -3009,7 +3042,7 @@ fn main() {
                 match load_bank_forks(
                     arg_matches,
                     &genesis_config,
-                    &blockstore,
+                    blockstore.clone(),
                     ProcessOptions {
                         new_hard_forks,
                         halt_at_slot: Some(snapshot_slot),
@@ -3351,7 +3384,7 @@ fn main() {
                 let (bank_forks, ..) = load_bank_forks(
                     arg_matches,
                     &genesis_config,
-                    &blockstore,
+                    Arc::new(blockstore),
                     process_options,
                     snapshot_archive_path,
                     incremental_snapshot_archive_path,
@@ -3439,7 +3472,7 @@ fn main() {
                 match load_bank_forks(
                     arg_matches,
                     &genesis_config,
-                    &blockstore,
+                    Arc::new(blockstore),
                     process_options,
                     snapshot_archive_path,
                     incremental_snapshot_archive_path,