Extrinsic to restore corrupt staking ledgers (#3706)

This PR adds a new extrinsic `Call::restore_ledger ` gated by
`StakingAdmin` origin that restores a corrupted staking ledger. This
extrinsic will be used to recover ledgers that were affected by the
issue discussed in
https://github.com/paritytech/polkadot-sdk/issues/3245.

The extrinsic will re-write the storage items associated with a stash
account provided as input parameter. The data used to reset the ledger
can be either i) fetched on-chain or ii) partially/totally set by the
input parameters of the call.

In order to use on-chain data to restore the staking locks, we need a
way to read the current lock in the balances pallet. This PR adds a
`InspectLockableCurrency` trait and implements it in the pallet
balances. An alternative would be to tightly couple staking with the
pallet balances but that's inelegant (an example of how it would look
like in [this
branch](https://github.com/paritytech/polkadot-sdk/tree/gpestana/ledger-badstate-clean_tightly)).

More details on the type of corruptions and corresponding fixes
https://hackmd.io/DLb5jEYWSmmvqXC9ae4yRg?view#/

We verified that the `Call::restore_ledger` does fix all current
corrupted ledgers in Polkadot and Kusama. You can verify it here
https://hackmd.io/v-XNrEoGRpe7APR-EZGhOA.

**Changes introduced**
- Adds `Call::restore_ledger ` extrinsic to recover a corrupted ledger;
- Adds trait `frame_support::traits::currency::InspectLockableCurrency`
to allow external pallets to read current locks given an account and
lock ID;
- Implements the `InspectLockableCurrency` in the pallet-balances.
- Adds staking locks try-runtime checks
(https://github.com/paritytech/polkadot-sdk/issues/3751)

**Todo**
- [x] benchmark `Call::restore_ledger`
- [x] throughout testing of all ledger recovering cases
- [x] consider adding the staking locks try-runtime checks to this PR
(https://github.com/paritytech/polkadot-sdk/issues/3751)
- [x] simulate restoring all ledgers
(https://hackmd.io/Dsa2tvhISNSs7zcqriTaxQ?view) in Polkadot and Kusama
using chopsticks -- https://hackmd.io/v-XNrEoGRpe7APR-EZGhOA

Related to https://github.com/paritytech/polkadot-sdk/issues/3245
Closes https://github.com/paritytech/polkadot-sdk/issues/3751

---------

Co-authored-by: command-bot <>

substrate/frame/staking/src/pallet/impls.rs

@@ -27,8 +27,8 @@ use frame_support::{
 	dispatch::WithPostDispatchInfo,
 	pallet_prelude::*,
 	traits::{
-		Currency, Defensive, DefensiveSaturating, EstimateNextNewSession, Get, Imbalance, Len,
-		OnUnbalanced, TryCollect, UnixTime,
+		Currency, Defensive, DefensiveSaturating, EstimateNextNewSession, Get, Imbalance,
+		InspectLockableCurrency, Len, OnUnbalanced, TryCollect, UnixTime,
 	},
 	weights::Weight,
 };
@@ -50,8 +50,8 @@ use sp_std::prelude::*;
 use crate::{
 	election_size_tracker::StaticTracker, log, slashing, weights::WeightInfo, ActiveEraInfo,
 	BalanceOf, EraInfo, EraPayout, Exposure, ExposureOf, Forcing, IndividualExposure,
-	MaxNominationsOf, MaxWinnersOf, Nominations, NominationsQuota, PositiveImbalanceOf,
-	RewardDestination, SessionInterface, StakingLedger, ValidatorPrefs,
+	LedgerIntegrityState, MaxNominationsOf, MaxWinnersOf, Nominations, NominationsQuota,
+	PositiveImbalanceOf, RewardDestination, SessionInterface, StakingLedger, ValidatorPrefs,
 };
 
 use super::pallet::*;
@@ -84,6 +84,38 @@ impl<T: Config> Pallet<T> {
 		StakingLedger::<T>::paired_account(Stash(stash.clone()))
 	}
 
+	/// Inspects and returns the corruption state of a ledger and bond, if any.
+	///
+	/// Note: all operations in this method access directly the `Bonded` and `Ledger` storage maps
+	/// instead of using the [`StakingLedger`] API since the bond and/or ledger may be corrupted.
+	pub(crate) fn inspect_bond_state(
+		stash: &T::AccountId,
+	) -> Result<LedgerIntegrityState, Error<T>> {
+		let lock = T::Currency::balance_locked(crate::STAKING_ID, &stash);
+
+		let controller = <Bonded<T>>::get(stash).ok_or_else(|| {
+			if lock == Zero::zero() {
+				Error::<T>::NotStash
+			} else {
+				Error::<T>::BadState
+			}
+		})?;
+
+		match Ledger::<T>::get(controller) {
+			Some(ledger) =>
+				if ledger.stash != *stash {
+					Ok(LedgerIntegrityState::Corrupted)
+				} else {
+					if lock != ledger.total {
+						Ok(LedgerIntegrityState::LockCorrupted)
+					} else {
+						Ok(LedgerIntegrityState::Ok)
+					}
+				},
+			None => Ok(LedgerIntegrityState::CorruptedKilled),
+		}
+	}
+
 	/// The total balance that can be slashed from a stash account as of right now.
 	pub fn slashable_balance_of(stash: &T::AccountId) -> BalanceOf<T> {
 		// Weight note: consider making the stake accessible through stash.
@@ -1837,12 +1869,12 @@ impl<T: Config> Pallet<T> {
 			"VoterList contains non-staker"
 		);
 
+		Self::check_ledgers()?;
 		Self::check_bonded_consistency()?;
 		Self::check_payees()?;
 		Self::check_nominators()?;
 		Self::check_exposures()?;
 		Self::check_paged_exposures()?;
-		Self::check_ledgers()?;
 		Self::check_count()
 	}
 
@@ -1851,6 +1883,7 @@ impl<T: Config> Pallet<T> {
 	/// * A bonded (stash, controller) pair should have only one associated ledger. I.e. if the
 	///   ledger is bonded by stash, the controller account must not bond a different ledger.
 	/// * A bonded (stash, controller) pair must have an associated ledger.
+	///
 	/// NOTE: these checks result in warnings only. Once
 	/// <https://github.com/paritytech/polkadot-sdk/issues/3245> is resolved, turn warns into check
 	/// failures.
@@ -1945,19 +1978,18 @@ impl<T: Config> Pallet<T> {
 	}
 
 	/// Invariants:
-	/// * `ledger.controller` is not stored in the storage (but populated at retrieval).
 	/// * Stake consistency: ledger.total == ledger.active + sum(ledger.unlocking).
-	/// * The controller keying the ledger and the ledger stash matches the state of the `Bonded`
-	/// storage.
+	/// * The ledger's controller and stash matches the associated `Bonded` tuple.
+	/// * Staking locked funds for every bonded stash should be the same as its ledger's total.
+	/// * Staking ledger and bond are not corrupted.
 	fn check_ledgers() -> Result<(), TryRuntimeError> {
 		Bonded::<T>::iter()
 			.map(|(stash, ctrl)| {
-				// `ledger.controller` is never stored in raw storage.
-				let raw = Ledger::<T>::get(stash).unwrap_or_else(|| {
-					Ledger::<T>::get(ctrl.clone())
-						.expect("try_check: bonded stash/ctrl does not have an associated ledger")
-				});
-				ensure!(raw.controller.is_none(), "raw storage controller should be None");
+				// ensure locks consistency.
+				ensure!(
+					Self::inspect_bond_state(&stash) == Ok(LedgerIntegrityState::Ok),
+					"bond, ledger and/or staking lock inconsistent for a bonded stash."
+				);
 
 				// ensure ledger consistency.
 				Self::ensure_ledger_consistent(ctrl)