Merge branch 'master' into shawntabrizi-crowdsale

Cargo.toml

@@ -42,10 +42,12 @@ members = [
 	"service",
 	"validation",
 
-	"node/messages",
+	"node/network/bridge",
 	"node/overseer",
 	"node/primitives",
 	"node/service",
+	"node/subsystem",
+	"node/test-helpers/subsystem",
 
 	"parachain/test-parachains",
 	"parachain/test-parachains/adder",

network/test/src/block_import.rs

@@ -60,7 +60,7 @@ fn import_single_good_block_works() {
 	let mut expected_aux = ImportedAux::default();
 	expected_aux.is_new_best = true;
 
-	match import_single_block(&mut polkadot_test_runtime_client::new(), BlockOrigin::File, block, &mut PassThroughVerifier(true)) {
+	match import_single_block(&mut polkadot_test_runtime_client::new(), BlockOrigin::File, block, &mut PassThroughVerifier::new(true)) {
 		Ok(BlockImportResult::ImportedUnknown(ref num, ref aux, ref org))
 			if *num == number as u32 && *aux == expected_aux && *org == Some(peer_id) => {}
 		r @ _ => panic!("{:?}", r)
@@ -70,7 +70,7 @@ fn import_single_good_block_works() {
 #[test]
 fn import_single_good_known_block_is_ignored() {
 	let (mut client, _hash, number, _, block) = prepare_good_block();
-	match import_single_block(&mut client, BlockOrigin::File, block, &mut PassThroughVerifier(true)) {
+	match import_single_block(&mut client, BlockOrigin::File, block, &mut PassThroughVerifier::new(true)) {
 		Ok(BlockImportResult::ImportedKnown(ref n)) if *n == number as u32 => {}
 		_ => panic!()
 	}
@@ -80,7 +80,7 @@ fn import_single_good_known_block_is_ignored() {
 fn import_single_good_block_without_header_fails() {
 	let (_, _, _, peer_id, mut block) = prepare_good_block();
 	block.header = None;
-	match import_single_block(&mut polkadot_test_runtime_client::new(), BlockOrigin::File, block, &mut PassThroughVerifier(true)) {
+	match import_single_block(&mut polkadot_test_runtime_client::new(), BlockOrigin::File, block, &mut PassThroughVerifier::new(true)) {
 		Err(BlockImportError::IncompleteHeader(ref org)) if *org == Some(peer_id) => {}
 		_ => panic!()
 	}
@@ -91,7 +91,7 @@ fn async_import_queue_drops() {
 	let executor = sp_core::testing::SpawnBlockingExecutor::new();
 	// Perform this test multiple times since it exhibits non-deterministic behavior.
 	for _ in 0..100 {
-		let verifier = PassThroughVerifier(true);
+		let verifier = PassThroughVerifier::new(true);
 
 		let queue = BasicQueue::new(
 			verifier,

network/test/src/lib.rs

@@ -600,7 +600,7 @@ pub trait TestNetFactory: Sized {
 			transaction_pool: Arc::new(EmptyTransactionPool),
 			protocol_id: ProtocolId::from(&b"test-protocol-name"[..]),
 			import_queue,
-			block_announce_validator: Box::new(DefaultBlockAnnounceValidator::new(client.clone())),
+			block_announce_validator: Box::new(DefaultBlockAnnounceValidator),
 			metrics_registry: None,
 		}).unwrap();
 
@@ -677,7 +677,7 @@ pub trait TestNetFactory: Sized {
 			transaction_pool: Arc::new(EmptyTransactionPool),
 			protocol_id: ProtocolId::from(&b"test-protocol-name"[..]),
 			import_queue,
-			block_announce_validator: Box::new(DefaultBlockAnnounceValidator::new(client.clone())),
+			block_announce_validator: Box::new(DefaultBlockAnnounceValidator),
 			metrics_registry: None,
 		}).unwrap();
 
@@ -804,7 +804,7 @@ impl TestNetFactory for TestNet {
 	fn make_verifier(&self, _client: PeersClient, _config: &ProtocolConfig, _peer_data: &())
 		-> Self::Verifier
 	{
-		PassThroughVerifier(false)
+		PassThroughVerifier::new(false)
 	}
 
 	fn peer(&mut self, i: usize) -> &mut Peer<()> {

node/network/bridge/Cargo.toml

+[package]
+name = "polkadot-network-bridge"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+edition = "2018"
+
+[dependencies]
+futures = "0.3.5"
+log = "0.4.8"
+futures-timer = "3.0.2"
+streamunordered = "0.5.1"
+polkadot-primitives = { path = "../../../primitives" }
+node-primitives = { package = "polkadot-node-primitives", path = "../../primitives" }
+parity-scale-codec = "1.3.0"
+sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
+sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" }
+polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../../subsystem" }
+
+[dev-dependencies]
+parking_lot = "0.10.0"
+subsystem-test = { package = "polkadot-subsystem-test-helpers", path = "../../test-helpers/subsystem" }
+assert_matches = "1.3.0"

node/overseer/Cargo.toml

@@ -11,7 +11,8 @@ futures-timer = "3.0.2"
 streamunordered = "0.5.1"
 polkadot-primitives = { path = "../../primitives" }
 client = { package = "sc-client-api", git = "https://github.com/paritytech/substrate", branch = "master" }
-messages = { package = "polkadot-node-messages", path = "../messages" }
+polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../subsystem" }
+async-trait = "0.1"
 
 [dev-dependencies]
 futures = { version = "0.3.5", features = ["thread-pool"] }

node/overseer/examples/minimal-example.rs

@@ -28,16 +28,17 @@ use futures_timer::Delay;
 use kv_log_macro as log;
 
 use polkadot_primitives::parachain::{BlockData, PoVBlock};
-use polkadot_overseer::{Overseer, Subsystem, SubsystemContext, SpawnedSubsystem};
+use polkadot_overseer::Overseer;
 
-use messages::{
-	AllMessages, CandidateBackingMessage, FromOverseer, CandidateValidationMessage
+use polkadot_subsystem::{Subsystem, SubsystemContext, SpawnedSubsystem, FromOverseer};
+use polkadot_subsystem::messages::{
+	AllMessages, CandidateBackingMessage, CandidateValidationMessage
 };
 
 struct Subsystem1;
 
 impl Subsystem1 {
-	async fn run(mut ctx: SubsystemContext<CandidateBackingMessage>)  {
+	async fn run(mut ctx: impl SubsystemContext<Message=CandidateBackingMessage>)  {
 		loop {
 			match ctx.try_recv().await {
 				Ok(Some(msg)) => {
@@ -56,7 +57,7 @@ impl Subsystem1 {
 			Delay::new(Duration::from_secs(1)).await;
 			let (tx, _) = oneshot::channel();
 
-			ctx.send_msg(AllMessages::CandidateValidation(
+			ctx.send_message(AllMessages::CandidateValidation(
 				CandidateValidationMessage::Validate(
 					Default::default(),
 					Default::default(),
@@ -70,8 +71,10 @@ impl Subsystem1 {
 	}
 }
 
-impl Subsystem<CandidateBackingMessage> for Subsystem1 {
-	fn start(&mut self, ctx: SubsystemContext<CandidateBackingMessage>) -> SpawnedSubsystem {
+impl<C> Subsystem<C> for Subsystem1
+	where C: SubsystemContext<Message=CandidateBackingMessage>
+{
+	fn start(&mut self, ctx: C) -> SpawnedSubsystem {
 		SpawnedSubsystem(Box::pin(async move {
 			Self::run(ctx).await;
 		}))
@@ -81,7 +84,7 @@ impl Subsystem<CandidateBackingMessage> for Subsystem1 {
 struct Subsystem2;
 
 impl Subsystem2 {
-	async fn run(mut ctx: SubsystemContext<CandidateValidationMessage>)  {
+	async fn run(mut ctx: impl SubsystemContext<Message=CandidateValidationMessage>)  {
 		ctx.spawn(Box::pin(async {
 			loop {
 				log::info!("Job tick");
@@ -105,8 +108,10 @@ impl Subsystem2 {
 	}
 }
 
-impl Subsystem<CandidateValidationMessage> for Subsystem2 {
-	fn start(&mut self, ctx: SubsystemContext<CandidateValidationMessage>) -> SpawnedSubsystem {
+impl<C> Subsystem<C> for Subsystem2
+	where C: SubsystemContext<Message=CandidateValidationMessage>
+{
+	fn start(&mut self, ctx: C) -> SpawnedSubsystem {
 		SpawnedSubsystem(Box::pin(async move {
 			Self::run(ctx).await;
 		}))

node/overseer/src/lib.rs

@@ -65,8 +65,8 @@ use futures::channel::{mpsc, oneshot};
 use futures::{
 	pending, poll, select,
 	future::{BoxFuture, RemoteHandle},
-	stream::FuturesUnordered,
-	task::{Spawn, SpawnError, SpawnExt},
+	stream::{self, FuturesUnordered},
+	task::{Spawn, SpawnExt},
 	Future, FutureExt, SinkExt, StreamExt,
 };
 use futures_timer::Delay;
@@ -75,50 +75,14 @@ use streamunordered::{StreamYield, StreamUnordered};
 use polkadot_primitives::{Block, BlockNumber, Hash};
 use client::{BlockImportNotification, BlockchainEvents, FinalityNotification};
 
-pub use messages::{
-	OverseerSignal, CandidateValidationMessage, CandidateBackingMessage, AllMessages,
-	FromOverseer,
+use polkadot_subsystem::messages::{
+	CandidateValidationMessage, CandidateBackingMessage, AllMessages
+};
+pub use polkadot_subsystem::{
+	Subsystem, SubsystemContext, OverseerSignal, FromOverseer, SubsystemError, SubsystemResult,
+	SpawnedSubsystem,
 };
 
-/// An error type that describes faults that may happen
-///
-/// These are:
-///   * Channels being closed
-///   * Subsystems dying when they are not expected to
-///   * Subsystems not dying when they are told to die
-///   * etc.
-#[derive(Debug)]
-pub struct SubsystemError;
-
-impl From<mpsc::SendError> for SubsystemError {
-	fn from(_: mpsc::SendError) -> Self {
-		Self
-	}
-}
-
-impl From<oneshot::Canceled> for SubsystemError {
-	fn from(_: oneshot::Canceled) -> Self {
-		Self
-	}
-}
-
-impl From<SpawnError> for SubsystemError {
-    fn from(_: SpawnError) -> Self {
-		Self
-    }
-}
-
-/// A `Result` type that wraps [`SubsystemError`].
-///
-/// [`SubsystemError`]: struct.SubsystemError.html
-pub type SubsystemResult<T> = Result<T, SubsystemError>;
-
-/// An asynchronous subsystem task that runs inside and being overseen by the [`Overseer`].
-///
-/// In essence it's just a newtype wrapping a `BoxFuture`.
-///
-/// [`Overseer`]: struct.Overseer.html
-pub struct SpawnedSubsystem(pub BoxFuture<'static, ()>);
 
 // A capacity of bounded channels inside the overseer.
 const CHANNEL_CAPACITY: usize = 1024;
@@ -278,7 +242,7 @@ impl Debug for ToOverseer {
 /// A running instance of some [`Subsystem`].
 ///
 /// [`Subsystem`]: trait.Subsystem.html
-struct SubsystemInstance<M: Debug> {
+struct SubsystemInstance<M> {
 	tx: mpsc::Sender<FromOverseer<M>>,
 }
 
@@ -289,17 +253,17 @@ struct SubsystemInstance<M: Debug> {
 /// [`Overseer`]: struct.Overseer.html
 /// [`Subsystem`]: trait.Subsystem.html
 /// [`SubsystemJob`]: trait.SubsystemJob.html
-pub struct SubsystemContext<M: Debug>{
+#[derive(Debug)]
+pub struct OverseerSubsystemContext<M>{
 	rx: mpsc::Receiver<FromOverseer<M>>,
 	tx: mpsc::Sender<ToOverseer>,
 }
 
-impl<M: Debug> SubsystemContext<M> {
-	/// Try to asyncronously receive a message.
-	///
-	/// This has to be used with caution, if you loop over this without
-	/// using `pending!()` macro you will end up with a busy loop!
-	pub async fn try_recv(&mut self) -> Result<Option<FromOverseer<M>>, ()> {
+#[async_trait::async_trait]
+impl<M: Send + 'static> SubsystemContext for OverseerSubsystemContext<M> {
+	type Message = M;
+
+	async fn try_recv(&mut self) -> Result<Option<FromOverseer<M>>, ()> {
 		match poll!(self.rx.next()) {
 			Poll::Ready(Some(msg)) => Ok(Some(msg)),
 			Poll::Ready(None) => Err(()),
@@ -307,13 +271,11 @@ impl<M: Debug> SubsystemContext<M> {
 		}
 	}
 
-	/// Receive a message.
-	pub async fn recv(&mut self) -> SubsystemResult<FromOverseer<M>> {
+	async fn recv(&mut self) -> SubsystemResult<FromOverseer<M>> {
 		self.rx.next().await.ok_or(SubsystemError)
 	}
 
-	/// Spawn a child task on the executor.
-	pub async fn spawn(&mut self, s: Pin<Box<dyn Future<Output = ()> + Send>>) -> SubsystemResult<()> {
+	async fn spawn(&mut self, s: Pin<Box<dyn Future<Output = ()> + Send>>) -> SubsystemResult<()> {
 		let (tx, rx) = oneshot::channel();
 		self.tx.send(ToOverseer::SpawnJob {
 			s,
@@ -323,33 +285,25 @@ impl<M: Debug> SubsystemContext<M> {
 		rx.await?
 	}
 
-	/// Send a direct message to some other `Subsystem`, routed based on message type.
-	pub async fn send_msg(&mut self, msg: AllMessages) -> SubsystemResult<()> {
+	async fn send_message(&mut self, msg: AllMessages) -> SubsystemResult<()> {
 		self.tx.send(ToOverseer::SubsystemMessage(msg)).await?;
 
 		Ok(())
 	}
 
-	fn new(rx: mpsc::Receiver<FromOverseer<M>>, tx: mpsc::Sender<ToOverseer>) -> Self {
-		Self {
-			rx,
-			tx,
-		}
+	async fn send_messages<T>(&mut self, msgs: T) -> SubsystemResult<()>
+		where T: IntoIterator<Item = AllMessages> + Send, T::IntoIter: Send
+	{
+		let mut msgs = stream::iter(msgs.into_iter().map(ToOverseer::SubsystemMessage).map(Ok));
+		self.tx.send_all(&mut msgs).await?;
+
+		Ok(())
 	}
 }
 
-/// A trait that describes the [`Subsystem`]s that can run on the [`Overseer`].
-///
-/// It is generic over the message type circulating in the system.
-/// The idea that we want some type contaning persistent state that
-/// can spawn actually running subsystems when asked to.
-///
-/// [`Overseer`]: struct.Overseer.html
-/// [`Subsystem`]: trait.Subsystem.html
-pub trait Subsystem<M: Debug> {
-	/// Start this `Subsystem` and return `SpawnedSubsystem`.
-	fn start(&mut self, ctx: SubsystemContext<M>) -> SpawnedSubsystem;
-}
+/// A subsystem compatible with the overseer - one which can be run in the context of the
+/// overseer.
+pub type CompatibleSubsystem<M> = Box<dyn Subsystem<OverseerSubsystemContext<M>> + Send>;
 
 /// A subsystem that we oversee.
 ///
@@ -359,8 +313,8 @@ pub trait Subsystem<M: Debug> {
 ///
 /// [`Subsystem`]: trait.Subsystem.html
 #[allow(dead_code)]
-struct OverseenSubsystem<M: Debug> {
-	subsystem: Box<dyn Subsystem<M> + Send>,
+struct OverseenSubsystem<M> {
+	subsystem: CompatibleSubsystem<M>,
 	instance: Option<SubsystemInstance<M>>,
 }
 
@@ -441,16 +395,20 @@ where
 	/// # use std::time::Duration;
 	/// # use futures::{executor, pin_mut, select, FutureExt};
 	/// # use futures_timer::Delay;
-	/// # use polkadot_overseer::{
-	/// #     Overseer, Subsystem, SpawnedSubsystem, SubsystemContext,
-	/// #     CandidateValidationMessage, CandidateBackingMessage,
+	/// # use polkadot_overseer::Overseer;
+	/// # use polkadot_subsystem::{
+	/// #     Subsystem, SpawnedSubsystem, SubsystemContext,
+	/// #     messages::{CandidateValidationMessage, CandidateBackingMessage},
 	/// # };
 	///
 	/// struct ValidationSubsystem;
-	/// impl Subsystem<CandidateValidationMessage> for ValidationSubsystem {
+	///
+	/// impl<C> Subsystem<C> for ValidationSubsystem
+	/// 	where C: SubsystemContext<Message=CandidateValidationMessage>
+	/// {
 	///     fn start(
 	///         &mut self,
-	///         mut ctx: SubsystemContext<CandidateValidationMessage>,
+	///         mut ctx: C,
 	///     ) -> SpawnedSubsystem {
 	///         SpawnedSubsystem(Box::pin(async move {
 	///             loop {
@@ -461,10 +419,12 @@ where
 	/// }
 	///
 	/// struct CandidateBackingSubsystem;
-	/// impl Subsystem<CandidateBackingMessage> for CandidateBackingSubsystem {
+	/// impl<C> Subsystem<C> for CandidateBackingSubsystem
+	/// 	where C: SubsystemContext<Message=CandidateBackingMessage>
+	/// {
 	///     fn start(
 	///         &mut self,
-	///         mut ctx: SubsystemContext<CandidateBackingMessage>,
+	///         mut ctx: C,
 	///     ) -> SpawnedSubsystem {
 	///         SpawnedSubsystem(Box::pin(async move {
 	///             loop {
@@ -498,8 +458,8 @@ where
 	/// ```
 	pub fn new(
 		leaves: impl IntoIterator<Item = BlockInfo>,
-		validation: Box<dyn Subsystem<CandidateValidationMessage> + Send>,
-		candidate_backing: Box<dyn Subsystem<CandidateBackingMessage> + Send>,
+		validation: CompatibleSubsystem<CandidateValidationMessage>,
+		candidate_backing: CompatibleSubsystem<CandidateBackingMessage>,
 		mut s: S,
 	) -> SubsystemResult<(Self, OverseerHandler)> {
 		let (events_tx, events_rx) = mpsc::channel(CHANNEL_CAPACITY);
@@ -680,6 +640,12 @@ where
 					let _ = s.tx.send(FromOverseer::Communication { msg }).await;
 				}
 			}
+			_ => {
+				// TODO: temporary catch-all until all subsystems are integrated with overseer.
+				// The overseer is not complete until this is an exhaustive match with all
+				// messages targeting an included subsystem.
+				// https://github.com/paritytech/polkadot/issues/1317
+			}
 		}
 	}
 
@@ -688,15 +654,15 @@ where
 	}
 }
 
-fn spawn<S: Spawn, M: Debug>(
+fn spawn<S: Spawn, M: Send + 'static>(
 	spawner: &mut S,
 	futures: &mut FuturesUnordered<RemoteHandle<()>>,
 	streams: &mut StreamUnordered<mpsc::Receiver<ToOverseer>>,
-	mut s: Box<dyn Subsystem<M> + Send>,
+	mut s: CompatibleSubsystem<M>,
 ) -> SubsystemResult<OverseenSubsystem<M>> {
 	let (to_tx, to_rx) = mpsc::channel(CHANNEL_CAPACITY);
 	let (from_tx, from_rx) = mpsc::channel(CHANNEL_CAPACITY);
-	let ctx = SubsystemContext::new(to_rx, from_tx);
+	let ctx = OverseerSubsystemContext { rx: to_rx, tx: from_tx };
 	let f = s.start(ctx);
 
 	let handle = spawner.spawn_with_handle(f.0)?;
@@ -723,8 +689,10 @@ mod tests {
 
 	struct TestSubsystem1(mpsc::Sender<usize>);
 
-	impl Subsystem<CandidateValidationMessage> for TestSubsystem1 {
-		fn start(&mut self, mut ctx: SubsystemContext<CandidateValidationMessage>) -> SpawnedSubsystem {
+	impl<C> Subsystem<C> for TestSubsystem1
+		where C: SubsystemContext<Message=CandidateValidationMessage>
+	{
+		fn start(&mut self, mut ctx: C) -> SpawnedSubsystem {
 			let mut sender = self.0.clone();
 			SpawnedSubsystem(Box::pin(async move {
 				let mut i = 0;
@@ -746,14 +714,16 @@ mod tests {
 
 	struct TestSubsystem2(mpsc::Sender<usize>);
 
-	impl Subsystem<CandidateBackingMessage> for TestSubsystem2 {
-		fn start(&mut self, mut ctx: SubsystemContext<CandidateBackingMessage>) -> SpawnedSubsystem {
+	impl<C> Subsystem<C> for TestSubsystem2
+		where C: SubsystemContext<Message=CandidateBackingMessage>
+	{
+		fn start(&mut self, mut ctx: C) -> SpawnedSubsystem {
 			SpawnedSubsystem(Box::pin(async move {
 				let mut c: usize = 0;
 				loop {
 					if c < 10 {
 						let (tx, _) = oneshot::channel();
-						ctx.send_msg(
+						ctx.send_message(
 							AllMessages::CandidateValidation(
 								CandidateValidationMessage::Validate(
 									Default::default(),
@@ -786,8 +756,10 @@ mod tests {
 
 	struct TestSubsystem4;
 
-	impl Subsystem<CandidateBackingMessage> for TestSubsystem4 {
-		fn start(&mut self, mut _ctx: SubsystemContext<CandidateBackingMessage>) -> SpawnedSubsystem {
+	impl<C> Subsystem<C> for TestSubsystem4
+		where C: SubsystemContext<Message=CandidateBackingMessage>
+	{
+		fn start(&mut self, mut _ctx: C) -> SpawnedSubsystem {
 			SpawnedSubsystem(Box::pin(async move {
 				// Do nothing and exit.
 			}))
@@ -871,8 +843,10 @@ mod tests {
 
 	struct TestSubsystem5(mpsc::Sender<OverseerSignal>);
 
-	impl Subsystem<CandidateValidationMessage> for TestSubsystem5 {
-		fn start(&mut self, mut ctx: SubsystemContext<CandidateValidationMessage>) -> SpawnedSubsystem {
+	impl<C> Subsystem<C> for TestSubsystem5
+		where C: SubsystemContext<Message=CandidateValidationMessage>
+	{
+		fn start(&mut self, mut ctx: C) -> SpawnedSubsystem {
 			let mut sender = self.0.clone();
 
 			SpawnedSubsystem(Box::pin(async move {
@@ -895,8 +869,10 @@ mod tests {
 
 	struct TestSubsystem6(mpsc::Sender<OverseerSignal>);
 
-	impl Subsystem<CandidateBackingMessage> for TestSubsystem6 {
-		fn start(&mut self, mut ctx: SubsystemContext<CandidateBackingMessage>) -> SpawnedSubsystem {
+	impl<C> Subsystem<C> for TestSubsystem6
+		where C: SubsystemContext<Message=CandidateBackingMessage>
+	{
+		fn start(&mut self, mut ctx: C) -> SpawnedSubsystem {
 			let mut sender = self.0.clone();
 
 			SpawnedSubsystem(Box::pin(async move {

node/primitives/Cargo.toml

@@ -10,3 +10,4 @@ polkadot-primitives = { path = "../../primitives" }
 polkadot-statement-table = { path = "../../statement-table" }
 parity-scale-codec = { version = "1.3.0", default-features = false, features = ["derive"] }
 runtime_primitives = { package = "sp-runtime", git = "https://github.com/paritytech/substrate", branch = "master", default-features = false }
+async-trait = "0.1"

node/primitives/src/lib.rs

@@ -64,6 +64,7 @@ impl EncodeAs<CompactStatement> for Statement {
 pub type SignedFullStatement = Signed<Statement, CompactStatement>;
 
 /// A misbehaviour report.
+#[derive(Debug)]
 pub enum MisbehaviorReport {
 	/// These validator nodes disagree on this candidate's validity, please figure it out
 	///
@@ -79,3 +80,12 @@ pub enum MisbehaviorReport {
 	/// This peer has seconded more than one parachain candidate for this relay parent head
 	DoubleVote(CandidateReceipt, SignedFullStatement, SignedFullStatement),
 }
+
+/// A unique identifier for a network protocol.
+pub type ProtocolId = [u8; 4];
+
+/// A succinct representation of a peer's view. This consists of a bounded amount of chain heads.
+///
+/// Up to `N` (5?) chain heads.
+#[derive(Debug, Clone, PartialEq, Eq, Encode, Decode)]
+pub struct View(pub Vec<Hash>);

node/service/Cargo.toml

@@ -15,6 +15,7 @@ hex-literal = "0.2.1"
 polkadot-primitives = { path = "../../primitives" }
 polkadot-runtime = { path = "../../runtime/polkadot" }
 polkadot-overseer = { path = "../overseer" }
+polkadot-subsystem = { package = "polkadot-node-subsystem", path = "../subsystem" }
 kusama-runtime = { path = "../../runtime/kusama" }
 westend-runtime = { path = "../../runtime/westend" }
 polkadot-network = { path = "../../network", optional = true }

node/service/src/lib.rs

@@ -29,10 +29,10 @@ use grandpa::{self, FinalityProofProvider as GrandpaFinalityProofProvider};
 use sc_executor::native_executor_instance;
 use log::info;
 use sp_blockchain::HeaderBackend;
-use polkadot_overseer::{
-	self as overseer,
-	BlockInfo, Overseer, OverseerHandler, Subsystem, SubsystemContext, SpawnedSubsystem,
-	CandidateValidationMessage, CandidateBackingMessage,
+use polkadot_overseer::{self as overseer, BlockInfo, Overseer, OverseerHandler};
+use polkadot_subsystem::{
+	Subsystem, SubsystemContext, SpawnedSubsystem,
+	messages::{CandidateValidationMessage, CandidateBackingMessage},
 };
 pub use service::{
 	Role, PruningMode, TransactionPoolOptions, Error, RuntimeGenesis,
@@ -269,8 +269,10 @@ macro_rules! new_full_start {
 
 struct CandidateValidationSubsystem;
 
-impl Subsystem<CandidateValidationMessage> for CandidateValidationSubsystem {
-	fn start(&mut self, mut ctx: SubsystemContext<CandidateValidationMessage>) -> SpawnedSubsystem {
+impl<C> Subsystem<C> for CandidateValidationSubsystem
+	where C: SubsystemContext<Message = CandidateValidationMessage>
+{
+	fn start(&mut self, mut ctx: C) -> SpawnedSubsystem {
 		SpawnedSubsystem(Box::pin(async move {
 			while let Ok(_) = ctx.recv().await {}
 		}))
@@ -279,8 +281,10 @@ impl Subsystem<CandidateValidationMessage> for CandidateValidationSubsystem {
 
 struct CandidateBackingSubsystem;
 
-impl Subsystem<CandidateBackingMessage> for CandidateBackingSubsystem {
-	fn start(&mut self, mut ctx: SubsystemContext<CandidateBackingMessage>) -> SpawnedSubsystem {
+impl<C> Subsystem<C> for CandidateBackingSubsystem
+	where C: SubsystemContext<Message = CandidateBackingMessage>
+{
+	fn start(&mut self, mut ctx: C) -> SpawnedSubsystem {
 		SpawnedSubsystem(Box::pin(async move {
 			while let Ok(_) = ctx.recv().await {}
 		}))

node/messages/Cargo.toml → node/subsystem/Cargo.toml

 [package]
-name = "polkadot-node-messages"
+name = "polkadot-node-subsystem"
 version = "0.1.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 edition = "2018"
-description = "Message types used by Subsystems"
+description = "Subsystem traits and message definitions"
 
 [dependencies]
 polkadot-primitives = { path = "../../primitives" }
@@ -11,3 +11,4 @@ polkadot-statement-table = { path = "../../statement-table" }
 polkadot-node-primitives = { path = "../primitives" }
 sc-network = { git = "https://github.com/paritytech/substrate", branch = "master" }
 futures = "0.3.5"
+async-trait = "0.1"

node/subsystem/src/lib.rs

+// Copyright 2017-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