Make candidate validation bounded again

polkadot/node/core/candidate-validation/Cargo.toml

@@ -8,6 +8,7 @@ license.workspace = true
 
 [dependencies]
 async-trait = "0.1.57"
+async-semaphore = "1.2.0"
 futures = "0.3.21"
 futures-timer = "3.0.2"
 gum = { package = "tracing-gum", path = "../../gum" }

polkadot/node/core/candidate-validation/src/lib.rs

@@ -23,6 +23,7 @@
 #![deny(unused_crate_dependencies, unused_results)]
 #![warn(missing_docs)]
 
+use async_semaphore::Semaphore;
 use polkadot_node_core_pvf::{
 	InternalValidationError, InvalidCandidate as WasmInvalidCandidate, PrepareError,
 	PrepareJobKind, PrepareStats, PvfPrepData, ValidationError, ValidationHost,
@@ -55,7 +56,7 @@ use polkadot_primitives::{
 
 use parity_scale_codec::Encode;
 
-use futures::{channel::oneshot, prelude::*};
+use futures::{channel::oneshot, prelude::*, stream::FuturesUnordered};
 
 use std::{
 	path::PathBuf,
@@ -153,106 +154,112 @@ async fn run<Context>(
 	.await?;
 	ctx.spawn_blocking("pvf-validation-host", task.boxed())?;
 
+	let mut tasks = FuturesUnordered::new();
+	// The task queue size is chosen to be somewhat bigger than the PVF host incoming queue size
+	// to allow exhaustive validation messages to fall through in case the tasks are clogged with
+	// `ValidateFromChainState` messages awaiting data from the runtime
+	let semaphore = Arc::new(Semaphore::new(polkadot_node_core_pvf::HOST_MESSAGE_QUEUE_SIZE * 2));
+
 	loop {
-		match ctx.recv().await? {
-			FromOrchestra::Signal(OverseerSignal::ActiveLeaves(_)) => {},
-			FromOrchestra::Signal(OverseerSignal::BlockFinalized(..)) => {},
-			FromOrchestra::Signal(OverseerSignal::Conclude) => return Ok(()),
-			FromOrchestra::Communication { msg } => match msg {
-				CandidateValidationMessage::ValidateFromChainState {
-					candidate_receipt,
-					pov,
-					executor_params,
-					exec_timeout_kind,
-					response_sender,
-					..
-				} => {
-					let bg = {
-						let mut sender = ctx.sender().clone();
-						let metrics = metrics.clone();
-						let validation_host = validation_host.clone();
-
-						async move {
-							let _timer = metrics.time_validate_from_chain_state();
-							let res = validate_from_chain_state(
-								&mut sender,
-								validation_host,
-								candidate_receipt,
-								pov,
-								executor_params,
-								exec_timeout_kind,
-								&metrics,
-							)
-							.await;
-
-							metrics.on_validation_event(&res);
-							let _ = response_sender.send(res);
-						}
-					};
-
-					ctx.spawn("validate-from-chain-state", bg.boxed())?;
-				},
-				CandidateValidationMessage::ValidateFromExhaustive {
-					validation_data,
-					validation_code,
-					candidate_receipt,
-					pov,
-					executor_params,
-					exec_timeout_kind,
-					response_sender,
-					..
-				} => {
-					let bg = {
-						let metrics = metrics.clone();
-						let validation_host = validation_host.clone();
-
-						async move {
-							let _timer = metrics.time_validate_from_exhaustive();
-							let res = validate_candidate_exhaustive(
-								validation_host,
-								validation_data,
-								validation_code,
-								candidate_receipt,
-								pov,
-								executor_params,
-								exec_timeout_kind,
-								&metrics,
-							)
-							.await;
-
-							metrics.on_validation_event(&res);
-							let _ = response_sender.send(res);
-						}
-					};
-
-					ctx.spawn("validate-from-exhaustive", bg.boxed())?;
-				},
-				CandidateValidationMessage::PreCheck {
-					relay_parent,
-					validation_code_hash,
-					response_sender,
-					..
-				} => {
-					let bg = {
-						let mut sender = ctx.sender().clone();
-						let validation_host = validation_host.clone();
-
-						async move {
-							let precheck_result = precheck_pvf(
-								&mut sender,
-								validation_host,
-								relay_parent,
-								validation_code_hash,
-							)
-							.await;
-
-							let _ = response_sender.send(precheck_result);
+		futures::select! {
+			comm = ctx.recv().fuse() => {
+				match comm {
+					Ok(FromOrchestra::Signal(OverseerSignal::ActiveLeaves(_))) => {},
+					Ok(FromOrchestra::Signal(OverseerSignal::BlockFinalized(..))) => {},
+					Ok(FromOrchestra::Signal(OverseerSignal::Conclude)) => return Ok(()),
+					Ok(FromOrchestra::Communication { msg }) => match msg {
+						CandidateValidationMessage::ValidateFromChainState {
+							candidate_receipt,
+							pov,
+							executor_params,
+							exec_timeout_kind,
+							response_sender,
+							..
+						} => {
+							let mut sender = ctx.sender().clone();
+							let metrics = metrics.clone();
+							let validation_host = validation_host.clone();
+
+							let guard = semaphore.acquire_arc().await;
+							tasks.push(async move {
+								let _guard = guard;
+								let _timer = metrics.time_validate_from_chain_state();
+								let res = validate_from_chain_state(
+									&mut sender,
+									validation_host,
+									candidate_receipt,
+									pov,
+									executor_params,
+									exec_timeout_kind,
+									&metrics,
+								)
+								.await;
+
+								metrics.on_validation_event(&res);
+								let _ = response_sender.send(res);
+							}.boxed());
+						},
+						CandidateValidationMessage::ValidateFromExhaustive {
+							validation_data,
+							validation_code,
+							candidate_receipt,
+							pov,
+							executor_params,
+							exec_timeout_kind,
+							response_sender,
+							..
+						} => {
+							let metrics = metrics.clone();
+							let validation_host = validation_host.clone();
+
+							let guard = semaphore.acquire_arc().await;
+							tasks.push(async move {
+								let _guard = guard;
+								let _timer = metrics.time_validate_from_exhaustive();
+								let res = validate_candidate_exhaustive(
+									validation_host,
+									validation_data,
+									validation_code,
+									candidate_receipt,
+									pov,
+									executor_params,
+									exec_timeout_kind,
+									&metrics,
+								)
+								.await;
+
+								metrics.on_validation_event(&res);
+								let _ = response_sender.send(res);
+							}.boxed());
+						},
+						CandidateValidationMessage::PreCheck {
+							relay_parent,
+							validation_code_hash,
+							response_sender,
+							..
+						} => {
+							let mut sender = ctx.sender().clone();
+							let validation_host = validation_host.clone();
+
+							let guard = semaphore.acquire_arc().await;
+							tasks.push(async move {
+								let _guard = guard;
+								let precheck_result = precheck_pvf(
+									&mut sender,
+									validation_host,
+									relay_parent,
+									validation_code_hash,
+								)
+								.await;
+
+								let _ = response_sender.send(precheck_result);
+							}.boxed());
 						}
-					};
-
-					ctx.spawn("candidate-validation-pre-check", bg.boxed())?;
-				},
+					},
+					Err(e) => return Err(SubsystemError::from(e))
+				}
 			},
+			_ = tasks.select_next_some() => ()
 		}
 	}
 }

polkadot/node/core/pvf/src/host.rs

@@ -59,6 +59,9 @@ pub const PREPARE_BINARY_NAME: &str = "polkadot-prepare-worker";
 /// The name of binary spawned to execute a PVF
 pub const EXECUTE_BINARY_NAME: &str = "polkadot-execute-worker";
 
+/// The size of incoming message queue
+pub const HOST_MESSAGE_QUEUE_SIZE: usize = 10;
+
 /// An alias to not spell the type for the oneshot sender for the PVF execution result.
 pub(crate) type ResultSender = oneshot::Sender<Result<ValidationResult, ValidationError>>;
 
@@ -213,7 +216,7 @@ pub async fn start(
 	// Run checks for supported security features once per host startup. Warn here if not enabled.
 	let security_status = security::check_security_status(&config).await;
 
-	let (to_host_tx, to_host_rx) = mpsc::channel(10);
+	let (to_host_tx, to_host_rx) = mpsc::channel(HOST_MESSAGE_QUEUE_SIZE);
 
 	let validation_host = ValidationHost { to_host_tx };
 

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

@@ -104,7 +104,10 @@ mod worker_intf;
 pub mod testing;
 
 pub use error::{InvalidCandidate, ValidationError};
-pub use host::{start, Config, ValidationHost, EXECUTE_BINARY_NAME, PREPARE_BINARY_NAME};
+pub use host::{
+	start, Config, ValidationHost, EXECUTE_BINARY_NAME, HOST_MESSAGE_QUEUE_SIZE,
+	PREPARE_BINARY_NAME,
+};
 pub use metrics::Metrics;
 pub use priority::Priority;
 pub use worker_intf::{framed_recv, framed_send, JOB_TIMEOUT_WALL_CLOCK_FACTOR};