Collator protocol subsystem (#1659)

* WIP

* The initial implementation of the collator side.

* Improve comments

* Multiple collation requests

* Add more tests and comments to validator side

* Add comments, remove dead code

* Apply suggestions from code review

Co-authored-by: Peter Goodspeed-Niklaus <coriolinus@users.noreply.github.com>

* Fix build after suggested changes

* Also connect to the next validator group

* Remove a Future impl and move TimeoutExt to util

* Minor nits

* Fix build

* Change FetchCollations back to FetchCollation

* Try this

* Final fixes

* Fix build

Co-authored-by: Peter Goodspeed-Niklaus <coriolinus@users.noreply.github.com>

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

@@ -212,7 +212,12 @@ impl CandidateSelectionJob {
 	) {
 		if self.seconded_candidate.is_none() {
 			let (candidate_receipt, pov) =
-				match get_collation(relay_parent, para_id, self.sender.clone()).await {
+				match get_collation(
+					relay_parent,
+					para_id,
+					collator_id.clone(),
+					self.sender.clone(),
+				).await {
 					Ok(response) => response,
 					Err(err) => {
 						log::warn!(
@@ -296,12 +301,14 @@ impl CandidateSelectionJob {
 async fn get_collation(
 	relay_parent: Hash,
 	para_id: ParaId,
+	collator_id: CollatorId,
 	mut sender: mpsc::Sender<FromJob>,
 ) -> Result<(CandidateReceipt, PoV), Error> {
 	let (tx, rx) = oneshot::channel();
 	sender
 		.send(FromJob::Collator(CollatorProtocolMessage::FetchCollation(
 			relay_parent,
+			collator_id,
 			para_id,
 			tx,
 		)))
@@ -514,7 +521,7 @@ mod tests {
 						CandidateSelectionMessage::Collation(
 							relay_parent,
 							para_id,
-							collator_id_clone,
+							collator_id_clone.clone(),
 						),
 					))
 					.await
@@ -525,11 +532,13 @@ mod tests {
 					match msg {
 						FromJob::Collator(CollatorProtocolMessage::FetchCollation(
 							got_relay_parent,
+							collator_id,
 							got_para_id,
 							return_sender,
 						)) => {
 							assert_eq!(got_relay_parent, relay_parent);
 							assert_eq!(got_para_id, para_id);
+							assert_eq!(collator_id, collator_id_clone);
 
 							return_sender
 								.send((candidate_receipt.clone(), pov.clone()))

polkadot/node/network/availability-distribution/Cargo.toml

@@ -14,6 +14,7 @@ polkadot-erasure-coding = { path = "../../../erasure-coding" }
 polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
 polkadot-network-bridge = { path = "../../network/bridge" }
 polkadot-node-network-protocol = { path = "../../network/protocol" }
+polkadot-node-subsystem-util = { path = "../../subsystem-util" }
 sc-keystore = { git = "https://github.com/paritytech/substrate", branch = "master" }
 derive_more = "0.99.9"
 sp-core = { git = "https://github.com/paritytech/substrate", branch = "master", features = ["std"]  }

polkadot/node/network/availability-distribution/src/tests.rs

@@ -22,7 +22,8 @@ use polkadot_primitives::v1::{
 	GroupRotationInfo, HeadData, PersistedValidationData, OccupiedCore,
 	PoV, ScheduledCore, ValidatorPair,
 };
-use polkadot_subsystem_testhelpers::{self as test_helpers, TimeoutExt};
+use polkadot_subsystem_testhelpers::{self as test_helpers};
+use polkadot_node_subsystem_util::TimeoutExt;
 use polkadot_node_network_protocol::ObservedRole;
 
 use futures::{executor, future, Future};

polkadot/node/network/bitfield-distribution/Cargo.toml

@@ -13,6 +13,7 @@ codec = { package="parity-scale-codec", version = "1.3.4" }
 node-primitives = { package = "polkadot-node-primitives", path = "../../primitives" }
 polkadot-primitives = { path = "../../../primitives" }
 polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
+polkadot-node-subsystem-util = { package = "polkadot-node-subsystem-util", path = "../../subsystem-util" }
 polkadot-network-bridge = { path = "../../network/bridge" }
 polkadot-node-network-protocol = { path = "../../network/protocol" }
 sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }

polkadot/node/network/bitfield-distribution/src/lib.rs

@@ -614,7 +614,8 @@ mod test {
 	use futures::executor;
 	use maplit::hashmap;
 	use polkadot_primitives::v1::{Signed, ValidatorPair, AvailabilityBitfield};
-	use polkadot_node_subsystem_test_helpers::{make_subsystem_context, TimeoutExt};
+	use polkadot_node_subsystem_test_helpers::make_subsystem_context;
+	use polkadot_node_subsystem_util::TimeoutExt;
 	use sp_core::crypto::Pair;
 	use std::time::Duration;
 	use assert_matches::assert_matches;

polkadot/node/network/collator-protocol/Cargo.toml

@@ -0,0 +1,30 @@
+[package]
+name = "polkadot-collator-protocol"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+edition = "2018"
+
+[dependencies]
+futures = "0.3.5"
+log = "0.4.11"
+derive_more = "0.99.9"
+
+codec = { package="parity-scale-codec", version = "1.3.4", features = ["std"]  }
+
+polkadot-primitives = { path = "../../../primitives" }
+polkadot-network-bridge = { path = "../../network/bridge" }
+polkadot-node-network-protocol = { path = "../../network/protocol" }
+polkadot-node-subsystem-util = { path = "../../subsystem-util" }
+polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
+
+[dev-dependencies]
+env_logger = "0.7.1"
+assert_matches = "1.3.0"
+smol-timeout = "0.1.0"
+smallvec = "1.4.2"
+futures-timer = "3.0.2"
+
+sp-core = { git = "https://github.com/paritytech/substrate", branch = "master", features = ["std"] }
+sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master" }
+
+polkadot-subsystem-testhelpers = { package = "polkadot-node-subsystem-test-helpers", path = "../../subsystem-test-helpers" }

polkadot/node/network/collator-protocol/src/lib.rs

@@ -0,0 +1,136 @@
+// Copyright 2020 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/>.
+
+//! The Collator Protocol allows collators and validators talk to each other.
+//! This subsystem implements both sides of the collator protocol.
+
+#![deny(missing_docs)]
+
+use std::time::Duration;
+use futures::{channel::oneshot, FutureExt};
+use log::trace;
+
+use polkadot_subsystem::{
+	Subsystem, SubsystemContext, SubsystemError, SpawnedSubsystem,
+	errors::RuntimeApiError,
+	metrics::{self, prometheus},
+	messages::{
+		AllMessages, CollatorProtocolMessage, NetworkBridgeMessage,
+	},
+};
+use polkadot_node_network_protocol::{
+	PeerId, ReputationChange as Rep,
+};
+use polkadot_primitives::v1::CollatorId;
+use polkadot_node_subsystem_util as util;
+
+mod collator_side;
+mod validator_side;
+
+const TARGET: &'static str = "colp";
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(1);
+
+#[derive(Debug, derive_more::From)]
+enum Error {
+	#[from]
+	Subsystem(SubsystemError),
+	#[from]
+	Oneshot(oneshot::Canceled),
+	#[from]
+	RuntimeApi(RuntimeApiError),
+	#[from]
+	UtilError(util::Error),
+}
+
+type Result<T> = std::result::Result<T, Error>;
+
+enum ProtocolSide {
+	Validator,
+	Collator(CollatorId),
+}
+
+/// The collator protocol subsystem.
+pub struct CollatorProtocolSubsystem {
+	protocol_side: ProtocolSide, 
+}
+
+impl CollatorProtocolSubsystem {
+	/// Start the collator protocol.
+	/// If `id` is `Some` this is a collator side of the protocol.
+	/// If `id` is `None` this is a validator side of the protocol. 
+	pub fn new(id: Option<CollatorId>) -> Self {
+		let protocol_side = match id {
+			Some(id) => ProtocolSide::Collator(id),
+			None => ProtocolSide::Validator,
+		};
+
+		Self {
+			protocol_side,
+		}
+	}
+
+	async fn run<Context>(self, ctx: Context) -> Result<()>
+	where
+		Context: SubsystemContext<Message = CollatorProtocolMessage>,
+	{
+		match self.protocol_side {
+		    ProtocolSide::Validator => validator_side::run(ctx, REQUEST_TIMEOUT).await,
+		    ProtocolSide::Collator(id) => collator_side::run(ctx, id).await,
+		}
+	}
+}
+
+/// Collator protocol metrics.
+#[derive(Default, Clone)]
+pub struct Metrics;
+
+impl metrics::Metrics for Metrics {
+	fn try_register(_registry: &prometheus::Registry) 
+		-> std::result::Result<Self, prometheus::PrometheusError> {
+		Ok(Metrics)
+	}
+}
+
+impl<Context> Subsystem<Context> for CollatorProtocolSubsystem
+where
+	Context: SubsystemContext<Message = CollatorProtocolMessage> + Sync + Send,
+{
+	type Metrics = Metrics;
+
+	fn start(self, ctx: Context) -> SpawnedSubsystem {
+		SpawnedSubsystem {
+			name: "collator-protocol-subsystem",
+			future: Box::pin(async move { self.run(ctx) }.map(|_| ())),
+		}
+	}
+}
+
+/// Modify the reputation of a peer based on its behavior.
+async fn modify_reputation<Context>(ctx: &mut Context, peer: PeerId, rep: Rep) -> Result<()>
+where
+	Context: SubsystemContext<Message = CollatorProtocolMessage>,
+{
+	trace!(
+		target: TARGET,
+		"Reputation change of {:?} for peer {:?}", rep, peer,
+	);
+
+	ctx.send_message(AllMessages::NetworkBridge(
+		NetworkBridgeMessage::ReportPeer(peer, rep),
+	)).await?;
+
+	Ok(())
+}

polkadot/node/subsystem-test-helpers/Cargo.toml

@@ -16,6 +16,7 @@ parking_lot = "0.10.0"
 pin-project = "0.4.23"
 polkadot-node-primitives = { path = "../primitives" }
 polkadot-node-subsystem = { path = "../subsystem" }
+polkadot-node-subsystem-util = { path = "../subsystem-util" }
 polkadot-primitives = { path = "../../primitives" }
 polkadot-statement-table = { path = "../../statement-table" }
 sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }

polkadot/node/subsystem-test-helpers/src/lib.rs

@@ -18,13 +18,12 @@
 
 use polkadot_node_subsystem::messages::AllMessages;
 use polkadot_node_subsystem::{FromOverseer, SubsystemContext, SubsystemError, SubsystemResult};
+use polkadot_node_subsystem_util::TimeoutExt;
 
 use futures::channel::mpsc;
 use futures::poll;
 use futures::prelude::*;
-use futures_timer::Delay;
 use parking_lot::Mutex;
-use pin_project::pin_project;
 use sp_core::{testing::TaskExecutor, traits::SpawnNamed};
 
 use std::convert::Infallible;
@@ -284,45 +283,3 @@ pub fn subsystem_test_harness<M, OverseerFactory, Overseer, TestFactory, Test>(
 			.expect("test timed out instead of completing")
 	});
 }
-
-/// A future that wraps another future with a `Delay` allowing for time-limited futures.
-#[pin_project]
-pub struct Timeout<F: Future> {
-	#[pin]
-	future: F,
-	#[pin]
-	delay: Delay,
-}
-
-/// Extends `Future` to allow time-limited futures.
-pub trait TimeoutExt: Future {
-	fn timeout(self, duration: Duration) -> Timeout<Self>
-	where
-		Self: Sized,
-	{
-		Timeout {
-			future: self,
-			delay: Delay::new(duration),
-		}
-	}
-}
-
-impl<F: Future> TimeoutExt for F {}
-
-impl<F: Future> Future for Timeout<F> {
-	type Output = Option<F::Output>;
-
-	fn poll(self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Self::Output> {
-		let this = self.project();
-
-		if this.delay.poll(ctx).is_ready() {
-			return Poll::Ready(None);
-		}
-
-		if let Poll::Ready(output) = this.future.poll(ctx) {
-			return Poll::Ready(Some(output));
-		}
-
-		Poll::Pending
-	}
-}

polkadot/node/subsystem-util/src/lib.rs

@@ -50,6 +50,7 @@ use std::{
 	convert::{TryFrom, TryInto},
 	marker::Unpin,
 	pin::Pin,
+	task::{Poll, Context},
 	time::Duration,
 };
 use streamunordered::{StreamUnordered, StreamYield};
@@ -973,9 +974,51 @@ macro_rules! delegated_subsystem {
 	};
 }
 
+/// A future that wraps another future with a `Delay` allowing for time-limited futures.
+#[pin_project]
+pub struct Timeout<F: Future> {
+	#[pin]
+	future: F,
+	#[pin]
+	delay: Delay,
+}
+
+/// Extends `Future` to allow time-limited futures.
+pub trait TimeoutExt: Future {
+	fn timeout(self, duration: Duration) -> Timeout<Self>
+	where
+		Self: Sized,
+	{
+		Timeout {
+			future: self,
+			delay: Delay::new(duration),
+		}
+	}
+}
+
+impl<F: Future> TimeoutExt for F {}
+
+impl<F: Future> Future for Timeout<F> {
+	type Output = Option<F::Output>;
+
+	fn poll(self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Self::Output> {
+		let this = self.project();
+
+		if this.delay.poll(ctx).is_ready() {
+			return Poll::Ready(None);
+		}
+
+		if let Poll::Ready(output) = this.future.poll(ctx) {
+			return Poll::Ready(Some(output));
+		}
+
+		Poll::Pending
+	}
+}
+
 #[cfg(test)]
 mod tests {
-	use super::{Error as UtilError, JobManager, JobTrait, JobsError, ToJobTrait};
+	use super::{Error as UtilError, JobManager, JobTrait, JobsError, TimeoutExt, ToJobTrait};
 	use polkadot_node_subsystem::{
 		messages::{AllMessages, CandidateSelectionMessage},
 		ActiveLeavesUpdate, FromOverseer, OverseerSignal, SpawnedSubsystem, Subsystem,
@@ -988,7 +1031,7 @@ mod tests {
 		future, Future, FutureExt, SinkExt,
 	};
 	use polkadot_primitives::v1::Hash;
-	use polkadot_node_subsystem_test_helpers::{self as test_helpers, make_subsystem_context, TimeoutExt as _};
+	use polkadot_node_subsystem_test_helpers::{self as test_helpers, make_subsystem_context};
 	use std::{collections::HashMap, convert::TryFrom, pin::Pin, time::Duration};
 
 	// basic usage: in a nutshell, when you want to define a subsystem, just focus on what its jobs do;

polkadot/node/subsystem/src/messages.rs

@@ -160,7 +160,7 @@ pub enum CollatorProtocolMessage {
 	/// Provide a collation to distribute to validators.
 	DistributeCollation(CandidateReceipt, PoV),
 	/// Fetch a collation under the given relay-parent for the given ParaId.
-	FetchCollation(Hash, ParaId, oneshot::Sender<(CandidateReceipt, PoV)>),
+	FetchCollation(Hash, CollatorId, ParaId, oneshot::Sender<(CandidateReceipt, PoV)>),
 	/// Report a collator as having provided an invalid collation. This should lead to disconnect
 	/// and blacklist of the collator.
 	ReportCollator(CollatorId),
@@ -176,7 +176,7 @@ impl CollatorProtocolMessage {
 		match self {
 			Self::CollateOn(_) => None,
 			Self::DistributeCollation(receipt, _) => Some(receipt.descriptor().relay_parent),
-			Self::FetchCollation(relay_parent, _, _) => Some(*relay_parent),
+			Self::FetchCollation(relay_parent, _, _, _) => Some(*relay_parent),
 			Self::ReportCollator(_) => None,
 			Self::NoteGoodCollation(_) => None,
 			Self::NetworkBridgeUpdateV1(_) => None,