Request based PoV distribution

Cargo.toml

@@ -57,7 +57,6 @@ members = [
 	"node/core/runtime-api",
 	"node/network/approval-distribution",
 	"node/network/bridge",
-	"node/network/pov-distribution",
 	"node/network/protocol",
 	"node/network/statement-distribution",
 	"node/network/bitfield-distribution",

node/core/backing/Cargo.toml

@@ -8,6 +8,7 @@ edition = "2018"
 futures = "0.3.12"
 sp-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
 polkadot-primitives = { path = "../../../primitives" }
+polkadot-node-network-protocol = { path = "../../network/protocol" }
 polkadot-node-primitives = { path = "../../primitives" }
 polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
 polkadot-node-subsystem-util = { path = "../../subsystem-util" }

node/core/backing/src/lib.rs

@@ -29,20 +29,12 @@ use sp_keystore::SyncCryptoStorePtr;
 use polkadot_primitives::v1::{
 	AvailableData, BackedCandidate, CandidateCommitments, CandidateDescriptor, CandidateHash,
 	CandidateReceipt, CollatorId, CommittedCandidateReceipt, CoreIndex, CoreState, Hash, Id as ParaId,
-	PoV, SigningContext, ValidatorId, ValidatorIndex, ValidatorSignature, ValidityAttestation,
+	PoV, SigningContext, ValidatorId, ValidatorIndex, ValidatorSignature, ValidityAttestation, AuthorityDiscoveryId,
 };
 use polkadot_node_primitives::{
 	Statement, SignedFullStatement, ValidationResult,
 };
-use polkadot_subsystem::{
-	PerLeafSpan, Stage,
-	jaeger,
-	messages::{
-		AllMessages, AvailabilityStoreMessage, CandidateBackingMessage, CandidateSelectionMessage,
-		CandidateValidationMessage, PoVDistributionMessage, ProvisionableData,
-		ProvisionerMessage, StatementDistributionMessage, ValidationFailed, RuntimeApiRequest,
-	},
-};
+use polkadot_subsystem::{PerLeafSpan, Stage, errors::RuntimeApiError, jaeger, messages::{AllMessages, AvailabilityStoreMessage, CandidateBackingMessage, CandidateSelectionMessage, CandidateValidationMessage, IfDisconnected, NetworkBridgeMessage, PoVDistributionMessage, ProvisionableData, ProvisionerMessage, RuntimeApiRequest, StatementDistributionMessage, ValidationFailed}};
 use polkadot_node_subsystem_util::{
 	self as util,
 	request_session_index_for_child,
@@ -54,6 +46,10 @@ use polkadot_node_subsystem_util::{
 	FromJobCommand,
 	metrics::{self, prometheus},
 };
+use polkadot_node_subsystem_util::Error as UtilError;
+use polkadot_node_network_protocol::request_response::{Recipient, Requests, request::{OutgoingRequest, RequestError}, v1::{
+		PoVFetchingRequest, PoVFetchingResponse
+	}};
 use statement_table::{
 	generic::AttestedCandidate as TableAttestedCandidate,
 	Context as TableContextTrait,
@@ -65,6 +61,7 @@ use statement_table::{
 	},
 };
 use thiserror::Error;
+use util::request_session_info;
 
 const LOG_TARGET: &str = "parachain::candidate-backing";
 
@@ -76,8 +73,22 @@ enum Error {
 	InvalidSignature,
 	#[error("Failed to send candidates {0:?}")]
 	Send(Vec<BackedCandidate>),
+	#[error("FetchPoV request error")]
+	FetchPoV(#[source] RequestError),
 	#[error("FetchPoV channel closed before receipt")]
-	FetchPoV(#[source] oneshot::Canceled),
+	FetchPoVCanceled(#[source] oneshot::Canceled),
+	#[error("FetchPoV runtime request failed in utilities")]
+	FetchPoVUtil(#[source] UtilError),
+	#[error("FetchPoV runtime request failed")]
+	FetchPoVRuntime(#[source] RuntimeApiError),
+	#[error("FetchPoV could not find session")]
+	FetchPoVNoSuchSession,
+	#[error("FetchPoV could not find index of signing validator")]
+	FetchPoVInvalidValidatorIndex,
+	#[error("Invalid response")]
+	FetchPoVInvalidResponse,
+	#[error("Backer did not have PoV")]
+	FetchPoVNoSuchPoV,
 	#[error("ValidateFromChainState channel closed before receipt")]
 	ValidateFromChainState(#[source] oneshot::Canceled),
 	#[error("StoreAvailableData channel closed before receipt")]
@@ -94,6 +105,14 @@ enum Error {
 	UtilError(#[from] util::Error),
 }
 
+/// PoV data to validate.
+enum PoVData {
+	/// Allready available (from candidate selection).
+	Ready(Arc<PoV>),
+	/// Needs to be fetched from validator (we are checking a signed statement).
+	FetchFromValidator(ValidatorIndex),
+}
+
 enum ValidatedCandidateCommand {
 	// We were instructed to second the candidate.
 	Second(BackgroundValidationResult),
@@ -336,18 +355,51 @@ async fn make_pov_available(
 	Ok(Ok(()))
 }
 
-async fn request_pov_from_distribution(
+async fn request_pov(
 	tx_from: &mut mpsc::Sender<FromJobCommand>,
 	parent: Hash,
 	descriptor: CandidateDescriptor,
+	from_validator: ValidatorIndex,
 ) -> Result<Arc<PoV>, Error> {
-	let (tx, rx) = oneshot::channel();
-
-	tx_from.send(AllMessages::PoVDistribution(
-		PoVDistributionMessage::FetchPoV(parent, descriptor, tx)
-	).into()).await?;
-
-	rx.await.map_err(Error::FetchPoV)
+	let session_index = request_session_index_for_child(parent, tx_from)
+		.await.map_err(Error::FetchPoVUtil)?
+		.await.map_err(Error::FetchPoVCanceled)?
+		.map_err(Error::FetchPoVRuntime)?;
+	let session_info = request_session_info(parent, session_index, tx_from)
+		.await.map_err(Error::FetchPoVUtil)?
+		.await.map_err(Error::FetchPoVCanceled)?
+		.map_err(Error::FetchPoVRuntime)?
+		.ok_or(Error::FetchPoVNoSuchSession)?;
+	let authority_id = session_info.discovery_keys
+		.get(from_validator.0 as usize)
+		.ok_or(Error::FetchPoVInvalidValidatorIndex)?;
+
+	let (req, pending_response) = OutgoingRequest::new(
+		Recipient::Authority(authority_id.clone()),
+		PoVFetchingRequest {
+			relay_parent: parent,
+			descriptor,
+		},
+	);
+	let full_req = Requests::PoVFetching(req);
+
+	tx_from.send(FromJobCommand::SendMessage(AllMessages::NetworkBridge(
+		NetworkBridgeMessage::SendRequests(
+			vec![full_req],
+			IfDisconnected::ImmediateError
+		)
+	))).await?;
+
+	let response = pending_response.await.map_err(Error::FetchPoV)?;
+	let pov = match response {
+		PoVFetchingResponse::PoV(compressed) => {
+			compressed.decompress().map_err(|_| Error::FetchPoVInvalidResponse)?
+		}
+		PoVFetchingResponse::NoSuchPoV => {
+			return Err(Error::FetchPoVNoSuchPoV)
+		}
+	};
+	Ok(Arc::new(pov))
 }
 
 async fn request_candidate_validation(
@@ -380,7 +432,7 @@ struct BackgroundValidationParams<F> {
 	tx_command: mpsc::Sender<ValidatedCandidateCommand>,
 	candidate: CandidateReceipt,
 	relay_parent: Hash,
-	pov: Option<Arc<PoV>>,
+	pov: PoVData,
 	validator_index: Option<ValidatorIndex>,
 	n_validators: usize,
 	span: Option<jaeger::Span>,
@@ -403,13 +455,14 @@ async fn validate_and_make_available(
 	} = params;
 
 	let pov = match pov {
-		Some(pov) => pov,
-		None => {
+		PoVData::Ready(pov) => pov,
+		PoVData::FetchFromValidator(validator_index) => {
 			let _span = span.as_ref().map(|s| s.child("request-pov"));
-			request_pov_from_distribution(
+			request_pov(
 				&mut tx_from,
 				relay_parent,
 				candidate.descriptor.clone(),
+				validator_index,
 			).await?
 		}
 	};
@@ -544,7 +597,6 @@ impl CandidateBackingJob {
 							).await? {
 								self.issue_candidate_seconded_message(stmt).await?;
 							}
-							self.distribute_pov(candidate.descriptor, pov).await?;
 						}
 					}
 					Err(candidate) => {
@@ -645,7 +697,7 @@ impl CandidateBackingJob {
 			tx_command: self.background_validation_tx.clone(),
 			candidate: candidate.clone(),
 			relay_parent: self.parent,
-			pov: Some(pov),
+			pov: PoVData::Ready(pov),
 			validator_index: self.table_context.validator.as_ref().map(|v| v.index()),
 			n_validators: self.table_context.validators.len(),
 			span,
@@ -824,6 +876,7 @@ impl CandidateBackingJob {
 	async fn kick_off_validation_work(
 		&mut self,
 		summary: TableSummary,
+		statement: SignedFullStatement,
 		span: Option<jaeger::Span>,
 	) -> Result<(), Error> {
 		let candidate_hash = summary.candidate;
@@ -862,7 +915,7 @@ impl CandidateBackingJob {
 			tx_command: self.background_validation_tx.clone(),
 			candidate,
 			relay_parent: self.parent,
-			pov: None,
+			pov: PoVData::FetchFromValidator(statement.validator_index()),
 			validator_index: self.table_context.validator.as_ref().map(|v| v.index()),
 			n_validators: self.table_context.validators.len(),
 			span,
@@ -887,7 +940,11 @@ impl CandidateBackingJob {
 						summary.group_id,
 					);
 
-					self.kick_off_validation_work(summary, span).await?;
+					self.kick_off_validation_work(
+						summary,
+						statement,
+						span,
+					).await?;
 				}
 			}
 		}
@@ -988,16 +1045,6 @@ impl CandidateBackingJob {
 		Ok(())
 	}
 
-	async fn distribute_pov(
-		&mut self,
-		descriptor: CandidateDescriptor,
-		pov: Arc<PoV>,
-	) -> Result<(), Error> {
-		self.tx_from.send(AllMessages::from(
-			PoVDistributionMessage::DistributePoV(self.parent, descriptor, pov),
-		).into()).await.map_err(Into::into)
-	}
-
 	async fn distribute_signed_statement(&mut self, s: SignedFullStatement) -> Result<(), Error> {
 		let smsg = StatementDistributionMessage::Share(self.parent, s);
 

node/network/availability-distribution/src/error.rs

@@ -17,19 +17,25 @@
 
 //! Error handling related code and Error/Result definitions.
 
+use polkadot_node_network_protocol::request_response::request::RequestError;
 use thiserror::Error;
 
 use futures::channel::oneshot;
 
 use polkadot_node_subsystem_util::Error as UtilError;
-use polkadot_primitives::v1::SessionIndex;
+use polkadot_primitives::v1::{CompressedPoVError, SessionIndex};
 use polkadot_subsystem::{errors::RuntimeApiError, SubsystemError};
 
+use crate::LOG_TARGET;
+
 /// Errors of this subsystem.
 #[derive(Debug, Error)]
 pub enum Error {
-	#[error("Response channel to obtain QueryChunk failed")]
+	#[error("Response channel to obtain chunk failed")]
 	QueryChunkResponseChannel(#[source] oneshot::Canceled),
+
+	#[error("Response channel to obtain available data failed")]
+	QueryAvailableDataResponseChannel(#[source] oneshot::Canceled),
 
 	#[error("Receive channel closed")]
 	IncomingMessageChannel(#[source] SubsystemError),
@@ -53,24 +59,39 @@ pub enum Error {
 	/// Sending response failed.
 	#[error("Sending a request's response failed.")]
 	SendResponse,
-}
 
-/// Error that we should handle gracefully by logging it.
-#[derive(Debug)]
-pub enum NonFatalError {
 	/// Some request to utility functions failed.
 	/// This can be either `RuntimeRequestCanceled` or `RuntimeApiError`.
+	#[error("Utility request failed")]
 	UtilRequest(UtilError),
 
 	/// Runtime API subsystem is down, which means we're shutting down.
+	#[error("Runtime request canceled")]
 	RuntimeRequestCanceled(oneshot::Canceled),
 
 	/// Some request to the runtime failed.
 	/// For example if we prune a block we're requesting info about.
+	#[error("Runtime API error")]
 	RuntimeRequest(RuntimeApiError),
 
 	/// We tried fetching a session info which was not available.
+	#[error("There was no session with the given index")]
 	NoSuchSession(SessionIndex),
+
+	/// Decompressing PoV failed.
+	#[error("PoV could not be decompressed")]
+	PoVDecompression(CompressedPoVError),
+
+	/// Fetching PoV failed with `RequestError`.
+	#[error("FetchPoV request error")]
+	FetchPoV(#[source] RequestError),
+
+	#[error("Remote responded with `NoSuchPoV`")]
+	NoSuchPoV,
+
+	/// No validator with the index could be found in current session.
+	#[error("Given validator index could not be found")]
+	InvalidValidatorIndex,
 }
 
 pub type Result<T> = std::result::Result<T, Error>;
@@ -87,9 +108,20 @@ pub(crate) async fn recv_runtime<V>(
 		oneshot::Receiver<std::result::Result<V, RuntimeApiError>>,
 		UtilError,
 	>,
-) -> std::result::Result<V, NonFatalError> {
-	r.map_err(NonFatalError::UtilRequest)?
+) -> std::result::Result<V, Error> {
+	r.map_err(Error::UtilRequest)?
 		.await
-		.map_err(NonFatalError::RuntimeRequestCanceled)?
-		.map_err(NonFatalError::RuntimeRequest)
+		.map_err(Error::RuntimeRequestCanceled)?
+		.map_err(Error::RuntimeRequest)
+}
+
+
+/// 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<()>, ctx: &'static str) {
+	if let Err(error) = result {
+		tracing::warn!(target: LOG_TARGET, error = ?error, ctx);
+	}
 }