Asset Pallet: Support repeated destroys to safely destroy large asset…

…s (#12310)

* Support repeated destroys to safely destroy large assets

* require freezing accounts before destroying

* support only deleting asset as final stage when there's no assets left

* pre: introduce the RemoveKeyLimit config parameter

* debug_ensure empty account in the right if block

* update to having separate max values for accounts and approvals

* add tests and use RemoveKeyLimit constant

* add useful comments to the extrinsics, and calculate returned weight

* add benchmarking for start_destroy and finish destroy

* push failing benchmark logic

* add benchmark tests for new functions

* update weights via local benchmarks

* remove extra weight file

* Update frame/assets/src/lib.rs

Co-authored-by: joe petrowski <25483142+joepetrowski@users.noreply.github.com>

* Update frame/assets/src/types.rs

Co-authored-by: joe petrowski <25483142+joepetrowski@users.noreply.github.com>

* Update frame/assets/src/lib.rs

Co-authored-by: joe petrowski <25483142+joepetrowski@users.noreply.github.com>

* effect some changes from codereview

* use NotFrozen error

* remove origin checks, as anyone can complete destruction after owner has begun the process; Add live check for other extrinsics

* fix comments about Origin behaviour

* add AssetStatus docs

* modularize logic to allow calling logic in on_idle and on_initialize hooks

* introduce simple migration for assets details

* reintroduce logging in the migrations

* move deposit_Event out of the mutate block

* Update frame/assets/src/functions.rs

Co-authored-by: Muharem Ismailov <ismailov.m.h@gmail.com>

* Update frame/assets/src/migration.rs

Co-authored-by: Muharem Ismailov <ismailov.m.h@gmail.com>

* move AssetNotLive checkout out of the mutate blocks

* rename RemoveKeysLimit to RemoveItemsLimit

* update docs

* fix event name in benchmark

* fix cargo fmt.

* fix lint in benchmarking

* Empty commit to trigger CI

* Update frame/assets/src/lib.rs

Co-authored-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>

* Update frame/assets/src/lib.rs

Co-authored-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>

* Update frame/assets/src/functions.rs

Co-authored-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>

* Update frame/assets/src/functions.rs

Co-authored-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>

* Update frame/assets/src/functions.rs

Co-authored-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>

* Update frame/assets/src/lib.rs

Co-authored-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>

* Update frame/assets/src/functions.rs

Co-authored-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>

* effect change suggested during code review

* move limit to a single location

* Update frame/assets/src/functions.rs

Co-authored-by: joe petrowski <25483142+joepetrowski@users.noreply.github.com>

* rename events

* fix weight typo, using rocksdb instead of T::DbWeight. Pending generating weights

* switch to using dead_account.len()

* rename event in the benchmarks

* empty to retrigger CI

* trigger CI to check cumulus dependency

* trigger CI for dependent cumulus

* Update frame/assets/src/migration.rs

Co-authored-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>

* move is-frozen to the assetStatus enum (#12547)

* add pre and post migration hooks

* update do_transfer logic to add new assert for more correct error messages

* trigger CI

* switch checking AssetStatus from checking Destroying state to checking live state

* fix error type in tests from Frozen to AssetNotLive

* trigger CI

* change ensure check for fn reducible_balance()

* change the error type to Error:<T,I>::IncorrectStatus to be clearer

* Trigger CI

Co-authored-by: joe petrowski <25483142+joepetrowski@users.noreply.github.com>
Co-authored-by: parity-processbot <>
Co-authored-by: Muharem Ismailov <ismailov.m.h@gmail.com>
Co-authored-by: Oliver Tale-Yazdi <oliver.tale-yazdi@parity.io>

bin/node/runtime/src/lib.rs

@@ -1451,6 +1451,7 @@ impl pallet_assets::Config for Runtime {
 	type Freezer = ();
 	type Extra = ();
 	type WeightInfo = pallet_assets::weights::SubstrateWeight<Runtime>;
+	type RemoveItemsLimit = ConstU32<1000>;
 }
 
 parameter_types! {

frame/assets/src/benchmarking.rs

@@ -78,25 +78,6 @@ fn swap_is_sufficient<T: Config<I>, I: 'static>(s: &mut bool) {
 	});
 }
 
-fn add_consumers<T: Config<I>, I: 'static>(minter: T::AccountId, n: u32) {
-	let origin = SystemOrigin::Signed(minter);
-	let mut s = false;
-	swap_is_sufficient::<T, I>(&mut s);
-	for i in 0..n {
-		let target = account("consumer", i, SEED);
-		T::Currency::make_free_balance_be(&target, T::Currency::minimum_balance());
-		let target_lookup = T::Lookup::unlookup(target);
-		assert!(Assets::<T, I>::mint(
-			origin.clone().into(),
-			Default::default(),
-			target_lookup,
-			100u32.into()
-		)
-		.is_ok());
-	}
-	swap_is_sufficient::<T, I>(&mut s);
-}
-
 fn add_sufficients<T: Config<I>, I: 'static>(minter: T::AccountId, n: u32) {
 	let origin = SystemOrigin::Signed(minter);
 	let mut s = true;
@@ -168,18 +149,66 @@ benchmarks_instance_pallet! {
 		assert_last_event::<T, I>(Event::ForceCreated { asset_id: Default::default(), owner: caller }.into());
 	}
 
-	destroy {
-		let c in 0 .. 5_000;
-		let s in 0 .. 5_000;
-		let a in 0 .. 5_00;
+	start_destroy {
+		let (caller, caller_lookup) = create_default_minted_asset::<T, I>(true, 100u32.into());
+		Assets::<T, I>::freeze_asset(
+			SystemOrigin::Signed(caller.clone()).into(),
+			Default::default(),
+		)?;
+	}:_(SystemOrigin::Signed(caller), Default::default())
+	verify {
+		assert_last_event::<T, I>(Event::DestructionStarted { asset_id: Default::default() }.into());
+	}
+
+	destroy_accounts {
+		let c in 0 .. T::RemoveItemsLimit::get();
 		let (caller, _) = create_default_asset::<T, I>(true);
-		add_consumers::<T, I>(caller.clone(), c);
-		add_sufficients::<T, I>(caller.clone(), s);
+		add_sufficients::<T, I>(caller.clone(), c);
+		Assets::<T, I>::freeze_asset(
+			SystemOrigin::Signed(caller.clone()).into(),
+			Default::default(),
+		)?;
+		Assets::<T,I>::start_destroy(SystemOrigin::Signed(caller.clone()).into(), Default::default())?;
+	}:_(SystemOrigin::Signed(caller), Default::default())
+	verify {
+		assert_last_event::<T, I>(Event::AccountsDestroyed {
+			asset_id: Default::default() ,
+			accounts_destroyed: c,
+			accounts_remaining: 0,
+		}.into());
+	}
+
+	destroy_approvals {
+		let a in 0 .. T::RemoveItemsLimit::get();
+		let (caller, _) = create_default_minted_asset::<T, I>(true, 100u32.into());
 		add_approvals::<T, I>(caller.clone(), a);
-		let witness = Asset::<T, I>::get(T::AssetId::default()).unwrap().destroy_witness();
-	}: _(SystemOrigin::Signed(caller), Default::default(), witness)
+		Assets::<T, I>::freeze_asset(
+			SystemOrigin::Signed(caller.clone()).into(),
+			Default::default(),
+		)?;
+		Assets::<T,I>::start_destroy(SystemOrigin::Signed(caller.clone()).into(), Default::default())?;
+	}:_(SystemOrigin::Signed(caller), Default::default())
 	verify {
-		assert_last_event::<T, I>(Event::Destroyed { asset_id: Default::default() }.into());
+		assert_last_event::<T, I>(Event::ApprovalsDestroyed {
+			asset_id: Default::default() ,
+			approvals_destroyed: a,
+			approvals_remaining: 0,
+		}.into());
+	}
+
+	finish_destroy {
+		let (caller, caller_lookup) = create_default_asset::<T, I>(true);
+		Assets::<T, I>::freeze_asset(
+			SystemOrigin::Signed(caller.clone()).into(),
+			Default::default(),
+		)?;
+		Assets::<T,I>::start_destroy(SystemOrigin::Signed(caller.clone()).into(), Default::default())?;
+	}:_(SystemOrigin::Signed(caller), Default::default())
+	verify {
+		assert_last_event::<T, I>(Event::Destroyed {
+			asset_id: Default::default() ,
+		}.into()
+		);
 	}
 
 	mint {

frame/assets/src/functions.rs

@@ -157,7 +157,7 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 		if details.supply.checked_sub(&amount).is_none() {
 			return Underflow
 		}
-		if details.is_frozen {
+		if details.status == AssetStatus::Frozen {
 			return Frozen
 		}
 		if amount.is_zero() {
@@ -205,7 +205,7 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 		keep_alive: bool,
 	) -> Result<T::Balance, DispatchError> {
 		let details = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
-		ensure!(!details.is_frozen, Error::<T, I>::Frozen);
+		ensure!(details.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 
 		let account = Account::<T, I>::get(id, who).ok_or(Error::<T, I>::NoAccount)?;
 		ensure!(!account.is_frozen, Error::<T, I>::Frozen);
@@ -300,6 +300,7 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 		ensure!(!Account::<T, I>::contains_key(id, &who), Error::<T, I>::AlreadyExists);
 		let deposit = T::AssetAccountDeposit::get();
 		let mut details = Asset::<T, I>::get(&id).ok_or(Error::<T, I>::Unknown)?;
+		ensure!(details.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 		let reason = Self::new_account(&who, &mut details, Some(deposit))?;
 		T::Currency::reserve(&who, deposit)?;
 		Asset::<T, I>::insert(&id, details);
@@ -321,9 +322,8 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 		let mut account = Account::<T, I>::get(id, &who).ok_or(Error::<T, I>::NoDeposit)?;
 		let deposit = account.reason.take_deposit().ok_or(Error::<T, I>::NoDeposit)?;
 		let mut details = Asset::<T, I>::get(&id).ok_or(Error::<T, I>::Unknown)?;
-
+		ensure!(details.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 		ensure!(account.balance.is_zero() || allow_burn, Error::<T, I>::WouldBurn);
-		ensure!(!details.is_frozen, Error::<T, I>::Frozen);
 		ensure!(!account.is_frozen, Error::<T, I>::Frozen);
 
 		T::Currency::unreserve(&who, deposit);
@@ -390,7 +390,7 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 		Self::can_increase(id, beneficiary, amount, true).into_result()?;
 		Asset::<T, I>::try_mutate(id, |maybe_details| -> DispatchResult {
 			let details = maybe_details.as_mut().ok_or(Error::<T, I>::Unknown)?;
-
+			ensure!(details.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 			check(details)?;
 
 			Account::<T, I>::try_mutate(id, beneficiary, |maybe_account| -> DispatchResult {
@@ -430,6 +430,12 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 		maybe_check_admin: Option<T::AccountId>,
 		f: DebitFlags,
 	) -> Result<T::Balance, DispatchError> {
+		let d = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
+		ensure!(
+			d.status == AssetStatus::Live || d.status == AssetStatus::Frozen,
+			Error::<T, I>::AssetNotLive
+		);
+
 		let actual = Self::decrease_balance(id, target, amount, f, |actual, details| {
 			// Check admin rights.
 			if let Some(check_admin) = maybe_check_admin {
@@ -467,12 +473,14 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 			return Ok(amount)
 		}
 
+		let details = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
+		ensure!(details.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
+
 		let actual = Self::prep_debit(id, target, amount, f)?;
 		let mut target_died: Option<DeadConsequence> = None;
 
 		Asset::<T, I>::try_mutate(id, |maybe_details| -> DispatchResult {
 			let details = maybe_details.as_mut().ok_or(Error::<T, I>::Unknown)?;
-
 			check(actual, details)?;
 
 			Account::<T, I>::try_mutate(id, target, |maybe_account| -> DispatchResult {
@@ -540,6 +548,8 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 		if amount.is_zero() {
 			return Ok((amount, None))
 		}
+		let details = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
+		ensure!(details.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 
 		// Figure out the debit and credit, together with side-effects.
 		let debit = Self::prep_debit(id, source, amount, f.into())?;
@@ -651,72 +661,123 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 				accounts: 0,
 				sufficients: 0,
 				approvals: 0,
-				is_frozen: false,
+				status: AssetStatus::Live,
 			},
 		);
 		Self::deposit_event(Event::ForceCreated { asset_id: id, owner });
 		Ok(())
 	}
 
-	/// Destroy an existing asset.
-	///
-	/// * `id`: The asset you want to destroy.
-	/// * `witness`: Witness data needed about the current state of the asset, used to confirm
-	///   complexity of the operation.
-	/// * `maybe_check_owner`: An optional check before destroying the asset, if the provided
-	///   account is the owner of that asset. Can be used for authorization checks.
-	pub(super) fn do_destroy(
+	/// Start the process of destroying an asset, by setting the asset status to `Destroying`, and
+	/// emitting the `DestructionStarted` event.
+	pub(super) fn do_start_destroy(
 		id: T::AssetId,
-		witness: DestroyWitness,
 		maybe_check_owner: Option<T::AccountId>,
-	) -> Result<DestroyWitness, DispatchError> {
-		let mut dead_accounts: Vec<T::AccountId> = vec![];
+	) -> DispatchResult {
+		Asset::<T, I>::try_mutate_exists(id, |maybe_details| -> Result<(), DispatchError> {
+			let mut details = maybe_details.as_mut().ok_or(Error::<T, I>::Unknown)?;
+			if let Some(check_owner) = maybe_check_owner {
+				ensure!(details.owner == check_owner, Error::<T, I>::NoPermission);
+			}
+			details.status = AssetStatus::Destroying;
 
-		let result_witness: DestroyWitness = Asset::<T, I>::try_mutate_exists(
-			id,
-			|maybe_details| -> Result<DestroyWitness, DispatchError> {
-				let mut details = maybe_details.take().ok_or(Error::<T, I>::Unknown)?;
-				if let Some(check_owner) = maybe_check_owner {
-					ensure!(details.owner == check_owner, Error::<T, I>::NoPermission);
-				}
-				ensure!(details.accounts <= witness.accounts, Error::<T, I>::BadWitness);
-				ensure!(details.sufficients <= witness.sufficients, Error::<T, I>::BadWitness);
-				ensure!(details.approvals <= witness.approvals, Error::<T, I>::BadWitness);
+			Self::deposit_event(Event::DestructionStarted { asset_id: id });
+			Ok(())
+		})
+	}
+
+	/// Destroy accounts associated with a given asset up to the max (T::RemoveItemsLimit).
+	///
+	/// Each call emits the `Event::DestroyedAccounts` event.
+	/// Returns the number of destroyed accounts.
+	pub(super) fn do_destroy_accounts(
+		id: T::AssetId,
+		max_items: u32,
+	) -> Result<u32, DispatchError> {
+		let mut dead_accounts: Vec<T::AccountId> = vec![];
+		let mut remaining_accounts = 0;
+		let _ =
+			Asset::<T, I>::try_mutate_exists(id, |maybe_details| -> Result<(), DispatchError> {
+				let mut details = maybe_details.as_mut().ok_or(Error::<T, I>::Unknown)?;
+				// Should only destroy accounts while the asset is in a destroying state
+				ensure!(details.status == AssetStatus::Destroying, Error::<T, I>::IncorrectStatus);
 
 				for (who, v) in Account::<T, I>::drain_prefix(id) {
-					// We have to force this as it's destroying the entire asset class.
-					// This could mean that some accounts now have irreversibly reserved
-					// funds.
 					let _ = Self::dead_account(&who, &mut details, &v.reason, true);
 					dead_accounts.push(who);
+					if dead_accounts.len() >= (max_items as usize) {
+						break
+					}
 				}
-				debug_assert_eq!(details.accounts, 0);
-				debug_assert_eq!(details.sufficients, 0);
+				remaining_accounts = details.accounts;
+				Ok(())
+			})?;
+
+		for who in &dead_accounts {
+			T::Freezer::died(id, &who);
+		}
 
-				let metadata = Metadata::<T, I>::take(&id);
-				T::Currency::unreserve(
-					&details.owner,
-					details.deposit.saturating_add(metadata.deposit),
-				);
+		Self::deposit_event(Event::AccountsDestroyed {
+			asset_id: id,
+			accounts_destroyed: dead_accounts.len() as u32,
+			accounts_remaining: remaining_accounts as u32,
+		});
+		Ok(dead_accounts.len() as u32)
+	}
 
-				for ((owner, _), approval) in Approvals::<T, I>::drain_prefix((&id,)) {
+	/// Destroy approvals associated with a given asset up to the max (T::RemoveItemsLimit).
+	///
+	/// Each call emits the `Event::DestroyedApprovals` event
+	/// Returns the number of destroyed approvals.
+	pub(super) fn do_destroy_approvals(
+		id: T::AssetId,
+		max_items: u32,
+	) -> Result<u32, DispatchError> {
+		let mut removed_approvals = 0;
+		let _ =
+			Asset::<T, I>::try_mutate_exists(id, |maybe_details| -> Result<(), DispatchError> {
+				let mut details = maybe_details.as_mut().ok_or(Error::<T, I>::Unknown)?;
+
+				// Should only destroy accounts while the asset is in a destroying state.
+				ensure!(details.status == AssetStatus::Destroying, Error::<T, I>::IncorrectStatus);
+
+				for ((owner, _), approval) in Approvals::<T, I>::drain_prefix((id,)) {
 					T::Currency::unreserve(&owner, approval.deposit);
+					removed_approvals = removed_approvals.saturating_add(1);
+					details.approvals = details.approvals.saturating_sub(1);
+					if removed_approvals >= max_items {
+						break
+					}
 				}
-				Self::deposit_event(Event::Destroyed { asset_id: id });
+				Self::deposit_event(Event::ApprovalsDestroyed {
+					asset_id: id,
+					approvals_destroyed: removed_approvals as u32,
+					approvals_remaining: details.approvals as u32,
+				});
+				Ok(())
+			})?;
+		Ok(removed_approvals)
+	}
 
-				Ok(DestroyWitness {
-					accounts: details.accounts,
-					sufficients: details.sufficients,
-					approvals: details.approvals,
-				})
-			},
-		)?;
+	/// Complete destroying an asset and unreserve the deposit.
+	///
+	/// On success, the `Event::Destroyed` event is emitted.
+	pub(super) fn do_finish_destroy(id: T::AssetId) -> DispatchResult {
+		Asset::<T, I>::try_mutate_exists(id, |maybe_details| -> Result<(), DispatchError> {
+			let details = maybe_details.take().ok_or(Error::<T, I>::Unknown)?;
+			ensure!(details.status == AssetStatus::Destroying, Error::<T, I>::IncorrectStatus);
+			ensure!(details.accounts == 0, Error::<T, I>::InUse);
+			ensure!(details.approvals == 0, Error::<T, I>::InUse);
+
+			let metadata = Metadata::<T, I>::take(&id);
+			T::Currency::unreserve(
+				&details.owner,
+				details.deposit.saturating_add(metadata.deposit),
+			);
+			Self::deposit_event(Event::Destroyed { asset_id: id });
 
-		// Execute hooks outside of `mutate`.
-		for who in dead_accounts {
-			T::Freezer::died(id, &who);
-		}
-		Ok(result_witness)
+			Ok(())
+		})
 	}
 
 	/// Creates an approval from `owner` to spend `amount` of asset `id` tokens by 'delegate'
@@ -730,7 +791,7 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 		amount: T::Balance,
 	) -> DispatchResult {
 		let mut d = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
-		ensure!(!d.is_frozen, Error::<T, I>::Frozen);
+		ensure!(d.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 		Approvals::<T, I>::try_mutate(
 			(id, &owner, &delegate),
 			|maybe_approved| -> DispatchResult {
@@ -780,6 +841,9 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 	) -> DispatchResult {
 		let mut owner_died: Option<DeadConsequence> = None;
 
+		let d = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
+		ensure!(d.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
+
 		Approvals::<T, I>::try_mutate_exists(
 			(id, &owner, delegate),
 			|maybe_approved| -> DispatchResult {
@@ -826,6 +890,7 @@ impl<T: Config<I>, I: 'static> Pallet<T, I> {
 			symbol.clone().try_into().map_err(|_| Error::<T, I>::BadMetadata)?;
 
 		let d = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
+		ensure!(d.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 		ensure!(from == &d.owner, Error::<T, I>::NoPermission);
 
 		Metadata::<T, I>::try_mutate_exists(id, |metadata| {