Move artifacts states into memory in PVF validation host

node/core/pvf/src/artifacts.rs

@@ -23,33 +23,44 @@ use std::{
 	time::{Duration, SystemTime},
 };
 
-/// A final product of preparation process. Contains either a ready to run compiled artifact or
-/// a description what went wrong.
-#[derive(Encode, Decode)]
-pub enum Artifact {
+/// An error that occurred during the prepare part of the PVF pipeline.
+#[derive(Debug, Clone, Encode, Decode)]
+pub enum PrepareError {
 	/// During the prevalidation stage of preparation an issue was found with the PVF.
-	PrevalidationErr(String),
+	Prevalidation(String),
 	/// Compilation failed for the given PVF.
-	PreparationErr(String),
+	Preparation(String),
 	/// This state indicates that the process assigned to prepare the artifact wasn't responsible
 	/// or were killed. This state is reported by the validation host (not by the worker).
-	DidntMakeIt,
-	/// The PVF passed all the checks and is ready for execution.
-	Compiled { compiled_artifact: Vec<u8> },
+	DidNotMakeIt,
 }
 
-impl Artifact {
-	/// Serializes this struct into a byte buffer.
-	pub fn serialize(&self) -> Vec<u8> {
-		self.encode()
+/// A wrapper for the compiled PVF code.
+#[derive(Encode, Decode)]
+pub struct CompiledArtifact(Vec<u8>);
+
+impl CompiledArtifact {
+	pub fn new(code: Vec<u8>) -> Self {
+		Self(code)
 	}
+}
 
-	/// Deserialize the given byte buffer to an artifact.
-	pub fn deserialize(mut bytes: &[u8]) -> Result<Self, String> {
-		Artifact::decode(&mut bytes).map_err(|e| format!("{:?}", e))
+impl AsRef<[u8]> for CompiledArtifact {
+	fn as_ref(&self) -> &[u8] {
+		self.0.as_slice()
 	}
 }
 
+/// A final product of preparation process. Contains either a ready to run compiled artifact or
+/// a description what went wrong.
+#[derive(Encode, Decode)]
+pub enum Artifact {
+	/// An error occurred during the prepare part of the PVF pipeline.
+	Error(PrepareError),
+	/// The PVF passed all the checks and is ready for execution.
+	Compiled(CompiledArtifact),
+}
+
 /// Identifier of an artifact. Right now it only encodes a code hash of the PVF. But if we get to
 /// multiple engine implementations the artifact ID should include the engine type as well.
 #[derive(Clone, Debug, PartialEq, Eq, PartialOrd, Ord, Hash)]
@@ -117,6 +128,9 @@ pub enum ArtifactState {
 	},
 	/// A task to prepare this artifact is scheduled.
 	Preparing,
+	/// The code couldn't be compiled due to an error. Such artifacts
+	/// never reach the executor and stay in the host's memory.
+	FailedToProcess(PrepareError),
 }
 
 /// A container of all known artifact ids and their states.

node/core/pvf/src/error.rs

@@ -14,6 +14,8 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
+use crate::artifacts::PrepareError;
+
 /// A error raised during validation of the candidate.
 #[derive(Debug, Clone)]
 pub enum ValidationError {
@@ -54,3 +56,14 @@ pub enum InvalidCandidate {
 	/// PVF execution (compilation is not included) took more time than was allotted.
 	HardTimeout,
 }
+
+impl From<PrepareError> for ValidationError {
+	fn from(error: PrepareError) -> Self {
+		let error_str = match error {
+			PrepareError::Prevalidation(err) => err,
+			PrepareError::Preparation(err) => err,
+			PrepareError::DidNotMakeIt => "preparation timeout".to_owned(),
+		};
+		ValidationError::InvalidCandidate(InvalidCandidate::WorkerReportedError(error_str))
+	}
+}

node/core/pvf/src/execute/worker.rs

@@ -15,7 +15,7 @@
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
 use crate::{
-	artifacts::{Artifact, ArtifactPathId},
+	artifacts::{ArtifactPathId, CompiledArtifact},
 	executor_intf::TaskExecutor,
 	worker_common::{
 		bytes_to_path, framed_recv, framed_send, path_to_bytes, spawn_with_program_path,
@@ -217,18 +217,12 @@ async fn validate_using_artifact(
 		Ok(b) => b,
 	};
 
-	let artifact = match Artifact::deserialize(&artifact_bytes) {
+	let artifact = match CompiledArtifact::decode(&mut artifact_bytes.as_slice()) {
 		Err(e) => return Response::InternalError(format!("artifact deserialization: {:?}", e)),
 		Ok(a) => a,
 	};
 
-	let compiled_artifact = match &artifact {
-		Artifact::PrevalidationErr(msg) => return Response::format_invalid("prevalidation", msg),
-		Artifact::PreparationErr(msg) => return Response::format_invalid("preparation", msg),
-		Artifact::DidntMakeIt => return Response::format_invalid("preparation timeout", ""),
-
-		Artifact::Compiled { compiled_artifact } => compiled_artifact,
-	};
+	let compiled_artifact = artifact.as_ref();
 
 	let validation_started_at = Instant::now();
 	let descriptor_bytes = match unsafe {

node/core/pvf/src/host.rs

@@ -327,8 +327,7 @@ async fn run(
 				.await);
 			},
 			from_prepare_queue = from_prepare_queue_rx.next() => {
-				let prepare::FromQueue::Prepared(artifact_id)
-					= break_if_fatal!(from_prepare_queue.ok_or(Fatal));
+				let from_queue = break_if_fatal!(from_prepare_queue.ok_or(Fatal));
 
 				// Note that preparation always succeeds.
 				//
@@ -344,7 +343,7 @@ async fn run(
 					&mut artifacts,
 					&mut to_execute_queue_tx,
 					&mut awaiting_prepare,
-					artifact_id,
+					from_queue,
 				).await);
 			},
 		}
@@ -419,6 +418,9 @@ async fn handle_execute_pvf(
 
 				awaiting_prepare.add(artifact_id, params, result_tx);
 			},
+			ArtifactState::FailedToProcess(error) => {
+				let _ = result_tx.send(Err(ValidationError::from(error.clone())));
+			},
 		}
 	} else {
 		// Artifact is unknown: register it and enqueue a job with the corresponding priority and
@@ -450,6 +452,7 @@ async fn handle_heads_up(
 					// Already preparing. We don't need to send a priority amend either because
 					// it can't get any lower than the background.
 				},
+				ArtifactState::FailedToProcess(_) => {},
 			}
 		} else {
 			// The artifact is unknown: register it and put a background job into the prepare queue.
@@ -471,8 +474,10 @@ async fn handle_prepare_done(
 	artifacts: &mut Artifacts,
 	execute_queue: &mut mpsc::Sender<execute::ToQueue>,
 	awaiting_prepare: &mut AwaitingPrepare,
-	artifact_id: ArtifactId,
+	from_queue: prepare::FromQueue,
 ) -> Result<(), Fatal> {
+	let prepare::FromQueue { artifact_id, result } = from_queue;
+
 	// Make some sanity checks and extract the current state.
 	let state = match artifacts.artifact_state_mut(&artifact_id) {
 		None => {
@@ -493,6 +498,12 @@ async fn handle_prepare_done(
 			never!("the artifact is already prepared: {:?}", artifact_id);
 			return Ok(())
 		},
+		Some(ArtifactState::FailedToProcess(_)) => {
+			// The reasoning is similar to the above, the artifact cannot be
+			// processed at this point.
+			never!("the artifact is already processed unsuccessfully: {:?}", artifact_id);
+			return Ok(())
+		},
 		Some(state @ ArtifactState::Preparing) => state,
 	};
 
@@ -506,6 +517,13 @@ async fn handle_prepare_done(
 			continue
 		}
 
+		// Don't send failed artifacts to the execution's queue.
+		if let Err(ref error) = result {
+			*state = ArtifactState::FailedToProcess(error.clone());
+			let _ = result_tx.send(Err(ValidationError::from(error.clone())));
+			continue
+		}
+
 		send_execute(
 			execute_queue,
 			execute::ToQueue::Enqueue {
@@ -895,7 +913,7 @@ mod tests {
 		);
 
 		test.from_prepare_queue_tx
-			.send(prepare::FromQueue::Prepared(artifact_id(1)))
+			.send(prepare::FromQueue { artifact_id: artifact_id(1), result: Ok(()) })
 			.await
 			.unwrap();
 		let result_tx_pvf_1_1 = assert_matches!(
@@ -908,7 +926,7 @@ mod tests {
 		);
 
 		test.from_prepare_queue_tx
-			.send(prepare::FromQueue::Prepared(artifact_id(2)))
+			.send(prepare::FromQueue { artifact_id: artifact_id(2), result: Ok(()) })
 			.await
 			.unwrap();
 		let result_tx_pvf_2 = assert_matches!(
@@ -957,7 +975,7 @@ mod tests {
 		);
 
 		test.from_prepare_queue_tx
-			.send(prepare::FromQueue::Prepared(artifact_id(1)))
+			.send(prepare::FromQueue { artifact_id: artifact_id(1), result: Ok(()) })
 			.await
 			.unwrap();
 

node/core/pvf/src/prepare/pool.rs

@@ -16,6 +16,7 @@
 
 use super::worker::{self, Outcome};
 use crate::{
+	artifacts::PrepareError,
 	metrics::Metrics,
 	worker_common::{IdleWorker, WorkerHandle},
 	LOG_TARGET,
@@ -80,7 +81,13 @@ pub enum FromPool {
 
 	/// The given worker either succeeded or failed the given job. Under any circumstances the
 	/// artifact file has been written. The `bool` says whether the worker ripped.
-	Concluded(Worker, bool),
+	Concluded {
+		worker: Worker,
+		rip: bool,
+		/// [`Ok`] indicates that compiled artifact is successfully stored on disk.
+		/// Otherwise, an [error](PrepareError) is supplied.
+		result: Result<(), PrepareError>,
+	},
 
 	/// The given worker ceased to exist.
 	Rip(Worker),
@@ -295,7 +302,7 @@ fn handle_mux(
 		},
 		PoolEvent::StartWork(worker, outcome) => {
 			match outcome {
-				Outcome::Concluded(idle) => {
+				Outcome::Concluded { worker: idle, result } => {
 					let data = match spawned.get_mut(worker) {
 						None => {
 							// Perhaps the worker was killed meanwhile and the result is no longer
@@ -310,7 +317,7 @@ fn handle_mux(
 					let old = data.idle.replace(idle);
 					assert_matches!(old, None, "attempt to overwrite an idle worker");
 
-					reply(from_pool, FromPool::Concluded(worker, false))?;
+					reply(from_pool, FromPool::Concluded { worker, rip: false, result })?;
 
 					Ok(())
 				},
@@ -321,9 +328,16 @@ fn handle_mux(
 
 					Ok(())
 				},
-				Outcome::DidntMakeIt => {
+				Outcome::DidNotMakeIt => {
 					if attempt_retire(metrics, spawned, worker) {
-						reply(from_pool, FromPool::Concluded(worker, true))?;
+						reply(
+							from_pool,
+							FromPool::Concluded {
+								worker,
+								rip: true,
+								result: Err(PrepareError::DidNotMakeIt),
+							},
+						)?;
 					}
 
 					Ok(())