geyser: Uses scan_accounts_for_geyser() at startup (anza-xyz#5069)

Author: brooksprumo

accounts-db/src/accounts_db/geyser_plugin_utils.rs

@@ -1,5 +1,5 @@
 use {
-    crate::{account_storage::meta::StoredAccountMeta, accounts_db::AccountsDb},
+    crate::{accounts_db::AccountsDb, accounts_update_notifier_interface::AccountForGeyser},
     solana_account::AccountSharedData,
     solana_clock::Slot,
     solana_measure::measure::Measure,
@@ -92,7 +92,7 @@ impl AccountsDb {
         let mut pubkeys = HashSet::new();
 
         // populate `accounts_duplicate` for any pubkeys that are in this storage twice.
-        // Storages cannot return `StoredAccountMeta<'_>` for more than 1 account at a time, so we have to do 2 passes to make sure
+        // Storages cannot return `AccountForGeyser` for more than 1 account at a time, so we have to do 2 passes to make sure
         // we don't have duplicate pubkeys.
         let mut i = 0;
         storage_entry.accounts.scan_pubkeys(|pubkey| {
@@ -104,14 +104,14 @@ impl AccountsDb {
 
         // now, actually notify geyser
         let mut i = 0;
-        storage_entry.accounts.scan_accounts(|account| {
+        storage_entry.accounts.scan_accounts_for_geyser(|account| {
             i += 1;
             account_len += 1;
-            if notified_accounts.contains(account.pubkey()) {
+            if notified_accounts.contains(account.pubkey) {
                 notify_stats.skipped_accounts += 1;
                 return;
             }
-            if let Some(highest_i) = accounts_duplicate.get(account.pubkey()) {
+            if let Some(highest_i) = accounts_duplicate.get(account.pubkey) {
                 if highest_i != &i {
                     // this pubkey is in this storage twice and the current instance is not the last one, so we skip it.
                     // We only send unique accounts in this slot to `notify_filtered_accounts`
@@ -140,7 +140,7 @@ impl AccountsDb {
         slot: Slot,
         write_version: u64,
         notified_accounts: &mut HashSet<Pubkey>,
-        accounts_to_stream: impl Iterator<Item = StoredAccountMeta<'a>>,
+        accounts_to_stream: impl Iterator<Item = AccountForGeyser<'a>>,
         notify_stats: &mut GeyserPluginNotifyAtSnapshotRestoreStats,
     ) {
         let notifier = self.accounts_update_notifier.as_ref().unwrap();
@@ -153,7 +153,7 @@ impl AccountsDb {
             notify_stats.total_pure_notify += measure_pure_notify.as_us() as usize;
 
             let mut measure_bookkeep = Measure::start("accountsdb-plugin-notifying-bookeeeping");
-            notified_accounts.insert(*account.pubkey());
+            notified_accounts.insert(*account.pubkey);
             measure_bookkeep.stop();
             notify_stats.total_pure_bookeeping += measure_bookkeep.as_us() as usize;
 
@@ -167,18 +167,12 @@ impl AccountsDb {
 #[cfg(test)]
 pub mod tests {
     use {
-        crate::{
-            account_storage::meta::StoredAccountMeta,
-            accounts_db::AccountsDb,
-            accounts_update_notifier_interface::{
-                AccountsUpdateNotifier, AccountsUpdateNotifierInterface,
-            },
+        super::*,
+        crate::accounts_update_notifier_interface::{
+            AccountsUpdateNotifier, AccountsUpdateNotifierInterface,
         },
         dashmap::DashMap,
-        solana_account::{AccountSharedData, ReadableAccount},
-        solana_clock::Slot,
-        solana_pubkey::Pubkey,
-        solana_transaction::sanitized::SanitizedTransaction,
+        solana_account::ReadableAccount as _,
         std::sync::{
             atomic::{AtomicBool, Ordering},
             Arc,
@@ -223,10 +217,10 @@ pub mod tests {
             &self,
             slot: Slot,
             _write_version: u64,
-            account: &StoredAccountMeta,
+            account: &AccountForGeyser<'_>,
         ) {
             self.accounts_notified
-                .entry(*account.pubkey())
+                .entry(*account.pubkey)
                 .or_default()
                 .push((slot, account.to_account_shared_data()));
         }

accounts-db/src/accounts_file.rs

@@ -3,13 +3,14 @@ use {
         account_info::AccountInfo,
         account_storage::meta::StoredAccountMeta,
         accounts_db::AccountsFileId,
+        accounts_update_notifier_interface::AccountForGeyser,
         append_vec::{AppendVec, AppendVecError, IndexInfo},
         storable_accounts::StorableAccounts,
         tiered_storage::{
             error::TieredStorageError, hot::HOT_FORMAT, index::IndexOffset, TieredStorage,
         },
     },
-    solana_account::AccountSharedData,
+    solana_account::{AccountSharedData, ReadableAccount as _},
     solana_clock::Slot,
     solana_pubkey::Pubkey,
     std::{
@@ -233,6 +234,25 @@ impl AccountsFile {
         }
     }
 
+    /// Iterate over all accounts and call `callback` with each account.
+    /// Only intended to be used by Geyser.
+    pub fn scan_accounts_for_geyser(
+        &self,
+        mut callback: impl for<'local> FnMut(AccountForGeyser<'local>),
+    ) {
+        self.scan_accounts(|stored_account_meta| {
+            let account_for_geyser = AccountForGeyser {
+                pubkey: stored_account_meta.pubkey(),
+                lamports: stored_account_meta.lamports(),
+                owner: stored_account_meta.owner(),
+                executable: stored_account_meta.executable(),
+                rent_epoch: stored_account_meta.rent_epoch(),
+                data: stored_account_meta.data(),
+            };
+            callback(account_for_geyser)
+        })
+    }
+
     /// for each offset in `sorted_offsets`, return the account size
     pub(crate) fn get_account_sizes(&self, sorted_offsets: &[usize]) -> Vec<usize> {
         match self {

accounts-db/src/accounts_update_notifier_interface.rs

@@ -1,6 +1,8 @@
 use {
-    crate::account_storage::meta::StoredAccountMeta, solana_account::AccountSharedData,
-    solana_clock::Slot, solana_pubkey::Pubkey, solana_transaction::sanitized::SanitizedTransaction,
+    solana_account::{AccountSharedData, ReadableAccount},
+    solana_clock::{Epoch, Slot},
+    solana_pubkey::Pubkey,
+    solana_transaction::sanitized::SanitizedTransaction,
     std::sync::Arc,
 };
 
@@ -24,11 +26,40 @@ pub trait AccountsUpdateNotifierInterface: std::fmt::Debug {
         &self,
         slot: Slot,
         write_version: u64,
-        account: &StoredAccountMeta,
+        account: &AccountForGeyser<'_>,
     );
 
     /// Notified when all accounts have been notified when restoring from a snapshot.
     fn notify_end_of_restore_from_snapshot(&self);
 }
 
 pub type AccountsUpdateNotifier = Arc<dyn AccountsUpdateNotifierInterface + Sync + Send>;
+
+/// Account type with only the fields necessary for Geyser
+#[derive(Debug, Clone)]
+pub struct AccountForGeyser<'a> {
+    pub pubkey: &'a Pubkey,
+    pub lamports: u64,
+    pub owner: &'a Pubkey,
+    pub executable: bool,
+    pub rent_epoch: Epoch,
+    pub data: &'a [u8],
+}
+
+impl ReadableAccount for AccountForGeyser<'_> {
+    fn lamports(&self) -> u64 {
+        self.lamports
+    }
+    fn data(&self) -> &[u8] {
+        self.data
+    }
+    fn owner(&self) -> &Pubkey {
+        self.owner
+    }
+    fn executable(&self) -> bool {
+        self.executable
+    }
+    fn rent_epoch(&self) -> Epoch {
+        self.rent_epoch
+    }
+}

geyser-plugin-manager/src/accounts_update_notifier.rs

@@ -6,9 +6,8 @@ use {
     },
     log::*,
     solana_account::{AccountSharedData, ReadableAccount},
-    solana_accounts_db::{
-        account_storage::meta::StoredAccountMeta,
-        accounts_update_notifier_interface::AccountsUpdateNotifierInterface,
+    solana_accounts_db::accounts_update_notifier_interface::{
+        AccountForGeyser, AccountsUpdateNotifierInterface,
     },
     solana_clock::Slot,
     solana_measure::measure::Measure,
@@ -45,12 +44,12 @@ impl AccountsUpdateNotifierInterface for AccountsUpdateNotifierImpl {
         &self,
         slot: Slot,
         write_version: u64,
-        account: &StoredAccountMeta,
+        account: &AccountForGeyser<'_>,
     ) {
         let mut measure_all = Measure::start("geyser-plugin-notify-account-restore-all");
         let mut measure_copy = Measure::start("geyser-plugin-copy-stored-account-info");
 
-        let mut account = self.accountinfo_from_stored_account_meta(account);
+        let mut account = self.accountinfo_from_account_for_geyser(account);
         account.write_version = write_version;
         measure_copy.stop();
 
@@ -135,17 +134,17 @@ impl AccountsUpdateNotifierImpl {
         }
     }
 
-    fn accountinfo_from_stored_account_meta<'a>(
+    fn accountinfo_from_account_for_geyser<'a>(
         &self,
-        stored_account_meta: &'a StoredAccountMeta,
+        account: &'a AccountForGeyser<'_>,
     ) -> ReplicaAccountInfoV3<'a> {
         ReplicaAccountInfoV3 {
-            pubkey: stored_account_meta.pubkey().as_ref(),
-            lamports: stored_account_meta.lamports(),
-            owner: stored_account_meta.owner().as_ref(),
-            executable: stored_account_meta.executable(),
-            rent_epoch: stored_account_meta.rent_epoch(),
-            data: stored_account_meta.data(),
+            pubkey: account.pubkey.as_ref(),
+            lamports: account.lamports(),
+            owner: account.owner().as_ref(),
+            executable: account.executable(),
+            rent_epoch: account.rent_epoch(),
+            data: account.data(),
             write_version: 0, // can/will be populated afterwards
             txn: None,
         }