Enable conservative out-of-bound snapshot cleaning (#8811)

* Enable conservative out-of-bound snapshot cleaning

* Add tests

(cherry picked from commit 4bbf09f)

runtime/src/accounts.rs

@@ -555,10 +555,6 @@ impl Accounts {
             .for_each(|(tx, result)| self.unlock_account(tx, result, &mut account_locks));
     }
 
-    pub fn has_accounts(&self, slot: Slot) -> bool {
-        self.accounts_db.has_accounts(slot)
-    }
-
     /// Store the accounts into the DB
     pub fn store_accounts(
         &self,

runtime/src/accounts_db.rs

@@ -249,6 +249,10 @@ impl AccountStorageEntry {
         self.count_and_status.read().unwrap().0
     }
 
+    pub fn has_accounts(&self) -> bool {
+        self.count() > 0
+    }
+
     pub fn slot_id(&self) -> Slot {
         self.slot_id
     }
@@ -627,17 +631,6 @@ impl AccountsDB {
         )
     }
 
-    pub fn has_accounts(&self, slot: Slot) -> bool {
-        if let Some(storage_slots) = self.storage.read().unwrap().0.get(&slot) {
-            for x in storage_slots.values() {
-                if x.count() > 0 {
-                    return true;
-                }
-            }
-        }
-        false
-    }
-
     // Reclaim older states of rooted non-zero lamport accounts as a general
     // AccountsDB bloat mitigation and preprocess for better zero-lamport purging.
     fn clean_old_rooted_accounts(&self, purges_in_root: Vec<Pubkey>) {
@@ -1415,10 +1408,17 @@ impl AccountsDB {
         r_storage
             .0
             .iter()
-            .filter(|(slot, storage)| {
-                **slot <= snapshot_slot && !storage.is_empty() && accounts_index.is_root(**slot)
+            .filter(|(slot, _slot_stores)| {
+                **slot <= snapshot_slot && accounts_index.is_root(**slot)
+            })
+            .map(|(_slot, slot_stores)| {
+                slot_stores
+                    .values()
+                    .filter(|x| x.has_accounts())
+                    .cloned()
+                    .collect()
             })
-            .map(|(_, slot_store)| slot_store.values().cloned().collect())
+            .filter(|snapshot_storage: &SnapshotStorage| !snapshot_storage.is_empty())
             .collect()
     }
 
@@ -1494,7 +1494,7 @@ impl AccountsDB {
         }
         // Need to add these last, otherwise older updates will be cleaned
         for slot_id in slots {
-            accounts_index.roots.insert(slot_id);
+            accounts_index.add_root(slot_id);
         }
 
         let mut counts = HashMap::new();
@@ -2550,12 +2550,12 @@ pub mod tests {
 
         print_accounts("post_f", &accounts);
 
-        accounts.verify_bank_hash(4, &HashMap::default()).unwrap();
-
         assert_load_account(&accounts, current_slot, pubkey, some_lamport);
         assert_load_account(&accounts, current_slot, purged_pubkey1, 0);
         assert_load_account(&accounts, current_slot, purged_pubkey2, 0);
         assert_load_account(&accounts, current_slot, dummy_pubkey, dummy_lamport);
+
+        accounts.verify_bank_hash(4, &HashMap::default()).unwrap();
     }
 
     #[test]
@@ -2951,6 +2951,24 @@ pub mod tests {
         assert_eq!(1, db.get_snapshot_storages(after_slot).len());
     }
 
+    #[test]
+    fn test_get_snapshot_storages_exclude_empty() {
+        let db = AccountsDB::new(Vec::new());
+
+        let key = Pubkey::default();
+        let account = Account::new(1, 0, &key);
+        let base_slot = 0;
+        let after_slot = base_slot + 1;
+
+        db.store(base_slot, &[(&key, &account)]);
+        db.add_root(base_slot);
+        assert_eq!(1, db.get_snapshot_storages(after_slot).len());
+
+        let storage = db.storage.read().unwrap();
+        storage.0[&0].values().next().unwrap().remove_account();
+        assert!(db.get_snapshot_storages(after_slot).is_empty());
+    }
+
     #[test]
     #[should_panic(expected = "double remove of account in slot: 0/store: 0!!")]
     fn test_storage_remove_account_double_remove() {
@@ -2965,10 +2983,9 @@ pub mod tests {
     }
 
     #[test]
-    fn test_accounts_purge_chained_purge_after_snapshot_restore_complex() {
+    fn test_accounts_purge_long_chained_after_snapshot_restore() {
         solana_logger::setup();
         let old_lamport = 223;
-        let latest_lamport = 777;
         let zero_lamport = 0;
         let no_data = 0;
         let owner = Account::default().owner;
@@ -2977,24 +2994,28 @@ pub mod tests {
         let account2 = Account::new(old_lamport + 100_001, no_data, &owner);
         let account3 = Account::new(old_lamport + 100_002, no_data, &owner);
         let dummy_account = Account::new(99999999, no_data, &owner);
-        let latest_account = Account::new(latest_lamport, no_data, &owner);
         let zero_lamport_account = Account::new(zero_lamport, no_data, &owner);
 
         let pubkey = Pubkey::new_rand();
-        let zero_lamport_pubkey = Pubkey::new_rand();
         let dummy_pubkey = Pubkey::new_rand();
         let purged_pubkey1 = Pubkey::new_rand();
         let purged_pubkey2 = Pubkey::new_rand();
 
         let mut current_slot = 0;
         let accounts = AccountsDB::new_single();
 
+        // create intermidiate updates to purged_pubkey1 so that
+        // generate_index must add slots as root last at once
         current_slot += 1;
         accounts.store(current_slot, &[(&pubkey, &account)]);
-        accounts.store(
-            current_slot,
-            &[(&zero_lamport_pubkey, &zero_lamport_account)],
-        );
+        accounts.store(current_slot, &[(&purged_pubkey1, &account2)]);
+        accounts.add_root(current_slot);
+
+        current_slot += 1;
+        accounts.store(current_slot, &[(&purged_pubkey1, &account2)]);
+        accounts.add_root(current_slot);
+
+        current_slot += 1;
         accounts.store(current_slot, &[(&purged_pubkey1, &account2)]);
         accounts.add_root(current_slot);
 
@@ -3008,21 +3029,16 @@ pub mod tests {
         accounts.add_root(current_slot);
 
         current_slot += 1;
-        accounts.store(current_slot, &[(&pubkey, &latest_account)]);
         accounts.store(current_slot, &[(&dummy_pubkey, &dummy_account)]);
         accounts.add_root(current_slot);
 
-        current_slot += 1;
-        accounts.store(current_slot, &[(&pubkey, &latest_account)]);
-        accounts.add_root(current_slot);
-
         print_count_and_status("before reconstruct", &accounts);
         let accounts = reconstruct_accounts_db_via_serialization(&accounts, current_slot);
         print_count_and_status("before purge zero", &accounts);
         accounts.clean_accounts();
         print_count_and_status("after purge zero", &accounts);
 
-        assert_load_account(&accounts, current_slot, pubkey, latest_lamport);
+        assert_load_account(&accounts, current_slot, pubkey, old_lamport);
         assert_load_account(&accounts, current_slot, purged_pubkey1, 0);
         assert_load_account(&accounts, current_slot, purged_pubkey2, 0);
     }