Supports accounts lt hash in snapshots

frozen abi diff
```
                    field accounts_lt_hash: core::option::Option<solana_runtime::serde_snapshot::types::SerdeAccountsLtHash>
                        enum Option (variants = 2)
                            variant(0) None (unit)
                            variant(1) Some(solana_runtime::serde_snapshot::types::SerdeAccountsLtHash) (newtype)
                                struct SerdeAccountsLtHash(solana_runtime::serde_snapshot::types::_::<impl serde::ser::Serialize for solana_runtime::serde_snapshot::types::SerdeAccountsLtHash>::serialize::__SerializeWith) (newtype)
                                    tuple (elements = 1024)
                                        element serde_with::ser::SerializeAsWrap<u16, serde_with::Same>
                                            primitive u16
```

runtime/src/bank.rs

@@ -459,6 +459,8 @@ pub struct BankFieldsToDeserialize {
     pub(crate) accounts_data_len: u64,
     pub(crate) incremental_snapshot_persistence: Option<BankIncrementalSnapshotPersistence>,
     pub(crate) epoch_accounts_hash: Option<Hash>,
+    // Once the accounts lt hash feature gate is enabled, remove the Option wrapper
+    pub(crate) accounts_lt_hash: Option<AccountsLtHash>,
 }
 
 /// Bank's common fields shared by all supported snapshot versions for serialization.
@@ -502,6 +504,8 @@ pub struct BankFieldsToSerialize {
     pub is_delta: bool,
     pub accounts_data_len: u64,
     pub versioned_epoch_stakes: HashMap<u64, VersionedEpochStakes>,
+    // Once the accounts lt hash feature gate is enabled, remove the Option wrapper
+    pub accounts_lt_hash: Option<AccountsLtHash>,
 }
 
 // Can't derive PartialEq because RwLock doesn't implement PartialEq
@@ -553,6 +557,7 @@ impl PartialEq for Bank {
             is_delta,
             #[cfg(feature = "dev-context-only-utils")]
             hash_overrides,
+            accounts_lt_hash,
             // TODO: Confirm if all these fields are intentionally ignored!
             rewards: _,
             cluster_type: _,
@@ -577,7 +582,6 @@ impl PartialEq for Bank {
             compute_budget: _,
             transaction_account_lock_limit: _,
             fee_structure: _,
-            accounts_lt_hash: _,
             cache_for_accounts_lt_hash: _,
             // Ignore new fields explicitly if they do not impact PartialEq.
             // Adding ".." will remove compile-time checks that if a new field
@@ -616,6 +620,8 @@ impl PartialEq for Bank {
             // different Mutexes.
             && (Arc::ptr_eq(hash_overrides, &other.hash_overrides) ||
                 *hash_overrides.lock().unwrap() == *other.hash_overrides.lock().unwrap())
+            && !(self.is_accounts_lt_hash_enabled() && other.is_accounts_lt_hash_enabled()
+                && *accounts_lt_hash.lock().unwrap() != *other.accounts_lt_hash.lock().unwrap())
     }
 }
 
@@ -656,6 +662,7 @@ impl BankFieldsToSerialize {
             is_delta: bool::default(),
             accounts_data_len: u64::default(),
             versioned_epoch_stakes: HashMap::default(),
+            accounts_lt_hash: Some(AccountsLtHash(LtHash([0x7E57; LtHash::NUM_ELEMENTS]))),
         }
     }
 }
@@ -1716,16 +1723,21 @@ impl Bank {
             .fill_missing_sysvar_cache_entries(&bank);
         bank.rebuild_skipped_rewrites();
 
-        let calculate_accounts_lt_hash_duration = bank.is_accounts_lt_hash_enabled().then(|| {
-            let (_, duration) = meas_dur!({
-                *bank.accounts_lt_hash.get_mut().unwrap() = bank
-                    .rc
-                    .accounts
-                    .accounts_db
-                    .calculate_accounts_lt_hash_at_startup(&bank.ancestors, bank.slot());
-            });
-            duration
-        });
+        let mut calculate_accounts_lt_hash_duration = None;
+        if bank.is_accounts_lt_hash_enabled() {
+            if let Some(accounts_lt_hash) = fields.accounts_lt_hash {
+                *bank.accounts_lt_hash.get_mut().unwrap() = accounts_lt_hash
+            } else {
+                let (_, duration) = meas_dur!(
+                    *bank.accounts_lt_hash.get_mut().unwrap() = bank
+                        .rc
+                        .accounts
+                        .accounts_db
+                        .calculate_accounts_lt_hash_at_startup(&bank.ancestors, bank.slot())
+                );
+                calculate_accounts_lt_hash_duration = Some(duration);
+            }
+        }
 
         // Sanity assertions between bank snapshot and genesis config
         // Consider removing from serializable bank state
@@ -1815,6 +1827,9 @@ impl Bank {
             is_delta: self.is_delta.load(Relaxed),
             accounts_data_len: self.load_accounts_data_size(),
             versioned_epoch_stakes,
+            accounts_lt_hash: self
+                .is_accounts_lt_hash_enabled()
+                .then(|| self.accounts_lt_hash.lock().unwrap().clone()),
         }
     }
 

runtime/src/bank/serde_snapshot.rs

@@ -25,7 +25,7 @@ mod tests {
             account_storage::{AccountStorageMap, AccountStorageReference},
             accounts_db::{
                 get_temp_accounts_paths, AccountShrinkThreshold, AccountStorageEntry, AccountsDb,
-                AtomicAccountsFileId, ACCOUNTS_DB_CONFIG_FOR_TESTING,
+                AccountsDbConfig, AtomicAccountsFileId, ACCOUNTS_DB_CONFIG_FOR_TESTING,
             },
             accounts_file::{AccountsFile, AccountsFileError, StorageAccess},
             accounts_hash::{AccountsDeltaHash, AccountsHash},
@@ -97,27 +97,41 @@ mod tests {
         let storage_access_iter = [StorageAccess::Mmap, StorageAccess::File].into_iter();
         let has_incremental_snapshot_persistence_iter = [false, true].into_iter();
         let has_epoch_accounts_hash_iter = [false, true].into_iter();
+        let has_accounts_lt_hash_iter = [false, true].into_iter();
 
-        for (storage_access, has_incremental_snapshot_persistence, has_epoch_accounts_hash) in itertools::iproduct!(
+        for (
+            storage_access,
+            has_incremental_snapshot_persistence,
+            has_epoch_accounts_hash,
+            has_accounts_lt_hash,
+        ) in itertools::iproduct!(
             storage_access_iter,
             has_incremental_snapshot_persistence_iter,
-            has_epoch_accounts_hash_iter
+            has_epoch_accounts_hash_iter,
+            has_accounts_lt_hash_iter
         ) {
             do_serialize_bank_snapshot(
                 storage_access,
                 has_incremental_snapshot_persistence,
                 has_epoch_accounts_hash,
+                has_accounts_lt_hash,
             );
         }
 
         fn do_serialize_bank_snapshot(
             storage_access: StorageAccess,
             has_incremental_snapshot_persistence: bool,
             has_epoch_accounts_hash: bool,
+            has_accounts_lt_hash: bool,
         ) {
             let (mut genesis_config, _) = create_genesis_config(500);
             genesis_config.epoch_schedule = EpochSchedule::custom(400, 400, false);
             let bank0 = Arc::new(Bank::new_for_tests(&genesis_config));
+            bank0
+                .rc
+                .accounts
+                .accounts_db
+                .set_is_experimental_accumulator_hash_enabled(has_accounts_lt_hash);
             let deposit_amount = bank0.get_minimum_balance_for_rent_exemption(0);
             let eah_start_slot = epoch_accounts_hash_utils::calculation_start(&bank0);
             let bank1 = Bank::new_from_parent(bank0.clone(), &Pubkey::default(), 1);
@@ -167,6 +181,8 @@ mod tests {
                     .set_valid(epoch_accounts_hash, eah_start_slot);
                 epoch_accounts_hash
             });
+            let expected_accounts_lt_hash =
+                has_accounts_lt_hash.then(|| bank2.accounts_lt_hash.lock().unwrap().clone());
 
             // Only if a bank was recently recreated from a snapshot will it have an epoch stakes entry
             // of type "delegations" which cannot be serialized into the versioned epoch stakes map. Simulate
@@ -203,6 +219,7 @@ mod tests {
                 assert!(!bank_fields.versioned_epoch_stakes.is_empty());
 
                 let versioned_epoch_stakes = mem::take(&mut bank_fields.versioned_epoch_stakes);
+                let accounts_lt_hash = bank_fields.accounts_lt_hash.clone().map(Into::into);
                 serde_snapshot::serialize_bank_snapshot_into(
                     &mut writer,
                     bank_fields,
@@ -216,6 +233,7 @@ mod tests {
                             .as_ref(),
                         epoch_accounts_hash: expected_epoch_accounts_hash,
                         versioned_epoch_stakes,
+                        accounts_lt_hash,
                     },
                     accounts_db.write_version.load(Ordering::Acquire),
                 )
@@ -238,6 +256,10 @@ mod tests {
                 full_snapshot_stream: &mut reader,
                 incremental_snapshot_stream: None,
             };
+            let accounts_db_config = AccountsDbConfig {
+                enable_experimental_accumulator_hash: has_accounts_lt_hash,
+                ..ACCOUNTS_DB_CONFIG_FOR_TESTING
+            };
             let dbank = serde_snapshot::bank_from_streams(
                 &mut snapshot_streams,
                 &dbank_paths,
@@ -250,7 +272,7 @@ mod tests {
                 None,
                 AccountShrinkThreshold::default(),
                 false,
-                Some(ACCOUNTS_DB_CONFIG_FOR_TESTING),
+                Some(accounts_db_config),
                 None,
                 Arc::default(),
             )
@@ -279,6 +301,14 @@ mod tests {
                 dbank.get_epoch_accounts_hash_to_serialize(),
                 expected_epoch_accounts_hash,
             );
+            assert_eq!(
+                dbank.is_accounts_lt_hash_enabled().then(|| dbank
+                    .accounts_lt_hash
+                    .lock()
+                    .unwrap()
+                    .clone()),
+                expected_accounts_lt_hash,
+            );
 
             assert_eq!(dbank, bank2);
         }
@@ -519,8 +549,10 @@ mod tests {
             super::*,
             solana_accounts_db::{
                 account_storage::meta::StoredMetaWriteVersion, accounts_db::BankHashStats,
+                accounts_hash::AccountsLtHash,
             },
             solana_frozen_abi::abi_example::AbiExample,
+            solana_lattice_hash::lt_hash::LtHash,
             solana_sdk::clock::Slot,
             std::marker::PhantomData,
         };
@@ -546,7 +578,7 @@ mod tests {
         #[cfg_attr(
             feature = "frozen-abi",
             derive(AbiExample),
-            frozen_abi(digest = "8hwm4YsQXJWGZdp762SkJnDok29LXKwFtmW9oQ2KSzrN")
+            frozen_abi(digest = "CKb4Lp4AEJKBCLqHCFSSsntQU3MpcphA38RcAXASeDuG")
         )]
         #[derive(Serialize)]
         pub struct BankAbiTestWrapper {
@@ -559,6 +591,10 @@ mod tests {
             S: serde::Serializer,
         {
             let bank = Bank::default_for_tests();
+            bank.rc
+                .accounts
+                .accounts_db
+                .set_is_experimental_accumulator_hash_enabled(true);
             let snapshot_storages = AccountsDb::example().get_snapshot_storages(0..1).0;
             // ensure there is at least one snapshot storage example for ABI digesting
             assert!(!snapshot_storages.is_empty());
@@ -585,6 +621,7 @@ mod tests {
                     incremental_snapshot_persistence: Some(&incremental_snapshot_persistence),
                     epoch_accounts_hash: Some(EpochAccountsHash::new(Hash::new_unique())),
                     versioned_epoch_stakes,
+                    accounts_lt_hash: Some(AccountsLtHash(LtHash::identity()).into()),
                 },
                 StoredMetaWriteVersion::default(),
             )

runtime/src/serde_snapshot.rs

@@ -203,6 +203,7 @@ impl From<DeserializableVersionedBank> for BankFieldsToDeserialize {
             is_delta: dvb.is_delta,
             incremental_snapshot_persistence: None,
             epoch_accounts_hash: None,
+            accounts_lt_hash: None, // populated from ExtraFieldsToDeserialize
         }
     }
 }
@@ -403,7 +404,6 @@ struct ExtraFieldsToDeserialize {
     #[serde(deserialize_with = "default_on_eof")]
     versioned_epoch_stakes: HashMap<u64, VersionedEpochStakes>,
     #[serde(deserialize_with = "default_on_eof")]
-    #[allow(dead_code)]
     accounts_lt_hash: Option<SerdeAccountsLtHash>,
 }
 
@@ -421,6 +421,7 @@ pub struct ExtraFieldsToSerialize<'a> {
     pub incremental_snapshot_persistence: Option<&'a BankIncrementalSnapshotPersistence>,
     pub epoch_accounts_hash: Option<EpochAccountsHash>,
     pub versioned_epoch_stakes: HashMap<u64, VersionedEpochStakes>,
+    pub accounts_lt_hash: Option<SerdeAccountsLtHash>,
 }
 
 fn deserialize_bank_fields<R>(
@@ -446,7 +447,7 @@ where
         incremental_snapshot_persistence,
         epoch_accounts_hash,
         versioned_epoch_stakes,
-        accounts_lt_hash: _,
+        accounts_lt_hash,
     } = extra_fields;
 
     bank_fields.fee_rate_governor = bank_fields
@@ -463,6 +464,8 @@ where
             .map(|(epoch, versioned_epoch_stakes)| (epoch, versioned_epoch_stakes.into())),
     );
 
+    bank_fields.accounts_lt_hash = accounts_lt_hash.map(Into::into);
+
     Ok((bank_fields, accounts_db_fields))
 }
 
@@ -699,6 +702,7 @@ impl<'a> Serialize for SerializableBankAndStorage<'a> {
         let write_version = accounts_db.write_version.load(Ordering::Acquire);
         let lamports_per_signature = bank_fields.fee_rate_governor.lamports_per_signature;
         let versioned_epoch_stakes = std::mem::take(&mut bank_fields.versioned_epoch_stakes);
+        let accounts_lt_hash = bank_fields.accounts_lt_hash.clone().map(Into::into);
         let bank_fields_to_serialize = (
             SerializableVersionedBank::from(bank_fields),
             SerializableAccountsDb::<'_> {
@@ -714,6 +718,7 @@ impl<'a> Serialize for SerializableBankAndStorage<'a> {
                 incremental_snapshot_persistence: None,
                 epoch_accounts_hash: self.bank.get_epoch_accounts_hash_to_serialize(),
                 versioned_epoch_stakes,
+                accounts_lt_hash,
             },
         );
         bank_fields_to_serialize.serialize(serializer)

runtime/src/snapshot_bank_utils.rs

@@ -913,7 +913,19 @@ fn bank_to_full_snapshot_archive_with(
         .accounts_db
         .set_latest_full_snapshot_slot(bank.slot());
     bank.squash(); // Bank may not be a root
-    bank.rehash(); // Bank accounts may have been manually modified by the caller
+
+    // Rehashing is not currently supported when the accounts lt hash is enabled.
+    // This is because rehashing will *re-mix-in* all the accounts stored in this bank into the
+    // accounts lt hash!  This is incorrect, as the accounts lt hash would change, even if the bank
+    // was *not* manually modified by the caller.
+    // We can re-allow rehasing if we change the Bank to hold its parent's accounts lt hash plus a
+    // *delta* accounts lt hash, and then Bank::hash_internal_state() will only recalculate the
+    // delta accounts lt hash.
+    // Another option is to consider if manual modification should even be allowed in the first
+    // place. Disallowing it would solve these issues.
+    if !bank.is_accounts_lt_hash_enabled() {
+        bank.rehash(); // Bank accounts may have been manually modified by the caller
+    }
     bank.force_flush_accounts_cache();
     bank.clean_accounts();
     let calculated_accounts_hash =
@@ -976,7 +988,11 @@ pub fn bank_to_incremental_snapshot_archive(
         .accounts_db
         .set_latest_full_snapshot_slot(full_snapshot_slot);
     bank.squash(); // Bank may not be a root
-    bank.rehash(); // Bank accounts may have been manually modified by the caller
+
+    // See the comment in bank_to_full_snapshot_archive() when calling rehash()
+    if !bank.is_accounts_lt_hash_enabled() {
+        bank.rehash(); // Bank accounts may have been manually modified by the caller
+    }
     bank.force_flush_accounts_cache();
     bank.clean_accounts();
     let calculated_incremental_accounts_hash =

runtime/src/snapshot_utils.rs

@@ -884,6 +884,7 @@ fn serialize_snapshot(
                 incremental_snapshot_persistence: bank_incremental_snapshot_persistence,
                 epoch_accounts_hash,
                 versioned_epoch_stakes,
+                accounts_lt_hash: bank_fields.accounts_lt_hash.clone().map(Into::into),
             };
             serde_snapshot::serialize_bank_snapshot_into(
                 stream,