Asset Pallet: Support repeated destroys to safely destroy large assets (#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>

substrate/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! {

substrate/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 {
+	start_destroy {
-		let c in 0 .. 5_000;
+		let (caller, caller_lookup) = create_default_minted_asset::<T, I>(true, 100u32.into());
-		let s in 0 .. 5_000;
+		Assets::<T, I>::freeze_asset(
-		let a in 0 .. 5_00;
+			SystemOrigin::Signed(caller.clone()).into(),
-		let (caller, _) = create_default_asset::<T, I>(true);
+			Default::default(),
-		add_consumers::<T, I>(caller.clone(), c);
+		)?;
-		add_sufficients::<T, I>(caller.clone(), s);
+	}:_(SystemOrigin::Signed(caller), Default::default())
-		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)
 	verify {
-		assert_last_event::<T, I>(Event::Destroyed { asset_id: Default::default() }.into());
+		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_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);
+		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::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 {

substrate/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.
+	/// Start the process of destroying an asset, by setting the asset status to `Destroying`, and
-	///
+	/// emitting the `DestructionStarted` event.
-	/// * `id`: The asset you want to destroy.
+	pub(super) fn do_start_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(
 		id: T::AssetId,
-		witness: DestroyWitness,
 		maybe_check_owner: Option<T::AccountId>,
-	) -> Result<DestroyWitness, DispatchError> {
+	) -> DispatchResult {
-		let mut dead_accounts: Vec<T::AccountId> = vec![];
+		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(
+			Self::deposit_event(Event::DestructionStarted { asset_id: id });
-			id,
+			Ok(())
-			|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);
+	/// Destroy accounts associated with a given asset up to the max (T::RemoveItemsLimit).
-				}
+	///
-				ensure!(details.accounts <= witness.accounts, Error::<T, I>::BadWitness);
+	/// Each call emits the `Event::DestroyedAccounts` event.
-				ensure!(details.sufficients <= witness.sufficients, Error::<T, I>::BadWitness);
+	/// Returns the number of destroyed accounts.
-				ensure!(details.approvals <= witness.approvals, Error::<T, I>::BadWitness);
+	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);
+				remaining_accounts = details.accounts;
-				debug_assert_eq!(details.sufficients, 0);
+				Ok(())
+			})?;
 
-				let metadata = Metadata::<T, I>::take(&id);
+		for who in &dead_accounts {
-				T::Currency::unreserve(
-					&details.owner,
-					details.deposit.saturating_add(metadata.deposit),
-				);
-
-				for ((owner, _), approval) in Approvals::<T, I>::drain_prefix((&id,)) {
-					T::Currency::unreserve(&owner, approval.deposit);
-				}
-				Self::deposit_event(Event::Destroyed { asset_id: id });
-
-				Ok(DestroyWitness {
-					accounts: details.accounts,
-					sufficients: details.sufficients,
-					approvals: details.approvals,
-				})
-			},
-		)?;
-
-		// Execute hooks outside of `mutate`.
-		for who in dead_accounts {
 			T::Freezer::died(id, &who);
 		}
-		Ok(result_witness)
+
+		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)
+	}
+
+	/// 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::ApprovalsDestroyed {
+					asset_id: id,
+					approvals_destroyed: removed_approvals as u32,
+					approvals_remaining: details.approvals as u32,
+				});
+				Ok(())
+			})?;
+		Ok(removed_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 });
+
+			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| {

substrate/frame/assets/src/impl_fungibles.rs

@@ -179,22 +179,6 @@ impl<T: Config<I>, I: 'static> fungibles::Create<T::AccountId> for Pallet<T, I>
 	}
 }
 
-impl<T: Config<I>, I: 'static> fungibles::Destroy<T::AccountId> for Pallet<T, I> {
-	type DestroyWitness = DestroyWitness;
-
-	fn get_destroy_witness(asset: &T::AssetId) -> Option<Self::DestroyWitness> {
-		Asset::<T, I>::get(asset).map(|asset_details| asset_details.destroy_witness())
-	}
-
-	fn destroy(
-		id: T::AssetId,
-		witness: Self::DestroyWitness,
-		maybe_check_owner: Option<T::AccountId>,
-	) -> Result<Self::DestroyWitness, DispatchError> {
-		Self::do_destroy(id, witness, maybe_check_owner)
-	}
-}
-
 impl<T: Config<I>, I: 'static> fungibles::metadata::Inspect<<T as SystemConfig>::AccountId>
 	for Pallet<T, I>
 {

substrate/frame/assets/src/lib.rs

@@ -121,11 +121,15 @@
 //! * [`System`](../frame_system/index.html)
 //! * [`Support`](../frame_support/index.html)
 
+// This recursion limit is needed because we have too many benchmarks and benchmarking will fail if
+// we add more without this limit.
+#![recursion_limit = "1024"]
 // Ensure we're `no_std` when compiling for Wasm.
 #![cfg_attr(not(feature = "std"), no_std)]
 
 #[cfg(feature = "runtime-benchmarks")]
 mod benchmarking;
+pub mod migration;
 #[cfg(test)]
 pub mod mock;
 #[cfg(test)]
@@ -174,8 +178,12 @@ pub mod pallet {
 	use frame_support::pallet_prelude::*;
 	use frame_system::pallet_prelude::*;
 
+	/// The current storage version.
+	const STORAGE_VERSION: StorageVersion = StorageVersion::new(1);
+
 	#[pallet::pallet]
 	#[pallet::generate_store(pub(super) trait Store)]
+	#[pallet::storage_version(STORAGE_VERSION)]
 	pub struct Pallet<T, I = ()>(_);
 
 	#[pallet::config]
@@ -195,6 +203,12 @@ pub mod pallet {
 			+ MaxEncodedLen
 			+ TypeInfo;
 
+		/// Max number of items to destroy per `destroy_accounts` and `destroy_approvals` call.
+		///
+		/// Must be configured to result in a weight that makes each call fit in a block.
+		#[pallet::constant]
+		type RemoveItemsLimit: Get<u32>;
+
 		/// Identifier for the class of asset.
 		type AssetId: Member
 			+ Parameter
@@ -342,7 +356,7 @@ pub mod pallet {
 						accounts: 0,
 						sufficients: 0,
 						approvals: 0,
-						is_frozen: false,
+						status: AssetStatus::Live,
 					},
 				);
 			}
@@ -417,6 +431,16 @@ pub mod pallet {
 		AssetFrozen { asset_id: T::AssetId },
 		/// Some asset `asset_id` was thawed.
 		AssetThawed { asset_id: T::AssetId },
+		/// Accounts were destroyed for given asset.
+		AccountsDestroyed { asset_id: T::AssetId, accounts_destroyed: u32, accounts_remaining: u32 },
+		/// Approvals were destroyed for given asset.
+		ApprovalsDestroyed {
+			asset_id: T::AssetId,
+			approvals_destroyed: u32,
+			approvals_remaining: u32,
+		},
+		/// An asset class is in the process of being destroyed.
+		DestructionStarted { asset_id: T::AssetId },
 		/// An asset class was destroyed.
 		Destroyed { asset_id: T::AssetId },
 		/// Some asset class was force-created.
@@ -487,6 +511,15 @@ pub mod pallet {
 		NoDeposit,
 		/// The operation would result in funds being burned.
 		WouldBurn,
+		/// The asset is a live asset and is actively being used. Usually emit for operations such
+		/// as `start_destroy` which require the asset to be in a destroying state.
+		LiveAsset,
+		/// The asset is not live, and likely being destroyed.
+		AssetNotLive,
+		/// The asset status is not the expected status.
+		IncorrectStatus,
+		/// The asset should be frozen before the given operation.
+		NotFrozen,
 	}
 
 	#[pallet::call]
@@ -540,7 +573,7 @@ pub mod pallet {
 					accounts: 0,
 					sufficients: 0,
 					approvals: 0,
-					is_frozen: false,
+					status: AssetStatus::Live,
 				},
 			);
 			Self::deposit_event(Event::Created { asset_id: id, creator: owner, owner: admin });
@@ -579,45 +612,89 @@ pub mod pallet {
 			Self::do_force_create(id, owner, is_sufficient, min_balance)
 		}
 
-		/// Destroy a class of fungible assets.
+		/// Start the process of destroying a class of fungible asset
+		/// start_destroy is the first in a series of extrinsics that should be called, to allow
+		/// destroying an asset.
 		///
 		/// The origin must conform to `ForceOrigin` or must be Signed and the sender must be the
 		/// owner of the asset `id`.
 		///
 		/// - `id`: The identifier of the asset to be destroyed. This must identify an existing
-		/// asset.
+		///   asset.
 		///
-		/// Emits `Destroyed` event when successful.
+		/// Assets must be freezed before calling start_destroy.
-		///
+		#[pallet::weight(T::WeightInfo::start_destroy())]
-		/// NOTE: It can be helpful to first freeze an asset before destroying it so that you
+		pub fn start_destroy(
-		/// can provide accurate witness information and prevent users from manipulating state
-		/// in a way that can make it harder to destroy.
-		///
-		/// Weight: `O(c + p + a)` where:
-		/// - `c = (witness.accounts - witness.sufficients)`
-		/// - `s = witness.sufficients`
-		/// - `a = witness.approvals`
-		#[pallet::weight(T::WeightInfo::destroy(
-			witness.accounts.saturating_sub(witness.sufficients),
- 			witness.sufficients,
- 			witness.approvals,
- 		))]
-		pub fn destroy(
 			origin: OriginFor<T>,
 			#[pallet::compact] id: T::AssetId,
-			witness: DestroyWitness,
+		) -> DispatchResult {
-		) -> DispatchResultWithPostInfo {
 			let maybe_check_owner = match T::ForceOrigin::try_origin(origin) {
 				Ok(_) => None,
 				Err(origin) => Some(ensure_signed(origin)?),
 			};
-			let details = Self::do_destroy(id, witness, maybe_check_owner)?;
+			Self::do_start_destroy(id, maybe_check_owner)
-			Ok(Some(T::WeightInfo::destroy(
+		}
-				details.accounts.saturating_sub(details.sufficients),
+
-				details.sufficients,
+		/// Destroy all accounts associated with a given asset.
-				details.approvals,
+		/// `destroy_accounts` should only be called after `start_destroy` has been called, and the
-			))
+		/// asset is in a `Destroying` state
-			.into())
+		///
+		/// Due to weight restrictions, this function may need to be called multiple
+		/// times to fully destroy all accounts. It will destroy `RemoveItemsLimit` accounts at a
+		/// time.
+		///
+		/// - `id`: The identifier of the asset to be destroyed. This must identify an existing
+		///   asset.
+		///
+		/// Each call Emits the `Event::DestroyedAccounts` event.
+		#[pallet::weight(T::WeightInfo::destroy_accounts(T::RemoveItemsLimit::get()))]
+		pub fn destroy_accounts(
+			origin: OriginFor<T>,
+			#[pallet::compact] id: T::AssetId,
+		) -> DispatchResultWithPostInfo {
+			let _ = ensure_signed(origin)?;
+			let removed_accounts = Self::do_destroy_accounts(id, T::RemoveItemsLimit::get())?;
+			Ok(Some(T::WeightInfo::destroy_accounts(removed_accounts)).into())
+		}
+
+		/// Destroy all approvals associated with a given asset up to the max (T::RemoveItemsLimit),
+		/// `destroy_approvals` should only be called after `start_destroy` has been called, and the
+		/// asset is in a `Destroying` state
+		///
+		/// Due to weight restrictions, this function may need to be called multiple
+		/// times to fully destroy all approvals. It will destroy `RemoveItemsLimit` approvals at a
+		/// time.
+		///
+		/// - `id`: The identifier of the asset to be destroyed. This must identify an existing
+		///   asset.
+		///
+		/// Each call Emits the `Event::DestroyedApprovals` event.
+		#[pallet::weight(T::WeightInfo::destroy_approvals(T::RemoveItemsLimit::get()))]
+		pub fn destroy_approvals(
+			origin: OriginFor<T>,
+			#[pallet::compact] id: T::AssetId,
+		) -> DispatchResultWithPostInfo {
+			let _ = ensure_signed(origin)?;
+			let removed_approvals = Self::do_destroy_approvals(id, T::RemoveItemsLimit::get())?;
+			Ok(Some(T::WeightInfo::destroy_approvals(removed_approvals)).into())
+		}
+
+		/// Complete destroying asset and unreserve currency.
+		/// `finish_destroy` should only be called after `start_destroy` has been called, and the
+		/// asset is in a `Destroying` state. All accounts or approvals should be destroyed before
+		/// hand.
+		///
+		/// - `id`: The identifier of the asset to be destroyed. This must identify an existing
+		///   asset.
+		///
+		/// Each successful call Emits the `Event::Destroyed` event.
+		#[pallet::weight(T::WeightInfo::finish_destroy())]
+		pub fn finish_destroy(
+			origin: OriginFor<T>,
+			#[pallet::compact] id: T::AssetId,
+		) -> DispatchResult {
+			let _ = ensure_signed(origin)?;
+			Self::do_finish_destroy(id)
 		}
 
 		/// Mint assets of a particular class.
@@ -793,6 +870,10 @@ pub mod pallet {
 			let origin = ensure_signed(origin)?;
 
 			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
+			);
 			ensure!(origin == d.freezer, Error::<T, I>::NoPermission);
 			let who = T::Lookup::lookup(who)?;
 
@@ -824,6 +905,10 @@ pub mod pallet {
 			let origin = ensure_signed(origin)?;
 
 			let details = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
+			ensure!(
+				details.status == AssetStatus::Live || details.status == AssetStatus::Frozen,
+				Error::<T, I>::AssetNotLive
+			);
 			ensure!(origin == details.admin, Error::<T, I>::NoPermission);
 			let who = T::Lookup::lookup(who)?;
 
@@ -854,9 +939,10 @@ pub mod pallet {
 
 			Asset::<T, I>::try_mutate(id, |maybe_details| {
 				let d = maybe_details.as_mut().ok_or(Error::<T, I>::Unknown)?;
+				ensure!(d.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 				ensure!(origin == d.freezer, Error::<T, I>::NoPermission);
 
-				d.is_frozen = true;
+				d.status = AssetStatus::Frozen;
 
 				Self::deposit_event(Event::<T, I>::AssetFrozen { asset_id: id });
 				Ok(())
@@ -882,8 +968,9 @@ pub mod pallet {
 			Asset::<T, I>::try_mutate(id, |maybe_details| {
 				let d = maybe_details.as_mut().ok_or(Error::<T, I>::Unknown)?;
 				ensure!(origin == d.admin, Error::<T, I>::NoPermission);
+				ensure!(d.status == AssetStatus::Frozen, Error::<T, I>::NotFrozen);
 
-				d.is_frozen = false;
+				d.status = AssetStatus::Live;
 
 				Self::deposit_event(Event::<T, I>::AssetThawed { asset_id: id });
 				Ok(())
@@ -911,6 +998,7 @@ pub mod pallet {
 
 			Asset::<T, I>::try_mutate(id, |maybe_details| {
 				let details = maybe_details.as_mut().ok_or(Error::<T, I>::Unknown)?;
+				ensure!(details.status == AssetStatus::Live, Error::<T, I>::LiveAsset);
 				ensure!(origin == details.owner, Error::<T, I>::NoPermission);
 				if details.owner == owner {
 					return Ok(())
@@ -956,6 +1044,7 @@ pub mod pallet {
 
 			Asset::<T, I>::try_mutate(id, |maybe_details| {
 				let details = maybe_details.as_mut().ok_or(Error::<T, I>::Unknown)?;
+				ensure!(details.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 				ensure!(origin == details.owner, Error::<T, I>::NoPermission);
 
 				details.issuer = issuer.clone();
@@ -1014,6 +1103,7 @@ pub mod pallet {
 			let origin = ensure_signed(origin)?;
 
 			let d = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
+			ensure!(d.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 			ensure!(origin == d.owner, Error::<T, I>::NoPermission);
 
 			Metadata::<T, I>::try_mutate_exists(id, |metadata| {
@@ -1142,13 +1232,18 @@ pub mod pallet {
 
 			Asset::<T, I>::try_mutate(id, |maybe_asset| {
 				let mut asset = maybe_asset.take().ok_or(Error::<T, I>::Unknown)?;
+				ensure!(asset.status != AssetStatus::Destroying, Error::<T, I>::AssetNotLive);
 				asset.owner = T::Lookup::lookup(owner)?;
 				asset.issuer = T::Lookup::lookup(issuer)?;
 				asset.admin = T::Lookup::lookup(admin)?;
 				asset.freezer = T::Lookup::lookup(freezer)?;
 				asset.min_balance = min_balance;
 				asset.is_sufficient = is_sufficient;
-				asset.is_frozen = is_frozen;
+				if is_frozen {
+					asset.status = AssetStatus::Frozen;
+				} else {
+					asset.status = AssetStatus::Live;
+				}
 				*maybe_asset = Some(asset);
 
 				Self::deposit_event(Event::AssetStatusChanged { asset_id: id });
@@ -1210,6 +1305,7 @@ pub mod pallet {
 			let owner = ensure_signed(origin)?;
 			let delegate = T::Lookup::lookup(delegate)?;
 			let mut d = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
+			ensure!(d.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 			let approval =
 				Approvals::<T, I>::take((id, &owner, &delegate)).ok_or(Error::<T, I>::Unknown)?;
 			T::Currency::unreserve(&owner, approval.deposit);
@@ -1242,6 +1338,7 @@ pub mod pallet {
 			delegate: AccountIdLookupOf<T>,
 		) -> DispatchResult {
 			let mut d = Asset::<T, I>::get(id).ok_or(Error::<T, I>::Unknown)?;
+			ensure!(d.status == AssetStatus::Live, Error::<T, I>::AssetNotLive);
 			T::ForceOrigin::try_origin(origin)
 				.map(|_| ())
 				.or_else(|origin| -> DispatchResult {

substrate/frame/assets/src/migration.rs

@@ -0,0 +1,122 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2017-2022 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: Apache-2.0
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// 	http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use frame_support::{log, traits::OnRuntimeUpgrade};
+
+pub mod v1 {
+	use frame_support::{pallet_prelude::*, weights::Weight};
+
+	use super::*;
+
+	#[derive(Decode)]
+	pub struct OldAssetDetails<Balance, AccountId, DepositBalance> {
+		pub owner: AccountId,
+		pub issuer: AccountId,
+		pub admin: AccountId,
+		pub freezer: AccountId,
+		pub supply: Balance,
+		pub deposit: DepositBalance,
+		pub min_balance: Balance,
+		pub is_sufficient: bool,
+		pub accounts: u32,
+		pub sufficients: u32,
+		pub approvals: u32,
+		pub is_frozen: bool,
+	}
+
+	impl<Balance, AccountId, DepositBalance> OldAssetDetails<Balance, AccountId, DepositBalance> {
+		fn migrate_to_v1(self) -> AssetDetails<Balance, AccountId, DepositBalance> {
+			let status = if self.is_frozen { AssetStatus::Frozen } else { AssetStatus::Live };
+
+			AssetDetails {
+				owner: self.owner,
+				issuer: self.issuer,
+				admin: self.admin,
+				freezer: self.freezer,
+				supply: self.supply,
+				deposit: self.deposit,
+				min_balance: self.min_balance,
+				is_sufficient: self.is_sufficient,
+				accounts: self.accounts,
+				sufficients: self.sufficients,
+				approvals: self.approvals,
+				status,
+			}
+		}
+	}
+
+	pub struct MigrateToV1<T>(sp_std::marker::PhantomData<T>);
+	impl<T: Config> OnRuntimeUpgrade for MigrateToV1<T> {
+		fn on_runtime_upgrade() -> Weight {
+			let current_version = Pallet::<T>::current_storage_version();
+			let onchain_version = Pallet::<T>::on_chain_storage_version();
+			if onchain_version == 0 && current_version == 1 {
+				let mut translated = 0u64;
+				Asset::<T>::translate::<
+					OldAssetDetails<T::Balance, T::AccountId, DepositBalanceOf<T>>,
+					_,
+				>(|_key, old_value| {
+					translated.saturating_inc();
+					Some(old_value.migrate_to_v1())
+				});
+				current_version.put::<Pallet<T>>();
+				log::info!(target: "runtime::assets", "Upgraded {} pools, storage to version {:?}", translated, current_version);
+				T::DbWeight::get().reads_writes(translated + 1, translated + 1)
+			} else {
+				log::info!(target: "runtime::assets",  "Migration did not execute. This probably should be removed");
+				T::DbWeight::get().reads(1)
+			}
+		}
+
+		#[cfg(feature = "try-runtime")]
+		fn pre_upgrade() -> Result<Vec<u8>, &'static str> {
+			frame_support::ensure!(
+				Pallet::<T>::on_chain_storage_version() == 0,
+				"must upgrade linearly"
+			);
+			let prev_count = Asset::<T>::iter().count();
+			Ok((prev_count as u32).encode())
+		}
+
+		#[cfg(feature = "try-runtime")]
+		fn post_upgrade(prev_count: Vec<u8>) -> Result<(), &'static str> {
+			let prev_count: u32 = Decode::decode(&mut prev_count.as_slice()).expect(
+				"the state parameter should be something that was generated by pre_upgrade",
+			);
+			let post_count = Asset::<T>::iter().count() as u32;
+			assert_eq!(
+				prev_count, post_count,
+				"the asset count before and after the migration should be the same"
+			);
+
+			let current_version = Pallet::<T>::current_storage_version();
+			let onchain_version = Pallet::<T>::on_chain_storage_version();
+
+			frame_support::ensure!(current_version == 1, "must_upgrade");
+			assert_eq!(
+				current_version, onchain_version,
+				"after migration, the current_version and onchain_version should be the same"
+			);
+
+			Asset::<T>::iter().for_each(|(_id, asset)| {
+				assert!(asset.status == AssetStatus::Live || asset.status == AssetStatus::Frozen, "assets should only be live or frozen. None should be in destroying status, or undefined state")
+			});
+			Ok(())
+		}
+	}
+}

substrate/frame/assets/src/mock.rs

@@ -100,6 +100,7 @@ impl Config for Test {
 	type Freezer = TestFreezer;
 	type WeightInfo = ();
 	type Extra = ();
+	type RemoveItemsLimit = ConstU32<5>;
 }
 
 use std::collections::HashMap;

substrate/frame/assets/src/tests.rs

@@ -328,8 +328,12 @@ fn lifecycle_should_work() {
 		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 20, 100));
 		assert_eq!(Account::<Test>::iter_prefix(0).count(), 2);
 
-		let w = Asset::<Test>::get(0).unwrap().destroy_witness();
+		assert_ok!(Assets::freeze_asset(RuntimeOrigin::signed(1), 0));
-		assert_ok!(Assets::destroy(RuntimeOrigin::signed(1), 0, w));
+		assert_ok!(Assets::start_destroy(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::destroy_accounts(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::destroy_approvals(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::finish_destroy(RuntimeOrigin::signed(1), 0));
+
 		assert_eq!(Balances::reserved_balance(&1), 0);
 
 		assert!(!Asset::<Test>::contains_key(0));
@@ -348,8 +352,12 @@ fn lifecycle_should_work() {
 		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 20, 100));
 		assert_eq!(Account::<Test>::iter_prefix(0).count(), 2);
 
-		let w = Asset::<Test>::get(0).unwrap().destroy_witness();
+		assert_ok!(Assets::freeze_asset(RuntimeOrigin::signed(1), 0));
-		assert_ok!(Assets::destroy(RuntimeOrigin::root(), 0, w));
+		assert_ok!(Assets::start_destroy(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::destroy_accounts(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::destroy_approvals(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::finish_destroy(RuntimeOrigin::signed(1), 0));
+
 		assert_eq!(Balances::reserved_balance(&1), 0);
 
 		assert!(!Asset::<Test>::contains_key(0));
@@ -358,24 +366,6 @@ fn lifecycle_should_work() {
 	});
 }
 
-#[test]
-fn destroy_with_bad_witness_should_not_work() {
-	new_test_ext().execute_with(|| {
-		Balances::make_free_balance_be(&1, 100);
-		assert_ok!(Assets::force_create(RuntimeOrigin::root(), 0, 1, true, 1));
-		let mut w = Asset::<Test>::get(0).unwrap().destroy_witness();
-		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 10, 100));
-		assert_eq!(asset_ids(), vec![0, 999]);
-		// witness too low
-		assert_noop!(Assets::destroy(RuntimeOrigin::signed(1), 0, w), Error::<Test>::BadWitness);
-		// witness too high is okay though
-		w.accounts += 2;
-		w.sufficients += 2;
-		assert_ok!(Assets::destroy(RuntimeOrigin::signed(1), 0, w));
-		assert_eq!(asset_ids(), vec![999]);
-	});
-}
-
 #[test]
 fn destroy_should_refund_approvals() {
 	new_test_ext().execute_with(|| {
@@ -388,8 +378,13 @@ fn destroy_should_refund_approvals() {
 		assert_eq!(Balances::reserved_balance(&1), 3);
 		assert_eq!(asset_ids(), vec![0, 999]);
 
-		let w = Asset::<Test>::get(0).unwrap().destroy_witness();
+		assert_ok!(Assets::freeze_asset(RuntimeOrigin::signed(1), 0));
-		assert_ok!(Assets::destroy(RuntimeOrigin::signed(1), 0, w));
+
+		assert_ok!(Assets::start_destroy(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::destroy_accounts(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::destroy_approvals(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::finish_destroy(RuntimeOrigin::signed(1), 0));
+
 		assert_eq!(Balances::reserved_balance(&1), 0);
 		assert_eq!(asset_ids(), vec![999]);
 
@@ -398,6 +393,58 @@ fn destroy_should_refund_approvals() {
 	});
 }
 
+#[test]
+fn partial_destroy_should_work() {
+	new_test_ext().execute_with(|| {
+		assert_ok!(Assets::force_create(RuntimeOrigin::root(), 0, 1, true, 1));
+		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 1, 10));
+		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 2, 10));
+		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 3, 10));
+		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 4, 10));
+		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 5, 10));
+		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 6, 10));
+		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 7, 10));
+		assert_ok!(Assets::freeze_asset(RuntimeOrigin::signed(1), 0));
+
+		assert_ok!(Assets::start_destroy(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::destroy_accounts(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::destroy_approvals(RuntimeOrigin::signed(1), 0));
+		// Asset is in use, as all the accounts have not yet been destroyed.
+		// We need to call destroy_accounts or destroy_approvals again until asset is completely
+		// cleaned up.
+		assert_noop!(Assets::finish_destroy(RuntimeOrigin::signed(1), 0), Error::<Test>::InUse);
+
+		System::assert_has_event(RuntimeEvent::Assets(crate::Event::AccountsDestroyed {
+			asset_id: 0,
+			accounts_destroyed: 5,
+			accounts_remaining: 2,
+		}));
+		System::assert_has_event(RuntimeEvent::Assets(crate::Event::ApprovalsDestroyed {
+			asset_id: 0,
+			approvals_destroyed: 0,
+			approvals_remaining: 0,
+		}));
+		// Partially destroyed Asset should continue to exist
+		assert!(Asset::<Test>::contains_key(0));
+
+		// Second call to destroy on PartiallyDestroyed asset
+		assert_ok!(Assets::destroy_accounts(RuntimeOrigin::signed(1), 0));
+		System::assert_has_event(RuntimeEvent::Assets(crate::Event::AccountsDestroyed {
+			asset_id: 0,
+			accounts_destroyed: 2,
+			accounts_remaining: 0,
+		}));
+		assert_ok!(Assets::destroy_approvals(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::destroy_approvals(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::finish_destroy(RuntimeOrigin::signed(1), 0));
+
+		System::assert_has_event(RuntimeEvent::Assets(crate::Event::Destroyed { asset_id: 0 }));
+
+		// Destroyed Asset should not exist
+		assert!(!Asset::<Test>::contains_key(0));
+	})
+}
+
 #[test]
 fn non_providing_should_work() {
 	new_test_ext().execute_with(|| {
@@ -540,7 +587,10 @@ fn transferring_frozen_asset_should_not_work() {
 		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 1, 100));
 		assert_eq!(Assets::balance(0, 1), 100);
 		assert_ok!(Assets::freeze_asset(RuntimeOrigin::signed(1), 0));
-		assert_noop!(Assets::transfer(RuntimeOrigin::signed(1), 0, 2, 50), Error::<Test>::Frozen);
+		assert_noop!(
+			Assets::transfer(RuntimeOrigin::signed(1), 0, 2, 50),
+			Error::<Test>::AssetNotLive
+		);
 		assert_ok!(Assets::thaw_asset(RuntimeOrigin::signed(1), 0));
 		assert_ok!(Assets::transfer(RuntimeOrigin::signed(1), 0, 2, 50));
 	});
@@ -556,7 +606,7 @@ fn approve_transfer_frozen_asset_should_not_work() {
 		assert_ok!(Assets::freeze_asset(RuntimeOrigin::signed(1), 0));
 		assert_noop!(
 			Assets::approve_transfer(RuntimeOrigin::signed(1), 0, 2, 50),
-			Error::<Test>::Frozen
+			Error::<Test>::AssetNotLive
 		);
 		assert_ok!(Assets::thaw_asset(RuntimeOrigin::signed(1), 0));
 		assert_ok!(Assets::approve_transfer(RuntimeOrigin::signed(1), 0, 2, 50));
@@ -590,8 +640,10 @@ fn origin_guards_should_work() {
 			Assets::force_transfer(RuntimeOrigin::signed(2), 0, 1, 2, 100),
 			Error::<Test>::NoPermission
 		);
-		let w = Asset::<Test>::get(0).unwrap().destroy_witness();
+		assert_noop!(
-		assert_noop!(Assets::destroy(RuntimeOrigin::signed(2), 0, w), Error::<Test>::NoPermission);
+			Assets::start_destroy(RuntimeOrigin::signed(2), 0),
+			Error::<Test>::NoPermission
+		);
 	});
 }
 
@@ -803,22 +855,37 @@ fn set_metadata_should_work() {
 
 /// Destroying an asset calls the `FrozenBalance::died` hooks of all accounts.
 #[test]
-fn destroy_calls_died_hooks() {
+fn destroy_accounts_calls_died_hooks() {
 	new_test_ext().execute_with(|| {
 		assert_ok!(Assets::force_create(RuntimeOrigin::root(), 0, 1, true, 50));
 		// Create account 1 and 2.
 		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 1, 100));
 		assert_ok!(Assets::mint(RuntimeOrigin::signed(1), 0, 2, 100));
-		// Destroy the asset.
+		// Destroy the accounts.
-		let w = Asset::<Test>::get(0).unwrap().destroy_witness();
+		assert_ok!(Assets::freeze_asset(RuntimeOrigin::signed(1), 0));
-		assert_ok!(Assets::destroy(RuntimeOrigin::signed(1), 0, w));
+		assert_ok!(Assets::start_destroy(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::destroy_accounts(RuntimeOrigin::signed(1), 0));
 
-		// Asset is gone and accounts 1 and 2 died.
+		// Accounts 1 and 2 died.
-		assert!(Asset::<Test>::get(0).is_none());
 		assert_eq!(hooks(), vec![Hook::Died(0, 1), Hook::Died(0, 2)]);
 	})
 }
 
+/// Destroying an asset calls the `FrozenBalance::died` hooks of all accounts.
+#[test]
+fn finish_destroy_asset_destroys_asset() {
+	new_test_ext().execute_with(|| {
+		assert_ok!(Assets::force_create(RuntimeOrigin::root(), 0, 1, true, 50));
+		// Destroy the accounts.
+		assert_ok!(Assets::freeze_asset(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::start_destroy(RuntimeOrigin::signed(1), 0));
+		assert_ok!(Assets::finish_destroy(RuntimeOrigin::signed(1), 0));
+
+		// Asset is gone
+		assert!(Asset::<Test>::get(0).is_none());
+	})
+}
+
 #[test]
 fn freezer_should_work() {
 	new_test_ext().execute_with(|| {

substrate/frame/assets/src/types.rs

@@ -29,6 +29,19 @@ pub(super) type DepositBalanceOf<T, I = ()> =
 pub(super) type AssetAccountOf<T, I> =
 	AssetAccount<<T as Config<I>>::Balance, DepositBalanceOf<T, I>, <T as Config<I>>::Extra>;
 
+/// AssetStatus holds the current state of the asset. It could either be Live and available for use,
+/// or in a Destroying state.
+#[derive(Clone, Encode, Decode, Eq, PartialEq, RuntimeDebug, MaxEncodedLen, TypeInfo)]
+pub(super) enum AssetStatus {
+	/// The asset is active and able to be used.
+	Live,
+	/// Whether the asset is frozen for non-admin transfers.
+	Frozen,
+	/// The asset is currently being destroyed, and all actions are no longer permitted on the
+	/// asset. Once set to `Destroying`, the asset can never transition back to a `Live` state.
+	Destroying,
+}
+
 #[derive(Clone, Encode, Decode, Eq, PartialEq, RuntimeDebug, MaxEncodedLen, TypeInfo)]
 pub struct AssetDetails<Balance, AccountId, DepositBalance> {
 	/// Can change `owner`, `issuer`, `freezer` and `admin` accounts.
@@ -54,18 +67,8 @@ pub struct AssetDetails<Balance, AccountId, DepositBalance> {
 	pub(super) sufficients: u32,
 	/// The total number of approvals.
 	pub(super) approvals: u32,
-	/// Whether the asset is frozen for non-admin transfers.
+	/// The status of the asset
-	pub(super) is_frozen: bool,
+	pub(super) status: AssetStatus,
-}
-
-impl<Balance, AccountId, DepositBalance> AssetDetails<Balance, AccountId, DepositBalance> {
-	pub fn destroy_witness(&self) -> DestroyWitness {
-		DestroyWitness {
-			accounts: self.accounts,
-			sufficients: self.sufficients,
-			approvals: self.approvals,
-		}
-	}
 }
 
 /// Data concerning an approval.
@@ -139,20 +142,6 @@ pub struct AssetMetadata<DepositBalance, BoundedString> {
 	pub(super) is_frozen: bool,
 }
 
-/// Witness data for the destroy transactions.
-#[derive(Copy, Clone, Encode, Decode, Eq, PartialEq, RuntimeDebug, MaxEncodedLen, TypeInfo)]
-pub struct DestroyWitness {
-	/// The number of accounts holding the asset.
-	#[codec(compact)]
-	pub(super) accounts: u32,
-	/// The number of accounts holding the asset with a self-sufficient reference.
-	#[codec(compact)]
-	pub(super) sufficients: u32,
-	/// The number of transfer-approvals of the asset.
-	#[codec(compact)]
-	pub(super) approvals: u32,
-}
-
 /// Trait for allowing a minimum balance on the account to be specified, beyond the
 /// `minimum_balance` of the asset. This is additive - the `minimum_balance` of the asset must be
 /// met *and then* anything here in addition.

substrate/frame/assets/src/weights.rs

@@ -49,7 +49,10 @@ use sp_std::marker::PhantomData;
 pub trait WeightInfo {
 	fn create() -> Weight;
 	fn force_create() -> Weight;
-	fn destroy(c: u32, s: u32, a: u32, ) -> Weight;
+	fn start_destroy() -> Weight;
+	fn destroy_accounts(c: u32) -> Weight;
+	fn destroy_approvals(m: u32) -> Weight;
+	fn finish_destroy() -> Weight;
 	fn mint() -> Weight;
 	fn burn() -> Weight;
 	fn transfer() -> Weight;
@@ -89,30 +92,49 @@ impl<T: frame_system::Config> WeightInfo for SubstrateWeight<T> {
 			.saturating_add(T::DbWeight::get().reads(1 as u64))
 			.saturating_add(T::DbWeight::get().writes(1 as u64))
 	}
+
 	// Storage: Assets Asset (r:1 w:1)
-	// Storage: Assets Account (r:5002 w:5001)
+	fn start_destroy() -> Weight {
-	// Storage: System Account (r:5000 w:5000)
+		Weight::from_ref_time(31_000_000 as u64)
-	// Storage: Assets Metadata (r:1 w:0)
+			.saturating_add(T::DbWeight::get().reads(1 as u64))
-	// Storage: Assets Approvals (r:501 w:500)
+			.saturating_add(T::DbWeight::get().writes(1 as u64))
-	/// The range of component `c` is `[0, 5000]`.
+	}
-	/// The range of component `s` is `[0, 5000]`.
+
-	/// The range of component `a` is `[0, 500]`.
+	// Storage: Assets Asset (r:1 w:1)
-	fn destroy(c: u32, s: u32, a: u32, ) -> Weight {
+	// Storage: Assets Account (r:1 w:0)
-		// Minimum execution time: 76_222_544 nanoseconds.
+	// Storage: System Account (r:20 w:20)
-		Weight::from_ref_time(76_864_587_000 as u64)
+	/// The range of component `c` is `[0, 1000]`.
-			// Standard Error: 127_086
+	fn destroy_accounts(c: u32, ) -> Weight {
-			.saturating_add(Weight::from_ref_time(8_645_143 as u64).saturating_mul(c as u64))
+		Weight::from_ref_time(37_000_000 as u64)
-			// Standard Error: 127_086
+			// Standard Error: 19_301
-			.saturating_add(Weight::from_ref_time(11_281_301 as u64).saturating_mul(s as u64))
+			.saturating_add(Weight::from_ref_time(25_467_908 as u64).saturating_mul(c as u64))
-			.saturating_add(T::DbWeight::get().reads(5 as u64))
+			.saturating_add(T::DbWeight::get().reads(2 as u64))
 			.saturating_add(T::DbWeight::get().reads((2 as u64).saturating_mul(c as u64)))
-			.saturating_add(T::DbWeight::get().reads((2 as u64).saturating_mul(s as u64)))
+			.saturating_add(T::DbWeight::get().writes(1 as u64))
-			.saturating_add(T::DbWeight::get().reads((1 as u64).saturating_mul(a as u64)))
-			.saturating_add(T::DbWeight::get().writes(2 as u64))
 			.saturating_add(T::DbWeight::get().writes((2 as u64).saturating_mul(c as u64)))
-			.saturating_add(T::DbWeight::get().writes((2 as u64).saturating_mul(s as u64)))
+	}
+
+	// Storage: Assets Asset (r:1 w:1)
+	// Storage: Assets Approvals (r:1 w:0)
+	/// The range of component `a` is `[0, 1000]`.
+	fn destroy_approvals(a: u32, ) -> Weight {
+		Weight::from_ref_time(39_000_000 as u64)
+			// Standard Error: 14_298
+			.saturating_add(Weight::from_ref_time(27_632_144 as u64).saturating_mul(a as u64))
+			.saturating_add(T::DbWeight::get().reads(2 as u64))
+			.saturating_add(T::DbWeight::get().reads((1 as u64).saturating_mul(a as u64)))
+			.saturating_add(T::DbWeight::get().writes(1 as u64))
 			.saturating_add(T::DbWeight::get().writes((1 as u64).saturating_mul(a as u64)))
 	}
+
+	// Storage: Assets Asset (r:1 w:1)
+	// Storage: Assets Metadata (r:1 w:0)
+	fn finish_destroy() -> Weight {
+		Weight::from_ref_time(33_000_000 as u64)
+			.saturating_add(T::DbWeight::get().reads(2 as u64))
+			.saturating_add(T::DbWeight::get().writes(1 as u64))
+	}
+
 	// Storage: Assets Asset (r:1 w:1)
 	// Storage: Assets Account (r:1 w:1)
 	fn mint() -> Weight {
@@ -302,30 +324,49 @@ impl WeightInfo for () {
 			.saturating_add(RocksDbWeight::get().reads(1 as u64))
 			.saturating_add(RocksDbWeight::get().writes(1 as u64))
 	}
+
 	// Storage: Assets Asset (r:1 w:1)
-	// Storage: Assets Account (r:5002 w:5001)
+	fn start_destroy() -> Weight {
-	// Storage: System Account (r:5000 w:5000)
+		Weight::from_ref_time(31_000_000 as u64)
-	// Storage: Assets Metadata (r:1 w:0)
+			.saturating_add(RocksDbWeight::get().reads(1 as u64))
-	// Storage: Assets Approvals (r:501 w:500)
+			.saturating_add(RocksDbWeight::get().writes(1 as u64))
-	/// The range of component `c` is `[0, 5000]`.
+	}
-	/// The range of component `s` is `[0, 5000]`.
+
-	/// The range of component `a` is `[0, 500]`.
+	// Storage: Assets Asset (r:1 w:1)
-	fn destroy(c: u32, s: u32, a: u32, ) -> Weight {
+	// Storage: Assets Account (r:1 w:0)
-		// Minimum execution time: 76_222_544 nanoseconds.
+	// Storage: System Account (r:20 w:20)
-		Weight::from_ref_time(76_864_587_000 as u64)
+	/// The range of component `c` is `[0, 1000]`.
-			// Standard Error: 127_086
+	fn destroy_accounts(c: u32, ) -> Weight {
-			.saturating_add(Weight::from_ref_time(8_645_143 as u64).saturating_mul(c as u64))
+		Weight::from_ref_time(37_000_000 as u64)
-			// Standard Error: 127_086
+			// Standard Error: 19_301
-			.saturating_add(Weight::from_ref_time(11_281_301 as u64).saturating_mul(s as u64))
+			.saturating_add(Weight::from_ref_time(25_467_908 as u64).saturating_mul(c as u64))
-			.saturating_add(RocksDbWeight::get().reads(5 as u64))
+			.saturating_add(RocksDbWeight::get().reads(2 as u64))
 			.saturating_add(RocksDbWeight::get().reads((2 as u64).saturating_mul(c as u64)))
-			.saturating_add(RocksDbWeight::get().reads((2 as u64).saturating_mul(s as u64)))
+			.saturating_add(RocksDbWeight::get().writes(1 as u64))
-			.saturating_add(RocksDbWeight::get().reads((1 as u64).saturating_mul(a as u64)))
-			.saturating_add(RocksDbWeight::get().writes(2 as u64))
 			.saturating_add(RocksDbWeight::get().writes((2 as u64).saturating_mul(c as u64)))
-			.saturating_add(RocksDbWeight::get().writes((2 as u64).saturating_mul(s as u64)))
+	}
+
+	// Storage: Assets Asset (r:1 w:1)
+	// Storage: Assets Approvals (r:1 w:0)
+	/// The range of component `a` is `[0, 1000]`.
+	fn destroy_approvals(a: u32, ) -> Weight {
+		Weight::from_ref_time(39_000_000 as u64)
+			// Standard Error: 14_298
+			.saturating_add(Weight::from_ref_time(27_632_144 as u64).saturating_mul(a as u64))
+			.saturating_add(RocksDbWeight::get().reads(2 as u64))
+			.saturating_add(RocksDbWeight::get().reads((1 as u64).saturating_mul(a as u64)))
+			.saturating_add(RocksDbWeight::get().writes(1 as u64))
 			.saturating_add(RocksDbWeight::get().writes((1 as u64).saturating_mul(a as u64)))
 	}
+
+	// Storage: Assets Asset (r:1 w:1)
+	// Storage: Assets Metadata (r:1 w:0)
+	fn finish_destroy() -> Weight {
+		Weight::from_ref_time(33_000_000 as u64)
+			.saturating_add(RocksDbWeight::get().reads(2 as u64))
+			.saturating_add(RocksDbWeight::get().writes(1 as u64))
+	}
+
 	// Storage: Assets Asset (r:1 w:1)
 	// Storage: Assets Account (r:1 w:1)
 	fn mint() -> Weight {

substrate/frame/transaction-payment/asset-tx-payment/src/tests.rs

@@ -168,6 +168,7 @@ impl pallet_assets::Config for Runtime {
 	type Freezer = ();
 	type Extra = ();
 	type WeightInfo = ();
+	type RemoveItemsLimit = ConstU32<1000>;
 }
 
 pub struct HardcodedAuthor;