ledger-tool: verify: add --verify-slots and --verify-slots-details

This adds:

    --verify-slots <FILENAME>
        If the file does not exist, write the slot hashes to this file;
        if the file does exist, verify slot hashes against this file.

    --verify-slots-details none|bank|tx|bank-tx
	Store the bank (=accounts) and/or transaction batches in the
	json file. For each tx batch, the bank hash is included.

The first case can be used to dump a list of (slot, hash) to a json file
during a replay. The second case can be used to check slot hashes against
previously recorded values.

This is useful for debugging consensus failures, eg:

    # on good commit/branch
    ledger-tool verify --verify-slots good.json --verify-slots-details=tx

    # on bad commit or potentially consensus breaking branch
    ledger-tool verify --verify-slots good.json --verify-slots-details=tx

On a hash mismatch an error will be logged with the expected hash vs the
computed hash.

accounts-db/src/transaction_results.rs

@@ -74,7 +74,7 @@ impl TransactionExecutionResult {
     }
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct TransactionExecutionDetails {
     pub status: transaction::Result<()>,
     pub log_messages: Option<Vec<String>>,
@@ -87,7 +87,7 @@ pub struct TransactionExecutionDetails {
     pub accounts_data_len_delta: i64,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
 pub enum DurableNonceFee {
     Valid(u64),
     Invalid,

core/src/banking_stage/consumer.rs

@@ -1680,6 +1680,7 @@ mod tests {
             let (replay_vote_sender, _replay_vote_receiver) = unbounded();
             let committer = Committer::new(
                 Some(TransactionStatusSender {
+                    bank_hash: false,
                     sender: transaction_status_sender,
                 }),
                 replay_vote_sender,
@@ -1825,6 +1826,7 @@ mod tests {
             let (replay_vote_sender, _replay_vote_receiver) = unbounded();
             let committer = Committer::new(
                 Some(TransactionStatusSender {
+                    bank_hash: false,
                     sender: transaction_status_sender,
                 }),
                 replay_vote_sender,

core/src/validator.rs

@@ -2191,6 +2191,7 @@ fn initialize_rpc_transaction_history_services(
     let max_complete_transaction_status_slot = Arc::new(AtomicU64::new(blockstore.max_root()));
     let (transaction_status_sender, transaction_status_receiver) = unbounded();
     let transaction_status_sender = Some(TransactionStatusSender {
+        bank_hash: false,
         sender: transaction_status_sender,
     });
     let transaction_status_service = Some(TransactionStatusService::new(

ledger-tool/Cargo.toml

@@ -17,6 +17,7 @@ crossbeam-channel = { workspace = true }
 csv = { workspace = true }
 dashmap = { workspace = true }
 futures = { workspace = true }
+hex = { workspace = true }
 histogram = { workspace = true }
 itertools = { workspace = true }
 log = { workspace = true }

ledger-tool/src/ledger_utils.rs

@@ -114,6 +114,7 @@ pub fn load_and_process_ledger_or_exit(
     blockstore: Arc<Blockstore>,
     process_options: ProcessOptions,
     snapshot_archive_path: Option<PathBuf>,
+    transaction_status_sender: Option<TransactionStatusSender>,
     incremental_snapshot_archive_path: Option<PathBuf>,
 ) -> (Arc<RwLock<BankForks>>, Option<StartingSnapshotHashes>) {
     load_and_process_ledger(
@@ -122,6 +123,7 @@ pub fn load_and_process_ledger_or_exit(
         blockstore,
         process_options,
         snapshot_archive_path,
+        transaction_status_sender,
         incremental_snapshot_archive_path,
     )
     .unwrap_or_else(|err| {
@@ -130,12 +132,13 @@ pub fn load_and_process_ledger_or_exit(
     })
 }
 
-pub fn load_and_process_ledger(
+fn load_and_process_ledger(
     arg_matches: &ArgMatches,
     genesis_config: &GenesisConfig,
     blockstore: Arc<Blockstore>,
     process_options: ProcessOptions,
     snapshot_archive_path: Option<PathBuf>,
+    transaction_status_sender: Option<TransactionStatusSender>,
     incremental_snapshot_archive_path: Option<PathBuf>,
 ) -> Result<(Arc<RwLock<BankForks>>, Option<StartingSnapshotHashes>), LoadAndProcessLedgerError> {
     let bank_snapshots_dir = if blockstore.is_primary_access() {
@@ -397,12 +400,13 @@ pub fn load_and_process_ledger(
             );
             (
                 Some(TransactionStatusSender {
+                    bank_hash: false,
                     sender: transaction_status_sender,
                 }),
                 Some(transaction_status_service),
             )
         } else {
-            (None, None)
+            (transaction_status_sender, None)
         };
 
     let result = blockstore_processor::process_blockstore_from_root(
@@ -423,6 +427,8 @@ pub fn load_and_process_ledger(
     accounts_hash_verifier.join().unwrap();
     if let Some(service) = transaction_status_service {
         service.join().unwrap();
+    } else if let Some(transaction_status_sender) = transaction_status_sender {
+        drop(transaction_status_sender.sender);
     }
 
     result