handle write_version correctly in geyser api (#29224)

geyser-plugin-manager/src/accounts_update_notifier.rs

@@ -9,11 +9,12 @@ use {
     solana_metrics::*,
     solana_runtime::{
         accounts_update_notifier_interface::AccountsUpdateNotifierInterface,
-        append_vec::{StoredAccountMeta, StoredMeta},
+        append_vec::StoredAccountMeta,
     },
     solana_sdk::{
         account::{AccountSharedData, ReadableAccount},
         clock::Slot,
+        pubkey::Pubkey,
         signature::Signature,
     },
     std::sync::{Arc, RwLock},
@@ -27,12 +28,13 @@ impl AccountsUpdateNotifierInterface for AccountsUpdateNotifierImpl {
     fn notify_account_update(
         &self,
         slot: Slot,
-        meta: &StoredMeta,
         account: &AccountSharedData,
         txn_signature: &Option<&Signature>,
+        pubkey: &Pubkey,
+        write_version: u64,
     ) {
         if let Some(account_info) =
-            self.accountinfo_from_shared_account_data(meta, account, txn_signature)
+            self.accountinfo_from_shared_account_data(account, txn_signature, pubkey, write_version)
         {
             self.notify_plugins_of_account_update(account_info, slot, false);
         }
@@ -104,18 +106,19 @@ impl AccountsUpdateNotifierImpl {
 
     fn accountinfo_from_shared_account_data<'a>(
         &self,
-        meta: &'a StoredMeta,
         account: &'a AccountSharedData,
         txn_signature: &'a Option<&'a Signature>,
+        pubkey: &'a Pubkey,
+        write_version: u64,
     ) -> Option<ReplicaAccountInfoV2<'a>> {
         Some(ReplicaAccountInfoV2 {
-            pubkey: meta.pubkey.as_ref(),
+            pubkey: pubkey.as_ref(),
             lamports: account.lamports(),
             owner: account.owner().as_ref(),
             executable: account.executable(),
             rent_epoch: account.rent_epoch(),
             data: account.data(),
-            write_version: meta.write_version,
+            write_version,
             txn_signature: *txn_signature,
         })
     }

runtime/src/accounts_db.rs

@@ -42,7 +42,7 @@ use {
         },
         append_vec::{
             aligned_stored_size, AppendVec, StorableAccountsWithHashesAndWriteVersions,
-            StoredAccountMeta, StoredMeta, StoredMetaWriteVersion, APPEND_VEC_MMAPPED_FILES_OPEN,
+            StoredAccountMeta, StoredMetaWriteVersion, APPEND_VEC_MMAPPED_FILES_OPEN,
             STORE_META_OVERHEAD,
         },
         cache_hash_data::{CacheHashData, CacheHashDataFile},
@@ -6614,13 +6614,17 @@ impl AccountsDb {
         }
     }
 
-    fn write_accounts_to_cache<'a, 'b, T: ReadableAccount + Sync>(
+    fn write_accounts_to_cache<'a, 'b, T: ReadableAccount + Sync, P>(
         &self,
         slot: Slot,
         accounts_and_meta_to_store: &impl StorableAccounts<'b, T>,
         txn_signatures_iter: Box<dyn std::iter::Iterator<Item = &Option<&Signature>> + 'a>,
         include_slot_in_hash: IncludeSlotInHash,
-    ) -> Vec<AccountInfo> {
+        mut write_version_producer: P,
+    ) -> Vec<AccountInfo>
+    where
+        P: Iterator<Item = u64>,
+    {
         txn_signatures_iter
             .enumerate()
             .map(|(i, signature)| {
@@ -6634,13 +6638,13 @@ impl AccountsDb {
                     account.lamports(),
                 );
 
-                let meta = StoredMeta {
+                self.notify_account_at_accounts_update(
-                    pubkey: *accounts_and_meta_to_store.pubkey(i),
+                    slot,
-                    data_len: account.data().len() as u64,
+                    &account,
-                    write_version: 0,
+                    signature,
-                };
+                    accounts_and_meta_to_store.pubkey(i),
-
+                    &mut write_version_producer,
-                self.notify_account_at_accounts_update(slot, &meta, &account, signature);
+                );
 
                 let cached_account = self.accounts_cache.store(
                     slot,
@@ -6703,6 +6707,7 @@ impl AccountsDb {
                 accounts,
                 signature_iter,
                 accounts.include_slot_in_hash(),
+                write_version_producer,
             )
         } else if accounts.has_hash_and_write_version() {
             self.write_accounts_to_storage(
@@ -9657,7 +9662,7 @@ pub mod tests {
             accounts_index::{
                 tests::*, AccountSecondaryIndexesIncludeExclude, ReadAccountMapEntry, RefCount,
             },
-            append_vec::{test_utils::TempFile, AccountMeta},
+            append_vec::{test_utils::TempFile, AccountMeta, StoredMeta},
             cache_hash_data_stats::CacheHashDataStats,
             inline_spl_token,
         },

runtime/src/accounts_db/geyser_plugin_utils.rs

@@ -1,8 +1,5 @@
 use {
-    crate::{
+    crate::{accounts_db::AccountsDb, append_vec::StoredAccountMeta},
-        accounts_db::AccountsDb,
-        append_vec::{StoredAccountMeta, StoredMeta},
-    },
     solana_measure::measure::Measure,
     solana_metrics::*,
     solana_sdk::{account::AccountSharedData, clock::Slot, pubkey::Pubkey, signature::Signature},
@@ -59,16 +56,25 @@ impl AccountsDb {
         notify_stats.report();
     }
 
-    pub fn notify_account_at_accounts_update(
+    pub fn notify_account_at_accounts_update<P>(
         &self,
         slot: Slot,
-        meta: &StoredMeta,
         account: &AccountSharedData,
         txn_signature: &Option<&Signature>,
-    ) {
+        pubkey: &Pubkey,
+        write_version_producer: &mut P,
+    ) where
+        P: Iterator<Item = u64>,
+    {
         if let Some(accounts_update_notifier) = &self.accounts_update_notifier {
             let notifier = &accounts_update_notifier.read().unwrap();
-            notifier.notify_account_update(slot, meta, account, txn_signature);
+            notifier.notify_account_update(
+                slot,
+                account,
+                txn_signature,
+                pubkey,
+                write_version_producer.next().unwrap(),
+            );
         }
     }
 
@@ -154,7 +160,7 @@ pub mod tests {
             accounts_update_notifier_interface::{
                 AccountsUpdateNotifier, AccountsUpdateNotifierInterface,
             },
-            append_vec::{StoredAccountMeta, StoredMeta},
+            append_vec::StoredAccountMeta,
         },
         dashmap::DashMap,
         solana_sdk::{
@@ -186,12 +192,13 @@ pub mod tests {
         fn notify_account_update(
             &self,
             slot: Slot,
-            meta: &StoredMeta,
             account: &AccountSharedData,
             _txn_signature: &Option<&Signature>,
+            pubkey: &Pubkey,
+            _write_version: u64,
         ) {
             self.accounts_notified
-                .entry(meta.pubkey)
+                .entry(*pubkey)
                 .or_default()
                 .push((slot, account.clone()));
         }

runtime/src/accounts_update_notifier_interface.rs

@@ -1,6 +1,6 @@
 use {
-    crate::append_vec::{StoredAccountMeta, StoredMeta},
+    crate::append_vec::StoredAccountMeta,
-    solana_sdk::{account::AccountSharedData, clock::Slot, signature::Signature},
+    solana_sdk::{account::AccountSharedData, clock::Slot, pubkey::Pubkey, signature::Signature},
     std::sync::{Arc, RwLock},
 };
 
@@ -9,9 +9,10 @@ pub trait AccountsUpdateNotifierInterface: std::fmt::Debug {
     fn notify_account_update(
         &self,
         slot: Slot,
-        meta: &StoredMeta,
         account: &AccountSharedData,
         txn_signature: &Option<&Signature>,
+        pubkey: &Pubkey,
+        write_version: u64,
     );
 
     /// Notified when the AccountsDb is initialized at start when restored