[NPoS] Check if staker is exposed in paged exposure storage entries (#2369)

Addresses a bug caused by
https://github.com/paritytech/polkadot-sdk/pull/1189. The changes are
still not released yet, so would like to push the fix soon so it can go
together with the release of the above PR.

`fast_unstake` checks if a staker is exposed in an era. However, this fn
is still returning whether the staker is exposed based on the old
storage item. This PR fixes that by looking in both old and new exposure
storages.

Also adds some integrity tests for paged exposures.

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

@@ -794,7 +794,7 @@ impl<T: Config> Pallet<T> {
 		stash: T::AccountId,
 		exposure: Exposure<T::AccountId, BalanceOf<T>>,
 	) {
-		<ErasStakers<T>>::insert(&current_era, &stash, &exposure);
+		EraInfo::<T>::set_exposure(current_era, &stash, exposure);
 	}
 
 	#[cfg(feature = "runtime-benchmarks")]
@@ -1745,9 +1745,16 @@ impl<T: Config> StakingInterface for Pallet<T> {
 	}
 
 	fn is_exposed_in_era(who: &Self::AccountId, era: &EraIndex) -> bool {
+		// look in the non paged exposures
+		// FIXME: Can be cleaned up once non paged exposures are cleared (https://github.com/paritytech/polkadot-sdk/issues/433)
 		ErasStakers::<T>::iter_prefix(era).any(|(validator, exposures)| {
 			validator == *who || exposures.others.iter().any(|i| i.who == *who)
 		})
+			||
+		// look in the paged exposures
+		ErasStakersPaged::<T>::iter_prefix((era,)).any(|((validator, _), exposure_page)| {
+			validator == *who || exposure_page.others.iter().any(|i| i.who == *who)
+		})
 	}
 	fn status(
 		who: &Self::AccountId,
@@ -1812,6 +1819,7 @@ impl<T: Config> Pallet<T> {
 
 		Self::check_nominators()?;
 		Self::check_exposures()?;
+		Self::check_paged_exposures()?;
 		Self::check_ledgers()?;
 		Self::check_count()
 	}
@@ -1860,6 +1868,70 @@ impl<T: Config> Pallet<T> {
 			.collect::<Result<(), TryRuntimeError>>()
 	}
 
+	fn check_paged_exposures() -> Result<(), TryRuntimeError> {
+		use sp_staking::PagedExposureMetadata;
+		use sp_std::collections::btree_map::BTreeMap;
+
+		// Sanity check for the paged exposure of the active era.
+		let mut exposures: BTreeMap<T::AccountId, PagedExposureMetadata<BalanceOf<T>>> =
+			BTreeMap::new();
+		let era = Self::active_era().unwrap().index;
+		let accumulator_default = PagedExposureMetadata {
+			total: Zero::zero(),
+			own: Zero::zero(),
+			nominator_count: 0,
+			page_count: 0,
+		};
+
+		ErasStakersPaged::<T>::iter_prefix((era,))
+			.map(|((validator, _page), expo)| {
+				ensure!(
+					expo.page_total ==
+						expo.others.iter().map(|e| e.value).fold(Zero::zero(), |acc, x| acc + x),
+					"wrong total exposure for the page.",
+				);
+
+				let metadata = exposures.get(&validator).unwrap_or(&accumulator_default);
+				exposures.insert(
+					validator,
+					PagedExposureMetadata {
+						total: metadata.total + expo.page_total,
+						own: metadata.own,
+						nominator_count: metadata.nominator_count + expo.others.len() as u32,
+						page_count: metadata.page_count + 1,
+					},
+				);
+
+				Ok(())
+			})
+			.collect::<Result<(), TryRuntimeError>>()?;
+
+		exposures
+			.iter()
+			.map(|(validator, metadata)| {
+				let actual_overview = ErasStakersOverview::<T>::get(era, validator);
+
+				ensure!(actual_overview.is_some(), "No overview found for a paged exposure");
+				let actual_overview = actual_overview.unwrap();
+
+				ensure!(
+					actual_overview.total == metadata.total + actual_overview.own,
+					"Exposure metadata does not have correct total exposed stake."
+				);
+				ensure!(
+					actual_overview.nominator_count == metadata.nominator_count,
+					"Exposure metadata does not have correct count of nominators."
+				);
+				ensure!(
+					actual_overview.page_count == metadata.page_count,
+					"Exposure metadata does not have correct count of pages."
+				);
+
+				Ok(())
+			})
+			.collect::<Result<(), TryRuntimeError>>()
+	}
+
 	fn check_nominators() -> Result<(), TryRuntimeError> {
 		// a check per nominator to ensure their entire stake is correctly distributed. Will only
 		// kick-in if the nomination was submitted before the current era.

substrate/frame/staking/src/tests.rs

@@ -6637,6 +6637,14 @@ fn test_validator_exposure_is_backward_compatible_with_non_paged_rewards_payout(
 		);
 		assert_eq!(EraInfo::<Test>::get_page_count(1, &11), 2);
 
+		// validator is exposed
+		assert!(<Staking as sp_staking::StakingInterface>::is_exposed_in_era(&11, &1));
+		// nominators are exposed
+		for i in 10..15 {
+			let who: AccountId = 1000 + i;
+			assert!(<Staking as sp_staking::StakingInterface>::is_exposed_in_era(&who, &1));
+		}
+
 		// case 2: exposure exist in ErasStakers and ErasStakersClipped (legacy).
 		// delete paged storage and add exposure to clipped storage
 		<ErasStakersPaged<Test>>::remove((1, 11, 0));
@@ -6672,6 +6680,14 @@ fn test_validator_exposure_is_backward_compatible_with_non_paged_rewards_payout(
 		assert_eq!(actual_exposure_full.own, 1000);
 		assert_eq!(actual_exposure_full.total, total_exposure);
 
+		// validator is exposed
+		assert!(<Staking as sp_staking::StakingInterface>::is_exposed_in_era(&11, &1));
+		// nominators are exposed
+		for i in 10..15 {
+			let who: AccountId = 1000 + i;
+			assert!(<Staking as sp_staking::StakingInterface>::is_exposed_in_era(&who, &1));
+		}
+
 		// for pages other than 0, clipped storage returns empty exposure
 		assert_eq!(EraInfo::<Test>::get_paged_exposure(1, &11, 1), None);
 		// page size is 1 for clipped storage