build(deps): bump memmap2 from 0.9.8 to 0.9.9

Cargo.toml

@@ -317,7 +317,7 @@ light-poseidon-0-2 = { package = "light-poseidon", version = "0.2.0" }
 log = "0.4.28"
 lru = "0.7.7"
 lz4 = "1.28.1"
-memmap2 = "0.9.8"
+memmap2 = "0.9.9"
 memoffset = "0.9"
 merlin = { version = "3", default-features = false }
 min-max-heap = "1.3.0"

core/src/completed_data_sets_service.rs

@@ -7,13 +7,25 @@
 use {
     crossbeam_channel::{Receiver, RecvTimeoutError, Sender},
     solana_entry::entry::Entry,
-    solana_ledger::blockstore::{Blockstore, CompletedDataSetInfo},
+    solana_ledger::{
+        blockstore::{Blockstore, CompletedDataSetInfo},
+        deshred_transaction_notifier_interface::DeshredTransactionNotifierArc,
+    },
+    solana_measure::measure::Measure,
+    solana_message::{v0::LoadedAddresses, VersionedMessage},
+    solana_metrics::*,
     solana_rpc::{max_slots::MaxSlots, rpc_subscriptions::RpcSubscriptions},
+    solana_runtime::bank_forks::BankForks,
     solana_signature::Signature,
+    solana_svm_transaction::message_address_table_lookup::SVMMessageAddressTableLookup,
+    solana_transaction::{
+        simple_vote_transaction_checker::is_simple_vote_transaction_impl,
+        versioned::VersionedTransaction,
+    },
     std::{
         sync::{
             atomic::{AtomicBool, Ordering},
-            Arc,
+            Arc, RwLock,
         },
         thread::{self, Builder, JoinHandle},
         time::Duration,
@@ -23,6 +35,46 @@ use {
 pub type CompletedDataSetsReceiver = Receiver<Vec<CompletedDataSetInfo>>;
 pub type CompletedDataSetsSender = Sender<Vec<CompletedDataSetInfo>>;
 
+/// Check if a versioned transaction is a simple vote transaction.
+/// This avoids cloning by extracting the required data directly.
+fn is_simple_vote_transaction(tx: &VersionedTransaction) -> bool {
+    let is_legacy = matches!(&tx.message, VersionedMessage::Legacy(_));
+    let (account_keys, instructions) = match &tx.message {
+        VersionedMessage::Legacy(msg) => (&msg.account_keys[..], &msg.instructions[..]),
+        VersionedMessage::V0(msg) => (&msg.account_keys[..], &msg.instructions[..]),
+    };
+    let instruction_programs = instructions
+        .iter()
+        .filter_map(|ix| account_keys.get(ix.program_id_index as usize));
+    is_simple_vote_transaction_impl(&tx.signatures, is_legacy, instruction_programs)
+}
+
+/// Load addresses from address lookup tables for a versioned transaction.
+/// Returns None for legacy transactions or if address resolution fails.
+/// Takes a Bank reference to avoid repeated lock acquisition.
+fn load_transaction_addresses(
+    tx: &VersionedTransaction,
+    bank: &solana_runtime::bank::Bank,
+) -> Option<LoadedAddresses> {
+    let VersionedMessage::V0(message) = &tx.message else {
+        // Legacy transactions don't have address table lookups
+        return None;
+    };
+
+    if message.address_table_lookups.is_empty() {
+        return None;
+    }
+
+    bank.load_addresses_from_ref(
+        message
+            .address_table_lookups
+            .iter()
+            .map(SVMMessageAddressTableLookup::from),
+    )
+    .ok()
+    .map(|(addresses, _deactivation_slot)| addresses)
+}
+
 pub struct CompletedDataSetsService {
     thread_hdl: JoinHandle<()>,
 }
@@ -32,8 +84,10 @@ impl CompletedDataSetsService {
         completed_sets_receiver: CompletedDataSetsReceiver,
         blockstore: Arc<Blockstore>,
         rpc_subscriptions: Arc<RpcSubscriptions>,
+        deshred_transaction_notifier: Option<DeshredTransactionNotifierArc>,
         exit: Arc<AtomicBool>,
         max_slots: Arc<MaxSlots>,
+        bank_forks: Arc<RwLock<BankForks>>,
     ) -> Self {
         let thread_hdl = Builder::new()
             .name("solComplDataSet".to_string())
@@ -47,7 +101,9 @@ impl CompletedDataSetsService {
                         &completed_sets_receiver,
                         &blockstore,
                         &rpc_subscriptions,
+                        &deshred_transaction_notifier,
                         &max_slots,
+                        &bank_forks,
                     ) {
                         break;
                     }
@@ -62,13 +118,78 @@ impl CompletedDataSetsService {
         completed_sets_receiver: &CompletedDataSetsReceiver,
         blockstore: &Blockstore,
         rpc_subscriptions: &RpcSubscriptions,
+        deshred_transaction_notifier: &Option<DeshredTransactionNotifierArc>,
         max_slots: &Arc<MaxSlots>,
+        bank_forks: &RwLock<BankForks>,
     ) -> Result<(), RecvTimeoutError> {
         const RECV_TIMEOUT: Duration = Duration::from_secs(1);
-        let handle_completed_data_set_info = |completed_data_set_info| {
+
+        let mut batch_measure = Measure::start("deshred_geyser_batch");
+
+        // Get root bank once per batch to minimize lock contention
+        let root_bank = deshred_transaction_notifier
+            .as_ref()
+            .and_then(|_| bank_forks.read().ok())
+            .map(|forks| forks.root_bank());
+
+        // Metrics accumulators
+        let mut total_lut_load_us: u64 = 0;
+        let mut total_notify_us: u64 = 0;
+        let mut total_transactions: u64 = 0;
+        let mut total_entries: u64 = 0;
+        let mut total_data_sets: u64 = 0;
+        let mut lut_transactions: u64 = 0;
+
+        let handle_completed_data_set_info = |completed_data_set_info: CompletedDataSetInfo,
+                                               total_lut_load_us: &mut u64,
+                                               total_notify_us: &mut u64,
+                                               total_transactions: &mut u64,
+                                               total_entries: &mut u64,
+                                               total_data_sets: &mut u64,
+                                               lut_transactions: &mut u64| {
             let CompletedDataSetInfo { slot, indices } = completed_data_set_info;
             match blockstore.get_entries_in_data_block(slot, indices, /*slot_meta:*/ None) {
                 Ok(entries) => {
+                    *total_data_sets += 1;
+                    *total_entries += entries.len() as u64;
+
+                    // Notify deshred transactions if notifier is enabled
+                    if let Some(notifier) = deshred_transaction_notifier {
+                        for entry in entries.iter() {
+                            for tx in &entry.transactions {
+                                if let Some(signature) = tx.signatures.first() {
+                                    *total_transactions += 1;
+                                    let is_vote = is_simple_vote_transaction(tx);
+
+                                    // Measure LUT loading time
+                                    let mut lut_measure = Measure::start("load_lut");
+                                    let loaded_addresses = root_bank
+                                        .as_ref()
+                                        .and_then(|bank| load_transaction_addresses(tx, bank));
+                                    lut_measure.stop();
+
+                                    if loaded_addresses.is_some() {
+                                        *lut_transactions += 1;
+                                        *total_lut_load_us += lut_measure.as_us();
+                                    }
+
+                                    // Measure notification time
+                                    let mut notify_measure = Measure::start("notify_deshred");
+                                    notifier.notify_deshred_transaction(
+                                        slot,
+                                        signature,
+                                        is_vote,
+                                        tx,
+                                        loaded_addresses.as_ref(),
+                                    );
+                                    notify_measure.stop();
+                                    *total_notify_us += notify_measure.as_us();
+                                }
+                            }
+                        }
+                    }
+
+                    // Existing: notify signatures for RPC subscriptions
                     let transactions = Self::get_transaction_signatures(entries);
                     if !transactions.is_empty() {
                         rpc_subscriptions.notify_signatures_received((slot, transactions));
@@ -78,15 +199,49 @@ impl CompletedDataSetsService {
             }
             slot
         };
+
         let slots = completed_sets_receiver
             .recv_timeout(RECV_TIMEOUT)
             .map(std::iter::once)?
             .chain(completed_sets_receiver.try_iter())
             .flatten()
-            .map(handle_completed_data_set_info);
+            .map(|info| {
+                handle_completed_data_set_info(
+                    info,
+                    &mut total_lut_load_us,
+                    &mut total_notify_us,
+                    &mut total_transactions,
+                    &mut total_entries,
+                    &mut total_data_sets,
+                    &mut lut_transactions,
+                )
+            });
+
         if let Some(slot) = slots.max() {
             max_slots.shred_insert.fetch_max(slot, Ordering::Relaxed);
         }
+
+        batch_measure.stop();
+
+        // Report metrics if we processed any transactions
+        if total_transactions > 0 {
+            datapoint_info!(
+                "deshred_geyser_timing",
+                ("batch_total_us", batch_measure.as_us() as i64, i64),
+                ("notify_total_us", total_notify_us as i64, i64),
+                ("lut_load_total_us", total_lut_load_us as i64, i64),
+                ("transactions_count", total_transactions as i64, i64),
+                ("lut_transactions_count", lut_transactions as i64, i64),
+                ("entries_count", total_entries as i64, i64),
+                ("data_sets_count", total_data_sets as i64, i64),
+                (
+                    "avg_notify_us",
+                    (total_notify_us / total_transactions) as i64,
+                    i64
+                ),
+            );
+        }
+
         Ok(())
     }
 

core/src/validator.rs

@@ -822,26 +822,29 @@ impl Validator {
         let (
             accounts_update_notifier,
             transaction_notifier,
+            deshred_transaction_notifier,
             entry_notifier,
             block_metadata_notifier,
             slot_status_notifier,
         ) = if let Some(service) = &geyser_plugin_service {
             (
                 service.get_accounts_update_notifier(),
                 service.get_transaction_notifier(),
+                service.get_deshred_transaction_notifier(),
                 service.get_entry_notifier(),
                 service.get_block_metadata_notifier(),
                 service.get_slot_status_notifier(),
             )
         } else {
-            (None, None, None, None, None)
+            (None, None, None, None, None, None)
         };
 
         info!(
             "Geyser plugin: accounts_update_notifier: {}, transaction_notifier: {}, \
-             entry_notifier: {}",
+             deshred_transaction_notifier: {}, entry_notifier: {}",
             accounts_update_notifier.is_some(),
             transaction_notifier.is_some(),
+            deshred_transaction_notifier.is_some(),
             entry_notifier.is_some()
         );
 
@@ -1307,8 +1310,10 @@ impl Validator {
                         completed_data_sets_receiver,
                         blockstore.clone(),
                         rpc_subscriptions.clone(),
+                        deshred_transaction_notifier.clone(),
                         exit.clone(),
                         max_slots.clone(),
+                        bank_forks.clone(),
                     );
                     (
                         Some(completed_data_sets_sender),

geyser-plugin-interface/Cargo.toml

@@ -19,6 +19,7 @@ agave-unstable-api = []
 log = { workspace = true, features = ["std"] }
 solana-clock = { workspace = true }
 solana-hash = { workspace = true }
+solana-message = { workspace = true }
 solana-signature = { workspace = true }
 solana-transaction = { workspace = true }
 solana-transaction-status = { workspace = true }