add additional assurances to create_inherent (#4349)

* minor: move checks into separate fn

* add additional validity checks

* simplify shuffling

* Closes potential OOB weight

* improve docs

* fooo

* remove obsolete comment

* move filtering into the rollback-transaction

Technically this is not necessary but avoids future footguns.

* move check up and avoid duplicate checks

* refactor: make sure backed candidates are sane, even more

* doc wording

Co-authored-by: Zeke Mostov <z.mostov@gmail.com>

* refactor: avoid const generics for sake of wasm size

`true` -> `FullCheck::Skip`, `false` -> `FullCheck::Yes`.

* chore: unify `CandidateCheckContext` instance names

* refactor: introduce `IndexedRetain` for `Vec<T>`

* chore: make tests prefix free

* doc: re-introduce removed comment

* refactor: remove another const generic to save some wasm size

Co-authored-by: Zeke Mostov <z.mostov@gmail.com>

polkadot/roadmap/implementers-guide/src/runtime/inclusion.md

@@ -51,13 +51,14 @@ All failed checks should lead to an unrecoverable error making the block invalid
   1. For each applied bit of each availability-bitfield, set the bit for the validator in the `CandidatePendingAvailability`'s `availability_votes` bitfield. Track all candidates that now have >2/3 of bits set in their `availability_votes`. These candidates are now available and can be enacted.
   1. For all now-available candidates, invoke the `enact_candidate` routine with the candidate and relay-parent number.
   1. Return a list of `(CoreIndex, CandidateHash)` from freed cores consisting of the cores where candidates have become available.
-* `sanitize_bitfields<T: crate::inclusion::Config, const CHECK_SIGS: bool>(
+* `sanitize_bitfields<T: crate::inclusion::Config>(
     unchecked_bitfields: UncheckedSignedAvailabilityBitfields,
     disputed_bitfield: DisputedBitfield,
     expected_bits: usize,
     parent_hash: T::Hash,
     session_index: SessionIndex,
     validators: &[ValidatorId],
+    full_check: FullCheck,
   )`:
   1. check that `disputed_bitfield` has the same number of bits as the `expected_bits`, iff not return early with an empty vec.
   1. each of the below checks is for each bitfield. If a check does not pass the bitfield will be skipped.
@@ -65,7 +66,7 @@ All failed checks should lead to an unrecoverable error making the block invalid
   1. check that the number of bits is equal to `expected_bits`.
   1. check that the validator index is strictly increasing (and thus also unique).
   1. check that the validator bit index is not out of bounds.
-  1. check the validators signature, iff `CHECK_SIGS=true`.
+  1. check the validators signature, iff `full_check=FullCheck::Yes`.
 
 * `sanitize_backed_candidates<T: crate::inclusion::Config, F: Fn(CandidateHash) -> bool>(
     relay_parent: T::Hash,

polkadot/runtime/parachains/src/inclusion.rs

@@ -22,7 +22,7 @@
 
 use crate::{
 	configuration, disputes, dmp, hrmp, paras,
-	paras_inherent::{sanitize_bitfields, DisputedBitfield, VERIFY_SIGS},
+	paras_inherent::{sanitize_bitfields, DisputedBitfield},
 	scheduler::CoreAssignment,
 	shared, ump,
 };
@@ -56,6 +56,19 @@ pub struct AvailabilityBitfieldRecord<N> {
 	submitted_at: N,                // for accounting, as meaning of bits may change over time.
 }
 
+/// Determines if all checks should be applied or if a subset was already completed
+/// in a code path that will be executed afterwards or was already executed before.
+#[derive(Encode, Decode, PartialEq, Eq, RuntimeDebug, TypeInfo)]
+pub(crate) enum FullCheck {
+	/// Yes, do a full check, skip nothing.
+	Yes,
+	/// Skip a subset of checks that are already completed before.
+	///
+	/// Attention: Should only be used when absolutely sure that the required
+	/// checks are completed before.
+	Skip,
+}
+
 /// A backed candidate pending availability.
 #[derive(Encode, Decode, PartialEq, TypeInfo)]
 #[cfg_attr(test, derive(Debug, Default))]
@@ -403,13 +416,14 @@ impl<T: Config> Pallet<T> {
 		let session_index = shared::Pallet::<T>::session_index();
 		let parent_hash = frame_system::Pallet::<T>::parent_hash();
 
-		let checked_bitfields = sanitize_bitfields::<T, VERIFY_SIGS>(
+		let checked_bitfields = sanitize_bitfields::<T>(
 			signed_bitfields,
 			disputed_bitfield,
 			expected_bits,
 			parent_hash,
 			session_index,
 			&validators[..],
+			FullCheck::Yes,
 		);
 
 		let freed_cores = Self::update_pending_availability_and_get_freed_cores::<_, true>(
@@ -427,12 +441,16 @@ impl<T: Config> Pallet<T> {
 	///
 	/// Both should be sorted ascending by core index, and the candidates should be a subset of
 	/// scheduled cores. If these conditions are not met, the execution of the function fails.
-	pub(crate) fn process_candidates(
+	pub(crate) fn process_candidates<GV>(
 		parent_storage_root: T::Hash,
 		candidates: Vec<BackedCandidate<T::Hash>>,
 		scheduled: Vec<CoreAssignment>,
-		group_validators: impl Fn(GroupIndex) -> Option<Vec<ValidatorIndex>>,
+		group_validators: GV,
-	) -> Result<ProcessedCandidates<T::Hash>, DispatchError> {
+		full_check: FullCheck,
+	) -> Result<ProcessedCandidates<T::Hash>, DispatchError>
+	where
+		GV: Fn(GroupIndex) -> Option<Vec<ValidatorIndex>>,
+	{
 		ensure!(candidates.len() <= scheduled.len(), Error::<T>::UnscheduledCandidate);
 
 		if scheduled.is_empty() {
@@ -446,7 +464,7 @@ impl<T: Config> Pallet<T> {
 		// before of the block where we include a candidate (i.e. this code path).
 		let now = <frame_system::Pallet<T>>::block_number();
 		let relay_parent_number = now - One::one();
-		let check_cx = CandidateCheckContext::<T>::new(now, relay_parent_number);
+		let check_ctx = CandidateCheckContext::<T>::new(now, relay_parent_number);
 
 		// Collect candidate receipts with backers.
 		let mut candidate_receipt_with_backing_validator_indices =
@@ -481,54 +499,20 @@ impl<T: Config> Pallet<T> {
 			//
 			// In the meantime, we do certain sanity checks on the candidates and on the scheduled
 			// list.
-			'a: for (candidate_idx, backed_candidate) in candidates.iter().enumerate() {
+			'next_backed_candidate: for (candidate_idx, backed_candidate) in
+				candidates.iter().enumerate()
+			{
+				if let FullCheck::Yes = full_check {
+					check_ctx.verify_backed_candidate(
+						parent_hash,
+						candidate_idx,
+						backed_candidate,
+					)?;
+				}
+
 				let para_id = backed_candidate.descriptor().para_id;
 				let mut backers = bitvec::bitvec![BitOrderLsb0, u8; 0; validators.len()];
 
-				// we require that the candidate is in the context of the parent block.
-				ensure!(
-					backed_candidate.descriptor().relay_parent == parent_hash,
-					Error::<T>::CandidateNotInParentContext,
-				);
-				ensure!(
-					backed_candidate.descriptor().check_collator_signature().is_ok(),
-					Error::<T>::NotCollatorSigned,
-				);
-
-				let validation_code_hash =
-					<paras::Pallet<T>>::validation_code_hash_at(para_id, now, None)
-						// A candidate for a parachain without current validation code is not scheduled.
-						.ok_or_else(|| Error::<T>::UnscheduledCandidate)?;
-				ensure!(
-					backed_candidate.descriptor().validation_code_hash == validation_code_hash,
-					Error::<T>::InvalidValidationCodeHash,
-				);
-
-				ensure!(
-					backed_candidate.descriptor().para_head ==
-						backed_candidate.candidate.commitments.head_data.hash(),
-					Error::<T>::ParaHeadMismatch,
-				);
-
-				if let Err(err) = check_cx.check_validation_outputs(
-					para_id,
-					&backed_candidate.candidate.commitments.head_data,
-					&backed_candidate.candidate.commitments.new_validation_code,
-					backed_candidate.candidate.commitments.processed_downward_messages,
-					&backed_candidate.candidate.commitments.upward_messages,
-					T::BlockNumber::from(backed_candidate.candidate.commitments.hrmp_watermark),
-					&backed_candidate.candidate.commitments.horizontal_messages,
-				) {
-					log::debug!(
-						target: LOG_TARGET,
-						"Validation outputs checking during inclusion of a candidate {} for parachain `{}` failed: {:?}",
-						candidate_idx,
-						u32::from(para_id),
-						err,
-					);
-					Err(err.strip_into_dispatch_err::<T>())?;
-				};
-
 				for (i, assignment) in scheduled[skip..].iter().enumerate() {
 					check_assignment_in_order(assignment)?;
 
@@ -631,7 +615,7 @@ impl<T: Config> Pallet<T> {
 							backers,
 							assignment.group_idx,
 						));
-						continue 'a
+						continue 'next_backed_candidate
 					}
 				}
 
@@ -682,7 +666,7 @@ impl<T: Config> Pallet<T> {
 					availability_votes,
 					relay_parent_number,
 					backers: backers.to_bitvec(),
-					backed_in_number: check_cx.now,
+					backed_in_number: check_ctx.now,
 					backing_group: group,
 				},
 			);
@@ -704,9 +688,9 @@ impl<T: Config> Pallet<T> {
 		// `relay_parent_number` is equal to `now`.
 		let now = <frame_system::Pallet<T>>::block_number();
 		let relay_parent_number = now;
-		let check_cx = CandidateCheckContext::<T>::new(now, relay_parent_number);
+		let check_ctx = CandidateCheckContext::<T>::new(now, relay_parent_number);
 
-		if let Err(err) = check_cx.check_validation_outputs(
+		if let Err(err) = check_ctx.check_validation_outputs(
 			para_id,
 			&validation_outputs.head_data,
 			&validation_outputs.new_validation_code,
@@ -941,17 +925,78 @@ impl<BlockNumber> AcceptanceCheckErr<BlockNumber> {
 }
 
 /// A collection of data required for checking a candidate.
-struct CandidateCheckContext<T: Config> {
+pub(crate) struct CandidateCheckContext<T: Config> {
 	config: configuration::HostConfiguration<T::BlockNumber>,
 	now: T::BlockNumber,
 	relay_parent_number: T::BlockNumber,
 }
 
 impl<T: Config> CandidateCheckContext<T> {
-	fn new(now: T::BlockNumber, relay_parent_number: T::BlockNumber) -> Self {
+	pub(crate) fn new(now: T::BlockNumber, relay_parent_number: T::BlockNumber) -> Self {
 		Self { config: <configuration::Pallet<T>>::config(), now, relay_parent_number }
 	}
 
+	/// Execute verification of the candidate.
+	///
+	/// Assures:
+	///  * correct expected relay parent reference
+	///  * collator signature check passes
+	///  * code hash of commitments matches current code hash
+	///  * para head in the descriptor and commitments match
+	pub(crate) fn verify_backed_candidate(
+		&self,
+		parent_hash: <T as frame_system::Config>::Hash,
+		candidate_idx: usize,
+		backed_candidate: &BackedCandidate<<T as frame_system::Config>::Hash>,
+	) -> Result<(), Error<T>> {
+		let para_id = backed_candidate.descriptor().para_id;
+		let now = self.now;
+
+		// we require that the candidate is in the context of the parent block.
+		ensure!(
+			backed_candidate.descriptor().relay_parent == parent_hash,
+			Error::<T>::CandidateNotInParentContext,
+		);
+		ensure!(
+			backed_candidate.descriptor().check_collator_signature().is_ok(),
+			Error::<T>::NotCollatorSigned,
+		);
+
+		let validation_code_hash = <paras::Pallet<T>>::validation_code_hash_at(para_id, now, None)
+			// A candidate for a parachain without current validation code is not scheduled.
+			.ok_or_else(|| Error::<T>::UnscheduledCandidate)?;
+		ensure!(
+			backed_candidate.descriptor().validation_code_hash == validation_code_hash,
+			Error::<T>::InvalidValidationCodeHash,
+		);
+
+		ensure!(
+			backed_candidate.descriptor().para_head ==
+				backed_candidate.candidate.commitments.head_data.hash(),
+			Error::<T>::ParaHeadMismatch,
+		);
+
+		if let Err(err) = self.check_validation_outputs(
+			para_id,
+			&backed_candidate.candidate.commitments.head_data,
+			&backed_candidate.candidate.commitments.new_validation_code,
+			backed_candidate.candidate.commitments.processed_downward_messages,
+			&backed_candidate.candidate.commitments.upward_messages,
+			T::BlockNumber::from(backed_candidate.candidate.commitments.hrmp_watermark),
+			&backed_candidate.candidate.commitments.horizontal_messages,
+		) {
+			log::debug!(
+				target: LOG_TARGET,
+				"Validation outputs checking during inclusion of a candidate {} for parachain `{}` failed: {:?}",
+				candidate_idx,
+				u32::from(para_id),
+				err,
+			);
+			Err(err.strip_into_dispatch_err::<T>())?;
+		};
+		Ok(())
+	}
+
 	/// Check the given outputs after candidate validation on whether it passes the acceptance
 	/// criteria.
 	fn check_validation_outputs(
@@ -1935,6 +1980,7 @@ pub(crate) mod tests {
 						vec![backed],
 						vec![chain_b_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::UnscheduledCandidate
 				);
@@ -1990,6 +2036,7 @@ pub(crate) mod tests {
 						vec![backed_b, backed_a],
 						vec![chain_a_assignment.clone(), chain_b_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::UnscheduledCandidate
 				);
@@ -2023,6 +2070,7 @@ pub(crate) mod tests {
 						vec![backed],
 						vec![chain_a_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::InsufficientBacking
 				);
@@ -2058,6 +2106,7 @@ pub(crate) mod tests {
 						vec![backed],
 						vec![chain_a_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::CandidateNotInParentContext
 				);
@@ -2097,6 +2146,7 @@ pub(crate) mod tests {
 							thread_a_assignment.clone(),
 						],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::WrongCollator,
 				);
@@ -2135,6 +2185,7 @@ pub(crate) mod tests {
 						vec![backed],
 						vec![thread_a_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::NotCollatorSigned
 				);
@@ -2185,6 +2236,7 @@ pub(crate) mod tests {
 						vec![backed],
 						vec![chain_a_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::CandidateScheduledBeforeParaFree
 				);
@@ -2228,6 +2280,7 @@ pub(crate) mod tests {
 						vec![backed],
 						vec![chain_a_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::CandidateScheduledBeforeParaFree
 				);
@@ -2279,6 +2332,7 @@ pub(crate) mod tests {
 						vec![backed],
 						vec![chain_a_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::PrematureCodeUpgrade
 				);
@@ -2313,6 +2367,7 @@ pub(crate) mod tests {
 						vec![backed],
 						vec![chain_a_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Err(Error::<Test>::ValidationDataHashMismatch.into()),
 				);
@@ -2348,6 +2403,7 @@ pub(crate) mod tests {
 						vec![backed],
 						vec![chain_a_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::InvalidValidationCodeHash
 				);
@@ -2383,6 +2439,7 @@ pub(crate) mod tests {
 						vec![backed],
 						vec![chain_a_assignment.clone()],
 						&group_validators,
+						FullCheck::Yes,
 					),
 					Error::<Test>::ParaHeadMismatch
 				);
@@ -2552,6 +2609,7 @@ pub(crate) mod tests {
 					thread_a_assignment.clone(),
 				],
 				&group_validators,
+				FullCheck::Yes,
 			)
 			.expect("candidates scheduled, in order, and backed");
 
@@ -2742,6 +2800,7 @@ pub(crate) mod tests {
 					vec![backed_a],
 					vec![chain_a_assignment.clone()],
 					&group_validators,
+					FullCheck::Yes,
 				)
 				.expect("candidates scheduled, in order, and backed");
 

polkadot/runtime/parachains/src/paras_inherent.rs

@@ -23,7 +23,9 @@
 
 use crate::{
 	disputes::DisputesHandler,
-	inclusion, initializer,
+	inclusion,
+	inclusion::{CandidateCheckContext, FullCheck},
+	initializer,
 	scheduler::{self, CoreAssignment, FreedReason},
 	shared, ump,
 };
@@ -42,21 +44,21 @@ use primitives::v1::{
 	UncheckedSignedAvailabilityBitfields, ValidatorId, ValidatorIndex,
 	PARACHAINS_INHERENT_IDENTIFIER,
 };
-use rand::{Rng, SeedableRng};
+use rand::{seq::SliceRandom, SeedableRng};
+
 use scale_info::TypeInfo;
-use sp_runtime::traits::Header as HeaderT;
+use sp_runtime::traits::{Header as HeaderT, One};
 use sp_std::{
 	cmp::Ordering,
 	collections::{btree_map::BTreeMap, btree_set::BTreeSet},
 	prelude::*,
 	vec::Vec,
 };
+
 #[cfg(feature = "runtime-benchmarks")]
 mod benchmarking;
 
 const LOG_TARGET: &str = "runtime::inclusion-inherent";
-const SKIP_SIG_VERIFY: bool = false;
-pub(crate) const VERIFY_SIGS: bool = true;
 
 pub trait WeightInfo {
 	/// Variant over `v`, the count of dispute statements in a dispute statement set. This gives the
@@ -158,6 +160,29 @@ fn backed_candidates_weight<T: frame_system::Config + Config>(
 		.fold(0, |acc, x| acc.saturating_add(x))
 }
 
+/// A helper trait to allow calling retain while getting access
+/// to the index of the item in the `vec`.
+trait IndexedRetain<T> {
+	/// Retains only the elements specified by the predicate.
+	///
+	/// In other words, remove all elements `e` residing at
+	/// index `i` such that `f(i, &e)` returns `false`. This method
+	/// operates in place, visiting each element exactly once in the
+	/// original order, and preserves the order of the retained elements.
+	fn indexed_retain(&mut self, f: impl FnMut(usize, &T) -> bool);
+}
+
+impl<T> IndexedRetain<T> for Vec<T> {
+	fn indexed_retain(&mut self, mut f: impl FnMut(usize, &T) -> bool) {
+		let mut idx = 0_usize;
+		self.retain(move |item| {
+			let ret = f(idx, item);
+			idx += 1_usize;
+			ret
+		})
+	}
+}
+
 /// A bitfield concerning concluded disputes for candidates
 /// associated to the core index equivalent to the bit position.
 #[derive(Default, PartialEq, Eq, Clone, Encode, Decode, RuntimeDebug, TypeInfo)]
@@ -252,8 +277,7 @@ pub mod pallet {
 			// (`enter`) and the off-chain checks by the block author (this function). Once we are confident
 			// in all the logic in this module this check should be removed to optimize performance.
 
-			let inherent_data =
+			let inherent_data = match Self::enter_inner(inherent_data.clone(), FullCheck::Skip) {
-				match Self::enter(frame_system::RawOrigin::None.into(), inherent_data.clone()) {
 				Ok(_) => inherent_data,
 				Err(err) => {
 					log::error!(
@@ -329,6 +353,16 @@ pub mod pallet {
 			ensure!(!Included::<T>::exists(), Error::<T>::TooManyInclusionInherents);
 			Included::<T>::set(Some(()));
 
+			Self::enter_inner(data, FullCheck::Yes)
+		}
+	}
+}
+
+impl<T: Config> Pallet<T> {
+	pub(crate) fn enter_inner(
+		data: ParachainsInherentData<T::Header>,
+		full_check: FullCheck,
+	) -> DispatchResultWithPostInfo {
 		let ParachainsInherentData {
 			bitfields: mut signed_bitfields,
 			mut backed_candidates,
@@ -351,6 +385,8 @@ pub mod pallet {
 			Error::<T>::InvalidParentHeader,
 		);
 
+		let now = <frame_system::Pallet<T>>::block_number();
+
 		let mut candidate_weight = backed_candidates_weight::<T>(&backed_candidates);
 		let mut bitfields_weight = signed_bitfields_weight::<T>(signed_bitfields.len());
 		let disputes_weight = dispute_statements_weight::<T>(&disputes);
@@ -454,7 +490,6 @@ pub mod pallet {
 		);
 
 		// Inform the disputes module of all included candidates.
-			let now = <frame_system::Pallet<T>>::block_number();
 		for (_, candidate_hash) in &freed_concluded {
 			T::DisputesHandler::note_included(current_session, *candidate_hash, now);
 		}
@@ -468,8 +503,9 @@ pub mod pallet {
 		let backed_candidates = sanitize_backed_candidates::<T, _>(
 			parent_hash,
 			backed_candidates,
-				move |candidate_hash: CandidateHash| -> bool {
+			move |_candidate_index: usize, backed_candidate: &BackedCandidate<T::Hash>| -> bool {
-					<T>::DisputesHandler::concluded_invalid(current_session, candidate_hash)
+				<T>::DisputesHandler::concluded_invalid(current_session, backed_candidate.hash())
+				// `fn process_candidates` does the verification checks
 			},
 			&scheduled[..],
 		);
@@ -484,6 +520,7 @@ pub mod pallet {
 			backed_candidates,
 			scheduled,
 			<scheduler::Pallet<T>>::group_validators,
+			full_check,
 		)?;
 
 		// The number of disputes included in a block is
@@ -503,7 +540,6 @@ pub mod pallet {
 
 		Ok(Some(total_weight).into())
 	}
-	}
 }
 
 impl<T: Config> Pallet<T> {
@@ -548,7 +584,7 @@ impl<T: Config> Pallet<T> {
 
 		T::DisputesHandler::filter_multi_dispute_data(&mut disputes);
 
-		let (concluded_invalid_disputes, mut bitfields, scheduled) =
+		let (mut backed_candidates, mut bitfields) =
 			frame_support::storage::with_transaction(|| {
 				// we don't care about fresh or not disputes
 				// this writes them to storage, so let's query it via those means
@@ -563,9 +599,13 @@ impl<T: Config> Pallet<T> {
 						e
 					});
 
-				// current concluded invalid disputes, only including the current block's votes
+				// Contains the disputes that are concluded in the current session only,
-				// TODO why does this say "only including the current block's votes"? This can include
+				// since these are the only ones that are relevant for the occupied cores
-				// remote disputes, right?
+				// and lightens the load on `collect_disputed` significantly.
+				// Cores can't be occupied with candidates of the previous sessions, and only
+				// things with new votes can have just concluded. We only need to collect
+				// cores with disputes that conclude just now, because disputes that
+				// concluded longer ago have already had any corresponding cores cleaned up.
 				let current_concluded_invalid_disputes = disputes
 					.iter()
 					.filter(|dss| dss.session == current_session)
@@ -576,8 +616,8 @@ impl<T: Config> Pallet<T> {
 					.map(|(_session, candidate)| candidate)
 					.collect::<BTreeSet<CandidateHash>>();
 
-				// all concluded invalid disputes, that are relevant for the set of candidates
+				// All concluded invalid disputes, that are relevant for the set of candidates
-				// the inherent provided
+				// the inherent provided.
 				let concluded_invalid_disputes = backed_candidates
 					.iter()
 					.map(|backed_candidate| backed_candidate.hash())
@@ -604,13 +644,14 @@ impl<T: Config> Pallet<T> {
 
 				// The following 3 calls are equiv to a call to `process_bitfields`
 				// but we can retain access to `bitfields`.
-				let bitfields = sanitize_bitfields::<T, SKIP_SIG_VERIFY>(
+				let bitfields = sanitize_bitfields::<T>(
 					bitfields,
 					disputed_bitfield,
 					expected_bits,
 					parent_hash,
 					current_session,
 					&validator_public[..],
+					FullCheck::Skip,
 				);
 
 				let freed_concluded =
@@ -627,31 +668,45 @@ impl<T: Config> Pallet<T> {
 				let freed = collect_all_freed_cores::<T, _>(freed_concluded.iter().cloned());
 
 				<scheduler::Pallet<T>>::clear();
-				<scheduler::Pallet<T>>::schedule(freed, <frame_system::Pallet<T>>::block_number());
+				let now = <frame_system::Pallet<T>>::block_number();
+				<scheduler::Pallet<T>>::schedule(freed, now);
 
 				let scheduled = <scheduler::Pallet<T>>::scheduled();
 
-				frame_support::storage::TransactionOutcome::Rollback((
+				let relay_parent_number = now - One::one();
-					// concluded disputes for backed candidates in this block
-					concluded_invalid_disputes,
-					// filtered bitfields,
-					bitfields,
-					// updated schedule
-					scheduled,
-				))
-			});
 
-		let mut backed_candidates = sanitize_backed_candidates::<T, _>(
+				let check_ctx = CandidateCheckContext::<T>::new(now, relay_parent_number);
+				let backed_candidates = sanitize_backed_candidates::<T, _>(
 					parent_hash,
 					backed_candidates,
-			move |candidate_hash: CandidateHash| -> bool {
+					move |candidate_idx: usize,
-				concluded_invalid_disputes.contains(&candidate_hash)
+					      backed_candidate: &BackedCandidate<<T as frame_system::Config>::Hash>|
+					      -> bool {
+						// never include a concluded-invalid candidate
+						concluded_invalid_disputes.contains(&backed_candidate.hash()) ||
+							// Instead of checking the candidates with code upgrades twice
+							// move the checking up here and skip it in the training wheels fallback.
+							// That way we avoid possible duplicate checks while assuring all
+							// backed candidates fine to pass on.
+							check_ctx
+								.verify_backed_candidate(parent_hash, candidate_idx, backed_candidate)
+								.is_err()
 					},
 					&scheduled[..],
 				);
 
+				frame_support::storage::TransactionOutcome::Rollback((
+					// filtered backed candidates
+					backed_candidates,
+					// filtered bitfields
+					bitfields,
+				))
+			});
+
 		let entropy = compute_entropy::<T>(parent_hash);
 		let mut rng = rand_chacha::ChaChaRng::from_seed(entropy.into());
+
+		// Assure the maximum block weight is adhered.
 		let max_block_weight = <T as frame_system::Config>::BlockWeights::get().max_block;
 		let _consumed_weight = apply_weight_limit::<T>(
 			&mut backed_candidates,
@@ -714,15 +769,11 @@ fn random_sel<X, F: Fn(&X) -> Weight>(
 
 	let mut weight_acc = 0 as Weight;
 
-	while !preferred_indices.is_empty() {
+	preferred_indices.shuffle(rng);
-		// randomly pick an index from the preferred set
+	for preferred_idx in preferred_indices {
-		let pick = rng.gen_range(0..preferred_indices.len());
-		// remove the index from the available set of preferred indices
-		let preferred_idx = preferred_indices.swap_remove(pick);
-
 		// preferred indices originate from outside
 		if let Some(item) = selectables.get(preferred_idx) {
-			let updated = weight_acc + weight_fn(item);
+			let updated = weight_acc.saturating_add(weight_fn(item));
 			if updated > weight_limit {
 				continue
 			}
@@ -731,14 +782,10 @@ fn random_sel<X, F: Fn(&X) -> Weight>(
 		}
 	}
 
-	while !indices.is_empty() {
+	indices.shuffle(rng);
-		// randomly pick an index
+	for idx in indices {
-		let pick = rng.gen_range(0..indices.len());
-		// remove the index from the available set of indices
-		let idx = indices.swap_remove(pick);
-
 		let item = &selectables[idx];
-		let updated = weight_acc + weight_fn(item);
+		let updated = weight_acc.saturating_add(weight_fn(item));
 
 		if updated > weight_limit {
 			continue
@@ -754,14 +801,18 @@ fn random_sel<X, F: Fn(&X) -> Weight>(
 	(weight_acc, picked_indices)
 }
 
-/// Considers an upper threshold that the candidates must not exceed.
+/// Considers an upper threshold that the inherent data must not exceed.
 ///
-/// If there is sufficient space, all bitfields and candidates will be included.
+/// If there is sufficient space, all disputes, all bitfields and all candidates
+/// will be included.
 ///
-/// Otherwise tries to include all bitfields, and fills in the remaining weight with candidates.
+/// Otherwise tries to include all disputes, and then tries to fill the remaining space with bitfields and then candidates.
 ///
-/// If even the bitfields are too large to fit into the `max_weight` limit, bitfields are randomly
+/// The selection process is random. For candidates, there is an exception for code upgrades as they are preferred.
-/// picked and _no_ candidates will be included.
+/// And for disputes, local and older disputes are preferred (see `limit_disputes`).
+/// for backed candidates, since with a increasing number of parachains their chances of
+/// inclusion become slim. All backed candidates  are checked beforehands in `fn create_inherent_inner`
+/// which guarantees sanity.
 fn apply_weight_limit<T: Config + inclusion::Config>(
 	candidates: &mut Vec<BackedCandidate<<T>::Hash>>,
 	bitfields: &mut UncheckedSignedAvailabilityBitfields,
@@ -804,12 +855,7 @@ fn apply_weight_limit<T: Config + inclusion::Config>(
 				|c| backed_candidate_weight::<T>(c),
 				remaining_weight,
 			);
-		let mut idx = 0_usize;
+		candidates.indexed_retain(|idx, _backed_candidate| indices.binary_search(&idx).is_ok());
-		candidates.retain(|_backed_candidate| {
-			let exists = indices.binary_search(&idx).is_ok();
-			idx += 1;
-			exists
-		});
 		// pick all bitfields, and
 		// fill the remaining space with candidates
 		let total = acc_candidate_weight.saturating_add(total_bitfields_weight);
@@ -828,12 +874,7 @@ fn apply_weight_limit<T: Config + inclusion::Config>(
 		remaining_weight,
 	);
 
-	let mut idx = 0_usize;
+	bitfields.indexed_retain(|idx, _bitfield| indices.binary_search(&idx).is_ok());
-	bitfields.retain(|_bitfield| {
-		let exists = indices.binary_search(&idx).is_ok();
-		idx += 1;
-		exists
-	});
 
 	total
 }
@@ -854,15 +895,16 @@ fn apply_weight_limit<T: Config + inclusion::Config>(
 /// they were actually checked and filtered to allow using it in both
 /// cases, as `filtering` and `checking` stage.
 ///
-/// `CHECK_SIGS` determines if validator signatures are checked. If true, bitfields that have an
+/// `full_check` determines if validator signatures are checked. If `::Yes`,
-/// invalid signature will be filtered out.
+/// bitfields that have an invalid signature will be filtered out.
-pub(crate) fn sanitize_bitfields<T: crate::inclusion::Config, const CHECK_SIGS: bool>(
+pub(crate) fn sanitize_bitfields<T: crate::inclusion::Config>(
 	unchecked_bitfields: UncheckedSignedAvailabilityBitfields,
 	disputed_bitfield: DisputedBitfield,
 	expected_bits: usize,
 	parent_hash: T::Hash,
 	session_index: SessionIndex,
 	validators: &[ValidatorId],
+	full_check: FullCheck,
 ) -> UncheckedSignedAvailabilityBitfields {
 	let mut bitfields = Vec::with_capacity(unchecked_bitfields.len());
 
@@ -882,8 +924,8 @@ pub(crate) fn sanitize_bitfields<T: crate::inclusion::Config, const CHECK_SIGS:
 		if unchecked_bitfield.unchecked_payload().0.len() != expected_bits {
 			log::trace!(
 				target: LOG_TARGET,
-				"[CHECK_SIGS: {}] bad bitfield length: {} != {:?}",
+				"[{:?}] bad bitfield length: {} != {:?}",
-				CHECK_SIGS,
+				full_check,
 				unchecked_bitfield.unchecked_payload().0.len(),
 				expected_bits,
 			);
@@ -895,8 +937,8 @@ pub(crate) fn sanitize_bitfields<T: crate::inclusion::Config, const CHECK_SIGS:
 		{
 			log::trace!(
 				target: LOG_TARGET,
-				"[CHECK_SIGS: {}] bitfield contains disputed cores: {:?}",
+				"[{:?}] bitfield contains disputed cores: {:?}",
-				CHECK_SIGS,
+				full_check,
 				unchecked_bitfield.unchecked_payload().0.clone() & disputed_bitfield.0.clone()
 			);
 			continue
@@ -907,8 +949,8 @@ pub(crate) fn sanitize_bitfields<T: crate::inclusion::Config, const CHECK_SIGS:
 		if !last_index.map_or(true, |last_index: ValidatorIndex| last_index < validator_index) {
 			log::trace!(
 				target: LOG_TARGET,
-				"[CHECK_SIGS: {}] bitfield validator index is not greater than last: !({:?} < {})",
+				"[{:?}] bitfield validator index is not greater than last: !({:?} < {})",
-				CHECK_SIGS,
+				full_check,
 				last_index.as_ref().map(|x| x.0),
 				validator_index.0
 			);
@@ -918,8 +960,8 @@ pub(crate) fn sanitize_bitfields<T: crate::inclusion::Config, const CHECK_SIGS:
 		if unchecked_bitfield.unchecked_validator_index().0 as usize >= validators.len() {
 			log::trace!(
 				target: LOG_TARGET,
-				"[CHECK_SIGS: {}] bitfield validator index is out of bounds: {} >= {}",
+				"[{:?}] bitfield validator index is out of bounds: {} >= {}",
-				CHECK_SIGS,
+				full_check,
 				validator_index.0,
 				validators.len(),
 			);
@@ -928,7 +970,7 @@ pub(crate) fn sanitize_bitfields<T: crate::inclusion::Config, const CHECK_SIGS:
 
 		let validator_public = &validators[validator_index.0 as usize];
 
-		if CHECK_SIGS {
+		if let FullCheck::Yes = full_check {
 			if let Ok(signed_bitfield) =
 				unchecked_bitfield.try_into_checked(&signing_context, validator_public)
 			{
@@ -954,15 +996,18 @@ pub(crate) fn sanitize_bitfields<T: crate::inclusion::Config, const CHECK_SIGS:
 ///
 /// `candidate_has_concluded_invalid_dispute` must return `true` if the candidate
 /// is disputed, false otherwise
-fn sanitize_backed_candidates<T: crate::inclusion::Config, F: Fn(CandidateHash) -> bool>(
+fn sanitize_backed_candidates<
+	T: crate::inclusion::Config,
+	F: FnMut(usize, &BackedCandidate<T::Hash>) -> bool,
+>(
 	relay_parent: T::Hash,
 	mut backed_candidates: Vec<BackedCandidate<T::Hash>>,
-	candidate_has_concluded_invalid_dispute: F,
+	mut candidate_has_concluded_invalid_dispute_or_is_invalid: F,
 	scheduled: &[CoreAssignment],
 ) -> Vec<BackedCandidate<T::Hash>> {
 	// Remove any candidates that were concluded invalid.
-	backed_candidates.retain(|backed_candidate| {
+	backed_candidates.indexed_retain(move |idx, backed_candidate| {
-		!candidate_has_concluded_invalid_dispute(backed_candidate.candidate.hash())
+		!candidate_has_concluded_invalid_dispute_or_is_invalid(idx, backed_candidate)
 	});
 
 	// Assure the backed candidate's `ParaId`'s core is free.
@@ -1657,7 +1702,7 @@ mod tests {
 
 		#[test]
 		// Ensure that when a block is over weight due to disputes and bitfields, we abort
-		fn limit_candidates_over_weight() {
+		fn limit_candidates_over_weight_1() {
 			new_test_ext(MockGenesisConfig::default()).execute_with(|| {
 				// Create the inherent data for this block
 				let mut dispute_statements = BTreeMap::new();
@@ -1728,7 +1773,7 @@ mod tests {
 
 		#[test]
 		// Ensure that when a block is over weight due to disputes and bitfields, we abort
-		fn limit_candidates_over_weight_overweight() {
+		fn limit_candidates_over_weight_0() {
 			new_test_ext(MockGenesisConfig::default()).execute_with(|| {
 				// Create the inherent data for this block
 				let mut dispute_statements = BTreeMap::new();
@@ -1871,24 +1916,26 @@ mod tests {
 
 			{
 				assert_eq!(
-					sanitize_bitfields::<Test, VERIFY_SIGS>(
+					sanitize_bitfields::<Test>(
 						unchecked_bitfields.clone(),
 						disputed_bitfield.clone(),
 						expected_bits,
 						parent_hash,
 						session_index,
-						&validator_public[..]
+						&validator_public[..],
+						FullCheck::Skip,
 					),
 					unchecked_bitfields.clone()
 				);
 				assert_eq!(
-					sanitize_bitfields::<Test, SKIP_SIG_VERIFY>(
+					sanitize_bitfields::<Test>(
 						unchecked_bitfields.clone(),
 						disputed_bitfield.clone(),
 						expected_bits,
 						parent_hash,
 						session_index,
-						&validator_public[..]
+						&validator_public[..],
+						FullCheck::Yes
 					),
 					unchecked_bitfields.clone()
 				);
@@ -1902,25 +1949,27 @@ mod tests {
 				disputed_bitfield.0.set(0, true);
 
 				assert_eq!(
-					sanitize_bitfields::<Test, VERIFY_SIGS>(
+					sanitize_bitfields::<Test>(
 						unchecked_bitfields.clone(),
 						disputed_bitfield.clone(),
 						expected_bits,
 						parent_hash,
 						session_index,
-						&validator_public[..]
+						&validator_public[..],
+						FullCheck::Yes
 					)
 					.len(),
 					1
 				);
 				assert_eq!(
-					sanitize_bitfields::<Test, SKIP_SIG_VERIFY>(
+					sanitize_bitfields::<Test>(
 						unchecked_bitfields.clone(),
 						disputed_bitfield.clone(),
 						expected_bits,
 						parent_hash,
 						session_index,
-						&validator_public[..]
+						&validator_public[..],
+						FullCheck::Skip
 					)
 					.len(),
 					1
@@ -1929,22 +1978,24 @@ mod tests {
 
 			// bitfield size mismatch
 			{
-				assert!(sanitize_bitfields::<Test, VERIFY_SIGS>(
+				assert!(sanitize_bitfields::<Test>(
 					unchecked_bitfields.clone(),
 					disputed_bitfield.clone(),
 					expected_bits + 1,
 					parent_hash,
 					session_index,
-					&validator_public[..]
+					&validator_public[..],
+					FullCheck::Yes
 				)
 				.is_empty());
-				assert!(sanitize_bitfields::<Test, SKIP_SIG_VERIFY>(
+				assert!(sanitize_bitfields::<Test>(
 					unchecked_bitfields.clone(),
 					disputed_bitfield.clone(),
 					expected_bits + 1,
 					parent_hash,
 					session_index,
-					&validator_public[..]
+					&validator_public[..],
+					FullCheck::Skip
 				)
 				.is_empty());
 			}
@@ -1953,24 +2004,26 @@ mod tests {
 			{
 				let shortened = validator_public.len() - 2;
 				assert_eq!(
-					&sanitize_bitfields::<Test, VERIFY_SIGS>(
+					&sanitize_bitfields::<Test>(
 						unchecked_bitfields.clone(),
 						disputed_bitfield.clone(),
 						expected_bits,
 						parent_hash,
 						session_index,
-						&validator_public[..shortened]
+						&validator_public[..shortened],
+						FullCheck::Yes,
 					)[..],
 					&unchecked_bitfields[..shortened]
 				);
 				assert_eq!(
-					&sanitize_bitfields::<Test, SKIP_SIG_VERIFY>(
+					&sanitize_bitfields::<Test>(
 						unchecked_bitfields.clone(),
 						disputed_bitfield.clone(),
 						expected_bits,
 						parent_hash,
 						session_index,
-						&validator_public[..shortened]
+						&validator_public[..shortened],
+						FullCheck::Skip,
 					)[..],
 					&unchecked_bitfields[..shortened]
 				);
@@ -1982,24 +2035,26 @@ mod tests {
 				let x = unchecked_bitfields.swap_remove(0);
 				unchecked_bitfields.push(x);
 				assert_eq!(
-					&sanitize_bitfields::<Test, VERIFY_SIGS>(
+					&sanitize_bitfields::<Test>(
 						unchecked_bitfields.clone(),
 						disputed_bitfield.clone(),
 						expected_bits,
 						parent_hash,
 						session_index,
-						&validator_public[..]
+						&validator_public[..],
+						FullCheck::Yes
 					)[..],
 					&unchecked_bitfields[..(unchecked_bitfields.len() - 2)]
 				);
 				assert_eq!(
-					&sanitize_bitfields::<Test, SKIP_SIG_VERIFY>(
+					&sanitize_bitfields::<Test>(
 						unchecked_bitfields.clone(),
 						disputed_bitfield.clone(),
 						expected_bits,
 						parent_hash,
 						session_index,
-						&validator_public[..]
+						&validator_public[..],
+						FullCheck::Skip
 					)[..],
 					&unchecked_bitfields[..(unchecked_bitfields.len() - 2)]
 				);
@@ -2017,24 +2072,26 @@ mod tests {
 					.and_then(|u| Some(u.set_signature(ValidatorSignature::default())))
 					.expect("we are accessing a valid index");
 				assert_eq!(
-					&sanitize_bitfields::<Test, VERIFY_SIGS>(
+					&sanitize_bitfields::<Test>(
 						unchecked_bitfields.clone(),
 						disputed_bitfield.clone(),
 						expected_bits,
 						parent_hash,
 						session_index,
-						&validator_public[..]
+						&validator_public[..],
+						FullCheck::Yes
 					)[..],
 					&unchecked_bitfields[..last_bit_idx]
 				);
 				assert_eq!(
-					&sanitize_bitfields::<Test, SKIP_SIG_VERIFY>(
+					&sanitize_bitfields::<Test>(
 						unchecked_bitfields.clone(),
 						disputed_bitfield.clone(),
 						expected_bits,
 						parent_hash,
 						session_index,
-						&validator_public[..]
+						&validator_public[..],
+						FullCheck::Skip
 					)[..],
 					&unchecked_bitfields[..]
 				);
@@ -2068,7 +2125,8 @@ mod tests {
 				.unwrap();
 			}
 
-			let has_concluded_invalid = |_candidate: CandidateHash| -> bool { false };
+			let has_concluded_invalid =
+				|_idx: usize, _backed_candidate: &BackedCandidate| -> bool { false };
 
 			let scheduled = (0_usize..2)
 				.into_iter()
@@ -2168,7 +2226,8 @@ mod tests {
 					}
 					set
 				};
-				let has_concluded_invalid = |candidate: CandidateHash| set.contains(&candidate);
+				let has_concluded_invalid =
+					|_idx: usize, candidate: &BackedCandidate| set.contains(&candidate.hash());
 				assert_eq!(
 					sanitize_backed_candidates::<Test, _>(
 						relay_parent,