provisioner: allow multiple cores assigned to the same para (#3233)

https://github.com/paritytech/polkadot-sdk/issues/3130

builds on top of https://github.com/paritytech/polkadot-sdk/pull/3160

Processes the availability cores and builds a record of how many
candidates it should request from prospective-parachains and their
predecessors.
Tries to supply as many candidates as the runtime can back. Note that
the runtime changes to back multiple candidates per para are not yet
done, but this paves the way for it.

The following backing/inclusion policy is assumed:
1. the runtime will never back candidates of the same para which don't
form a chain with the already backed candidates. Even if the others are
still pending availability. We're optimistic that they won't time out
and we don't want to back parachain forks (as the complexity would be
huge).
2. if a candidate is timed out of the core before being included, all of
its successors occupying a core will be evicted.
3. only the candidates which are made available and form a chain
starting from the on-chain para head may be included/enacted and cleared
from the cores. In other words, if para head is at A and the cores are
occupied by B->C->D, and B and D are made available, only B will be
included and its core cleared. C and D will remain on the cores awaiting
for C to be made available or timed out. As point (2) above already
says, if C is timed out, D will also be dropped.
4. The runtime will deduplicate candidates which form a cycle. For
example if the provisioner supplies candidates A->B->A, the runtime will
only back A (as the state output will be the same)

Note that if a candidate is timed out, we don't guarantee that in the
next relay chain block the block author will be able to fill all of the
timed out cores of the para. That increases complexity by a lot.
Instead, the provisioner will supply N candidates where N is the number
of candidates timed out, but doesn't include their successors which will
be also deleted by the runtime. This'll be backfilled in the next relay
chain block.

Adjacent changes:
- Also fixes: https://github.com/paritytech/polkadot-sdk/issues/3141
- For non prospective-parachains, don't supply multiple candidates per
para (we can't have elastic scaling without prospective parachains
enabled). paras_inherent should already sanitise this input but it's
more efficient this way.

Note: all of these changes are backwards-compatible with the
non-elastic-scaling scenario (one core per para).

polkadot/node/core/provisioner/src/error.rs

@@ -44,14 +44,17 @@ pub enum Error {
 	#[error("failed to get block number")]
 	CanceledBlockNumber(#[source] oneshot::Canceled),
 
+	#[error("failed to get session index")]
+	CanceledSessionIndex(#[source] oneshot::Canceled),
+
 	#[error("failed to get backed candidates")]
 	CanceledBackedCandidates(#[source] oneshot::Canceled),
 
 	#[error("failed to get votes on dispute")]
 	CanceledCandidateVotes(#[source] oneshot::Canceled),
 
-	#[error("failed to get backable candidate from prospective parachains")]
-	CanceledBackableCandidate(#[source] oneshot::Canceled),
+	#[error("failed to get backable candidates from prospective parachains")]
+	CanceledBackableCandidates(#[source] oneshot::Canceled),
 
 	#[error(transparent)]
 	ChainApi(#[from] ChainApiError),
@@ -71,11 +74,6 @@ pub enum Error {
 	#[error("failed to send return message with Inherents")]
 	InherentDataReturnChannel,
 
-	#[error(
-		"backed candidate does not correspond to selected candidate; check logic in provisioner"
-	)]
-	BackedCandidateOrderingProblem,
-
 	#[fatal]
 	#[error("Failed to spawn background task")]
 	FailedToSpawnBackgroundTask,

polkadot/node/core/provisioner/src/lib.rs

@@ -24,26 +24,29 @@ use futures::{
 	channel::oneshot, future::BoxFuture, prelude::*, stream::FuturesUnordered, FutureExt,
 };
 use futures_timer::Delay;
+use schnellru::{ByLength, LruMap};
 
 use polkadot_node_subsystem::{
 	jaeger,
 	messages::{
-		CandidateBackingMessage, ChainApiMessage, ProspectiveParachainsMessage, ProvisionableData,
-		ProvisionerInherentData, ProvisionerMessage, RuntimeApiRequest,
+		Ancestors, CandidateBackingMessage, ChainApiMessage, ProspectiveParachainsMessage,
+		ProvisionableData, ProvisionerInherentData, ProvisionerMessage, RuntimeApiRequest,
 	},
 	overseer, ActivatedLeaf, ActiveLeavesUpdate, FromOrchestra, OverseerSignal, PerLeafSpan,
 	SpawnedSubsystem, SubsystemError,
 };
 use polkadot_node_subsystem_util::{
 	has_required_runtime, request_availability_cores, request_persisted_validation_data,
-	runtime::{prospective_parachains_mode, ProspectiveParachainsMode},
+	request_session_index_for_child,
+	runtime::{prospective_parachains_mode, request_node_features, ProspectiveParachainsMode},
 	TimeoutExt,
 };
 use polkadot_primitives::{
-	BackedCandidate, BlockNumber, CandidateHash, CandidateReceipt, CoreState, Hash, Id as ParaId,
-	OccupiedCoreAssumption, SignedAvailabilityBitfield, ValidatorIndex,
+	vstaging::{node_features::FeatureIndex, NodeFeatures},
+	BackedCandidate, BlockNumber, CandidateHash, CandidateReceipt, CoreIndex, CoreState, Hash,
+	Id as ParaId, OccupiedCoreAssumption, SessionIndex, SignedAvailabilityBitfield, ValidatorIndex,
 };
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, HashSet};
 
 mod disputes;
 mod error;
@@ -77,11 +80,18 @@ impl ProvisionerSubsystem {
 	}
 }
 
+/// Per-session info we need for the provisioner subsystem.
+pub struct PerSession {
+	prospective_parachains_mode: ProspectiveParachainsMode,
+	elastic_scaling_mvp: bool,
+}
+
 /// A per-relay-parent state for the provisioning subsystem.
 pub struct PerRelayParent {
 	leaf: ActivatedLeaf,
 	backed_candidates: Vec<CandidateReceipt>,
 	prospective_parachains_mode: ProspectiveParachainsMode,
+	elastic_scaling_mvp: bool,
 	signed_bitfields: Vec<SignedAvailabilityBitfield>,
 	is_inherent_ready: bool,
 	awaiting_inherent: Vec<oneshot::Sender<ProvisionerInherentData>>,
@@ -89,13 +99,14 @@ pub struct PerRelayParent {
 }
 
 impl PerRelayParent {
-	fn new(leaf: ActivatedLeaf, prospective_parachains_mode: ProspectiveParachainsMode) -> Self {
+	fn new(leaf: ActivatedLeaf, per_session: &PerSession) -> Self {
 		let span = PerLeafSpan::new(leaf.span.clone(), "provisioner");
 
 		Self {
 			leaf,
 			backed_candidates: Vec::new(),
-			prospective_parachains_mode,
+			prospective_parachains_mode: per_session.prospective_parachains_mode,
+			elastic_scaling_mvp: per_session.elastic_scaling_mvp,
 			signed_bitfields: Vec::new(),
 			is_inherent_ready: false,
 			awaiting_inherent: Vec::new(),
@@ -124,10 +135,17 @@ impl<Context> ProvisionerSubsystem {
 async fn run<Context>(mut ctx: Context, metrics: Metrics) -> FatalResult<()> {
 	let mut inherent_delays = InherentDelays::new();
 	let mut per_relay_parent = HashMap::new();
+	let mut per_session = LruMap::new(ByLength::new(2));
 
 	loop {
-		let result =
-			run_iteration(&mut ctx, &mut per_relay_parent, &mut inherent_delays, &metrics).await;
+		let result = run_iteration(
+			&mut ctx,
+			&mut per_relay_parent,
+			&mut per_session,
+			&mut inherent_delays,
+			&metrics,
+		)
+		.await;
 
 		match result {
 			Ok(()) => break,
@@ -142,6 +160,7 @@ async fn run<Context>(mut ctx: Context, metrics: Metrics) -> FatalResult<()> {
 async fn run_iteration<Context>(
 	ctx: &mut Context,
 	per_relay_parent: &mut HashMap<Hash, PerRelayParent>,
+	per_session: &mut LruMap<SessionIndex, PerSession>,
 	inherent_delays: &mut InherentDelays,
 	metrics: &Metrics,
 ) -> Result<(), Error> {
@@ -151,7 +170,7 @@ async fn run_iteration<Context>(
 				// Map the error to ensure that the subsystem exits when the overseer is gone.
 				match from_overseer.map_err(Error::OverseerExited)? {
 					FromOrchestra::Signal(OverseerSignal::ActiveLeaves(update)) =>
-						handle_active_leaves_update(ctx.sender(), update, per_relay_parent, inherent_delays).await?,
+						handle_active_leaves_update(ctx.sender(), update, per_relay_parent, per_session, inherent_delays).await?,
 					FromOrchestra::Signal(OverseerSignal::BlockFinalized(..)) => {},
 					FromOrchestra::Signal(OverseerSignal::Conclude) => return Ok(()),
 					FromOrchestra::Communication { msg } => {
@@ -183,6 +202,7 @@ async fn handle_active_leaves_update(
 	sender: &mut impl overseer::ProvisionerSenderTrait,
 	update: ActiveLeavesUpdate,
 	per_relay_parent: &mut HashMap<Hash, PerRelayParent>,
+	per_session: &mut LruMap<SessionIndex, PerSession>,
 	inherent_delays: &mut InherentDelays,
 ) -> Result<(), Error> {
 	gum::trace!(target: LOG_TARGET, "Handle ActiveLeavesUpdate");
@@ -191,10 +211,31 @@ async fn handle_active_leaves_update(
 	}
 
 	if let Some(leaf) = update.activated {
+		let session_index = request_session_index_for_child(leaf.hash, sender)
+			.await
+			.await
+			.map_err(Error::CanceledSessionIndex)??;
+		if per_session.get(&session_index).is_none() {
+			let prospective_parachains_mode =
+				prospective_parachains_mode(sender, leaf.hash).await?;
+			let elastic_scaling_mvp = request_node_features(leaf.hash, session_index, sender)
+				.await?
+				.unwrap_or(NodeFeatures::EMPTY)
+				.get(FeatureIndex::ElasticScalingMVP as usize)
+				.map(|b| *b)
+				.unwrap_or(false);
+
+			per_session.insert(
+				session_index,
+				PerSession { prospective_parachains_mode, elastic_scaling_mvp },
+			);
+		}
+
+		let session_info = per_session.get(&session_index).expect("Just inserted");
+
 		gum::trace!(target: LOG_TARGET, leaf_hash=?leaf.hash, "Adding delay");
-		let prospective_parachains_mode = prospective_parachains_mode(sender, leaf.hash).await?;
 		let delay_fut = Delay::new(PRE_PROPOSE_TIMEOUT).map(move |_| leaf.hash).boxed();
-		per_relay_parent.insert(leaf.hash, PerRelayParent::new(leaf, prospective_parachains_mode));
+		per_relay_parent.insert(leaf.hash, PerRelayParent::new(leaf, session_info));
 		inherent_delays.push(delay_fut);
 	}
 
@@ -253,6 +294,7 @@ async fn send_inherent_data_bg<Context>(
 	let signed_bitfields = per_relay_parent.signed_bitfields.clone();
 	let backed_candidates = per_relay_parent.backed_candidates.clone();
 	let mode = per_relay_parent.prospective_parachains_mode;
+	let elastic_scaling_mvp = per_relay_parent.elastic_scaling_mvp;
 	let span = per_relay_parent.span.child("req-inherent-data");
 
 	let mut sender = ctx.sender().clone();
@@ -272,6 +314,7 @@ async fn send_inherent_data_bg<Context>(
 			&signed_bitfields,
 			&backed_candidates,
 			mode,
+			elastic_scaling_mvp,
 			return_senders,
 			&mut sender,
 			&metrics,
@@ -383,6 +426,7 @@ async fn send_inherent_data(
 	bitfields: &[SignedAvailabilityBitfield],
 	candidates: &[CandidateReceipt],
 	prospective_parachains_mode: ProspectiveParachainsMode,
+	elastic_scaling_mvp: bool,
 	return_senders: Vec<oneshot::Sender<ProvisionerInherentData>>,
 	from_job: &mut impl overseer::ProvisionerSenderTrait,
 	metrics: &Metrics,
@@ -434,6 +478,7 @@ async fn send_inherent_data(
 		&bitfields,
 		candidates,
 		prospective_parachains_mode,
+		elastic_scaling_mvp,
 		leaf.hash,
 		from_job,
 	)
@@ -558,6 +603,8 @@ async fn select_candidate_hashes_from_tracked(
 
 	let mut selected_candidates =
 		Vec::with_capacity(candidates.len().min(availability_cores.len()));
+	let mut selected_parachains =
+		HashSet::with_capacity(candidates.len().min(availability_cores.len()));
 
 	gum::debug!(
 		target: LOG_TARGET,
@@ -591,6 +638,12 @@ async fn select_candidate_hashes_from_tracked(
 			CoreState::Free => continue,
 		};
 
+		if selected_parachains.contains(&scheduled_core.para_id) {
+			// We already picked a candidate for this parachain. Elastic scaling only works with
+			// prospective parachains mode.
+			continue
+		}
+
 		let validation_data = match request_persisted_validation_data(
 			relay_parent,
 			scheduled_core.para_id,
@@ -624,6 +677,7 @@ async fn select_candidate_hashes_from_tracked(
 				"Selected candidate receipt",
 			);
 
+			selected_parachains.insert(candidate.descriptor.para_id);
 			selected_candidates.push((candidate_hash, candidate.descriptor.relay_parent));
 		}
 	}
@@ -637,70 +691,93 @@ async fn select_candidate_hashes_from_tracked(
 /// Should be called when prospective parachains are enabled.
 async fn request_backable_candidates(
 	availability_cores: &[CoreState],
+	elastic_scaling_mvp: bool,
 	bitfields: &[SignedAvailabilityBitfield],
 	relay_parent: Hash,
 	sender: &mut impl overseer::ProvisionerSenderTrait,
 ) -> Result<Vec<(CandidateHash, Hash)>, Error> {
 	let block_number = get_block_number_under_construction(relay_parent, sender).await?;
 
-	let mut selected_candidates = Vec::with_capacity(availability_cores.len());
+	// Record how many cores are scheduled for each paraid. Use a BTreeMap because
+	// we'll need to iterate through them.
+	let mut scheduled_cores: BTreeMap<ParaId, usize> = BTreeMap::new();
+	// The on-chain ancestors of a para present in availability-cores.
+	let mut ancestors: HashMap<ParaId, Ancestors> =
+		HashMap::with_capacity(availability_cores.len());
 
 	for (core_idx, core) in availability_cores.iter().enumerate() {
-		let (para_id, required_path) = match core {
+		let core_idx = CoreIndex(core_idx as u32);
+		match core {
 			CoreState::Scheduled(scheduled_core) => {
-				// The core is free, pick the first eligible candidate from
-				// the fragment tree.
-				(scheduled_core.para_id, Vec::new())
+				*scheduled_cores.entry(scheduled_core.para_id).or_insert(0) += 1;
 			},
 			CoreState::Occupied(occupied_core) => {
-				if bitfields_indicate_availability(core_idx, bitfields, &occupied_core.availability)
-				{
+				let is_available = bitfields_indicate_availability(
+					core_idx.0 as usize,
+					bitfields,
+					&occupied_core.availability,
+				);
+
+				if is_available {
+					ancestors
+						.entry(occupied_core.para_id())
+						.or_default()
+						.insert(occupied_core.candidate_hash);
+
 					if let Some(ref scheduled_core) = occupied_core.next_up_on_available {
-						// The candidate occupying the core is available, choose its
-						// child in the fragment tree.
-						//
-						// TODO: doesn't work for on-demand parachains. We lean hard on the
-						// assumption that cores are fixed to specific parachains within a session.
-						// https://github.com/paritytech/polkadot/issues/5492
-						(scheduled_core.para_id, vec![occupied_core.candidate_hash])
-					} else {
-						continue
-					}
-				} else {
-					if occupied_core.time_out_at != block_number {
-						continue
+						// Request a new backable candidate for the newly scheduled para id.
+						*scheduled_cores.entry(scheduled_core.para_id).or_insert(0) += 1;
 					}
+				} else if occupied_core.time_out_at <= block_number {
+					// Timed out before being available.
+
 					if let Some(ref scheduled_core) = occupied_core.next_up_on_time_out {
 						// Candidate's availability timed out, practically same as scheduled.
-						(scheduled_core.para_id, Vec::new())
-					} else {
-						continue
+						*scheduled_cores.entry(scheduled_core.para_id).or_insert(0) += 1;
 					}
+				} else {
+					// Not timed out and not available.
+					ancestors
+						.entry(occupied_core.para_id())
+						.or_default()
+						.insert(occupied_core.candidate_hash);
 				}
 			},
 			CoreState::Free => continue,
 		};
+	}
 
-		// We should be calling this once per para rather than per core.
-		// TODO: Will be fixed in https://github.com/paritytech/polkadot-sdk/pull/3233.
-		// For now, at least make sure we don't supply the same candidate multiple times in case a
-		// para has multiple cores scheduled.
-		let response = get_backable_candidate(relay_parent, para_id, required_path, sender).await?;
-		match response {
-			Some((hash, relay_parent)) => {
-				if !selected_candidates.iter().any(|bc| &(hash, relay_parent) == bc) {
-					selected_candidates.push((hash, relay_parent))
-				}
-			},
-			None => {
-				gum::debug!(
-					target: LOG_TARGET,
-					leaf_hash = ?relay_parent,
-					core = core_idx,
-					"No backable candidate returned by prospective parachains",
-				);
-			},
+	let mut selected_candidates: Vec<(CandidateHash, Hash)> =
+		Vec::with_capacity(availability_cores.len());
+
+	for (para_id, core_count) in scheduled_cores {
+		let para_ancestors = ancestors.remove(&para_id).unwrap_or_default();
+
+		// If elastic scaling MVP is disabled, only allow one candidate per parachain.
+		if !elastic_scaling_mvp && core_count > 1 {
+			continue
 		}
+
+		let response = get_backable_candidates(
+			relay_parent,
+			para_id,
+			para_ancestors,
+			core_count as u32,
+			sender,
+		)
+		.await?;
+
+		if response.is_empty() {
+			gum::debug!(
+				target: LOG_TARGET,
+				leaf_hash = ?relay_parent,
+				?para_id,
+				"No backable candidate returned by prospective parachains",
+			);
+			continue
+		}
+
+		selected_candidates.extend(response.into_iter().take(core_count));
 	}
 
 	Ok(selected_candidates)
@@ -713,6 +790,7 @@ async fn select_candidates(
 	bitfields: &[SignedAvailabilityBitfield],
 	candidates: &[CandidateReceipt],
 	prospective_parachains_mode: ProspectiveParachainsMode,
+	elastic_scaling_mvp: bool,
 	relay_parent: Hash,
 	sender: &mut impl overseer::ProvisionerSenderTrait,
 ) -> Result<Vec<BackedCandidate>, Error> {
@@ -722,7 +800,14 @@ async fn select_candidates(
 
 	let selected_candidates = match prospective_parachains_mode {
 		ProspectiveParachainsMode::Enabled { .. } =>
-			request_backable_candidates(availability_cores, bitfields, relay_parent, sender).await?,
+			request_backable_candidates(
+				availability_cores,
+				elastic_scaling_mvp,
+				bitfields,
+				relay_parent,
+				sender,
+			)
+			.await?,
 		ProspectiveParachainsMode::Disabled =>
 			select_candidate_hashes_from_tracked(
 				availability_cores,
@@ -745,24 +830,6 @@ async fn select_candidates(
 	gum::trace!(target: LOG_TARGET, leaf_hash=?relay_parent,
 				"Got {} backed candidates", candidates.len());
 
-	// `selected_candidates` is generated in ascending order by core index, and
-	// `GetBackedCandidates` _should_ preserve that property, but let's just make sure.
-	//
-	// We can't easily map from `BackedCandidate` to `core_idx`, but we know that every selected
-	// candidate maps to either 0 or 1 backed candidate, and the hashes correspond. Therefore, by
-	// checking them in order, we can ensure that the backed candidates are also in order.
-	let mut backed_idx = 0;
-	for selected in selected_candidates {
-		if selected.0 ==
-			candidates.get(backed_idx).ok_or(Error::BackedCandidateOrderingProblem)?.hash()
-		{
-			backed_idx += 1;
-		}
-	}
-	if candidates.len() != backed_idx {
-		Err(Error::BackedCandidateOrderingProblem)?;
-	}
-
 	// keep only one candidate with validation code.
 	let mut with_validation_code = false;
 	candidates.retain(|c| {
@@ -804,28 +871,27 @@ async fn get_block_number_under_construction(
 	}
 }
 
-/// Requests backable candidate from Prospective Parachains based on
-/// the given path in the fragment tree.
-async fn get_backable_candidate(
+/// Requests backable candidates from Prospective Parachains based on
+/// the given ancestors in the fragment tree. The ancestors may not be ordered.
+async fn get_backable_candidates(
 	relay_parent: Hash,
 	para_id: ParaId,
-	required_path: Vec<CandidateHash>,
+	ancestors: Ancestors,
+	count: u32,
 	sender: &mut impl overseer::ProvisionerSenderTrait,
-) -> Result<Option<(CandidateHash, Hash)>, Error> {
+) -> Result<Vec<(CandidateHash, Hash)>, Error> {
 	let (tx, rx) = oneshot::channel();
 	sender
 		.send_message(ProspectiveParachainsMessage::GetBackableCandidates(
 			relay_parent,
 			para_id,
-			1, // core count hardcoded to 1, until elastic scaling is implemented and enabled.
-			required_path,
+			count,
+			ancestors,
 			tx,
 		))
 		.await;
 
-	rx.await
-		.map_err(Error::CanceledBackableCandidate)
-		.map(|res| res.get(0).copied())
+	rx.await.map_err(Error::CanceledBackableCandidates)
 }
 
 /// The availability bitfield for a given core is the transpose

polkadot/node/core/provisioner/src/tests.rs

@@ -22,6 +22,9 @@ use polkadot_primitives::{OccupiedCore, ScheduledCore};
 const MOCK_GROUP_SIZE: usize = 5;
 
 pub fn occupied_core(para_id: u32) -> CoreState {
+	let mut candidate_descriptor = dummy_candidate_descriptor(dummy_hash());
+	candidate_descriptor.para_id = para_id.into();
+
 	CoreState::Occupied(OccupiedCore {
 		group_responsible: para_id.into(),
 		next_up_on_available: None,
@@ -29,7 +32,7 @@ pub fn occupied_core(para_id: u32) -> CoreState {
 		time_out_at: 200_u32,
 		next_up_on_time_out: None,
 		availability: bitvec![u8, bitvec::order::Lsb0; 0; 32],
-		candidate_descriptor: dummy_candidate_descriptor(dummy_hash()),
+		candidate_descriptor,
 		candidate_hash: Default::default(),
 	})
 }
@@ -254,10 +257,56 @@ mod select_candidates {
 	use polkadot_primitives::{
 		BlockNumber, CandidateCommitments, CommittedCandidateReceipt, PersistedValidationData,
 	};
+	use rstest::rstest;
 
 	const BLOCK_UNDER_PRODUCTION: BlockNumber = 128;
 
-	// For test purposes, we always return this set of availability cores:
+	fn dummy_candidate_template() -> CandidateReceipt {
+		let empty_hash = PersistedValidationData::<Hash, BlockNumber>::default().hash();
+
+		let mut descriptor_template = dummy_candidate_descriptor(dummy_hash());
+		descriptor_template.persisted_validation_data_hash = empty_hash;
+		CandidateReceipt {
+			descriptor: descriptor_template,
+			commitments_hash: CandidateCommitments::default().hash(),
+		}
+	}
+
+	fn make_candidates(
+		core_count: usize,
+		expected_backed_indices: Vec<usize>,
+	) -> (Vec<CandidateHash>, Vec<BackedCandidate>) {
+		let candidate_template = dummy_candidate_template();
+		let candidates: Vec<_> = std::iter::repeat(candidate_template)
+			.take(core_count)
+			.enumerate()
+			.map(|(idx, mut candidate)| {
+				candidate.descriptor.para_id = idx.into();
+				candidate
+			})
+			.collect();
+
+		let expected_backed = expected_backed_indices
+			.iter()
+			.map(|&idx| candidates[idx].clone())
+			.map(|c| {
+				BackedCandidate::new(
+					CommittedCandidateReceipt {
+						descriptor: c.descriptor.clone(),
+						commitments: Default::default(),
+					},
+					Vec::new(),
+					default_bitvec(MOCK_GROUP_SIZE),
+					None,
+				)
+			})
+			.collect();
+		let candidate_hashes = candidates.into_iter().map(|c| c.hash()).collect();
+
+		(candidate_hashes, expected_backed)
+	}
+
+	// For testing only one core assigned to a parachain, we return this set of availability cores:
 	//
 	//   [
 	//      0: Free,
@@ -273,7 +322,7 @@ mod select_candidates {
 	//     10: Occupied(both next_up set, not available, timeout),
 	//     11: Occupied(next_up_on_available and available, but different successor para_id)
 	//   ]
-	fn mock_availability_cores() -> Vec<CoreState> {
+	fn mock_availability_cores_one_per_para() -> Vec<CoreState> {
 		use std::ops::Not;
 		use CoreState::{Free, Scheduled};
 
@@ -292,6 +341,7 @@ mod select_candidates {
 			build_occupied_core(4, |core| {
 				core.next_up_on_available = Some(scheduled_core(4));
 				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(41));
 			}),
 			// 5: Occupied(next_up_on_time_out set but not timeout),
 			build_occupied_core(5, |core| {
@@ -307,12 +357,14 @@ mod select_candidates {
 			build_occupied_core(7, |core| {
 				core.next_up_on_time_out = Some(scheduled_core(7));
 				core.time_out_at = BLOCK_UNDER_PRODUCTION;
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(71));
 			}),
 			// 8: Occupied(both next_up set, available),
 			build_occupied_core(8, |core| {
 				core.next_up_on_available = Some(scheduled_core(8));
 				core.next_up_on_time_out = Some(scheduled_core(8));
 				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(81));
 			}),
 			// 9: Occupied(both next_up set, not available, no timeout),
 			build_occupied_core(9, |core| {
@@ -324,6 +376,7 @@ mod select_candidates {
 				core.next_up_on_available = Some(scheduled_core(10));
 				core.next_up_on_time_out = Some(scheduled_core(10));
 				core.time_out_at = BLOCK_UNDER_PRODUCTION;
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(101));
 			}),
 			// 11: Occupied(next_up_on_available and available, but different successor para_id)
 			build_occupied_core(11, |core| {
@@ -333,20 +386,189 @@ mod select_candidates {
 		]
 	}
 
+	// For test purposes with multiple possible cores assigned to a para, we always return this set
+	// of availability cores:
+	fn mock_availability_cores_multiple_per_para() -> Vec<CoreState> {
+		use std::ops::Not;
+		use CoreState::{Free, Scheduled};
+
+		vec![
+			// 0: Free,
+			Free,
+			// 1: Scheduled(default),
+			Scheduled(scheduled_core(1)),
+			// 2: Occupied(no next_up set),
+			occupied_core(2),
+			// 3: Occupied(next_up_on_available set but not available),
+			build_occupied_core(3, |core| {
+				core.next_up_on_available = Some(scheduled_core(3));
+			}),
+			// 4: Occupied(next_up_on_available set and available),
+			build_occupied_core(4, |core| {
+				core.next_up_on_available = Some(scheduled_core(4));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(41));
+			}),
+			// 5: Occupied(next_up_on_time_out set but not timeout),
+			build_occupied_core(5, |core| {
+				core.next_up_on_time_out = Some(scheduled_core(5));
+			}),
+			// 6: Occupied(next_up_on_time_out set and timeout but available),
+			build_occupied_core(6, |core| {
+				core.next_up_on_time_out = Some(scheduled_core(6));
+				core.time_out_at = BLOCK_UNDER_PRODUCTION;
+				core.availability = core.availability.clone().not();
+			}),
+			// 7: Occupied(next_up_on_time_out set and timeout and not available),
+			build_occupied_core(7, |core| {
+				core.next_up_on_time_out = Some(scheduled_core(7));
+				core.time_out_at = BLOCK_UNDER_PRODUCTION;
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(71));
+			}),
+			// 8: Occupied(both next_up set, available),
+			build_occupied_core(8, |core| {
+				core.next_up_on_available = Some(scheduled_core(8));
+				core.next_up_on_time_out = Some(scheduled_core(8));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(81));
+			}),
+			// 9: Occupied(both next_up set, not available, no timeout),
+			build_occupied_core(9, |core| {
+				core.next_up_on_available = Some(scheduled_core(9));
+				core.next_up_on_time_out = Some(scheduled_core(9));
+			}),
+			// 10: Occupied(both next_up set, not available, timeout),
+			build_occupied_core(10, |core| {
+				core.next_up_on_available = Some(scheduled_core(10));
+				core.next_up_on_time_out = Some(scheduled_core(10));
+				core.time_out_at = BLOCK_UNDER_PRODUCTION;
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(101));
+			}),
+			// 11: Occupied(next_up_on_available and available, but different successor para_id)
+			build_occupied_core(11, |core| {
+				core.next_up_on_available = Some(scheduled_core(12));
+				core.availability = core.availability.clone().not();
+			}),
+			// 12-14: Occupied(next_up_on_available and available, same para_id).
+			build_occupied_core(12, |core| {
+				core.next_up_on_available = Some(scheduled_core(12));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(121));
+			}),
+			build_occupied_core(12, |core| {
+				core.next_up_on_available = Some(scheduled_core(12));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(122));
+			}),
+			build_occupied_core(12, |core| {
+				core.next_up_on_available = Some(scheduled_core(12));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(123));
+			}),
+			// 15: Scheduled on same para_id as 12-14.
+			Scheduled(scheduled_core(12)),
+			// 16: Occupied(13, no next_up set, not available)
+			build_occupied_core(13, |core| {
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(131));
+			}),
+			// 17: Occupied(13, no next_up set, available)
+			build_occupied_core(13, |core| {
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(132));
+			}),
+			// 18: Occupied(13, next_up_on_available set to 13 but not available)
+			build_occupied_core(13, |core| {
+				core.next_up_on_available = Some(scheduled_core(13));
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(133));
+			}),
+			// 19: Occupied(13, next_up_on_available set to 13 and available)
+			build_occupied_core(13, |core| {
+				core.next_up_on_available = Some(scheduled_core(13));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(134));
+			}),
+			// 20: Occupied(13, next_up_on_time_out set to 13 but not timeout)
+			build_occupied_core(13, |core| {
+				core.next_up_on_time_out = Some(scheduled_core(13));
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(135));
+			}),
+			// 21: Occupied(13, next_up_on_available set to 14 and available)
+			build_occupied_core(13, |core| {
+				core.next_up_on_available = Some(scheduled_core(14));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(136));
+			}),
+			// 22: Occupied(13, next_up_on_available set to 14 but not available)
+			build_occupied_core(13, |core| {
+				core.next_up_on_available = Some(scheduled_core(14));
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(137));
+			}),
+			// 23: Occupied(13, both next_up set to 14, available)
+			build_occupied_core(13, |core| {
+				core.next_up_on_available = Some(scheduled_core(14));
+				core.next_up_on_time_out = Some(scheduled_core(14));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(138));
+			}),
+			// 24: Occupied(13, both next_up set to 14, not available, timeout)
+			build_occupied_core(13, |core| {
+				core.next_up_on_available = Some(scheduled_core(14));
+				core.next_up_on_time_out = Some(scheduled_core(14));
+				core.time_out_at = BLOCK_UNDER_PRODUCTION;
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(1399));
+			}),
+			// 25: Occupied(13, next_up_on_available and available, but successor para_id 15)
+			build_occupied_core(13, |core| {
+				core.next_up_on_available = Some(scheduled_core(15));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(139));
+			}),
+			// 26: Occupied(15, next_up_on_available and available, but successor para_id 13)
+			build_occupied_core(15, |core| {
+				core.next_up_on_available = Some(scheduled_core(13));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(151));
+			}),
+			// 27: Occupied(15, both next_up, both available and timed out)
+			build_occupied_core(15, |core| {
+				core.next_up_on_available = Some(scheduled_core(15));
+				core.availability = core.availability.clone().not();
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(152));
+				core.time_out_at = BLOCK_UNDER_PRODUCTION;
+			}),
+			// 28: Occupied(13, both next_up set to 13, not available)
+			build_occupied_core(13, |core| {
+				core.next_up_on_available = Some(scheduled_core(13));
+				core.next_up_on_time_out = Some(scheduled_core(13));
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(1398));
+			}),
+			// 29: Occupied(13, both next_up set to 13, not available, timeout)
+			build_occupied_core(13, |core| {
+				core.next_up_on_available = Some(scheduled_core(13));
+				core.next_up_on_time_out = Some(scheduled_core(13));
+				core.time_out_at = BLOCK_UNDER_PRODUCTION;
+				core.candidate_hash = CandidateHash(Hash::from_low_u64_be(1397));
+			}),
+		]
+	}
+
 	async fn mock_overseer(
 		mut receiver: mpsc::UnboundedReceiver<AllMessages>,
-		expected: Vec<BackedCandidate>,
+		mock_availability_cores: Vec<CoreState>,
+		mut expected: Vec<BackedCandidate>,
+		mut expected_ancestors: HashMap<Vec<CandidateHash>, Ancestors>,
 		prospective_parachains_mode: ProspectiveParachainsMode,
 	) {
 		use ChainApiMessage::BlockNumber;
 		use RuntimeApiMessage::Request;
 
+		let mut backed_iter = expected.clone().into_iter();
+
+		expected.sort_by_key(|c| c.candidate().descriptor.para_id);
 		let mut candidates_iter = expected
 			.iter()
 			.map(|candidate| (candidate.hash(), candidate.descriptor().relay_parent));
 
-		let mut backed_iter = expected.clone().into_iter();
-
 		while let Some(from_job) = receiver.next().await {
 			match from_job {
 				AllMessages::ChainApi(BlockNumber(_relay_parent, tx)) =>
@@ -356,7 +578,7 @@ mod select_candidates {
 					PersistedValidationDataReq(_para_id, _assumption, tx),
 				)) => tx.send(Ok(Some(Default::default()))).unwrap(),
 				AllMessages::RuntimeApi(Request(_parent_hash, AvailabilityCores(tx))) =>
-					tx.send(Ok(mock_availability_cores())).unwrap(),
+					tx.send(Ok(mock_availability_cores.clone())).unwrap(),
 				AllMessages::CandidateBacking(CandidateBackingMessage::GetBackedCandidates(
 					hashes,
 					sender,
@@ -373,35 +595,71 @@ mod select_candidates {
 					let _ = sender.send(response);
 				},
 				AllMessages::ProspectiveParachains(
-					ProspectiveParachainsMessage::GetBackableCandidates(_, _, count, _, tx),
-				) => {
-					assert_eq!(count, 1);
+					ProspectiveParachainsMessage::GetBackableCandidates(
+						_,
+						_para_id,
+						count,
+						actual_ancestors,
+						tx,
+					),
+				) => match prospective_parachains_mode {
+					ProspectiveParachainsMode::Enabled { .. } => {
+						assert!(count > 0);
+						let candidates =
+							(&mut candidates_iter).take(count as usize).collect::<Vec<_>>();
+						assert_eq!(candidates.len(), count as usize);
 
-					match prospective_parachains_mode {
-						ProspectiveParachainsMode::Enabled { .. } => {
-							let _ =
-								tx.send(candidates_iter.next().map_or_else(Vec::new, |c| vec![c]));
-						},
-						ProspectiveParachainsMode::Disabled =>
-							panic!("unexpected prospective parachains request"),
-					}
+						if !expected_ancestors.is_empty() {
+							if let Some(expected_required_ancestors) = expected_ancestors.remove(
+								&(candidates
+									.clone()
+									.into_iter()
+									.take(actual_ancestors.len())
+									.map(|(c_hash, _)| c_hash)
+									.collect::<Vec<_>>()),
+							) {
+								assert_eq!(expected_required_ancestors, actual_ancestors);
+							} else {
+								assert_eq!(actual_ancestors.len(), 0);
+							}
+						}
+
+						let _ = tx.send(candidates);
+					},
+					ProspectiveParachainsMode::Disabled =>
+						panic!("unexpected prospective parachains request"),
 				},
 				_ => panic!("Unexpected message: {:?}", from_job),
 			}
 		}
+
+		if let ProspectiveParachainsMode::Enabled { .. } = prospective_parachains_mode {
+			assert_eq!(candidates_iter.next(), None);
+		}
+		assert_eq!(expected_ancestors.len(), 0);
 	}
 
-	#[test]
-	fn can_succeed() {
+	#[rstest]
+	#[case(ProspectiveParachainsMode::Disabled)]
+	#[case(ProspectiveParachainsMode::Enabled {max_candidate_depth: 0, allowed_ancestry_len: 0})]
+	fn can_succeed(#[case] prospective_parachains_mode: ProspectiveParachainsMode) {
 		test_harness(
-			|r| mock_overseer(r, Vec::new(), ProspectiveParachainsMode::Disabled),
+			|r| {
+				mock_overseer(
+					r,
+					Vec::new(),
+					Vec::new(),
+					HashMap::new(),
+					prospective_parachains_mode,
+				)
+			},
 			|mut tx: TestSubsystemSender| async move {
-				let prospective_parachains_mode = ProspectiveParachainsMode::Disabled;
 				select_candidates(
 					&[],
 					&[],
 					&[],
 					prospective_parachains_mode,
+					false,
 					Default::default(),
 					&mut tx,
 				)
@@ -411,22 +669,22 @@ mod select_candidates {
 		)
 	}
 
-	// this tests that only the appropriate candidates get selected.
-	// To accomplish this, we supply a candidate list containing one candidate per possible core;
-	// the candidate selection algorithm must filter them to the appropriate set
-	#[test]
-	fn selects_correct_candidates() {
-		let mock_cores = mock_availability_cores();
-
-		let empty_hash = PersistedValidationData::<Hash, BlockNumber>::default().hash();
-
-		let mut descriptor_template = dummy_candidate_descriptor(dummy_hash());
-		descriptor_template.persisted_validation_data_hash = empty_hash;
-		let candidate_template = CandidateReceipt {
-			descriptor: descriptor_template,
-			commitments_hash: CandidateCommitments::default().hash(),
-		};
-
+	// Test candidate selection when prospective parachains mode is disabled.
+	// This tests that only the appropriate candidates get selected when prospective parachains mode
+	// is disabled. To accomplish this, we supply a candidate list containing one candidate per
+	// possible core; the candidate selection algorithm must filter them to the appropriate set
+	#[rstest]
+	// why those particular indices? see the comments on mock_availability_cores_*() functions.
+	#[case(mock_availability_cores_one_per_para(), vec![1, 4, 7, 8, 10], true)]
+	#[case(mock_availability_cores_one_per_para(), vec![1, 4, 7, 8, 10], false)]
+	#[case(mock_availability_cores_multiple_per_para(), vec![1, 4, 7, 8, 10, 12, 13, 14, 15], true)]
+	#[case(mock_availability_cores_multiple_per_para(), vec![1, 4, 7, 8, 10, 12, 13, 14, 15], false)]
+	fn test_in_subsystem_selection(
+		#[case] mock_cores: Vec<CoreState>,
+		#[case] expected_candidates: Vec<usize>,
+		#[case] elastic_scaling_mvp: bool,
+	) {
+		let candidate_template = dummy_candidate_template();
 		let candidates: Vec<_> = std::iter::repeat(candidate_template)
 			.take(mock_cores.len())
 			.enumerate()
@@ -453,9 +711,8 @@ mod select_candidates {
 			})
 			.collect();
 
-		// why those particular indices? see the comments on mock_availability_cores()
 		let expected_candidates: Vec<_> =
-			[1, 4, 7, 8, 10].iter().map(|&idx| candidates[idx].clone()).collect();
+			expected_candidates.into_iter().map(|idx| candidates[idx].clone()).collect();
 		let prospective_parachains_mode = ProspectiveParachainsMode::Disabled;
 
 		let expected_backed = expected_candidates
@@ -473,14 +730,24 @@ mod select_candidates {
 			})
 			.collect();
 
+		let mock_cores_clone = mock_cores.clone();
 		test_harness(
-			|r| mock_overseer(r, expected_backed, prospective_parachains_mode),
+			|r| {
+				mock_overseer(
+					r,
+					mock_cores_clone,
+					expected_backed,
+					HashMap::new(),
+					prospective_parachains_mode,
+				)
+			},
 			|mut tx: TestSubsystemSender| async move {
-				let result = select_candidates(
+				let result: Vec<BackedCandidate> = select_candidates(
 					&mock_cores,
 					&[],
 					&candidates,
 					prospective_parachains_mode,
+					elastic_scaling_mvp,
 					Default::default(),
 					&mut tx,
 				)
@@ -498,20 +765,24 @@ mod select_candidates {
 		)
 	}
 
-	#[test]
-	fn selects_max_one_code_upgrade() {
-		let mock_cores = mock_availability_cores();
+	#[rstest]
+	#[case(ProspectiveParachainsMode::Disabled)]
+	#[case(ProspectiveParachainsMode::Enabled {max_candidate_depth: 0, allowed_ancestry_len: 0})]
+	fn selects_max_one_code_upgrade(
+		#[case] prospective_parachains_mode: ProspectiveParachainsMode,
+	) {
+		let mock_cores = mock_availability_cores_one_per_para();
 
 		let empty_hash = PersistedValidationData::<Hash, BlockNumber>::default().hash();
 
 		// why those particular indices? see the comments on mock_availability_cores()
-		// the first candidate with code is included out of [1, 4, 7, 8, 10].
-		let cores = [1, 4, 7, 8, 10];
+		// the first candidate with code is included out of [1, 4, 7, 8, 10, 12].
+		let cores = [1, 4, 7, 8, 10, 12];
 		let cores_with_code = [1, 4, 8];
 
-		let expected_cores = [1, 7, 10];
+		let expected_cores = [1, 7, 10, 12];
 
-		let committed_receipts: Vec<_> = (0..mock_cores.len())
+		let committed_receipts: Vec<_> = (0..=mock_cores.len())
 			.map(|i| {
 				let mut descriptor = dummy_candidate_descriptor(dummy_hash());
 				descriptor.para_id = i.into();
@@ -552,23 +823,32 @@ mod select_candidates {
 		let expected_backed_filtered: Vec<_> =
 			expected_cores.iter().map(|&idx| candidates[idx].clone()).collect();
 
-		let prospective_parachains_mode = ProspectiveParachainsMode::Disabled;
+		let mock_cores_clone = mock_cores.clone();
 
 		test_harness(
-			|r| mock_overseer(r, expected_backed, prospective_parachains_mode),
+			|r| {
+				mock_overseer(
+					r,
+					mock_cores_clone,
+					expected_backed,
+					HashMap::new(),
+					prospective_parachains_mode,
+				)
+			},
 			|mut tx: TestSubsystemSender| async move {
 				let result = select_candidates(
 					&mock_cores,
 					&[],
 					&candidates,
 					prospective_parachains_mode,
+					false,
 					Default::default(),
 					&mut tx,
 				)
 				.await
 				.unwrap();
 
-				assert_eq!(result.len(), 3);
+				assert_eq!(result.len(), 4);
 
 				result.into_iter().for_each(|c| {
 					assert!(
@@ -581,66 +861,214 @@ mod select_candidates {
 		)
 	}
 
-	#[test]
-	fn request_from_prospective_parachains() {
-		let mock_cores = mock_availability_cores();
-		let empty_hash = PersistedValidationData::<Hash, BlockNumber>::default().hash();
-
-		let mut descriptor_template = dummy_candidate_descriptor(dummy_hash());
-		descriptor_template.persisted_validation_data_hash = empty_hash;
-		let candidate_template = CandidateReceipt {
-			descriptor: descriptor_template,
-			commitments_hash: CandidateCommitments::default().hash(),
-		};
-
-		let candidates: Vec<_> = std::iter::repeat(candidate_template)
-			.take(mock_cores.len())
-			.enumerate()
-			.map(|(idx, mut candidate)| {
-				candidate.descriptor.para_id = idx.into();
-				candidate
-			})
-			.collect();
+	#[rstest]
+	#[case(true)]
+	#[case(false)]
+	fn request_from_prospective_parachains_one_core_per_para(#[case] elastic_scaling_mvp: bool) {
+		let mock_cores = mock_availability_cores_one_per_para();
 
 		// why those particular indices? see the comments on mock_availability_cores()
-		let expected_candidates: Vec<_> =
-			[1, 4, 7, 8, 10].iter().map(|&idx| candidates[idx].clone()).collect();
+		let expected_candidates: Vec<_> = vec![1, 4, 7, 8, 10, 12];
+		let (candidates, expected_candidates) =
+			make_candidates(mock_cores.len() + 1, expected_candidates);
+
 		// Expect prospective parachains subsystem requests.
 		let prospective_parachains_mode =
 			ProspectiveParachainsMode::Enabled { max_candidate_depth: 0, allowed_ancestry_len: 0 };
 
-		let expected_backed = expected_candidates
-			.iter()
-			.map(|c| {
-				BackedCandidate::new(
-					CommittedCandidateReceipt {
-						descriptor: c.descriptor.clone(),
-						commitments: Default::default(),
-					},
-					Vec::new(),
-					default_bitvec(MOCK_GROUP_SIZE),
-					None,
-				)
-			})
-			.collect();
+		let mut required_ancestors: HashMap<Vec<CandidateHash>, Ancestors> = HashMap::new();
+		required_ancestors.insert(
+			vec![candidates[4]],
+			vec![CandidateHash(Hash::from_low_u64_be(41))].into_iter().collect(),
+		);
+		required_ancestors.insert(
+			vec![candidates[8]],
+			vec![CandidateHash(Hash::from_low_u64_be(81))].into_iter().collect(),
+		);
 
+		let mock_cores_clone = mock_cores.clone();
+		let expected_candidates_clone = expected_candidates.clone();
 		test_harness(
-			|r| mock_overseer(r, expected_backed, prospective_parachains_mode),
+			|r| {
+				mock_overseer(
+					r,
+					mock_cores_clone,
+					expected_candidates_clone,
+					required_ancestors,
+					prospective_parachains_mode,
+				)
+			},
 			|mut tx: TestSubsystemSender| async move {
 				let result = select_candidates(
 					&mock_cores,
 					&[],
 					&[],
 					prospective_parachains_mode,
+					elastic_scaling_mvp,
 					Default::default(),
 					&mut tx,
 				)
 				.await
 				.unwrap();
 
+				assert_eq!(result.len(), expected_candidates.len());
 				result.into_iter().for_each(|c| {
 					assert!(
-						expected_candidates.iter().any(|c2| c.candidate().corresponds_to(c2)),
+						expected_candidates
+							.iter()
+							.any(|c2| c.candidate().corresponds_to(&c2.receipt())),
+						"Failed to find candidate: {:?}",
+						c,
+					)
+				});
+			},
+		)
+	}
+
+	#[test]
+	fn request_from_prospective_parachains_multiple_cores_per_para_elastic_scaling_mvp() {
+		let mock_cores = mock_availability_cores_multiple_per_para();
+
+		// why those particular indices? see the comments on mock_availability_cores()
+		let expected_candidates: Vec<_> =
+			vec![1, 4, 7, 8, 10, 12, 12, 12, 12, 12, 13, 13, 13, 14, 14, 14, 15, 15];
+		// Expect prospective parachains subsystem requests.
+		let prospective_parachains_mode =
+			ProspectiveParachainsMode::Enabled { max_candidate_depth: 0, allowed_ancestry_len: 0 };
+
+		let (candidates, expected_candidates) =
+			make_candidates(mock_cores.len(), expected_candidates);
+
+		let mut required_ancestors: HashMap<Vec<CandidateHash>, Ancestors> = HashMap::new();
+		required_ancestors.insert(
+			vec![candidates[4]],
+			vec![CandidateHash(Hash::from_low_u64_be(41))].into_iter().collect(),
+		);
+		required_ancestors.insert(
+			vec![candidates[8]],
+			vec![CandidateHash(Hash::from_low_u64_be(81))].into_iter().collect(),
+		);
+		required_ancestors.insert(
+			[12, 12, 12].iter().map(|&idx| candidates[idx]).collect::<Vec<_>>(),
+			vec![
+				CandidateHash(Hash::from_low_u64_be(121)),
+				CandidateHash(Hash::from_low_u64_be(122)),
+				CandidateHash(Hash::from_low_u64_be(123)),
+			]
+			.into_iter()
+			.collect(),
+		);
+		required_ancestors.insert(
+			[13, 13, 13].iter().map(|&idx| candidates[idx]).collect::<Vec<_>>(),
+			(131..=139)
+				.map(|num| CandidateHash(Hash::from_low_u64_be(num)))
+				.chain(std::iter::once(CandidateHash(Hash::from_low_u64_be(1398))))
+				.collect(),
+		);
+
+		required_ancestors.insert(
+			[15, 15].iter().map(|&idx| candidates[idx]).collect::<Vec<_>>(),
+			vec![
+				CandidateHash(Hash::from_low_u64_be(151)),
+				CandidateHash(Hash::from_low_u64_be(152)),
+			]
+			.into_iter()
+			.collect(),
+		);
+
+		let mock_cores_clone = mock_cores.clone();
+		let expected_candidates_clone = expected_candidates.clone();
+		test_harness(
+			|r| {
+				mock_overseer(
+					r,
+					mock_cores_clone,
+					expected_candidates,
+					required_ancestors,
+					prospective_parachains_mode,
+				)
+			},
+			|mut tx: TestSubsystemSender| async move {
+				let result = select_candidates(
+					&mock_cores,
+					&[],
+					&[],
+					prospective_parachains_mode,
+					true,
+					Default::default(),
+					&mut tx,
+				)
+				.await
+				.unwrap();
+
+				assert_eq!(result.len(), expected_candidates_clone.len());
+				result.into_iter().for_each(|c| {
+					assert!(
+						expected_candidates_clone
+							.iter()
+							.any(|c2| c.candidate().corresponds_to(&c2.receipt())),
+						"Failed to find candidate: {:?}",
+						c,
+					)
+				});
+			},
+		)
+	}
+
+	#[test]
+	fn request_from_prospective_parachains_multiple_cores_per_para_elastic_scaling_mvp_disabled() {
+		let mock_cores = mock_availability_cores_multiple_per_para();
+
+		// why those particular indices? see the comments on mock_availability_cores()
+		let expected_candidates: Vec<_> = vec![1, 4, 7, 8, 10];
+		// Expect prospective parachains subsystem requests.
+		let prospective_parachains_mode =
+			ProspectiveParachainsMode::Enabled { max_candidate_depth: 0, allowed_ancestry_len: 0 };
+
+		let (candidates, expected_candidates) =
+			make_candidates(mock_cores.len(), expected_candidates);
+
+		let mut required_ancestors: HashMap<Vec<CandidateHash>, Ancestors> = HashMap::new();
+		required_ancestors.insert(
+			vec![candidates[4]],
+			vec![CandidateHash(Hash::from_low_u64_be(41))].into_iter().collect(),
+		);
+		required_ancestors.insert(
+			vec![candidates[8]],
+			vec![CandidateHash(Hash::from_low_u64_be(81))].into_iter().collect(),
+		);
+
+		let mock_cores_clone = mock_cores.clone();
+		let expected_candidates_clone = expected_candidates.clone();
+		test_harness(
+			|r| {
+				mock_overseer(
+					r,
+					mock_cores_clone,
+					expected_candidates,
+					required_ancestors,
+					prospective_parachains_mode,
+				)
+			},
+			|mut tx: TestSubsystemSender| async move {
+				let result = select_candidates(
+					&mock_cores,
+					&[],
+					&[],
+					prospective_parachains_mode,
+					false,
+					Default::default(),
+					&mut tx,
+				)
+				.await
+				.unwrap();
+
+				assert_eq!(result.len(), expected_candidates_clone.len());
+				result.into_iter().for_each(|c| {
+					assert!(
+						expected_candidates_clone
+							.iter()
+							.any(|c2| c.candidate().corresponds_to(&c2.receipt())),
 						"Failed to find candidate: {:?}",
 						c,
 					)
@@ -651,18 +1079,11 @@ mod select_candidates {
 
 	#[test]
 	fn request_receipts_based_on_relay_parent() {
-		let mock_cores = mock_availability_cores();
-		let empty_hash = PersistedValidationData::<Hash, BlockNumber>::default().hash();
-
-		let mut descriptor_template = dummy_candidate_descriptor(dummy_hash());
-		descriptor_template.persisted_validation_data_hash = empty_hash;
-		let candidate_template = CandidateReceipt {
-			descriptor: descriptor_template,
-			commitments_hash: CandidateCommitments::default().hash(),
-		};
+		let mock_cores = mock_availability_cores_one_per_para();
+		let candidate_template = dummy_candidate_template();
 
 		let candidates: Vec<_> = std::iter::repeat(candidate_template)
-			.take(mock_cores.len())
+			.take(mock_cores.len() + 1)
 			.enumerate()
 			.map(|(idx, mut candidate)| {
 				candidate.descriptor.para_id = idx.into();
@@ -673,7 +1094,7 @@ mod select_candidates {
 
 		// why those particular indices? see the comments on mock_availability_cores()
 		let expected_candidates: Vec<_> =
-			[1, 4, 7, 8, 10].iter().map(|&idx| candidates[idx].clone()).collect();
+			[1, 4, 7, 8, 10, 12].iter().map(|&idx| candidates[idx].clone()).collect();
 		// Expect prospective parachains subsystem requests.
 		let prospective_parachains_mode =
 			ProspectiveParachainsMode::Enabled { max_candidate_depth: 0, allowed_ancestry_len: 0 };
@@ -693,14 +1114,24 @@ mod select_candidates {
 			})
 			.collect();
 
+		let mock_cores_clone = mock_cores.clone();
 		test_harness(
-			|r| mock_overseer(r, expected_backed, prospective_parachains_mode),
+			|r| {
+				mock_overseer(
+					r,
+					mock_cores_clone,
+					expected_backed,
+					HashMap::new(),
+					prospective_parachains_mode,
+				)
+			},
 			|mut tx: TestSubsystemSender| async move {
 				let result = select_candidates(
 					&mock_cores,
 					&[],
 					&[],
 					prospective_parachains_mode,
+					false,
 					Default::default(),
 					&mut tx,
 				)