backing: refactor off of jobs system (#5377)

* backing: refactor off of jobs system

* rename FailedToSpawnBg

* refactor: backing uses fatality

* fix service compilation

polkadot/Cargo.lock

@@ -6704,6 +6704,7 @@ version = "0.9.19"
 dependencies = [
  "assert_matches",
  "bitvec",
+ "fatality",
  "futures 0.3.21",
  "polkadot-erasure-coding",
  "polkadot-node-primitives",

polkadot/node/core/backing/Cargo.toml

@@ -16,6 +16,7 @@ statement-table = { package = "polkadot-statement-table", path = "../../../state
 bitvec = { version = "1.0.0", default-features = false, features = ["alloc"] }
 gum = { package = "tracing-gum", path = "../../gum" }
 thiserror = "1.0.30"
+fatality = "0.0.6"
 
 [dev-dependencies]
 sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }

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

@@ -0,0 +1,94 @@
+// Copyright 2022 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+use fatality::Nested;
+use futures::channel::{mpsc, oneshot};
+
+use polkadot_node_subsystem_util::Error as UtilError;
+use polkadot_primitives::v2::BackedCandidate;
+use polkadot_subsystem::{messages::ValidationFailed, SubsystemError};
+
+use crate::LOG_TARGET;
+
+pub type Result<T> = std::result::Result<T, Error>;
+pub type FatalResult<T> = std::result::Result<T, FatalError>;
+
+/// Errors that can occur in candidate backing.
+#[allow(missing_docs)]
+#[fatality::fatality(splitable)]
+pub enum Error {
+	#[error("Candidate is not found")]
+	CandidateNotFound,
+
+	#[error("Signature is invalid")]
+	InvalidSignature,
+
+	#[error("Failed to send candidates {0:?}")]
+	Send(Vec<BackedCandidate>),
+
+	#[error("FetchPoV failed")]
+	FetchPoV,
+
+	#[fatal]
+	#[error("Failed to spawn background task")]
+	FailedToSpawnBackgroundTask,
+
+	#[error("ValidateFromChainState channel closed before receipt")]
+	ValidateFromChainState(#[source] oneshot::Canceled),
+
+	#[error("StoreAvailableData channel closed before receipt")]
+	StoreAvailableData(#[source] oneshot::Canceled),
+
+	#[error("a channel was closed before receipt in try_join!")]
+	JoinMultiple(#[source] oneshot::Canceled),
+
+	#[error("Obtaining erasure chunks failed")]
+	ObtainErasureChunks(#[from] erasure_coding::Error),
+
+	#[error(transparent)]
+	ValidationFailed(#[from] ValidationFailed),
+
+	#[fatal]
+	#[error(transparent)]
+	BackgroundValidationMpsc(#[from] mpsc::SendError),
+
+	#[error(transparent)]
+	UtilError(#[from] UtilError),
+
+	#[error(transparent)]
+	SubsystemError(#[from] SubsystemError),
+}
+
+/// Utility for eating top level errors and log them.
+///
+/// We basically always want to try and continue on error. This utility function is meant to
+/// consume top-level errors by simply logging them
+pub fn log_error(result: Result<()>) -> std::result::Result<(), FatalError> {
+	match result.into_nested()? {
+		Ok(()) => Ok(()),
+		Err(jfyi) => {
+			jfyi.log();
+			Ok(())
+		},
+	}
+}
+
+impl JfyiError {
+	/// Log a `JfyiError`.
+	pub fn log(self) {
+		gum::debug!(target: LOG_TARGET, error = ?self);
+	}
+}

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

@@ -28,7 +28,7 @@ use polkadot_primitives::v2::{
 	ScheduledCore,
 };
 use polkadot_subsystem::{
-	messages::{CollatorProtocolMessage, RuntimeApiMessage, RuntimeApiRequest},
+	messages::{CollatorProtocolMessage, RuntimeApiMessage, RuntimeApiRequest, ValidationFailed},
 	ActivatedLeaf, ActiveLeavesUpdate, FromOverseer, LeafStatus, OverseerSignal,
 };
 use sp_application_crypto::AppKey;
@@ -153,8 +153,11 @@ fn test_harness<T: Future<Output = VirtualOverseer>>(
 
 	let (context, virtual_overseer) = test_helpers::make_subsystem_context(pool.clone());
 
-	let subsystem =
-		CandidateBackingSubsystem::new(pool.clone(), keystore, Metrics(None)).run(context);
+	let subsystem = async move {
+		if let Err(e) = super::run(context, keystore, Metrics(None)).await {
+			panic!("{:?}", e);
+		}
+	};
 
 	let test_fut = test(virtual_overseer);
 

polkadot/node/service/src/overseer.rs

@@ -145,7 +145,7 @@ pub fn prepared_overseer_builder<'a, Spawner, RuntimeClient>(
 		Arc<RuntimeClient>,
 		CandidateValidationSubsystem,
 		PvfCheckerSubsystem,
-		CandidateBackingSubsystem<Spawner>,
+		CandidateBackingSubsystem,
 		StatementDistributionSubsystem<rand::rngs::StdRng>,
 		AvailabilityDistributionSubsystem,
 		AvailabilityRecoverySubsystem,
@@ -201,7 +201,6 @@ where
 			Metrics::register(registry)?,
 		))
 		.candidate_backing(CandidateBackingSubsystem::new(
-			spawner.clone(),
 			keystore.clone(),
 			Metrics::register(registry)?,
 		))