Safe and sane multi-item storage removal

client/db/src/bench.rs

@@ -393,10 +393,11 @@ impl<B: BlockT> StateBackend<HashFor<B>> for BenchmarkingState<B> {
 		&self,
 		child_info: Option<&ChildInfo>,
 		prefix: Option<&[u8]>,
+		start_at: Option<&[u8]>,
 		f: F,
 	) {
 		if let Some(ref state) = *self.state.borrow() {
-			state.apply_to_keys_while(child_info, prefix, f)
+			state.apply_to_keys_while(child_info, prefix, start_at, f)
 		}
 	}
 

client/db/src/lib.rs

@@ -226,9 +226,10 @@ impl<B: BlockT> StateBackend<HashFor<B>> for RefTrackingState<B> {
 		&self,
 		child_info: Option<&ChildInfo>,
 		prefix: Option<&[u8]>,
+		start_at: Option<&[u8]>,
 		f: F,
 	) {
-		self.state.apply_to_keys_while(child_info, prefix, f)
+		self.state.apply_to_keys_while(child_info, prefix, start_at, f)
 	}
 
 	fn for_child_keys_with_prefix<F: FnMut(&[u8])>(

client/db/src/storage_cache.rs

@@ -639,9 +639,10 @@ impl<S: StateBackend<HashFor<B>>, B: BlockT> StateBackend<HashFor<B>> for Cachin
 		&self,
 		child_info: Option<&ChildInfo>,
 		prefix: Option<&[u8]>,
+		start_at: Option<&[u8]>,
 		f: F,
 	) {
-		self.state.apply_to_keys_while(child_info, prefix, f)
+		self.state.apply_to_keys_while(child_info, prefix, start_at, f)
 	}
 
 	fn next_storage_key(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {
@@ -839,9 +840,10 @@ impl<S: StateBackend<HashFor<B>>, B: BlockT> StateBackend<HashFor<B>>
 		&self,
 		child_info: Option<&ChildInfo>,
 		prefix: Option<&[u8]>,
+		start_at: Option<&[u8]>,
 		f: F,
 	) {
-		self.caching_state().apply_to_keys_while(child_info, prefix, f)
+		self.caching_state().apply_to_keys_while(child_info, prefix, start_at, f)
 	}
 
 	fn next_storage_key(&self, key: &[u8]) -> Result<Option<Vec<u8>>, Self::Error> {

frame/bags-list/src/list/mod.rs

@@ -94,7 +94,9 @@ impl<T: Config<I>, I: 'static> List<T, I> {
 	/// this function should generally not be used in production as it could lead to a very large
 	/// number of storage accesses.
 	pub(crate) fn unsafe_clear() {
+		#[allow(deprecated)]
 		crate::ListBags::<T, I>::remove_all(None);
+		#[allow(deprecated)]
 		crate::ListNodes::<T, I>::remove_all();
 	}
 

frame/contracts/src/migration.rs

@@ -56,6 +56,7 @@ mod v4 {
 	use super::*;
 
 	pub fn migrate<T: Config>() -> Weight {
+		#[allow(deprecated)]
 		migration::remove_storage_prefix(<Pallet<T>>::name().as_bytes(), b"CurrentSchedule", b"");
 		T::DbWeight::get().writes(1)
 	}

frame/contracts/src/storage.rs

@@ -27,12 +27,12 @@ use crate::{
 use codec::{Decode, Encode, MaxEncodedLen};
 use frame_support::{
 	dispatch::{DispatchError, DispatchResult},
-	storage::child::{self, ChildInfo, KillStorageResult},
+	storage::child::{self, ChildInfo},
 	weights::Weight,
 };
 use scale_info::TypeInfo;
 use sp_core::crypto::UncheckedFrom;
-use sp_io::hashing::blake2_256;
+use sp_io::{hashing::blake2_256, KillStorageResult};
 use sp_runtime::{
 	traits::{Hash, Zero},
 	RuntimeDebug,
@@ -266,16 +266,16 @@ where
 		while !queue.is_empty() && remaining_key_budget > 0 {
 			// Cannot panic due to loop condition
 			let trie = &mut queue[0];
-			let outcome =
-				child::kill_storage(&child_trie_info(&trie.trie_id), Some(remaining_key_budget));
+			#[allow(deprecated)]
+			let outcome = child::kill_storage(&child_trie_info(&trie.trie_id), Some(remaining_key_budget));
 			let keys_removed = match outcome {
 				// This happens when our budget wasn't large enough to remove all keys.
-				KillStorageResult::SomeRemaining(count) => count,
-				KillStorageResult::AllRemoved(count) => {
+				KillStorageResult::SomeRemaining(c) => c,
+				KillStorageResult::AllRemoved(c) => {
 					// We do not care to preserve order. The contract is deleted already and
 					// no one waits for the trie to be deleted.
 					queue.swap_remove(0);
-					count
+					c
 				},
 			};
 			remaining_key_budget = remaining_key_budget.saturating_sub(keys_removed);

frame/elections-phragmen/src/benchmarking.rs

@@ -148,6 +148,7 @@ fn clean<T: Config>() {
 	<Members<T>>::kill();
 	<Candidates<T>>::kill();
 	<RunnersUp<T>>::kill();
+	#[allow(deprecated)]
 	<Voting<T>>::remove_all(None);
 }
 

frame/im-online/src/lib.rs

@@ -900,7 +900,9 @@ impl<T: Config> OneSessionHandler<T::AccountId> for Pallet<T> {
 		// Remove all received heartbeats and number of authored blocks from the
 		// current session, they have already been processed and won't be needed
 		// anymore.
+		#[allow(deprecated)]
 		ReceivedHeartbeats::<T>::remove_prefix(&T::ValidatorSet::session_index(), None);
+		#[allow(deprecated)]
 		AuthoredBlocks::<T>::remove_prefix(&T::ValidatorSet::session_index(), None);
 
 		if offenders.is_empty() {

frame/scheduler/src/lib.rs

@@ -565,6 +565,7 @@ impl<T: Config> Pallet<T> {
 			},
 		);
 
+		#[allow(deprecated)]
 		frame_support::storage::migration::remove_storage_prefix(
 			Self::name().as_bytes(),
 			b"StorageVersion",
@@ -601,6 +602,7 @@ impl<T: Config> Pallet<T> {
 			)
 		});
 
+		#[allow(deprecated)]
 		frame_support::storage::migration::remove_storage_prefix(
 			Self::name().as_bytes(),
 			b"StorageVersion",

frame/society/src/lib.rs

@@ -1067,6 +1067,7 @@ pub mod pallet {
 			Founder::<T, I>::kill();
 			Rules::<T, I>::kill();
 			Candidates::<T, I>::kill();
+			#[allow(deprecated)]
 			SuspendedCandidates::<T, I>::remove_all(None);
 			Self::deposit_event(Event::<T, I>::Unfounded { founder });
 			Ok(())
@@ -1511,6 +1512,7 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 				.collect::<Vec<_>>();
 
 			// Clean up all votes.
+			#[allow(deprecated)]
 			<Votes<T, I>>::remove_all(None);
 
 			// Reward one of the voters who voted the right way.
@@ -1695,6 +1697,7 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 				}
 
 				// Clean up all votes.
+				#[allow(deprecated)]
 				<DefenderVotes<T, I>>::remove_all(None);
 			}
 

frame/staking/src/pallet/impls.rs

@@ -585,8 +585,11 @@ impl<T: Config> Pallet<T> {
 
 	/// Clear all era information for given era.
 	pub(crate) fn clear_era_information(era_index: EraIndex) {
+		#[allow(deprecated)]
 		<ErasStakers<T>>::remove_prefix(era_index, None);
+		#[allow(deprecated)]
 		<ErasStakersClipped<T>>::remove_prefix(era_index, None);
+		#[allow(deprecated)]
 		<ErasValidatorPrefs<T>>::remove_prefix(era_index, None);
 		<ErasValidatorReward<T>>::remove(era_index);
 		<ErasRewardPoints<T>>::remove(era_index);
@@ -984,9 +987,13 @@ impl<T: Config> ElectionDataProvider for Pallet<T> {
 
 	#[cfg(feature = "runtime-benchmarks")]
 	fn clear() {
+		#[allow(deprecated)]
 		<Bonded<T>>::remove_all(None);
+		#[allow(deprecated)]
 		<Ledger<T>>::remove_all(None);
+		#[allow(deprecated)]
 		<Validators<T>>::remove_all();
+		#[allow(deprecated)]
 		<Nominators<T>>::remove_all();
 
 		T::VoterList::unsafe_clear();
@@ -1368,7 +1375,9 @@ impl<T: Config> SortedListProvider<T::AccountId> for UseNominatorsAndValidatorsM
 	fn unsafe_clear() {
 		// NOTE: Caller must ensure this doesn't lead to too many storage accesses. This is a
 		// condition of SortedListProvider::unsafe_clear.
+		#[allow(deprecated)]
 		Nominators::<T>::remove_all();
+		#[allow(deprecated)]
 		Validators::<T>::remove_all();
 	}
 }

frame/staking/src/slashing.rs

@@ -557,7 +557,9 @@ impl<'a, T: 'a + Config> Drop for InspectingSpans<'a, T> {
 
 /// Clear slashing metadata for an obsolete era.
 pub(crate) fn clear_era_metadata<T: Config>(obsolete_era: EraIndex) {
+	#[allow(deprecated)]
 	<Pallet<T> as Store>::ValidatorSlashInEra::remove_prefix(&obsolete_era, None);
+	#[allow(deprecated)]
 	<Pallet<T> as Store>::NominatorSlashInEra::remove_prefix(&obsolete_era, None);
 }
 

frame/staking/src/testing_utils.rs

@@ -36,9 +36,11 @@ const SEED: u32 = 0;
 
 /// This function removes all validators and nominators from storage.
 pub fn clear_validators_and_nominators<T: Config>() {
+	#[allow(deprecated)]
 	Validators::<T>::remove_all();
 
 	// whenever we touch nominators counter we should update `T::VoterList` as well.
+	#[allow(deprecated)]
 	Nominators::<T>::remove_all();
 
 	// NOTE: safe to call outside block production

frame/support/src/lib.rs

@@ -1097,8 +1097,12 @@ pub mod tests {
 			DoubleMap::insert(&(key1 + 1), &key2, &4u64);
 			DoubleMap::insert(&(key1 + 1), &(key2 + 1), &4u64);
 			assert!(matches!(
-				DoubleMap::remove_prefix(&key1, None),
-				sp_io::KillStorageResult::AllRemoved(0), // all in overlay
+				DoubleMap::clear_prefix(&key1, u32::max_value(), None),
+				// Note this is the incorrect answer (for now), since we are using v2 of
+				// `clear_prefix`.
+				// When we switch to v3, then this will become:
+				//   sp_io::MultiRemovalResults::NoneLeft { db: 0, total: 2 },
+				sp_io::MultiRemovalResults { maybe_cursor: None, backend: 0, unique: 0, loops: 0 },
 			));
 			assert_eq!(DoubleMap::get(&key1, &key2), 0u64);
 			assert_eq!(DoubleMap::get(&key1, &(key2 + 1)), 0u64);

frame/support/src/storage/child.rs

@@ -21,7 +21,7 @@
 // NOTE: could replace unhashed by having only one kind of storage (top trie being the child info
 // of null length parent storage key).
 
-pub use crate::sp_io::KillStorageResult;
+pub use crate::sp_io::{KillStorageResult, MultiRemovalResults};
 use crate::sp_std::prelude::*;
 use codec::{Codec, Decode, Encode};
 pub use sp_core::storage::{ChildInfo, ChildType, StateVersion};
@@ -136,13 +136,66 @@ pub fn exists(child_info: &ChildInfo, key: &[u8]) -> bool {
 /// not make much sense because it is not cumulative when called inside the same block.
 /// Use this function to distribute the deletion of a single child trie across multiple
 /// blocks.
+#[deprecated = "Use `clear_storage` instead"]
 pub fn kill_storage(child_info: &ChildInfo, limit: Option<u32>) -> KillStorageResult {
 	match child_info.child_type() {
 		ChildType::ParentKeyId =>
 			sp_io::default_child_storage::storage_kill(child_info.storage_key(), limit),
 	}
 }
 
+/// Partially clear the child storage of each key-value pair.
+///
+/// # Limit
+///
+/// A *limit* should always be provided through `maybe_limit`. This is one fewer than the
+/// maximum number of backend iterations which may be done by this operation and as such
+/// represents the maximum number of backend deletions which may happen. A *limit* of zero
+/// implies that no keys will be deleted, though there may be a single iteration done.
+///
+/// The limit can be used to partially delete storage items in case it is too large or costly
+/// to delete all in a single operation.
+///
+/// # Cursor
+///
+/// A *cursor* may be passed in to this operation with `maybe_cursor`. `None` should only be
+/// passed once (in the initial call) for any attempt to clear storage. In general, subsequent calls
+/// operating on the same prefix should pass `Some` and this value should be equal to the
+/// previous call result's `maybe_cursor` field. The only exception to this is when you can
+/// guarantee that the subsequent call is in a new block; in this case the previous call's result
+/// cursor need not be passed in an a `None` may be passed instead. This exception may be useful
+/// then making this call solely from a block-hook such as `on_initialize`.
+///
+/// Returns [`MultiRemovalResults`](sp_io::MultiRemovalResults) to inform about the result. Once the
+/// resultant `maybe_cursor` field is `None`, then no further items remain to be deleted.
+///
+/// NOTE: After the initial call for any given child storage, it is important that no keys further
+/// keys are inserted. If so, then they may or may not be deleted by subsequent calls.
+///
+/// # Note
+///
+/// Please note that keys which are residing in the overlay for the child are deleted without
+/// counting towards the `limit`.
+pub fn clear_storage(
+	child_info: &ChildInfo,
+	maybe_limit: Option<u32>,
+	_maybe_cursor: Option<&[u8]>,
+) -> MultiRemovalResults {
+	// TODO: Once the network has upgraded to include the new host functions, this code can be
+	// enabled.
+	// sp_io::default_child_storage::storage_kill(prefix, maybe_limit, maybe_cursor)
+	let r = match child_info.child_type() {
+		ChildType::ParentKeyId =>
+			sp_io::default_child_storage::storage_kill(child_info.storage_key(), maybe_limit),
+	};
+	use sp_io::KillStorageResult::*;
+	let (maybe_cursor, backend) = match r {
+		AllRemoved(db) => (None, db),
+		SomeRemaining(db) => (Some(child_info.storage_key().to_vec()), db),
+	};
+	MultiRemovalResults { maybe_cursor, backend, unique: backend, loops: backend }
+}
+
 /// Ensure `key` has no explicit entry in storage.
 pub fn kill(child_info: &ChildInfo, key: &[u8]) {
 	match child_info.child_type() {

frame/support/src/storage/generator/double_map.rs

@@ -202,11 +202,28 @@ where
 		unhashed::kill(&Self::storage_double_map_final_key(k1, k2))
 	}
 
-	fn remove_prefix<KArg1>(k1: KArg1, limit: Option<u32>) -> sp_io::KillStorageResult
+	fn remove_prefix<KArg1>(k1: KArg1, maybe_limit: Option<u32>) -> sp_io::KillStorageResult
 	where
 		KArg1: EncodeLike<K1>,
 	{
-		unhashed::kill_prefix(Self::storage_double_map_final_key1(k1).as_ref(), limit)
+		unhashed::clear_prefix(Self::storage_double_map_final_key1(k1).as_ref(), maybe_limit, None)
+			.into()
+	}
+
+	fn clear_prefix<KArg1>(
+		k1: KArg1,
+		limit: u32,
+		maybe_cursor: Option<&[u8]>,
+	) -> sp_io::MultiRemovalResults
+	where
+		KArg1: EncodeLike<K1>,
+	{
+		unhashed::clear_prefix(
+			Self::storage_double_map_final_key1(k1).as_ref(),
+			Some(limit),
+			maybe_cursor,
+		)
+		.into()
 	}
 
 	fn iter_prefix_values<KArg1>(k1: KArg1) -> storage::PrefixIterator<V>

frame/support/src/storage/generator/nmap.rs

@@ -183,7 +183,22 @@ where
 	where
 		K: HasKeyPrefix<KP>,
 	{
-		unhashed::kill_prefix(&Self::storage_n_map_partial_key(partial_key), limit)
+		unhashed::clear_prefix(&Self::storage_n_map_partial_key(partial_key), limit, None).into()
+	}
+
+	fn clear_prefix<KP>(
+		partial_key: KP,
+		limit: u32,
+		maybe_cursor: Option<&[u8]>,
+	) -> sp_io::MultiRemovalResults
+	where
+		K: HasKeyPrefix<KP>,
+	{
+		unhashed::clear_prefix(
+			&Self::storage_n_map_partial_key(partial_key),
+			Some(limit),
+			maybe_cursor,
+		)
 	}
 
 	fn iter_prefix_values<KP>(partial_key: KP) -> PrefixIterator<V>