CandidateBackingSubsystem (#1312)

* Updates guide for CandidateBacking

* Move assignment types to primitives

* Initial implementation.

* More functionality

* use assert_matches

* Changes to report misbehaviors

* Some fixes after a review

* Remove a blank line

* Update guide and some types

* Adds run_job function

* Some comments and refactorings

* Fix review

* Remove warnings

* Use summary in kicking off validation

* Parallelize requests

* Validation provides local and global validation params

* Test issued validity tracking

* Nits from review

polkadot/node/core/backing/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "polkadot-node-core-backing"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+edition = "2018"
+
+[dependencies]
+futures = "0.3.5"
+log = "0.4.8"
+sp-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "master" }
+keystore = { package = "sc-keystore", git = "https://github.com/paritytech/substrate", branch = "master" }
+primitives = { package = "sp-core", git = "https://github.com/paritytech/substrate", branch = "master" }
+
+polkadot-primitives = { path = "../../../primitives" }
+polkadot-node-primitives = { path = "../../primitives" }
+polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
+erasure-coding = { package = "polkadot-erasure-coding", path = "../../../erasure-coding" }
+statement-table = { package = "polkadot-statement-table", path = "../../../statement-table" }
+futures-timer = "3.0.2"
+streamunordered = "0.5.1"
+derive_more = "0.99.9"
+bitvec = { version = "0.17.4", default-features = false, features = ["alloc"] }
+
+[dev-dependencies]
+sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master" }
+futures = { version = "0.3.5", features = ["thread-pool"] }
+subsystem-test = { package = "polkadot-subsystem-test-helpers", path = "../../test-helpers/subsystem" }
+assert_matches = "1.3.0"

polkadot/node/overseer/examples/minimal-example.rs

@@ -59,6 +59,7 @@ impl Subsystem1 {
 
 			ctx.send_message(AllMessages::CandidateValidation(
 				CandidateValidationMessage::Validate(
+					Default::default(),
 					Default::default(),
 					Default::default(),
 					PoVBlock {

polkadot/node/overseer/src/lib.rs

@@ -726,6 +726,7 @@ mod tests {
 						ctx.send_message(
 							AllMessages::CandidateValidation(
 								CandidateValidationMessage::Validate(
+									Default::default(),
 									Default::default(),
 									Default::default(),
 									PoVBlock {

polkadot/node/primitives/src/lib.rs

@@ -23,10 +23,17 @@
 use parity_scale_codec::{Decode, Encode};
 use polkadot_primitives::{Hash,
 	parachain::{
-		AbridgedCandidateReceipt, CandidateReceipt, CompactStatement,
-		EncodeAs, Signed,
+		AbridgedCandidateReceipt, CompactStatement,
+		EncodeAs, Signed, SigningContext, ValidatorIndex, ValidatorId,
 	}
 };
+use polkadot_statement_table::{
+	generic::{
+		ValidityDoubleVote as TableValidityDoubleVote,
+		MultipleCandidates as TableMultipleCandidates,
+	},
+	Misbehavior as TableMisbehavior,
+};
 
 /// A statement, where the candidate receipt is included in the `Seconded` variant.
 #[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
@@ -77,11 +84,128 @@ pub enum MisbehaviorReport {
 	/// this message should be dispatched with all of them, in arbitrary order.
 	///
 	/// This variant is also used when our own validity checks disagree with others'.
-	CandidateValidityDisagreement(CandidateReceipt, Vec<SignedFullStatement>),
+	CandidateValidityDisagreement(AbridgedCandidateReceipt, Vec<SignedFullStatement>),
 	/// I've noticed a peer contradicting itself about a particular candidate
-	SelfContradiction(CandidateReceipt, SignedFullStatement, SignedFullStatement),
+	SelfContradiction(AbridgedCandidateReceipt, SignedFullStatement, SignedFullStatement),
 	/// This peer has seconded more than one parachain candidate for this relay parent head
-	DoubleVote(CandidateReceipt, SignedFullStatement, SignedFullStatement),
+	DoubleVote(SignedFullStatement, SignedFullStatement),
+}
+
+/// A utility struct used to convert `TableMisbehavior` to `MisbehaviorReport`s.
+pub struct FromTableMisbehavior {
+	/// Index of the validator.
+	pub id: ValidatorIndex,
+	/// The misbehavior reported by the table.
+	pub report: TableMisbehavior,
+	/// Signing context.
+	pub signing_context: SigningContext,
+	/// Misbehaving validator's public key.
+	pub key: ValidatorId,
+}
+
+/// Result of the validation of the candidate.
+#[derive(Debug)]
+pub enum ValidationResult {
+	/// Candidate is valid.
+	Valid,
+	/// Candidate is invalid.
+	Invalid,
+}
+
+impl std::convert::TryFrom<FromTableMisbehavior> for MisbehaviorReport {
+	type Error = ();
+
+	fn try_from(f: FromTableMisbehavior) -> Result<Self, Self::Error> {
+		match f.report {
+			TableMisbehavior::ValidityDoubleVote(
+				TableValidityDoubleVote::IssuedAndValidity((c, s1), (d, s2))
+			) => {
+				let receipt = c.clone();
+				let signed_1 = SignedFullStatement::new(
+					Statement::Seconded(c),
+					f.id,
+					s1,
+					&f.signing_context,
+					&f.key,
+				).ok_or(())?;
+				let signed_2 = SignedFullStatement::new(
+					Statement::Valid(d),
+					f.id,
+					s2,
+					&f.signing_context,
+					&f.key,
+				).ok_or(())?;
+
+				Ok(MisbehaviorReport::SelfContradiction(receipt, signed_1, signed_2))
+			}
+			TableMisbehavior::ValidityDoubleVote(
+				TableValidityDoubleVote::IssuedAndInvalidity((c, s1), (d, s2))
+			) => {
+				let receipt = c.clone();
+				let signed_1 = SignedFullStatement::new(
+					Statement::Seconded(c),
+					f.id,
+					s1,
+					&f.signing_context,
+					&f.key,
+				).ok_or(())?;
+				let signed_2 = SignedFullStatement::new(
+					Statement::Invalid(d),
+					f.id,
+					s2,
+					&f.signing_context,
+					&f.key,
+				).ok_or(())?;
+
+				Ok(MisbehaviorReport::SelfContradiction(receipt, signed_1, signed_2))
+			}
+			TableMisbehavior::ValidityDoubleVote(
+				TableValidityDoubleVote::ValidityAndInvalidity(c, s1, s2)
+			) => {
+				let signed_1 = SignedFullStatement::new(
+					Statement::Valid(c.hash()),
+					f.id,
+					s1,
+					&f.signing_context,
+					&f.key,
+				).ok_or(())?;
+				let signed_2 = SignedFullStatement::new(
+					Statement::Invalid(c.hash()),
+					f.id,
+					s2,
+					&f.signing_context,
+					&f.key,
+				).ok_or(())?;
+
+				Ok(MisbehaviorReport::SelfContradiction(c, signed_1, signed_2))
+			}
+			TableMisbehavior::MultipleCandidates(
+				TableMultipleCandidates {
+					first,
+					second,
+				}
+			) => {
+				let signed_1 = SignedFullStatement::new(
+					Statement::Seconded(first.0),
+					f.id,
+					first.1,
+					&f.signing_context,
+					&f.key,
+				).ok_or(())?;
+
+				let signed_2 = SignedFullStatement::new(
+					Statement::Seconded(second.0),
+					f.id,
+					second.1,
+					&f.signing_context,
+					&f.key,
+				).ok_or(())?;
+
+				Ok(MisbehaviorReport::DoubleVote(signed_1, signed_2))
+			}
+			_ => Err(()),
+		}
+	}
 }
 
 /// A unique identifier for a network protocol.

polkadot/node/subsystem/src/messages.rs

@@ -28,10 +28,11 @@ use polkadot_primitives::{BlockNumber, Hash, Signature};
 use polkadot_primitives::parachain::{
 	AbridgedCandidateReceipt, PoVBlock, ErasureChunk, BackedCandidate, Id as ParaId,
 	SignedAvailabilityBitfield, SigningContext, ValidatorId, ValidationCode, ValidatorIndex,
-	CandidateDescriptor,
+	CoreAssignment, CoreOccupied, HeadData, CandidateDescriptor, GlobalValidationSchedule,
+	LocalValidationData,
 };
 use polkadot_node_primitives::{
-	MisbehaviorReport, SignedFullStatement, View, ProtocolId,
+	MisbehaviorReport, SignedFullStatement, View, ProtocolId, ValidationResult,
 };
 
 use std::sync::Arc;
@@ -53,12 +54,12 @@ pub enum CandidateSelectionMessage {
 /// Messages received by the Candidate Backing subsystem.
 #[derive(Debug)]
 pub enum CandidateBackingMessage {
-	/// Registers a stream listener for updates to the set of backable candidates that could be backed
-	/// in a child of the given relay-parent, referenced by its hash.
-	RegisterBackingWatcher(Hash, mpsc::Sender<NewBackedCandidate>),
+	/// Requests a set of backable candidates that could be backed in a child of the given
+	/// relay-parent, referenced by its hash.
+	GetBackedCandidates(Hash, oneshot::Sender<Vec<NewBackedCandidate>>),
 	/// Note that the Candidate Backing subsystem should second the given candidate in the context of the
 	/// given relay-parent (ref. by hash). This candidate must be validated.
-	Second(Hash, AbridgedCandidateReceipt),
+	Second(Hash, AbridgedCandidateReceipt, PoVBlock),
 	/// Note a validator's statement about a particular candidate. Disagreements about validity must be escalated
 	/// to a broader check by Misbehavior Arbitration. Agreements are simply tallied until a quorum is reached.
 	Statement(Hash, SignedFullStatement),
@@ -78,8 +79,12 @@ pub enum CandidateValidationMessage {
 	Validate(
 		Hash,
 		AbridgedCandidateReceipt,
+		HeadData,
 		PoVBlock,
-		oneshot::Sender<Result<(), ValidationFailed>>,
+		oneshot::Sender<Result<
+			(ValidationResult, GlobalValidationSchedule, LocalValidationData),
+			ValidationFailed,
+		>>,
 	),
 }
 
@@ -151,17 +156,34 @@ pub enum AvailabilityStoreMessage {
 	StoreChunk(Hash, ValidatorIndex, ErasureChunk),
 }
 
+/// The information on scheduler assignments that some somesystems may be querying.
+#[derive(Debug, Clone)]
+pub struct SchedulerRoster {
+	/// Validator-to-groups assignments.
+	pub validator_groups: Vec<Vec<ValidatorIndex>>,
+	/// All scheduled paras.
+	pub scheduled: Vec<CoreAssignment>,
+	/// Upcoming paras (chains and threads).
+	pub upcoming: Vec<ParaId>,
+	/// Occupied cores.
+	pub availability_cores: Vec<Option<CoreOccupied>>,
+}
+
 /// A request to the Runtime API subsystem.
 #[derive(Debug)]
 pub enum RuntimeApiRequest {
 	/// Get the current validator set.
 	Validators(oneshot::Sender<Vec<ValidatorId>>),
+	/// Get the assignments of validators to cores.
+	ValidatorGroups(oneshot::Sender<SchedulerRoster>),
 	/// Get a signing context for bitfields and statements.
 	SigningContext(oneshot::Sender<SigningContext>),
 	/// Get the validation code for a specific para, assuming execution under given block number, and
 	/// an optional block number representing an intermediate parablock executed in the context of
 	/// that block.
 	ValidationCode(ParaId, BlockNumber, Option<BlockNumber>, oneshot::Sender<ValidationCode>),
+	/// Get head data for a specific para.
+	HeadData(ParaId, oneshot::Sender<HeadData>),
 }
 
 /// A message to the Runtime API subsystem.