[Contracts] Overflowing bounded DeletionQueue allows DoS against contract termination (#13702)

* [Contracts review] Overflowing bounded `DeletionQueue` allows DoS against contract termination

* wip

* wip

* wip

* wip

* wip

* fix doc

* wip

* PR review

* unbreak tests

* fixes

* update budget computation

* PR comment: use BlockWeights::get().max_block

* PR comment: Update queue_trie_for_deletion signature

* PR comment: update deletion budget docstring

* PR comment: impl Default with derive(DefaultNoBound)

* PR comment: Remove DeletedContract

* PR comment Add ring_buffer test

* remove missed comment

* misc comments

* contracts: add sr25519_recover

* Revert "contracts: add sr25519_recover"

This reverts commit d4600e00934b90e5882cf5288f36f98911b51722.

* ".git/.scripts/commands/bench/bench.sh" pallet dev pallet_contracts

* PR comments update print_schedule

* Update frame/contracts/src/benchmarking/mod.rs

* Update frame/contracts/src/storage.rs

* Update frame/contracts/src/storage.rs

* rm temporary fixes

* fix extra ;

* Update frame/contracts/src/storage.rs

Co-authored-by: juangirini <juangirini@gmail.com>

* Update frame/contracts/src/storage.rs

Co-authored-by: Alexander Theißen <alex.theissen@me.com>

* Update frame/contracts/src/lib.rs

Co-authored-by: Alexander Theißen <alex.theissen@me.com>

* Update frame/contracts/src/lib.rs

Co-authored-by: Alexander Theißen <alex.theissen@me.com>

* Support stable rust for compiling the runtime (#13580)

* Support stable rust for compiling the runtime

This pull request brings support for compiling the runtime with stable Rust. This requires at least
rust 1.68.0 to work on stable. The code is written in a way that it is backwards compatible and
should automatically work when someone compiles with 1.68.0+ stable.

* We always support nightlies!

* 🤦

* Sort by version

* Review feedback

* Review feedback

* Fix version parsing

* Apply suggestions from code review

Co-authored-by: Koute <koute@users.noreply.github.com>

---------

Co-authored-by: Koute <koute@users.noreply.github.com>

* github PR commit fixes

* Revert "Support stable rust for compiling the runtime (#13580)"

This reverts commit 0b985aa5ad114a42003519b712d25a6acc40b0ad.

* Restore DeletionQueueMap

* fix namings

* PR comment

* move comments

* Update frame/contracts/src/storage.rs

* Update frame/contracts/src/storage.rs

* fixes

---------

Co-authored-by: command-bot <>
Co-authored-by: juangirini <juangirini@gmail.com>
Co-authored-by: Alexander Theißen <alex.theissen@me.com>
Co-authored-by: Bastian Köcher <git@kchr.de>
Co-authored-by: Koute <koute@users.noreply.github.com>

substrate/bin/node/runtime/src/lib.rs

@@ -1198,13 +1198,6 @@ impl pallet_tips::Config for Runtime {
 parameter_types! {
 	pub const DepositPerItem: Balance = deposit(1, 0);
 	pub const DepositPerByte: Balance = deposit(0, 1);
-	pub const DeletionQueueDepth: u32 = 128;
-	// The lazy deletion runs inside on_initialize.
-	pub DeletionWeightLimit: Weight = RuntimeBlockWeights::get()
-		.per_class
-		.get(DispatchClass::Normal)
-		.max_total
-		.unwrap_or(RuntimeBlockWeights::get().max_block);
 	pub Schedule: pallet_contracts::Schedule<Runtime> = Default::default();
 }
 
@@ -1227,8 +1220,6 @@ impl pallet_contracts::Config for Runtime {
 	type WeightPrice = pallet_transaction_payment::Pallet<Self>;
 	type WeightInfo = pallet_contracts::weights::SubstrateWeight<Self>;
 	type ChainExtension = ();
-	type DeletionQueueDepth = DeletionQueueDepth;
-	type DeletionWeightLimit = DeletionWeightLimit;
 	type Schedule = Schedule;
 	type AddressGenerator = pallet_contracts::DefaultAddressGenerator;
 	type MaxCodeLen = ConstU32<{ 123 * 1024 }>;

substrate/frame/contracts/src/benchmarking/mod.rs

@@ -214,19 +214,7 @@ benchmarks! {
 	on_initialize_per_trie_key {
 		let k in 0..1024;
 		let instance = Contract::<T>::with_storage(WasmModule::dummy(), k, T::Schedule::get().limits.payload_len)?;
-		instance.info()?.queue_trie_for_deletion()?;
-	}: {
-		ContractInfo::<T>::process_deletion_queue_batch(Weight::MAX)
-	}
-
-	#[pov_mode = Measured]
-	on_initialize_per_queue_item {
-		let q in 0..1024.min(T::DeletionQueueDepth::get());
-		for i in 0 .. q {
-			let instance = Contract::<T>::with_index(i, WasmModule::dummy(), vec![])?;
-			instance.info()?.queue_trie_for_deletion()?;
-			ContractInfoOf::<T>::remove(instance.account_id);
-		}
+		instance.info()?.queue_trie_for_deletion();
 	}: {
 		ContractInfo::<T>::process_deletion_queue_batch(Weight::MAX)
 	}
@@ -3020,16 +3008,12 @@ benchmarks! {
 	print_schedule {
 		#[cfg(feature = "std")]
 		{
-			let weight_limit = T::DeletionWeightLimit::get();
-			let max_queue_depth = T::DeletionQueueDepth::get() as usize;
-			let empty_queue_throughput = ContractInfo::<T>::deletion_budget(0, weight_limit);
-			let full_queue_throughput = ContractInfo::<T>::deletion_budget(max_queue_depth, weight_limit);
+			let max_weight = <T as frame_system::Config>::BlockWeights::get().max_block;
+			let (weight_per_key, key_budget) = ContractInfo::<T>::deletion_budget(max_weight);
 			println!("{:#?}", Schedule::<T>::default());
 			println!("###############################################");
-			println!("Lazy deletion weight per key: {}", empty_queue_throughput.0);
-			println!("Lazy deletion throughput per block (empty queue, full queue): {}, {}",
-				empty_queue_throughput.1, full_queue_throughput.1,
-			);
+			println!("Lazy deletion weight per key: {weight_per_key}");
+			println!("Lazy deletion throughput per block: {key_budget}");
 		}
 		#[cfg(not(feature = "std"))]
 		Err("Run this bench with a native runtime in order to see the schedule.")?;

substrate/frame/contracts/src/exec.rs

@@ -1204,7 +1204,7 @@ where
 			T::Currency::reducible_balance(&frame.account_id, Expendable, Polite),
 			ExistenceRequirement::AllowDeath,
 		)?;
-		info.queue_trie_for_deletion()?;
+		info.queue_trie_for_deletion();
 		ContractInfoOf::<T>::remove(&frame.account_id);
 		E::remove_user(info.code_hash);
 		Contracts::<T>::deposit_event(

substrate/frame/contracts/src/lib.rs

@@ -102,7 +102,7 @@ mod tests;
 use crate::{
 	exec::{AccountIdOf, ErrorOrigin, ExecError, Executable, Key, Stack as ExecStack},
 	gas::GasMeter,
-	storage::{meter::Meter as StorageMeter, ContractInfo, DeletedContract},
+	storage::{meter::Meter as StorageMeter, ContractInfo, DeletionQueueManager},
 	wasm::{OwnerInfo, PrefabWasmModule, TryInstantiate},
 	weights::WeightInfo,
 };
@@ -245,33 +245,6 @@ pub mod pallet {
 		/// memory usage of your runtime.
 		type CallStack: Array<Item = Frame<Self>>;
 
-		/// The maximum number of contracts that can be pending for deletion.
-		///
-		/// When a contract is deleted by calling `seal_terminate` it becomes inaccessible
-		/// immediately, but the deletion of the storage items it has accumulated is performed
-		/// later. The contract is put into the deletion queue. This defines how many
-		/// contracts can be queued up at the same time. If that limit is reached `seal_terminate`
-		/// will fail. The action must be retried in a later block in that case.
-		///
-		/// The reasons for limiting the queue depth are:
-		///
-		/// 1. The queue is in storage in order to be persistent between blocks. We want to limit
-		/// 	the amount of storage that can be consumed.
-		/// 2. The queue is stored in a vector and needs to be decoded as a whole when reading
-		///		it at the end of each block. Longer queues take more weight to decode and hence
-		///		limit the amount of items that can be deleted per block.
-		#[pallet::constant]
-		type DeletionQueueDepth: Get<u32>;
-
-		/// The maximum amount of weight that can be consumed per block for lazy trie removal.
-		///
-		/// The amount of weight that is dedicated per block to work on the deletion queue. Larger
-		/// values allow more trie keys to be deleted in each block but reduce the amount of
-		/// weight that is left for transactions. See [`Self::DeletionQueueDepth`] for more
-		/// information about the deletion queue.
-		#[pallet::constant]
-		type DeletionWeightLimit: Get<Weight>;
-
 		/// The amount of balance a caller has to pay for each byte of storage.
 		///
 		/// # Note
@@ -329,25 +302,6 @@ pub mod pallet {
 				.saturating_add(T::WeightInfo::on_process_deletion_queue_batch())
 		}
 
-		fn on_initialize(_block: T::BlockNumber) -> Weight {
-			// We want to process the deletion_queue in the on_idle hook. Only in the case
-			// that the queue length has reached its maximal depth, we process it here.
-			let max_len = T::DeletionQueueDepth::get() as usize;
-			let queue_len = <DeletionQueue<T>>::decode_len().unwrap_or(0);
-			if queue_len >= max_len {
-				// We do not want to go above the block limit and rather avoid lazy deletion
-				// in that case. This should only happen on runtime upgrades.
-				let weight_limit = T::BlockWeights::get()
-					.max_block
-					.saturating_sub(System::<T>::block_weight().total())
-					.min(T::DeletionWeightLimit::get());
-				ContractInfo::<T>::process_deletion_queue_batch(weight_limit)
-					.saturating_add(T::WeightInfo::on_process_deletion_queue_batch())
-			} else {
-				T::WeightInfo::on_process_deletion_queue_batch()
-			}
-		}
-
 		fn integrity_test() {
 			// Total runtime memory is expected to have 128Mb upper limit
 			const MAX_RUNTIME_MEM: u32 = 1024 * 1024 * 128;
@@ -860,12 +814,6 @@ pub mod pallet {
 		/// in this error. Note that this usually  shouldn't happen as deploying such contracts
 		/// is rejected.
 		NoChainExtension,
-		/// Removal of a contract failed because the deletion queue is full.
-		///
-		/// This can happen when calling `seal_terminate`.
-		/// The queue is filled by deleting contracts and emptied by a fixed amount each block.
-		/// Trying again during another block is the only way to resolve this issue.
-		DeletionQueueFull,
 		/// A contract with the same AccountId already exists.
 		DuplicateContract,
 		/// A contract self destructed in its constructor.
@@ -949,10 +897,15 @@ pub mod pallet {
 	/// Evicted contracts that await child trie deletion.
 	///
 	/// Child trie deletion is a heavy operation depending on the amount of storage items
-	/// stored in said trie. Therefore this operation is performed lazily in `on_initialize`.
+	/// stored in said trie. Therefore this operation is performed lazily in `on_idle`.
 	#[pallet::storage]
-	pub(crate) type DeletionQueue<T: Config> =
-		StorageValue<_, BoundedVec<DeletedContract, T::DeletionQueueDepth>, ValueQuery>;
+	pub(crate) type DeletionQueue<T: Config> = StorageMap<_, Twox64Concat, u32, TrieId>;
+
+	/// A pair of monotonic counters used to track the latest contract marked for deletion
+	/// and the latest deleted contract in queue.
+	#[pallet::storage]
+	pub(crate) type DeletionQueueCounter<T: Config> =
+		StorageValue<_, DeletionQueueManager<T>, ValueQuery>;
 }
 
 /// Context of a contract invocation.

substrate/frame/contracts/src/storage.rs

@@ -22,15 +22,15 @@ pub mod meter;
 use crate::{
 	exec::{AccountIdOf, Key},
 	weights::WeightInfo,
-	AddressGenerator, BalanceOf, CodeHash, Config, ContractInfoOf, DeletionQueue, Error, Pallet,
-	TrieId, SENTINEL,
+	AddressGenerator, BalanceOf, CodeHash, Config, ContractInfoOf, DeletionQueue,
+	DeletionQueueCounter, Error, Pallet, TrieId, SENTINEL,
 };
 use codec::{Decode, Encode, MaxEncodedLen};
 use frame_support::{
-	dispatch::{DispatchError, DispatchResult},
+	dispatch::DispatchError,
 	storage::child::{self, ChildInfo},
 	weights::Weight,
-	RuntimeDebugNoBound,
+	DefaultNoBound, RuntimeDebugNoBound,
 };
 use scale_info::TypeInfo;
 use sp_io::KillStorageResult;
@@ -38,7 +38,7 @@ use sp_runtime::{
 	traits::{Hash, Saturating, Zero},
 	RuntimeDebug,
 };
-use sp_std::{ops::Deref, prelude::*};
+use sp_std::{marker::PhantomData, ops::Deref, prelude::*};
 
 /// Information for managing an account and its sub trie abstraction.
 /// This is the required info to cache for an account.
@@ -204,27 +204,21 @@ impl<T: Config> ContractInfo<T> {
 	/// Push a contract's trie to the deletion queue for lazy removal.
 	///
 	/// You must make sure that the contract is also removed when queuing the trie for deletion.
-	pub fn queue_trie_for_deletion(&self) -> DispatchResult {
-		<DeletionQueue<T>>::try_append(DeletedContract { trie_id: self.trie_id.clone() })
-			.map_err(|_| <Error<T>>::DeletionQueueFull.into())
+	pub fn queue_trie_for_deletion(&self) {
+		DeletionQueueManager::<T>::load().insert(self.trie_id.clone());
 	}
 
 	/// Calculates the weight that is necessary to remove one key from the trie and how many
-	/// of those keys can be deleted from the deletion queue given the supplied queue length
-	/// and weight limit.
-	pub fn deletion_budget(queue_len: usize, weight_limit: Weight) -> (Weight, u32) {
+	/// of those keys can be deleted from the deletion queue given the supplied weight limit.
+	pub fn deletion_budget(weight_limit: Weight) -> (Weight, u32) {
 		let base_weight = T::WeightInfo::on_process_deletion_queue_batch();
-		let weight_per_queue_item = T::WeightInfo::on_initialize_per_queue_item(1) -
-			T::WeightInfo::on_initialize_per_queue_item(0);
 		let weight_per_key = T::WeightInfo::on_initialize_per_trie_key(1) -
 			T::WeightInfo::on_initialize_per_trie_key(0);
-		let decoding_weight = weight_per_queue_item.saturating_mul(queue_len as u64);
 
 		// `weight_per_key` being zero makes no sense and would constitute a failure to
 		// benchmark properly. We opt for not removing any keys at all in this case.
 		let key_budget = weight_limit
 			.saturating_sub(base_weight)
-			.saturating_sub(decoding_weight)
 			.checked_div_per_component(&weight_per_key)
 			.unwrap_or(0) as u32;
 
@@ -235,13 +229,13 @@ impl<T: Config> ContractInfo<T> {
 	///
 	/// It returns the amount of weight used for that task.
 	pub fn process_deletion_queue_batch(weight_limit: Weight) -> Weight {
-		let queue_len = <DeletionQueue<T>>::decode_len().unwrap_or(0);
-		if queue_len == 0 {
+		let mut queue = <DeletionQueueManager<T>>::load();
+
+		if queue.is_empty() {
 			return Weight::zero()
 		}
 
-		let (weight_per_key, mut remaining_key_budget) =
-			Self::deletion_budget(queue_len, weight_limit);
+		let (weight_per_key, mut remaining_key_budget) = Self::deletion_budget(weight_limit);
 
 		// We want to check whether we have enough weight to decode the queue before
 		// proceeding. Too little weight for decoding might happen during runtime upgrades
@@ -250,30 +244,25 @@ impl<T: Config> ContractInfo<T> {
 			return weight_limit
 		}
 
-		let mut queue = <DeletionQueue<T>>::get();
+		while remaining_key_budget > 0 {
+			let Some(entry) = queue.next() else { break };
 
-		while !queue.is_empty() && remaining_key_budget > 0 {
-			// Cannot panic due to loop condition
-			let trie = &mut queue[0];
 			#[allow(deprecated)]
 			let outcome = child::kill_storage(
-				&ChildInfo::new_default(&trie.trie_id),
+				&ChildInfo::new_default(&entry.trie_id),
 				Some(remaining_key_budget),
 			);
-			let keys_removed = match outcome {
+
+			match outcome {
 				// This happens when our budget wasn't large enough to remove all keys.
-				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);
-					c
+				KillStorageResult::SomeRemaining(_) => return weight_limit,
+				KillStorageResult::AllRemoved(keys_removed) => {
+					entry.remove();
+					remaining_key_budget = remaining_key_budget.saturating_sub(keys_removed);
 				},
 			};
-			remaining_key_budget = remaining_key_budget.saturating_sub(keys_removed);
 		}
 
-		<DeletionQueue<T>>::put(queue);
 		weight_limit.saturating_sub(weight_per_key.saturating_mul(u64::from(remaining_key_budget)))
 	}
 
@@ -281,25 +270,9 @@ impl<T: Config> ContractInfo<T> {
 	pub fn load_code_hash(account: &AccountIdOf<T>) -> Option<CodeHash<T>> {
 		<ContractInfoOf<T>>::get(account).map(|i| i.code_hash)
 	}
-
-	/// Fill up the queue in order to exercise the limits during testing.
-	#[cfg(test)]
-	pub fn fill_queue_with_dummies() {
-		use frame_support::{traits::Get, BoundedVec};
-		let queue: Vec<DeletedContract> = (0..T::DeletionQueueDepth::get())
-			.map(|_| DeletedContract { trie_id: TrieId::default() })
-			.collect();
-		let bounded: BoundedVec<_, _> = queue.try_into().map_err(|_| ()).unwrap();
-		<DeletionQueue<T>>::put(bounded);
-	}
 }
 
-#[derive(Encode, Decode, TypeInfo, MaxEncodedLen)]
-pub struct DeletedContract {
-	pub(crate) trie_id: TrieId,
-}
-
-/// Information about what happended to the pre-existing value when calling [`ContractInfo::write`].
+/// Information about what happened to the pre-existing value when calling [`ContractInfo::write`].
 #[cfg_attr(test, derive(Debug, PartialEq))]
 pub enum WriteOutcome {
 	/// No value existed at the specified key.
@@ -352,3 +325,84 @@ impl<T: Config> Deref for DepositAccount<T> {
 		&self.0
 	}
 }
+
+/// Manage the removal of contracts storage that are marked for deletion.
+///
+/// When a contract is deleted by calling `seal_terminate` it becomes inaccessible
+/// immediately, but the deletion of the storage items it has accumulated is performed
+/// later by pulling the contract from the queue in the `on_idle` hook.
+#[derive(Encode, Decode, TypeInfo, MaxEncodedLen, DefaultNoBound, Clone)]
+#[scale_info(skip_type_params(T))]
+pub struct DeletionQueueManager<T: Config> {
+	/// Counter used as a key for inserting a new deleted contract in the queue.
+	/// The counter is incremented after each insertion.
+	insert_counter: u32,
+	/// The index used to read the next element to be deleted in the queue.
+	/// The counter is incremented after each deletion.
+	delete_counter: u32,
+
+	_phantom: PhantomData<T>,
+}
+
+/// View on a contract that is marked for deletion.
+struct DeletionQueueEntry<'a, T: Config> {
+	/// the trie id of the contract to delete.
+	trie_id: TrieId,
+
+	/// A mutable reference on the queue so that the contract can be removed, and none can be added
+	/// or read in the meantime.
+	queue: &'a mut DeletionQueueManager<T>,
+}
+
+impl<'a, T: Config> DeletionQueueEntry<'a, T> {
+	/// Remove the contract from the deletion queue.
+	fn remove(self) {
+		<DeletionQueue<T>>::remove(self.queue.delete_counter);
+		self.queue.delete_counter = self.queue.delete_counter.wrapping_add(1);
+		<DeletionQueueCounter<T>>::set(self.queue.clone());
+	}
+}
+
+impl<T: Config> DeletionQueueManager<T> {
+	/// Load the `DeletionQueueCounter`, so we can perform read or write operations on the
+	/// DeletionQueue storage.
+	fn load() -> Self {
+		<DeletionQueueCounter<T>>::get()
+	}
+
+	/// Returns `true` if the queue contains no elements.
+	fn is_empty(&self) -> bool {
+		self.insert_counter.wrapping_sub(self.delete_counter) == 0
+	}
+
+	/// Insert a contract in the deletion queue.
+	fn insert(&mut self, trie_id: TrieId) {
+		<DeletionQueue<T>>::insert(self.insert_counter, trie_id);
+		self.insert_counter = self.insert_counter.wrapping_add(1);
+		<DeletionQueueCounter<T>>::set(self.clone());
+	}
+
+	/// Fetch the next contract to be deleted.
+	///
+	/// Note:
+	/// we use the delete counter to get the next value to read from the queue and thus don't pay
+	/// the cost of an extra call to `sp_io::storage::next_key` to lookup the next entry in the map
+	fn next(&mut self) -> Option<DeletionQueueEntry<T>> {
+		if self.is_empty() {
+			return None
+		}
+
+		let entry = <DeletionQueue<T>>::get(self.delete_counter);
+		entry.map(|trie_id| DeletionQueueEntry { trie_id, queue: self })
+	}
+}
+
+#[cfg(test)]
+impl<T: Config> DeletionQueueManager<T> {
+	pub fn from_test_values(insert_counter: u32, delete_counter: u32) -> Self {
+		Self { insert_counter, delete_counter, _phantom: Default::default() }
+	}
+	pub fn as_test_tuple(&self) -> (u32, u32) {
+		(self.insert_counter, self.delete_counter)
+	}
+}

substrate/frame/contracts/src/tests.rs

@@ -22,26 +22,27 @@ use crate::{
 		Result as ExtensionResult, RetVal, ReturnFlags, SysConfig,
 	},
 	exec::{Frame, Key},
+	storage::DeletionQueueManager,
 	tests::test_utils::{get_contract, get_contract_checked},
 	wasm::{Determinism, PrefabWasmModule, ReturnCode as RuntimeReturnCode},
 	weights::WeightInfo,
 	BalanceOf, Code, CodeStorage, Config, ContractInfo, ContractInfoOf, DefaultAddressGenerator,
-	DeletionQueue, Error, Pallet, Schedule,
+	DeletionQueueCounter, Error, Pallet, Schedule,
 };
 use assert_matches::assert_matches;
 use codec::Encode;
 use frame_support::{
 	assert_err, assert_err_ignore_postinfo, assert_noop, assert_ok,
-	dispatch::{DispatchClass, DispatchErrorWithPostInfo, PostDispatchInfo},
+	dispatch::{DispatchErrorWithPostInfo, PostDispatchInfo},
 	parameter_types,
 	storage::child,
 	traits::{
-		ConstU32, ConstU64, Contains, Currency, ExistenceRequirement, Get, LockableCurrency,
-		OnIdle, OnInitialize, WithdrawReasons,
+		ConstU32, ConstU64, Contains, Currency, ExistenceRequirement, LockableCurrency, OnIdle,
+		OnInitialize, WithdrawReasons,
 	},
 	weights::{constants::WEIGHT_REF_TIME_PER_SECOND, Weight},
 };
-use frame_system::{self as system, EventRecord, Phase};
+use frame_system::{EventRecord, Phase};
 use pretty_assertions::{assert_eq, assert_ne};
 use sp_io::hashing::blake2_256;
 use sp_keystore::{testing::MemoryKeystore, KeystoreExt};
@@ -383,7 +384,6 @@ impl Contains<RuntimeCall> for TestFilter {
 }
 
 parameter_types! {
-	pub const DeletionWeightLimit: Weight = GAS_LIMIT;
 	pub static UnstableInterface: bool = true;
 }
 
@@ -399,8 +399,6 @@ impl Config for Test {
 	type WeightInfo = ();
 	type ChainExtension =
 		(TestExtension, DisabledExtension, RevertingExtension, TempStorageExtension);
-	type DeletionQueueDepth = ConstU32<1024>;
-	type DeletionWeightLimit = DeletionWeightLimit;
 	type Schedule = MySchedule;
 	type DepositPerByte = DepositPerByte;
 	type DepositPerItem = DepositPerItem;
@@ -1972,25 +1970,6 @@ fn lazy_removal_works() {
 	});
 }
 
-#[test]
-fn lazy_removal_on_full_queue_works_on_initialize() {
-	ExtBuilder::default().existential_deposit(50).build().execute_with(|| {
-		// Fill the deletion queue with dummy values, so that on_initialize attempts
-		// to clear the queue
-		ContractInfo::<Test>::fill_queue_with_dummies();
-
-		let queue_len_initial = <DeletionQueue<Test>>::decode_len().unwrap_or(0);
-
-		// Run the lazy removal
-		Contracts::on_initialize(System::block_number());
-
-		let queue_len_after_on_initialize = <DeletionQueue<Test>>::decode_len().unwrap_or(0);
-
-		// Queue length should be decreased after call of on_initialize()
-		assert!(queue_len_initial - queue_len_after_on_initialize > 0);
-	});
-}
-
 #[test]
 fn lazy_batch_removal_works() {
 	let (code, _hash) = compile_module::<Test>("self_destruct").unwrap();
@@ -2054,7 +2033,7 @@ fn lazy_removal_partial_remove_works() {
 	// We create a contract with some extra keys above the weight limit
 	let extra_keys = 7u32;
 	let weight_limit = Weight::from_parts(5_000_000_000, 0);
-	let (_, max_keys) = ContractInfo::<Test>::deletion_budget(1, weight_limit);
+	let (_, max_keys) = ContractInfo::<Test>::deletion_budget(weight_limit);
 	let vals: Vec<_> = (0..max_keys + extra_keys)
 		.map(|i| (blake2_256(&i.encode()), (i as u32), (i as u32).encode()))
 		.collect();
@@ -2139,33 +2118,6 @@ fn lazy_removal_partial_remove_works() {
 	});
 }
 
-#[test]
-fn lazy_removal_does_no_run_on_full_queue_and_full_block() {
-	ExtBuilder::default().existential_deposit(50).build().execute_with(|| {
-		// Fill up the block which should prevent the lazy storage removal from running.
-		System::register_extra_weight_unchecked(
-			<Test as system::Config>::BlockWeights::get().max_block,
-			DispatchClass::Mandatory,
-		);
-
-		// Fill the deletion queue with dummy values, so that on_initialize attempts
-		// to clear the queue
-		ContractInfo::<Test>::fill_queue_with_dummies();
-
-		// Check that on_initialize() tries to perform lazy removal but removes nothing
-		//  as no more weight is left for that.
-		let weight_used = Contracts::on_initialize(System::block_number());
-		let base = <<Test as Config>::WeightInfo as WeightInfo>::on_process_deletion_queue_batch();
-		assert_eq!(weight_used, base);
-
-		// Check that the deletion queue is still full after execution of the
-		// on_initialize() hook.
-		let max_len: u32 = <Test as Config>::DeletionQueueDepth::get();
-		let queue_len: u32 = <DeletionQueue<Test>>::decode_len().unwrap_or(0).try_into().unwrap();
-		assert_eq!(max_len, queue_len);
-	});
-}
-
 #[test]
 fn lazy_removal_does_no_run_on_low_remaining_weight() {
 	let (code, _hash) = compile_module::<Test>("self_destruct").unwrap();
@@ -2209,7 +2161,7 @@ fn lazy_removal_does_no_run_on_low_remaining_weight() {
 		// But value should be still there as the lazy removal did not run, yet.
 		assert_matches!(child::get(trie, &[99]), Some(42));
 
-		// Assign a remaining weight which is too low for a successfull deletion of the contract
+		// Assign a remaining weight which is too low for a successful deletion of the contract
 		let low_remaining_weight =
 			<<Test as Config>::WeightInfo as WeightInfo>::on_process_deletion_queue_batch();
 
@@ -2259,7 +2211,7 @@ fn lazy_removal_does_not_use_all_weight() {
 		.account_id;
 
 		let info = get_contract(&addr);
-		let (weight_per_key, max_keys) = ContractInfo::<Test>::deletion_budget(1, weight_limit);
+		let (weight_per_key, max_keys) = ContractInfo::<Test>::deletion_budget(weight_limit);
 
 		// We create a contract with one less storage item than we can remove within the limit
 		let vals: Vec<_> = (0..max_keys - 1)
@@ -2314,40 +2266,75 @@ fn lazy_removal_does_not_use_all_weight() {
 }
 
 #[test]
-fn deletion_queue_full() {
+fn deletion_queue_ring_buffer_overflow() {
 	let (code, _hash) = compile_module::<Test>("self_destruct").unwrap();
-	ExtBuilder::default().existential_deposit(50).build().execute_with(|| {
+	let mut ext = ExtBuilder::default().existential_deposit(50).build();
+
+	// setup the deletion queue with custom counters
+	ext.execute_with(|| {
+		let queue = DeletionQueueManager::from_test_values(u32::MAX - 1, u32::MAX - 1);
+		<DeletionQueueCounter<Test>>::set(queue);
+	});
+
+	// commit the changes to the storage
+	ext.commit_all().unwrap();
+
+	ext.execute_with(|| {
 		let min_balance = <Test as Config>::Currency::minimum_balance();
 		let _ = Balances::deposit_creating(&ALICE, 1000 * min_balance);
+		let mut tries: Vec<child::ChildInfo> = vec![];
 
-		let addr = Contracts::bare_instantiate(
-			ALICE,
-			min_balance * 100,
-			GAS_LIMIT,
-			None,
-			Code::Upload(code),
-			vec![],
-			vec![],
-			false,
-		)
-		.result
-		.unwrap()
-		.account_id;
+		// add 3 contracts to the deletion queue
+		for i in 0..3u8 {
+			let addr = Contracts::bare_instantiate(
+				ALICE,
+				min_balance * 100,
+				GAS_LIMIT,
+				None,
+				Code::Upload(code.clone()),
+				vec![],
+				vec![i],
+				false,
+			)
+			.result
+			.unwrap()
+			.account_id;
 
-		// fill the deletion queue up until its limit
-		ContractInfo::<Test>::fill_queue_with_dummies();
+			let info = get_contract(&addr);
+			let trie = &info.child_trie_info();
 
-		// Terminate the contract should fail
-		assert_err_ignore_postinfo!(
-			Contracts::call(RuntimeOrigin::signed(ALICE), addr.clone(), 0, GAS_LIMIT, None, vec![],),
-			Error::<Test>::DeletionQueueFull,
-		);
+			// Put value into the contracts child trie
+			child::put(trie, &[99], &42);
 
-		// Contract should exist because removal failed
-		get_contract(&addr);
-	});
+			// Terminate the contract. Contract info should be gone, but value should be still
+			// there as the lazy removal did not run, yet.
+			assert_ok!(Contracts::call(
+				RuntimeOrigin::signed(ALICE),
+				addr.clone(),
+				0,
+				GAS_LIMIT,
+				None,
+				vec![]
+			));
+
+			assert!(!<ContractInfoOf::<Test>>::contains_key(&addr));
+			assert_matches!(child::get(trie, &[99]), Some(42));
+
+			tries.push(trie.clone())
+		}
+
+		// Run single lazy removal
+		Contracts::on_idle(System::block_number(), Weight::MAX);
+
+		// The single lazy removal should have removed all queued tries
+		for trie in tries.iter() {
+			assert_matches!(child::get::<i32>(trie, &[99]), None);
+		}
+
+		// insert and delete counter values should go from u32::MAX - 1 to 1
+		assert_eq!(<DeletionQueueCounter<Test>>::get().as_test_tuple(), (1, 1));
+	})
 }
-
 #[test]
 fn refcounter() {
 	let (wasm, code_hash) = compile_module::<Test>("self_destruct").unwrap();