Backport Accounts Fixes #16838 and the test #17038 (#19412)

* reclaims unref accounts from index (#16838)

* Test account index and store alignment (#17038)

* Use ReclaimResult::Default() instead of building subtypes

* Add test to ensure account_db store and index are aligned

Co-authored-by: Jeff Washington (jwash) <75863576+jeffwashington@users.noreply.github.com>
Co-authored-by: steviez <steven@solana.com>

runtime/src/accounts_db.rs

@@ -1181,31 +1181,30 @@ impl AccountsDb {
     // Reclaim older states of rooted accounts for AccountsDb bloat mitigation
     fn clean_old_rooted_accounts(
         &self,
-        purges_in_root: Vec<Pubkey>,
+        purges: Vec<Pubkey>,
         max_clean_root: Option<Slot>,
     ) -> ReclaimResult {
-        if purges_in_root.is_empty() {
-            return (HashMap::new(), HashMap::new());
+        if purges.is_empty() {
+            return ReclaimResult::default();
         }
         // This number isn't carefully chosen; just guessed randomly such that
         // the hot loop will be the order of ~Xms.
         const INDEX_CLEAN_BULK_COUNT: usize = 4096;
 
         let mut clean_rooted = Measure::start("clean_old_root-ms");
-        let reclaim_vecs =
-            purges_in_root
-                .par_chunks(INDEX_CLEAN_BULK_COUNT)
-                .map(|pubkeys: &[Pubkey]| {
-                    let mut reclaims = Vec::new();
-                    for pubkey in pubkeys {
-                        self.accounts_index.clean_rooted_entries(
-                            &pubkey,
-                            &mut reclaims,
-                            max_clean_root,
-                        );
-                    }
-                    reclaims
-                });
+        let reclaim_vecs = purges
+            .par_chunks(INDEX_CLEAN_BULK_COUNT)
+            .map(|pubkeys: &[Pubkey]| {
+                let mut reclaims = Vec::new();
+                for pubkey in pubkeys {
+                    self.accounts_index.clean_rooted_entries(
+                        &pubkey,
+                        &mut reclaims,
+                        max_clean_root,
+                    );
+                }
+                reclaims
+            });
         let reclaims: Vec<_> = reclaim_vecs.flatten().collect();
         clean_rooted.stop();
         inc_new_counter_info!("clean-old-root-par-clean-ms", clean_rooted.as_ms() as usize);
@@ -1216,7 +1215,7 @@ impl AccountsDb {
         // and those stores may be used for background hashing.
         let reset_accounts = false;
 
-        let mut reclaim_result = (HashMap::new(), HashMap::new());
+        let mut reclaim_result = ReclaimResult::default();
         self.handle_reclaims(
             &reclaims,
             None,
@@ -1599,7 +1598,9 @@ impl AccountsDb {
         // Don't reset from clean, since the pubkeys in those stores may need to be unref'ed
         // and those stores may be used for background hashing.
         let reset_accounts = false;
-        self.handle_reclaims(&reclaims, None, false, None, reset_accounts);
+        let mut reclaim_result = ReclaimResult::default();
+        let reclaim_result = Some(&mut reclaim_result);
+        self.handle_reclaims(&reclaims, None, false, reclaim_result, reset_accounts);
 
         reclaims_time.stop();
 
@@ -5910,6 +5911,51 @@ pub mod tests {
         assert_eq!(accounts.alive_account_count_in_slot(1), 0);
     }
 
+    #[test]
+    fn test_clean_multiple_zero_lamport_decrements_index_ref_count() {
+        solana_logger::setup();
+
+        let accounts = AccountsDb::new(Vec::new(), &ClusterType::Development);
+        let pubkey1 = solana_sdk::pubkey::new_rand();
+        let pubkey2 = solana_sdk::pubkey::new_rand();
+        let zero_lamport_account =
+            AccountSharedData::new(0, 0, AccountSharedData::default().owner());
+
+        // Store 2 accounts in slot 0, then update account 1 in two more slots
+        accounts.store_uncached(0, &[(&pubkey1, &zero_lamport_account)]);
+        accounts.store_uncached(0, &[(&pubkey2, &zero_lamport_account)]);
+        accounts.store_uncached(1, &[(&pubkey1, &zero_lamport_account)]);
+        accounts.store_uncached(2, &[(&pubkey1, &zero_lamport_account)]);
+        // Root all slots
+        accounts.add_root(0);
+        accounts.add_root(1);
+        accounts.add_root(2);
+
+        // Account ref counts should match how many slots they were stored in
+        // Account 1 = 3 slots; account 2 = 1 slot
+        assert_eq!(accounts.accounts_index.ref_count_from_storage(&pubkey1), 3);
+        assert_eq!(accounts.accounts_index.ref_count_from_storage(&pubkey2), 1);
+
+        accounts.clean_accounts(None);
+        // Slots 0 and 1 should each have been cleaned because all of their
+        // accounts are zero lamports
+        assert!(accounts.storage.get_slot_stores(0).is_none());
+        assert!(accounts.storage.get_slot_stores(1).is_none());
+        // Slot 2 only has a zero lamport account as well. But, calc_delete_dependencies()
+        // should exclude slot 2 from the clean due to changes in other slots
+        assert!(accounts.storage.get_slot_stores(2).is_some());
+        // Index ref counts should be consistent with the slot stores. Account 1 ref count
+        // should be 1 since slot 2 is the only alive slot; account 2 should have a ref
+        // count of 0 due to slot 0 being dead
+        assert_eq!(accounts.accounts_index.ref_count_from_storage(&pubkey1), 1);
+        assert_eq!(accounts.accounts_index.ref_count_from_storage(&pubkey2), 0);
+
+        accounts.clean_accounts(None);
+        // Slot 2 will now be cleaned, which will leave account 1 with a ref count of 0
+        assert!(accounts.storage.get_slot_stores(2).is_none());
+        assert_eq!(accounts.accounts_index.ref_count_from_storage(&pubkey1), 0);
+    }
+
     #[test]
     fn test_clean_zero_lamport_and_old_roots() {
         solana_logger::setup();