Asynchronous Backing MegaPR

Cargo.toml

@@ -78,6 +78,7 @@ members = [
 	"node/core/chain-selection",
 	"node/core/dispute-coordinator",
 	"node/core/parachains-inherent",
+	"node/core/prospective-parachains",
 	"node/core/provisioner",
 	"node/core/pvf",
 	"node/core/pvf-checker",
@@ -207,6 +208,7 @@ fast-runtime = [ "polkadot-cli/fast-runtime" ]
 runtime-metrics = [ "polkadot-cli/runtime-metrics" ]
 pyroscope = ["polkadot-cli/pyroscope"]
 jemalloc-allocator = ["polkadot-node-core-pvf/jemalloc-allocator", "polkadot-overseer/jemalloc-allocator"]
+network-protocol-staging = ["polkadot-cli/network-protocol-staging"]
 
 # Configuration for building a .deb package - for use with `cargo-deb`
 [package.metadata.deb]

cli/Cargo.toml

@@ -75,3 +75,4 @@ rococo-native = ["service/rococo-native"]
 
 malus = ["full-node", "service/malus"]
 runtime-metrics = ["service/runtime-metrics", "polkadot-node-metrics/runtime-metrics"]
+network-protocol-staging = ["service/network-protocol-staging"]

node/collation-generation/src/lib.rs

@@ -198,15 +198,15 @@ async fn handle_new_activations<Context>(
 			let (scheduled_core, assumption) = match core {
 				CoreState::Scheduled(scheduled_core) =>
 					(scheduled_core, OccupiedCoreAssumption::Free),
-				CoreState::Occupied(_occupied_core) => {
-					// TODO: https://github.com/paritytech/polkadot/issues/1573
-					gum::trace!(
-						target: LOG_TARGET,
-						core_idx = %core_idx,
-						relay_parent = ?relay_parent,
-						"core is occupied. Keep going.",
-					);
-					continue
+				CoreState::Occupied(occupied_core) => {
+					// TODO [now]: this assumes that next up == current.
+					// in practice we should only set `OccupiedCoreAssumption::Included`
+					// when the candidate occupying the core is also of the same para.
+					if let Some(scheduled) = occupied_core.next_up_on_available {
+						(scheduled, OccupiedCoreAssumption::Included)
+					} else {
+						continue
+					}
 				},
 				CoreState::Free => {
 					gum::trace!(
@@ -286,6 +286,7 @@ async fn handle_new_activations<Context>(
 				"collation-builder",
 				Box::pin(async move {
 					let persisted_validation_data_hash = validation_data.hash();
+					let parent_head_data_hash = validation_data.parent_head.hash();
 
 					let (collation, result_sender) =
 						match (task_config.collator)(relay_parent, &validation_data).await {
@@ -385,8 +386,13 @@ async fn handle_new_activations<Context>(
 
 					if let Err(err) = task_sender
 						.send(
-							CollatorProtocolMessage::DistributeCollation(ccr, pov, result_sender)
-								.into(),
+							CollatorProtocolMessage::DistributeCollation(
+								ccr,
+								parent_head_data_hash,
+								pov,
+								result_sender,
+							)
+							.into(),
 						)
 						.await
 					{

node/core/approval-voting/src/criteria.rs

@@ -274,7 +274,7 @@ pub(crate) fn compute_assignments(
 	// Ignore any cores where the assigned group is our own.
 	let leaving_cores = leaving_cores
 		.into_iter()
-		.filter(|&(_, _, ref g)| !is_in_backing_group(&config.validator_groups, index, *g))
+		.filter(|(_, _, g)| !is_in_backing_group(&config.validator_groups, index, *g))
 		.map(|(c_hash, core, _)| (c_hash, core))
 		.collect::<Vec<_>>();
 
@@ -496,7 +496,7 @@ pub(crate) fn check_assignment_cert(
 		return Err(InvalidAssignment(Reason::IsInBackingGroup))
 	}
 
-	let &(ref vrf_output, ref vrf_proof) = &assignment.vrf;
+	let (vrf_output, vrf_proof) = &assignment.vrf;
 	match assignment.kind {
 		AssignmentCertKind::RelayVRFModulo { sample } => {
 			if sample >= config.relay_vrf_modulo_samples {

node/core/approval-voting/src/lib.rs

@@ -486,7 +486,7 @@ impl Wakeups {
 			.collect();
 
 		let mut pruned_wakeups = BTreeMap::new();
-		self.reverse_wakeups.retain(|&(ref h, ref c_h), tick| {
+		self.reverse_wakeups.retain(|(h, c_h), tick| {
 			let live = !pruned_blocks.contains(h);
 			if !live {
 				pruned_wakeups.entry(*tick).or_insert_with(HashSet::new).insert((*h, *c_h));

node/core/approval-voting/src/ops.rs

@@ -62,7 +62,7 @@ fn visit_and_remove_block_entry(
 	};
 
 	overlayed_db.delete_block_entry(&block_hash);
-	for &(_, ref candidate_hash) in block_entry.candidates() {
+	for (_, candidate_hash) in block_entry.candidates() {
 		let candidate = match visited_candidates.entry(*candidate_hash) {
 			Entry::Occupied(e) => e.into_mut(),
 			Entry::Vacant(e) => {
@@ -227,7 +227,7 @@ pub fn add_block_entry(
 
 	// read and write all updated entries.
 	{
-		for &(_, ref candidate_hash) in entry.candidates() {
+		for (_, candidate_hash) in entry.candidates() {
 			let NewCandidateInfo { candidate, backing_group, our_assignment } =
 				match candidate_info(candidate_hash) {
 					None => return Ok(Vec::new()),

node/core/backing/src/error.rs

@@ -17,9 +17,9 @@
 use fatality::Nested;
 use futures::channel::{mpsc, oneshot};
 
-use polkadot_node_subsystem::{messages::ValidationFailed, SubsystemError};
-use polkadot_node_subsystem_util::Error as UtilError;
-use polkadot_primitives::BackedCandidate;
+use polkadot_node_subsystem::{messages::ValidationFailed, RuntimeApiError, SubsystemError};
+use polkadot_node_subsystem_util::{runtime, Error as UtilError};
+use polkadot_primitives::{BackedCandidate, ValidationCodeHash};
 
 use crate::LOG_TARGET;
 
@@ -30,6 +30,18 @@ pub type FatalResult<T> = std::result::Result<T, FatalError>;
 #[allow(missing_docs)]
 #[fatality::fatality(splitable)]
 pub enum Error {
+	#[fatal]
+	#[error("Failed to spawn background task")]
+	FailedToSpawnBackgroundTask,
+
+	#[fatal(forward)]
+	#[error("Error while accessing runtime information")]
+	Runtime(#[from] runtime::Error),
+
+	#[fatal]
+	#[error(transparent)]
+	BackgroundValidationMpsc(#[from] mpsc::SendError),
+
 	#[error("Candidate is not found")]
 	CandidateNotFound,
 
@@ -42,16 +54,27 @@ pub enum Error {
 	#[error("FetchPoV failed")]
 	FetchPoV,
 
-	#[fatal]
-	#[error("Failed to spawn background task")]
-	FailedToSpawnBackgroundTask,
+	#[error("Fetching validation code by hash failed {0:?}, {1:?}")]
+	FetchValidationCode(ValidationCodeHash, RuntimeApiError),
+
+	#[error("Fetching Runtime API version failed {0:?}")]
+	FetchRuntimeApiVersion(RuntimeApiError),
+
+	#[error("No validation code {0:?}")]
+	NoValidationCode(ValidationCodeHash),
+
+	#[error("Candidate rejected by prospective parachains subsystem")]
+	RejectedByProspectiveParachains,
 
-	#[error("ValidateFromChainState channel closed before receipt")]
-	ValidateFromChainState(#[source] oneshot::Canceled),
+	#[error("ValidateFromExhaustive channel closed before receipt")]
+	ValidateFromExhaustive(#[source] oneshot::Canceled),
 
 	#[error("StoreAvailableData channel closed before receipt")]
 	StoreAvailableData(#[source] oneshot::Canceled),
 
+	#[error("RuntimeAPISubsystem channel closed before receipt")]
+	RuntimeApiUnavailable(#[source] oneshot::Canceled),
+
 	#[error("a channel was closed before receipt in try_join!")]
 	JoinMultiple(#[source] oneshot::Canceled),
 
@@ -61,10 +84,6 @@ pub enum Error {
 	#[error(transparent)]
 	ValidationFailed(#[from] ValidationFailed),
 
-	#[fatal]
-	#[error(transparent)]
-	BackgroundValidationMpsc(#[from] mpsc::SendError),
-
 	#[error(transparent)]
 	UtilError(#[from] UtilError),