contracts: Lazy storage removal (#7740)

* Do not evict a contract from within a call stack

We don't want to trigger contract eviction automatically when
a contract is called. This is because those changes can be
reverted due to how storage transactions are used at the moment.
More Information:
https://github.com/paritytech/substrate/issues/6439#issuecomment-648754324

It can be re-introduced once the linked issue is resolved. In the meantime
`claim_surcharge` must be called to evict a contract.

* Lazily delete storage in on_initialize instead of when removing the contract

* Add missing documentation of new error

* Make Module::claim_surcharge public

It being the only dispatchable that is private is an oversight.

* review: Add final newline

* review: Simplify assert statement

* Add test that checks that partial remove of a contract works

* Premote warning to error

* Added missing docs for seal_terminate

* Lazy deletion should only take AVERAGE_ON_INITIALIZE_RATIO of the block

* Added informational about the lazy deletion throughput

* Avoid lazy deletion in case the block is already full

* Prevent queue decoding in case of an already full block

* Add test that checks that on_initialize honors block limits

substrate/frame/contracts/src/storage.rs

@@ -20,20 +20,37 @@
 use crate::{
 	exec::{AccountIdOf, StorageKey},
 	AliveContractInfo, BalanceOf, CodeHash, ContractInfo, ContractInfoOf, Config, TrieId,
-	AccountCounter,
+	AccountCounter, DeletionQueue, Error,
+	weights::WeightInfo,
 };
+use codec::{Encode, Decode};
 use sp_std::prelude::*;
 use sp_std::marker::PhantomData;
 use sp_io::hashing::blake2_256;
 use sp_runtime::traits::Bounded;
 use sp_core::crypto::UncheckedFrom;
-use frame_support::{storage::child, StorageMap};
+use frame_support::{
+	dispatch::DispatchResult,
+	StorageMap,
+	debug,
+	storage::{child::{self, KillOutcome}, StorageValue},
+	traits::Get,
+	weights::Weight,
+};
 
 /// An error that means that the account requested either doesn't exist or represents a tombstone
 /// account.
 #[cfg_attr(test, derive(PartialEq, Eq, Debug))]
 pub struct ContractAbsentError;
 
+#[derive(Encode, Decode)]
+pub struct DeletedContract {
+	pair_count: u32,
+	trie_id: TrieId,
+}
+
+
+
 pub struct Storage<T>(PhantomData<T>);
 
 impl<T> Storage<T>
@@ -191,18 +208,105 @@ where
 		})
 	}
 
-	/// Removes the contract and all the storage associated with it.
+	/// Push a contract's trie to the deletion queue for lazy removal.
 	///
-	/// This function doesn't affect the account.
-	pub fn destroy_contract(address: &AccountIdOf<T>, trie_id: &TrieId) {
-		<ContractInfoOf<T>>::remove(address);
-		child::kill_storage(&crate::child_trie_info(&trie_id), None);
+	/// You must make sure that the contract is also removed or converted into a tombstone
+	/// when queuing the trie for deletion.
+	pub fn queue_trie_for_deletion(contract: &AliveContractInfo<T>) -> DispatchResult {
+		if DeletionQueue::decode_len().unwrap_or(0) >= T::DeletionQueueDepth::get() as usize {
+			Err(Error::<T>::DeletionQueueFull.into())
+		} else {
+			DeletionQueue::append(DeletedContract {
+				pair_count: contract.total_pair_count,
+				trie_id: contract.trie_id.clone(),
+			});
+			Ok(())
+		}
+	}
+
+	/// 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) -> (u64, u32) {
+		let base_weight = T::WeightInfo::on_initialize();
+		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 Weight);
+
+		// `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(weight_per_key)
+			.unwrap_or(0) as u32;
+
+		(weight_per_key, key_budget)
+	}
+
+	/// Delete as many items from the deletion queue possible within the supplied weight limit.
+	///
+	/// It returns the amount of weight used for that task or `None` when no weight was used
+	/// apart from the base weight.
+	pub fn process_deletion_queue_batch(weight_limit: Weight) -> Weight {
+		let queue_len = DeletionQueue::decode_len().unwrap_or(0);
+		if queue_len == 0 {
+			return weight_limit;
+		}
+
+		let (weight_per_key, mut remaining_key_budget) = Self::deletion_budget(
+			queue_len,
+			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
+		// which consume the whole block before the other `on_initialize` blocks are called.
+		if remaining_key_budget == 0 {
+			return weight_limit;
+		}
+
+		let mut queue = DeletionQueue::get();
+
+		while !queue.is_empty() && remaining_key_budget > 0 {
+			// Cannot panic due to loop condition
+			let trie = &mut queue[0];
+			let pair_count = trie.pair_count;
+			let outcome = child::kill_storage(
+				&crate::child_trie_info(&trie.trie_id),
+				Some(remaining_key_budget),
+			);
+			if pair_count > remaining_key_budget {
+				// Cannot underflow because of the if condition
+				trie.pair_count -= remaining_key_budget;
+			} else {
+				// We do not care to preserve order. The contract is deleted already and
+				// noone waits for the trie to be deleted.
+				let removed = queue.swap_remove(0);
+				match outcome {
+					// This should not happen as our budget was large enough to remove all keys.
+					KillOutcome::SomeRemaining => {
+						debug::error!(
+							"After deletion keys are remaining in this child trie: {:?}",
+							removed.trie_id,
+						);
+					},
+					KillOutcome::AllRemoved => (),
+				}
+			}
+			remaining_key_budget = remaining_key_budget
+				.saturating_sub(remaining_key_budget.min(pair_count));
+		}
+
+		DeletionQueue::put(queue);
+		weight_limit.saturating_sub(weight_per_key.saturating_mul(remaining_key_budget as Weight))
 	}
 
 	/// This generator uses inner counter for account id and applies the hash over `AccountId +
 	/// accountid_counter`.
 	pub fn generate_trie_id(account_id: &AccountIdOf<T>) -> TrieId {
-		use frame_support::StorageValue;
 		use sp_runtime::traits::Hash;
 		// Note that skipping a value due to error is not an issue here.
 		// We only need uniqueness, not sequence.
@@ -226,4 +330,15 @@ where
 			.and_then(|i| i.as_alive().map(|i| i.code_hash))
 			.ok_or(ContractAbsentError)
 	}
+
+	/// Fill up the queue in order to exercise the limits during testing.
+	#[cfg(test)]
+	pub fn fill_queue_with_dummies() {
+		let queue: Vec<_> = (0..T::DeletionQueueDepth::get()).map(|_| DeletedContract {
+			pair_count: 0,
+			trie_id: vec![],
+		})
+		.collect();
+		DeletionQueue::put(queue);
+	}
 }