Interchain message-passing (#117)

* compute ingress and routing in polkadot runtime

* extract parent candidates from block when beginning consensus

* fetch incoming messages when validating

* fix consensus tests

* parachain wasm execution uses messages

* update parachain tests to check if messages are executed

* abstract out network service to make room for network tests

* skeleton for incoming data fetch

* collate ingress from consensus-gossip

* keep track of validated candidates in the shared-table

* add some shared_table tests for new behavior

* broadcast egress messages on gossip

* test compute_ingress

* move network tests to module folder

* dummy network for consensus-network tests

* make consensus network generic over executor

* test egress broadcast and ingress fetch

* fix test compilation

* address some grumbles

* address grumbles and fix parachain shuffle

* remove broadcast parameter from consensus network trait

polkadot/network/src/consensus.rs

@@ -20,13 +20,14 @@
 //! each time consensus begins on a new chain head.
 
 use sr_primitives::traits::ProvideRuntimeApi;
-use substrate_network::consensus_gossip::ConsensusMessage;
-use polkadot_consensus::{Network, SharedTable, Collators, Statement, GenericStatement};
+use substrate_network::{consensus_gossip::ConsensusMessage, Context as NetContext};
+use polkadot_consensus::{Network as ParachainNetwork, SharedTable, Collators, Statement, GenericStatement};
 use polkadot_primitives::{AccountId, Block, Hash, SessionKey};
 use polkadot_primitives::parachain::{Id as ParaId, Collation, Extrinsic, ParachainHost, BlockData};
 use codec::Decode;
 
 use futures::prelude::*;
+use futures::future::Executor as FutureExecutor;
 use futures::sync::mpsc;
 
 use std::collections::HashMap;
@@ -36,22 +37,95 @@ use arrayvec::ArrayVec;
 use tokio::runtime::TaskExecutor;
 use parking_lot::Mutex;
 
-use super::NetworkService;
 use router::Router;
+use super::PolkadotProtocol;
+
+/// An executor suitable for dispatching async consensus tasks.
+pub trait Executor {
+	fn spawn<F: Future<Item=(),Error=()> + Send + 'static>(&self, f: F);
+}
+
+/// A wrapped futures::future::Executor.
+pub struct WrappedExecutor<T>(pub T);
+
+impl<T> Executor for WrappedExecutor<T>
+	where T: FutureExecutor<Box<Future<Item=(),Error=()> + Send + 'static>>
+{
+	fn spawn<F: Future<Item=(),Error=()> + Send + 'static>(&self, f: F) {
+		if let Err(e) = self.0.execute(Box::new(f)) {
+			warn!(target: "consensus", "could not spawn consensus task: {:?}", e);
+		}
+	}
+}
+
+impl Executor for TaskExecutor {
+	fn spawn<F: Future<Item=(),Error=()> + Send + 'static>(&self, f: F) {
+		TaskExecutor::spawn(self, f)
+	}
+}
+
+/// Basic functionality that a network has to fulfill.
+pub trait NetworkService: Send + Sync + 'static {
+	/// Get a stream of gossip messages for a given hash.
+	fn gossip_messages_for(&self, topic: Hash) -> mpsc::UnboundedReceiver<ConsensusMessage>;
+
+	/// Gossip a message on given topic.
+	fn gossip_message(&self, topic: Hash, message: Vec<u8>);
+
+	/// Drop a gossip topic.
+	fn drop_gossip(&self, topic: Hash);
+
+	/// Execute a closure with the polkadot protocol.
+	fn with_spec<F: Send + 'static>(&self, with: F)
+		where F: FnOnce(&mut PolkadotProtocol, &mut NetContext<Block>);
+}
+
+impl NetworkService for super::NetworkService {
+	fn gossip_messages_for(&self, topic: Hash) -> mpsc::UnboundedReceiver<ConsensusMessage> {
+		let (tx, rx) = std::sync::mpsc::channel();
+
+		self.with_gossip(move |gossip, _| {
+			let inner_rx = gossip.messages_for(topic);
+			let _ = tx.send(inner_rx);
+		});
+
+		match rx.recv() {
+			Ok(rx) => rx,
+			Err(_) => mpsc::unbounded().1, // return empty channel.
+		}
+	}
+
+	fn gossip_message(&self, topic: Hash, message: Vec<u8>) {
+		self.gossip_consensus_message(topic, message, false);
+	}
+
+	fn drop_gossip(&self, topic: Hash) {
+		self.with_gossip(move |gossip, _| {
+			gossip.collect_garbage_for_topic(topic);
+		})
+	}
+
+	fn with_spec<F: Send + 'static>(&self, with: F)
+		where F: FnOnce(&mut PolkadotProtocol, &mut NetContext<Block>)
+	{
+		super::NetworkService::with_spec(self, with)
+	}
+}
 
 // task that processes all gossipped consensus messages,
 // checking signatures
-struct MessageProcessTask<P, E> {
+struct MessageProcessTask<P, E, N: NetworkService, T> {
 	inner_stream: mpsc::UnboundedReceiver<ConsensusMessage>,
 	parent_hash: Hash,
-	table_router: Router<P>,
-	exit: E,
+	table_router: Router<P, E, N, T>,
 }
 
-impl<P, E> MessageProcessTask<P, E> where
+impl<P, E, N, T> MessageProcessTask<P, E, N, T> where
 	P: ProvideRuntimeApi + Send + Sync + 'static,
 	P::Api: ParachainHost<Block>,
 	E: Future<Item=(),Error=()> + Clone + Send + 'static,
+	N: NetworkService,
+	T: Clone + Executor + Send + 'static,
 {
 	fn process_message(&self, msg: ConsensusMessage) -> Option<Async<()>> {
 		use polkadot_consensus::SignedStatement;
@@ -64,7 +138,7 @@ impl<P, E> MessageProcessTask<P, E> where
 				statement.sender,
 				&self.parent_hash
 			) {
-				self.table_router.import_statement(statement, self.exit.clone());
+				self.table_router.import_statement(statement);
 			}
 		}
 
@@ -72,10 +146,12 @@ impl<P, E> MessageProcessTask<P, E> where
 	}
 }
 
-impl<P, E> Future for MessageProcessTask<P, E> where
+impl<P, E, N, T> Future for MessageProcessTask<P, E, N, T> where
 	P: ProvideRuntimeApi + Send + Sync + 'static,
 	P::Api: ParachainHost<Block>,
 	E: Future<Item=(),Error=()> + Clone + Send + 'static,
+	N: NetworkService,
+	T: Clone + Executor + Send + 'static,
 {
 	type Item = ();
 	type Error = ();
@@ -95,43 +171,45 @@ impl<P, E> Future for MessageProcessTask<P, E> where
 }
 
 /// Wrapper around the network service
-pub struct ConsensusNetwork<P, E> {
-	network: Arc<NetworkService>,
+pub struct ConsensusNetwork<P, E, N, T> {
+	network: Arc<N>,
 	api: Arc<P>,
+	executor: T,
 	exit: E,
 }
 
-impl<P, E> ConsensusNetwork<P, E> {
+impl<P, E, N, T> ConsensusNetwork<P, E, N, T> {
 	/// Create a new consensus networking object.
-	pub fn new(network: Arc<NetworkService>, exit: E, api: Arc<P>) -> Self {
-		ConsensusNetwork { network, exit, api }
+	pub fn new(network: Arc<N>, exit: E, api: Arc<P>, executor: T) -> Self {
+		ConsensusNetwork { network, exit, api, executor }
 	}
 }
 
-impl<P, E: Clone> Clone for ConsensusNetwork<P, E> {
+impl<P, E: Clone, N, T: Clone> Clone for ConsensusNetwork<P, E, N, T> {
 	fn clone(&self) -> Self {
 		ConsensusNetwork {
 			network: self.network.clone(),
 			exit: self.exit.clone(),
 			api: self.api.clone(),
+			executor: self.executor.clone(),
 		}
 	}
 }
 
 /// A long-lived network which can create parachain statement  routing processes on demand.
-impl<P, E> Network for ConsensusNetwork<P,E> where
+impl<P, E, N, T> ParachainNetwork for ConsensusNetwork<P, E, N, T> where
 	P: ProvideRuntimeApi + Send + Sync + 'static,
 	P::Api: ParachainHost<Block>,
 	E: Clone + Future<Item=(),Error=()> + Send + 'static,
+	N: NetworkService,
+	T: Clone + Executor + Send + 'static,
 {
-	type TableRouter = Router<P>;
+	type TableRouter = Router<P, E, N, T>;
 
-	/// Instantiate a table router using the given shared table.
 	fn communication_for(
 		&self,
-		_validators: &[SessionKey],
 		table: Arc<SharedTable>,
-		task_executor: TaskExecutor,
+		outgoing: polkadot_consensus::Outgoing,
 	) -> Self::TableRouter {
 		let parent_hash = table.consensus_parent_hash().clone();
 
@@ -142,41 +220,34 @@ impl<P, E> Network for ConsensusNetwork<P,E> where
 			table,
 			self.network.clone(),
 			self.api.clone(),
-			task_executor.clone(),
+			self.executor.clone(),
 			parent_hash,
 			knowledge.clone(),
+			self.exit.clone(),
 		);
 
-		let attestation_topic = table_router.gossip_topic();
-		let exit = self.exit.clone();
+		table_router.broadcast_egress(outgoing);
 
-		let (tx, rx) = std::sync::mpsc::channel();
-		self.network.with_gossip(move |gossip, _| {
-			let inner_rx = gossip.messages_for(attestation_topic);
-			let _ = tx.send(inner_rx);
-		});
+		let attestation_topic = table_router.gossip_topic();
 
 		let table_router_clone = table_router.clone();
-		let executor = task_executor.clone();
+		let executor = self.executor.clone();
 
 		// spin up a task in the background that processes all incoming statements
 		// TODO: propagate statements on a timer?
+		let inner_stream = self.network.gossip_messages_for(attestation_topic);
 		self.network
 			.with_spec(move |spec, ctx| {
 				spec.new_consensus(ctx, parent_hash, CurrentConsensus {
 					knowledge,
 					local_session_key,
 				});
-				let inner_stream = match rx.try_recv() {
-					Ok(inner_stream) => inner_stream,
-					_ => unreachable!("1. The with_gossip closure executed first, 2. the reply should be available")
-				};
 				let process_task = MessageProcessTask {
 					inner_stream,
 					parent_hash,
 					table_router: table_router_clone,
-					exit,
 				};
+
 				executor.spawn(process_task);
 		});
 
@@ -213,8 +284,10 @@ impl Future for AwaitingCollation {
 	}
 }
 
-impl<P: ProvideRuntimeApi + Send + Sync + 'static, E: Clone> Collators for ConsensusNetwork<P, E>
-	where P::Api: ParachainHost<Block>,
+impl<P, E: Clone, N, T: Clone> Collators for ConsensusNetwork<P, E, N, T> where
+	P: ProvideRuntimeApi + Send + Sync + 'static,
+	P::Api: ParachainHost<Block>,
+	N: NetworkService,
 {
 	type Error = NetworkDown;
 	type Collation = AwaitingCollation;

polkadot/network/src/lib.rs

@@ -16,9 +16,8 @@
 
 //! Polkadot-specific network implementation.
 //!
-//! This manages gossip of consensus messages for BFT and for parachain statements,
-//! parachain block and extrinsic data fetching, communication between collators and validators,
-//! and more.
+//! This manages routing for parachain statements, parachain block and extrinsic data fetching,
+//! communication between collators and validators, and more.
 
 extern crate parity_codec as codec;
 extern crate substrate_network;
@@ -30,16 +29,23 @@ extern crate polkadot_availability_store as av_store;
 extern crate polkadot_primitives;
 
 extern crate arrayvec;
-extern crate futures;
 extern crate parking_lot;
 extern crate tokio;
-extern crate rhododendron;
+extern crate slice_group_by;
 
+#[macro_use]
+extern crate futures;
 #[macro_use]
 extern crate log;
 #[macro_use]
 extern crate parity_codec_derive;
 
+#[cfg(test)]
+extern crate substrate_client;
+
+#[cfg(test)]
+extern crate substrate_keyring;
+
 mod collator_pool;
 mod local_collations;
 mod router;
@@ -256,7 +262,7 @@ impl PolkadotProtocol {
 						send_polkadot_message(
 							ctx,
 							who,
-							Message::RequestBlockData(req_id, parent, c_hash)
+							Message::RequestBlockData(req_id, parent, c_hash),
 						);
 
 						in_flight.insert((req_id, who), pending);

polkadot/network/src/router.rs

@@ -23,21 +23,28 @@
 //! and dispatch evaluation work as necessary when new statements come in.
 
 use sr_primitives::traits::{ProvideRuntimeApi, BlakeTwo256, Hash as HashT};
-use polkadot_consensus::{SharedTable, TableRouter, SignedStatement, GenericStatement, ParachainWork};
+use polkadot_consensus::{
+	SharedTable, TableRouter, SignedStatement, GenericStatement, ParachainWork, Incoming,
+	Validated, Outgoing,
+};
 use polkadot_primitives::{Block, Hash, SessionKey};
-use polkadot_primitives::parachain::{BlockData, Extrinsic, CandidateReceipt, ParachainHost};
+use polkadot_primitives::parachain::{
+	BlockData, Extrinsic, CandidateReceipt, ParachainHost, Id as ParaId, Message
+};
 
-use codec::Encode;
-use futures::prelude::*;
-use tokio::runtime::TaskExecutor;
+use codec::{Encode, Decode};
+use futures::{future, prelude::*};
+use futures::sync::oneshot::{self, Receiver};
 use parking_lot::Mutex;
 
-use std::collections::{HashMap, HashSet};
-use std::io;
+use std::collections::{hash_map::{Entry, HashMap}, HashSet};
+use std::{io, mem};
 use std::sync::Arc;
 
-use consensus::Knowledge;
-use super::NetworkService;
+use consensus::{NetworkService, Knowledge, Executor};
+
+type IngressPair = (ParaId, Vec<Message>);
+type IngressPairRef<'a> = (ParaId, &'a [Message]);
 
 fn attestation_topic(parent_hash: Hash) -> Hash {
 	let mut v = parent_hash.as_ref().to_vec();
@@ -46,26 +53,88 @@ fn attestation_topic(parent_hash: Hash) -> Hash {
 	BlakeTwo256::hash(&v[..])
 }
 
+fn incoming_message_topic(parent_hash: Hash, parachain: ParaId) -> Hash {
+	let mut v = parent_hash.as_ref().to_vec();
+	parachain.using_encoded(|s| v.extend(s));
+	v.extend(b"incoming");
+
+	BlakeTwo256::hash(&v[..])
+}
+
+/// Receiver for block data.
+pub struct BlockDataReceiver {
+	outer: Receiver<Receiver<BlockData>>,
+	inner: Option<Receiver<BlockData>>
+}
+
+impl Future for BlockDataReceiver {
+	type Item = BlockData;
+	type Error = io::Error;
+
+	fn poll(&mut self) -> Poll<BlockData, io::Error> {
+		let map_err = |_| io::Error::new(
+			io::ErrorKind::Other,
+			"Sending end of channel hung up",
+		);
+
+		if let Some(ref mut inner) = self.inner {
+			return inner.poll().map_err(map_err);
+		}
+		match self.outer.poll().map_err(map_err)? {
+			Async::Ready(mut inner) => {
+				let poll_result = inner.poll();
+				self.inner = Some(inner);
+				poll_result.map_err(map_err)
+			}
+			Async::NotReady => Ok(Async::NotReady),
+		}
+	}
+}
+/// receiver for incoming data.
+#[derive(Clone)]
+pub struct IncomingReceiver {
+	inner: future::Shared<Receiver<Incoming>>
+}
+
+impl Future for IncomingReceiver {
+	type Item = Incoming;
+	type Error = io::Error;
+
+	fn poll(&mut self) -> Poll<Incoming, io::Error> {
+		match self.inner.poll() {
+			Ok(Async::NotReady) => Ok(Async::NotReady),
+			Ok(Async::Ready(i)) => Ok(Async::Ready(Incoming::clone(&*i))),
+			Err(_) => Err(io::Error::new(
+				io::ErrorKind::Other,
+				"Sending end of channel hung up",
+			)),
+		}
+	}
+}
+
 /// Table routing implementation.
-pub struct Router<P> {
+pub struct Router<P, E, N: NetworkService, T> {
 	table: Arc<SharedTable>,
-	network: Arc<NetworkService>,
+	network: Arc<N>,
 	api: Arc<P>,
-	task_executor: TaskExecutor,
+	exit: E,
+	task_executor: T,
 	parent_hash: Hash,
 	attestation_topic: Hash,
 	knowledge: Arc<Mutex<Knowledge>>,
+	fetch_incoming: Arc<Mutex<HashMap<ParaId, IncomingReceiver>>>,
 	deferred_statements: Arc<Mutex<DeferredStatements>>,
 }
 
-impl<P> Router<P> {
+impl<P, E, N: NetworkService, T> Router<P, E, N, T> {
 	pub(crate) fn new(
 		table: Arc<SharedTable>,
-		network: Arc<NetworkService>,
+		network: Arc<N>,
 		api: Arc<P>,
-		task_executor: TaskExecutor,
+		task_executor: T,
 		parent_hash: Hash,
 		knowledge: Arc<Mutex<Knowledge>>,
+		exit: E,
 	) -> Self {
 		Router {
 			table,
@@ -75,7 +144,9 @@ impl<P> Router<P> {
 			parent_hash,
 			attestation_topic: attestation_topic(parent_hash),
 			knowledge,
+			fetch_incoming: Arc::new(Mutex::new(HashMap::new())),
 			deferred_statements: Arc::new(Mutex::new(DeferredStatements::new())),
+			exit,
 		}
 	}
 
@@ -85,7 +156,7 @@ impl<P> Router<P> {
 	}
 }
 
-impl<P> Clone for Router<P> {
+impl<P, E: Clone, N: NetworkService, T: Clone> Clone for Router<P, E, N, T> {
 	fn clone(&self) -> Self {
 		Router {
 			table: self.table.clone(),
@@ -95,18 +166,21 @@ impl<P> Clone for Router<P> {
 			parent_hash: self.parent_hash.clone(),
 			attestation_topic: self.attestation_topic.clone(),
 			deferred_statements: self.deferred_statements.clone(),
+			fetch_incoming: self.fetch_incoming.clone(),
 			knowledge: self.knowledge.clone(),
+			exit: self.exit.clone(),
 		}
 	}
 }
 
-impl<P: ProvideRuntimeApi + Send + Sync + 'static> Router<P>
-	where P::Api: ParachainHost<Block>
+impl<P: ProvideRuntimeApi + Send + Sync + 'static, E, N, T> Router<P, E, N, T> where
+	P::Api: ParachainHost<Block>,
+	N: NetworkService,
+	T: Clone + Executor + Send + 'static,
+	E: Future<Item=(),Error=()> + Clone + Send + 'static,
 {
 	/// Import a statement whose signature has been checked already.
-	pub(crate) fn import_statement<Exit>(&self, statement: SignedStatement, exit: Exit)
-		where Exit: Future<Item=(),Error=()> + Clone + Send + 'static
-	{
+	pub(crate) fn import_statement(&self, statement: SignedStatement) {
 		trace!(target: "p_net", "importing consensus statement {:?}", statement.statement);
 
 		// defer any statements for which we haven't imported the candidate yet
@@ -146,15 +220,47 @@ impl<P: ProvideRuntimeApi + Send + Sync + 'static> Router<P>
 
 			if let Some(work) = producer.map(|p| self.create_work(c_hash, p)) {
 				trace!(target: "consensus", "driving statement work to completion");
-				self.task_executor.spawn(work.select(exit.clone()).then(|_| Ok(())));
+				let work = work.select2(self.exit.clone()).then(|_| Ok(()));
+				self.task_executor.spawn(work);
+			}
+		}
+	}
+
+	/// Broadcast outgoing messages to peers.
+	pub(crate) fn broadcast_egress(&self, outgoing: Outgoing) {
+		use slice_group_by::LinearGroupBy;
+
+		let mut group_messages = Vec::new();
+		for egress in outgoing {
+			let source = egress.from;
+			let messages = egress.messages.outgoing_messages;
+
+			let groups = LinearGroupBy::new(&messages, |a, b| a.target == b.target);
+			for group in groups {
+				let target = match group.get(0) {
+					Some(msg) => msg.target,
+					None => continue, // skip empty.
+				};
+
+				group_messages.clear(); // reuse allocation from previous iterations.
+				group_messages.extend(group.iter().map(|msg| msg.data.clone()).map(Message));
+
+				debug!(target: "consensus", "Circulating messages from {:?} to {:?} at {}",
+					source, target, self.parent_hash);
+
+				// this is the ingress from source to target, with given messages.
+				let target_incoming = incoming_message_topic(self.parent_hash, target);
+				let ingress_for: IngressPairRef = (source, &group_messages[..]);
+
+				self.network.gossip_message(target_incoming, ingress_for.encode());
 			}
 		}
 	}
 
 	fn create_work<D>(&self, candidate_hash: Hash, producer: ParachainWork<D>)
-		-> impl Future<Item=(),Error=()>
+		-> impl Future<Item=(),Error=()> + Send + 'static
 		where
-		D: Future<Item=BlockData,Error=io::Error> + Send + 'static,
+		D: Future<Item=(BlockData, Incoming),Error=io::Error> + Send + 'static,
 	{
 		let table = self.table.clone();
 		let network = self.network.clone();
@@ -162,36 +268,97 @@ impl<P: ProvideRuntimeApi + Send + Sync + 'static> Router<P>
 		let attestation_topic = self.attestation_topic.clone();
 
 		producer.prime(self.api.clone())
-			.map(move |produced| {
+			.map(move |validated| {
 				// store the data before broadcasting statements, so other peers can fetch.
 				knowledge.lock().note_candidate(
 					candidate_hash,
-					Some(produced.block_data),
-					produced.extrinsic,
+					Some(validated.block_data().clone()),
+					validated.extrinsic().cloned(),
 				);
 
 				// propagate the statement.
 				// consider something more targeted than gossip in the future.
-				let signed = table.sign_and_import(produced.validity);
-				network.gossip_consensus_message(attestation_topic, signed.encode(), false);
+				let signed = table.import_validated(validated);
+				network.gossip_message(attestation_topic, signed.encode());
 			})
 			.map_err(|e| debug!(target: "p_net", "Failed to produce statements: {:?}", e))
 	}
+
 }
 
-impl<P: ProvideRuntimeApi + Send> TableRouter for Router<P>
-	where P::Api: ParachainHost<Block>
+impl<P: ProvideRuntimeApi, E, N, T> Router<P, E, N, T> where
+	P::Api: ParachainHost<Block>,
+	N: NetworkService,
+	T: Executor,
+	E: Future<Item=(),Error=()> + Clone + Send + 'static,
+{
+	fn do_fetch_incoming(&self, parachain: ParaId) -> IncomingReceiver {
+		use polkadot_primitives::BlockId;
+		let (tx, rx) = {
+			let mut fetching = self.fetch_incoming.lock();
+			match fetching.entry(parachain) {
+				Entry::Occupied(entry) => return entry.get().clone(),
+				Entry::Vacant(entry) => {
+					// has not been requested yet.
+					let (tx, rx) = oneshot::channel();
+					let rx = IncomingReceiver { inner: rx.shared() };
+					entry.insert(rx.clone());
+
+					(tx, rx)
+				}
+			}
+		};
+
+		let parent_hash = self.parent_hash;
+		let topic = incoming_message_topic(parent_hash, parachain);
+		let gossip_messages = self.network.gossip_messages_for(topic)
+			.map_err(|()| panic!("unbounded receivers do not throw errors; qed"))
+			.filter_map(|msg| IngressPair::decode(&mut msg.as_slice()));
+
+		let canon_roots = self.api.runtime_api().ingress(&BlockId::hash(parent_hash), parachain)
+			.map_err(|e| format!("Cannot fetch ingress for parachain {:?} at {:?}: {:?}",
+				parachain, parent_hash, e)
+			);
+
+		let work = canon_roots.into_future()
+			.and_then(move |ingress_roots| match ingress_roots {
+				None => Err(format!("No parachain {:?} registered at {}", parachain, parent_hash)),
+				Some(roots) => Ok(roots.into_iter().collect())
+			})
+			.and_then(move |ingress_roots| ComputeIngress {
+				inner: gossip_messages,
+				ingress_roots,
+				incoming: Vec::new(),
+			})
+			.map(move |incoming| if let Some(i) = incoming { let _ = tx.send(i); })
+			.select2(self.exit.clone())
+			.then(|_| Ok(()));
+
+		self.task_executor.spawn(work);
+
+		rx
+	}
+}
+
+impl<P: ProvideRuntimeApi + Send, E, N, T> TableRouter for Router<P, E, N, T> where
+	P::Api: ParachainHost<Block>,
+	N: NetworkService,
+	T: Clone + Executor + Send + 'static,
+	E: Future<Item=(),Error=()> + Clone + Send + 'static,
 {
 	type Error = io::Error;
 	type FetchCandidate = BlockDataReceiver;
+	type FetchIncoming = IncomingReceiver;
 
 	fn local_candidate(&self, receipt: CandidateReceipt, block_data: BlockData, extrinsic: Extrinsic) {
-		// give to network to make available.
+		// produce a signed statement
 		let hash = receipt.hash();
-		let candidate = self.table.sign_and_import(GenericStatement::Candidate(receipt));
+		let validated = Validated::collated_local(receipt, block_data.clone(), extrinsic.clone());
+		let statement = self.table.import_validated(validated);
 
+		// give to network to make available.
 		self.knowledge.lock().note_candidate(hash, Some(block_data), Some(extrinsic));
-		self.network.gossip_consensus_message(self.attestation_topic, candidate.encode(), false);
+		self.network.gossip_message(self.attestation_topic, statement.encode());
 	}
 
 	fn fetch_block_data(&self, candidate: &CandidateReceipt) -> BlockDataReceiver {
@@ -204,44 +371,27 @@ impl<P: ProvideRuntimeApi + Send> TableRouter for Router<P>
 		});
 		BlockDataReceiver { outer: rx, inner: None }
 	}
-}
 
-impl<P> Drop for Router<P> {
-	fn drop(&mut self) {
-		let parent_hash = self.parent_hash.clone();
-		self.network.with_spec(move |spec, _| spec.remove_consensus(&parent_hash));
+	fn fetch_incoming(&self, parachain: ParaId) -> Self::FetchIncoming {
+		self.do_fetch_incoming(parachain)
 	}
 }
 
-/// Receiver for block data.
-pub struct BlockDataReceiver {
-	outer: ::futures::sync::oneshot::Receiver<::futures::sync::oneshot::Receiver<BlockData>>,
-	inner: Option<::futures::sync::oneshot::Receiver<BlockData>>
-}
+impl<P, E, N: NetworkService, T> Drop for Router<P, E, N, T> {
+	fn drop(&mut self) {
+		let parent_hash = self.parent_hash.clone();
+		self.network.with_spec(move |spec, _| spec.remove_consensus(&parent_hash));
+		self.network.drop_gossip(self.attestation_topic);
 
-impl Future for BlockDataReceiver {
-	type Item = BlockData;
-	type Error = io::Error;
-
-	fn poll(&mut self) -> Poll<BlockData, io::Error> {
-		if let Some(ref mut inner) = self.inner {
-			return inner
-				.poll()
-				.map_err(|_| io::Error::new(
-					io::ErrorKind::Other,
-					"Sending end of channel hung up",
-			))
+		{
+			let mut incoming_fetched = self.fetch_incoming.lock();
+			for (para_id, _) in incoming_fetched.drain() {
+				self.network.drop_gossip(incoming_message_topic(
+					self.parent_hash,
+					para_id,
+				));
+			}
 		}
-		if let Ok(futures::Async::Ready(mut inner)) = self.outer.poll() {
-			let poll_result = inner.poll();
-			self.inner = Some(inner);
-			return poll_result
-				.map_err(|_| io::Error::new(
-					io::ErrorKind::Other,
-					"Sending end of channel hung up",
-			))
-		}
-		Ok(futures::Async::NotReady)
 	}
 }
 
@@ -300,10 +450,63 @@ impl DeferredStatements {
 	}
 }
 
+// computes ingress from incoming stream of messages.
+// returns `None` if the stream concludes too early.
+#[must_use = "futures do nothing unless polled"]
+struct ComputeIngress<S> {
+	ingress_roots: HashMap<ParaId, Hash>,
+	incoming: Vec<IngressPair>,
+	inner: S,
+}
+
+impl<S> Future for ComputeIngress<S> where S: Stream<Item=IngressPair> {
+	type Item = Option<Incoming>;
+	type Error = S::Error;
+
+	fn poll(&mut self) -> Poll<Option<Incoming>, Self::Error> {
+		loop {
+			if self.ingress_roots.is_empty() {
+				return Ok(Async::Ready(
+					Some(mem::replace(&mut self.incoming, Vec::new()))
+				))
+			}
+
+			let (para_id, messages) = match try_ready!(self.inner.poll()) {
+				None => return Ok(Async::Ready(None)),
+				Some(next) => next,
+			};
+
+			match self.ingress_roots.entry(para_id) {
+				Entry::Vacant(_) => continue,
+				Entry::Occupied(occupied) => {
+					let canon_root = occupied.get().clone();
+					let messages = messages.iter().map(|m| &m.0[..]);
+					if ::polkadot_consensus::message_queue_root(messages) != canon_root {
+						continue;
+					}
+
+					occupied.remove();
+				}
+			}
+
+			let pos = self.incoming.binary_search_by_key(
+				&para_id,
+				|&(id, _)| id,
+			)
+				.err()
+				.expect("incoming starts empty and only inserted when \
+					para_id not inserted before; qed");
+
+			self.incoming.insert(pos, (para_id, messages));
+		}
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;
 	use substrate_primitives::H512;
+	use futures::stream;
 
 	#[test]
 	fn deferred_statements_works() {
@@ -345,4 +548,70 @@ mod tests {
 			assert!(traces.is_empty());
 		}
 	}
+
+	#[test]
+	fn compute_ingress_works() {
+		let actual_messages = [
+			(
+				ParaId::from(1),
+				vec![Message(vec![1, 3, 5, 6]), Message(vec![4, 4, 4, 4])],
+			),
+			(
+				ParaId::from(2),
+				vec![
+					Message(vec![1, 3, 7, 9, 1, 2, 3, 4, 5, 6]),
+					Message(b"hello world".to_vec()),
+				],
+			),
+			(
+				ParaId::from(5),
+				vec![Message(vec![1, 2, 3, 4, 5]), Message(vec![6, 9, 6, 9])],
+			),
+		];
+
+		let roots: HashMap<_, _> = actual_messages.iter()
+			.map(|&(para_id, ref messages)| (
+				para_id,
+				::polkadot_consensus::message_queue_root(messages.iter().map(|m| &m.0)),
+			))
+			.collect();
+
+		let inputs = [
+			(
+				ParaId::from(1), // wrong message.
+				vec![Message(vec![1, 1, 2, 2]), Message(vec![3, 3, 4, 4])],
+			),
+			(
+				ParaId::from(1),
+				vec![Message(vec![1, 3, 5, 6]), Message(vec![4, 4, 4, 4])],
+			),
+			(
+				ParaId::from(1), // duplicate
+				vec![Message(vec![1, 3, 5, 6]), Message(vec![4, 4, 4, 4])],
+			),
+
+			(
+				ParaId::from(5), // out of order
+				vec![Message(vec![1, 2, 3, 4, 5]), Message(vec![6, 9, 6, 9])],
+			),
+			(
+				ParaId::from(1234), // un-routed parachain.
+				vec![Message(vec![9, 9, 9, 9])],
+			),
+			(
+				ParaId::from(2),
+				vec![
+					Message(vec![1, 3, 7, 9, 1, 2, 3, 4, 5, 6]),
+					Message(b"hello world".to_vec()),
+				],
+			),
+		];
+		let ingress = ComputeIngress {
+			ingress_roots: roots,
+			incoming: Vec::new(),
+			inner: stream::iter_ok::<_, ()>(inputs.iter().cloned()),
+		};
+
+		assert_eq!(ingress.wait().unwrap().unwrap(), actual_messages);
+	}
 }

polkadot/network/src/tests/consensus.rs

@@ -0,0 +1,466 @@
+// Copyright 2019 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/>.
+
+//! Tests and helpers for consensus networking.
+
+use consensus::NetworkService;
+use substrate_network::{consensus_gossip::ConsensusMessage, Context as NetContext};
+use substrate_primitives::{Ed25519AuthorityId, NativeOrEncoded};
+use substrate_keyring::Keyring;
+use {PolkadotProtocol};
+
+use polkadot_consensus::{SharedTable, MessagesFrom, Network, TableRouter};
+use polkadot_primitives::{AccountId, Block, Hash, Header, BlockId};
+use polkadot_primitives::parachain::{Id as ParaId, Chain, DutyRoster, ParachainHost, OutgoingMessage};
+use parking_lot::Mutex;
+use substrate_client::error::Result as ClientResult;
+use substrate_client::runtime_api::{Core, RuntimeVersion, ApiExt};
+use sr_primitives::ExecutionContext;
+use sr_primitives::traits::{ApiRef, ProvideRuntimeApi};
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use futures::{prelude::*, sync::mpsc};
+use tokio::runtime::{Runtime, TaskExecutor};
+
+use super::TestContext;
+
+#[derive(Clone, Copy)]
+struct NeverExit;
+
+impl Future for NeverExit {
+	type Item = ();
+	type Error = ();
+
+	fn poll(&mut self) -> Poll<(), ()> {
+		Ok(Async::NotReady)
+	}
+}
+
+struct GossipRouter {
+	incoming_messages: mpsc::UnboundedReceiver<(Hash, ConsensusMessage)>,
+	incoming_streams: mpsc::UnboundedReceiver<(Hash, mpsc::UnboundedSender<ConsensusMessage>)>,
+	outgoing: Vec<(Hash, mpsc::UnboundedSender<ConsensusMessage>)>,
+	messages: Vec<(Hash, ConsensusMessage)>,
+}
+
+impl GossipRouter {
+	fn add_message(&mut self, topic: Hash, message: ConsensusMessage) {
+		self.outgoing.retain(|&(ref o_topic, ref sender)| {
+			o_topic != &topic || sender.unbounded_send(message.clone()).is_ok()
+		});
+		self.messages.push((topic, message));
+	}
+
+	fn add_outgoing(&mut self, topic: Hash, sender: mpsc::UnboundedSender<ConsensusMessage>) {
+		for message in self.messages.iter()
+			.filter(|&&(ref t, _)| t == &topic)
+			.map(|&(_, ref msg)| msg.clone())
+		{
+			if let Err(_) = sender.unbounded_send(message) { return }
+		}
+
+		self.outgoing.push((topic, sender));
+	}
+}
+
+impl Future for GossipRouter {
+	type Item = ();
+	type Error = ();
+
+	fn poll(&mut self) -> Poll<(), ()> {
+		loop {
+			match self.incoming_messages.poll().unwrap() {
+				Async::Ready(Some((topic, message))) => self.add_message(topic, message),
+				Async::Ready(None) => panic!("ended early."),
+				Async::NotReady => break,
+			}
+		}
+
+		loop {
+			match self.incoming_streams.poll().unwrap() {
+				Async::Ready(Some((topic, sender))) => self.add_outgoing(topic, sender),
+				Async::Ready(None) => panic!("ended early."),
+				Async::NotReady => break,
+			}
+		}
+
+		Ok(Async::NotReady)
+	}
+}
+
+
+#[derive(Clone)]
+struct GossipHandle {
+	send_message: mpsc::UnboundedSender<(Hash, ConsensusMessage)>,
+	send_listener: mpsc::UnboundedSender<(Hash, mpsc::UnboundedSender<ConsensusMessage>)>,
+}
+
+fn make_gossip() -> (GossipRouter, GossipHandle) {
+	let (message_tx, message_rx) = mpsc::unbounded();
+	let (listener_tx, listener_rx) = mpsc::unbounded();
+
+	(
+		GossipRouter {
+			incoming_messages: message_rx,
+			incoming_streams: listener_rx,
+			outgoing: Vec::new(),
+			messages: Vec::new(),
+		},
+		GossipHandle { send_message: message_tx, send_listener: listener_tx },
+	)
+}
+
+struct TestNetwork {
+	proto: Arc<Mutex<PolkadotProtocol>>,
+	gossip: GossipHandle,
+}
+
+impl NetworkService for TestNetwork {
+	fn gossip_messages_for(&self, topic: Hash) -> mpsc::UnboundedReceiver<ConsensusMessage> {
+		let (tx, rx) = mpsc::unbounded();
+		let _  = self.gossip.send_listener.unbounded_send((topic, tx));
+		rx
+	}
+
+	fn gossip_message(&self, topic: Hash, message: ConsensusMessage) {
+		let _ = self.gossip.send_message.unbounded_send((topic, message));
+	}
+
+	fn drop_gossip(&self, _topic: Hash) {}
+
+	fn with_spec<F: Send + 'static>(&self, with: F)
+		where F: FnOnce(&mut PolkadotProtocol, &mut NetContext<Block>)
+	{
+		let mut context = TestContext::default();
+		let res = with(&mut *self.proto.lock(), &mut context);
+		// TODO: send context to worker for message routing.
+		res
+	}
+}
+
+#[derive(Default)]
+struct ApiData {
+	validators: Vec<AccountId>,
+	duties: Vec<Chain>,
+	active_parachains: Vec<ParaId>,
+	ingress: HashMap<ParaId, Vec<(ParaId, Hash)>>,
+}
+
+#[derive(Default, Clone)]
+struct TestApi {
+	data: Arc<Mutex<ApiData>>,
+}
+
+struct RuntimeApi {
+	data: Arc<Mutex<ApiData>>,
+}
+
+impl ProvideRuntimeApi for TestApi {
+	type Api = RuntimeApi;
+
+	fn runtime_api<'a>(&'a self) -> ApiRef<'a, Self::Api> {
+		RuntimeApi { data: self.data.clone() }.into()
+	}
+}
+
+impl Core<Block> for RuntimeApi {
+	fn version_runtime_api_impl(
+		&self,
+		_: &BlockId,
+		_: ExecutionContext,
+		_: Option<()>,
+		_: Vec<u8>,
+	) -> ClientResult<NativeOrEncoded<RuntimeVersion>> {
+		unimplemented!("Not required for testing!")
+	}
+
+	fn authorities_runtime_api_impl(
+		&self,
+		_: &BlockId,
+		_: ExecutionContext,
+		_: Option<()>,
+		_: Vec<u8>,
+	) -> ClientResult<NativeOrEncoded<Vec<Ed25519AuthorityId>>> {
+		unimplemented!("Not required for testing!")
+	}
+
+	fn execute_block_runtime_api_impl(
+		&self,
+		_: &BlockId,
+		_: ExecutionContext,
+		_: Option<(Block)>,
+		_: Vec<u8>,
+	) -> ClientResult<NativeOrEncoded<()>> {
+		unimplemented!("Not required for testing!")
+	}
+
+	fn initialise_block_runtime_api_impl(
+		&self,
+		_: &BlockId,
+		_: ExecutionContext,
+		_: Option<&Header>,
+		_: Vec<u8>,
+	) -> ClientResult<NativeOrEncoded<()>> {
+		unimplemented!("Not required for testing!")
+	}
+}
+
+impl ApiExt<Block> for RuntimeApi {
+	fn map_api_result<F: FnOnce(&Self) -> Result<R, E>, R, E>(
+		&self,
+		_: F
+	) -> Result<R, E> {
+		unimplemented!("Not required for testing!")
+	}
+
+	fn runtime_version_at(&self, _: &BlockId) -> ClientResult<RuntimeVersion> {
+		unimplemented!("Not required for testing!")
+	}
+}
+
+impl ParachainHost<Block> for RuntimeApi {
+	fn validators_runtime_api_impl(
+		&self,
+		_at: &BlockId,
+		_: ExecutionContext,
+		_: Option<()>,
+		_: Vec<u8>,
+	) -> ClientResult<NativeOrEncoded<Vec<AccountId>>> {
+		Ok(NativeOrEncoded::Native(self.data.lock().validators.clone()))
+	}
+
+	fn duty_roster_runtime_api_impl(
+		&self,
+		_at: &BlockId,
+		_: ExecutionContext,
+		_: Option<()>,
+		_: Vec<u8>,
+	) -> ClientResult<NativeOrEncoded<DutyRoster>> {
+
+		Ok(NativeOrEncoded::Native(DutyRoster {
+			validator_duty: self.data.lock().duties.clone(),
+		}))
+	}
+
+	fn active_parachains_runtime_api_impl(
+		&self,
+		_at: &BlockId,
+		_: ExecutionContext,
+		_: Option<()>,
+		_: Vec<u8>,
+	) -> ClientResult<NativeOrEncoded<Vec<ParaId>>> {
+		Ok(NativeOrEncoded::Native(self.data.lock().active_parachains.clone()))
+	}
+
+	fn parachain_head_runtime_api_impl(
+		&self,
+		_at: &BlockId,
+		_: ExecutionContext,
+		_: Option<ParaId>,
+		_: Vec<u8>,
+	) -> ClientResult<NativeOrEncoded<Option<Vec<u8>>>> {
+		Ok(NativeOrEncoded::Native(Some(Vec::new())))
+	}
+
+	fn parachain_code_runtime_api_impl(
+		&self,
+		_at: &BlockId,
+		_: ExecutionContext,
+		_: Option<ParaId>,
+		_: Vec<u8>,
+	) -> ClientResult<NativeOrEncoded<Option<Vec<u8>>>> {
+		Ok(NativeOrEncoded::Native(Some(Vec::new())))
+	}
+
+	fn ingress_runtime_api_impl(
+		&self,
+		_at: &BlockId,
+		_: ExecutionContext,
+		id: Option<ParaId>,
+		_: Vec<u8>,
+	) -> ClientResult<NativeOrEncoded<Option<Vec<(ParaId, Hash)>>>> {
+		let id = id.unwrap();
+		Ok(NativeOrEncoded::Native(self.data.lock().ingress.get(&id).cloned()))
+	}
+}
+
+type TestConsensusNetwork = ::consensus::ConsensusNetwork<
+	TestApi,
+	NeverExit,
+	TestNetwork,
+	TaskExecutor,
+>;
+
+struct Built {
+	gossip: GossipRouter,
+	api_handle: Arc<Mutex<ApiData>>,
+	networks: Vec<TestConsensusNetwork>,
+}
+
+fn build_network(n: usize, executor: TaskExecutor) -> Built {
+	let (gossip_router, gossip_handle) = make_gossip();
+	let api_handle = Arc::new(Mutex::new(Default::default()));
+	let runtime_api = Arc::new(TestApi { data: api_handle.clone() });
+
+	let networks = (0..n).map(|_| {
+		let net = Arc::new(TestNetwork {
+			proto: Arc::new(Mutex::new(PolkadotProtocol::new(None))),
+			gossip: gossip_handle.clone(),
+		});
+
+		TestConsensusNetwork::new(
+			net,
+			NeverExit,
+			runtime_api.clone(),
+			executor.clone(),
+		)
+	});
+
+	let networks: Vec<_> = networks.collect();
+
+	Built {
+		gossip: gossip_router,
+		api_handle,
+		networks,
+	}
+}
+
+#[derive(Default)]
+struct IngressBuilder {
+	egress: HashMap<(ParaId, ParaId), Vec<Vec<u8>>>,
+}
+
+impl IngressBuilder {
+	fn add_messages(&mut self, source: ParaId, messages: &[OutgoingMessage]) {
+		for message in messages {
+			let target = message.target;
+			self.egress.entry((source, target)).or_insert_with(Vec::new).push(message.data.clone());
+		}
+	}
+
+	fn build(self) -> HashMap<ParaId, Vec<(ParaId, Hash)>> {
+		let mut map = HashMap::new();
+		for ((source, target), messages) in self.egress {
+			map.entry(target).or_insert_with(Vec::new)
+				.push((source, polkadot_consensus::message_queue_root(&messages)));
+		}
+
+		for roots in map.values_mut() {
+			roots.sort_by_key(|&(para_id, _)| para_id);
+		}
+
+		map
+	}
+}
+
+fn make_table(data: &ApiData, local_key: &Keyring, parent_hash: Hash) -> Arc<SharedTable> {
+	use ::av_store::Store;
+
+	let store = Store::new_in_memory();
+	let authorities: Vec<_> = data.validators.iter().map(|v| v.to_fixed_bytes().into()).collect();
+	let (group_info, _) = ::polkadot_consensus::make_group_info(
+		DutyRoster { validator_duty: data.duties.clone() },
+		&authorities,
+		local_key.to_raw_public().into()
+	).unwrap();
+
+	Arc::new(SharedTable::new(
+		group_info,
+		Arc::new(local_key.pair()),
+		parent_hash,
+		store,
+	))
+}
+
+#[test]
+fn ingress_fetch_works() {
+	let mut runtime = Runtime::new().unwrap();
+	let built = build_network(3, runtime.executor());
+
+	let id_a: ParaId = 1.into();
+	let id_b: ParaId = 2.into();
+	let id_c: ParaId = 3.into();
+
+	let key_a = Keyring::Alice;
+	let key_b = Keyring::Bob;
+	let key_c = Keyring::Charlie;
+
+	let messages_from_a = vec![
+		OutgoingMessage { target: id_b, data: vec![1, 2, 3] },
+		OutgoingMessage { target: id_b, data: vec![3, 4, 5] },
+		OutgoingMessage { target: id_c, data: vec![9, 9, 9] },
+	];
+
+	let messages_from_b = vec![
+		OutgoingMessage { target: id_a, data: vec![1, 1, 1, 1, 1,] },
+		OutgoingMessage { target: id_c, data: b"hello world".to_vec() },
+	];
+
+	let messages_from_c = vec![
+		OutgoingMessage { target: id_a, data: b"dog42".to_vec() },
+		OutgoingMessage { target: id_b, data: b"dogglesworth".to_vec() },
+	];
+
+	let ingress = {
+		let mut builder = IngressBuilder::default();
+		builder.add_messages(id_a, &messages_from_a);
+		builder.add_messages(id_b, &messages_from_b);
+		builder.add_messages(id_c, &messages_from_c);
+
+		builder.build()
+	};
+
+	let parent_hash = [1; 32].into();
+
+	let (router_a, router_b, router_c) = {
+		let mut api_handle = built.api_handle.lock();
+		*api_handle = ApiData {
+			active_parachains: vec![id_a, id_b, id_c],
+			duties: vec![Chain::Parachain(id_a), Chain::Parachain(id_b), Chain::Parachain(id_c)],
+			validators: vec![
+				key_a.to_raw_public().into(),
+				key_b.to_raw_public().into(),
+				key_c.to_raw_public().into(),
+			],
+			ingress,
+		};
+
+		(
+			built.networks[0].communication_for(
+				make_table(&*api_handle, &key_a, parent_hash),
+				vec![MessagesFrom::from_messages(id_a, messages_from_a)],
+			),
+			built.networks[1].communication_for(
+				make_table(&*api_handle, &key_b, parent_hash),
+				vec![MessagesFrom::from_messages(id_b, messages_from_b)],
+			),
+			built.networks[2].communication_for(
+				make_table(&*api_handle, &key_c, parent_hash),
+				vec![MessagesFrom::from_messages(id_c, messages_from_c)],
+			),
+		)
+	};
+
+	// make sure everyone can get ingress for their own parachain.
+	let fetch_a = router_a.fetch_incoming(id_a).map_err(|_| format!("Could not fetch ingress_a"));
+	let fetch_b = router_b.fetch_incoming(id_b).map_err(|_| format!("Could not fetch ingress_b"));
+	let fetch_c = router_c.fetch_incoming(id_c).map_err(|_| format!("Could not fetch ingress_c"));
+
+	let work = fetch_a.join3(fetch_b, fetch_c);
+	runtime.spawn(built.gossip.then(|_| Ok(()))); // in background.
+	runtime.block_on(work).unwrap();
+}

polkadot/network/src/tests/mod.rs

@@ -34,6 +34,8 @@ use substrate_network::{
 use std::sync::Arc;
 use futures::Future;
 
+mod consensus;
+
 #[derive(Default)]
 struct TestContext {
 	disabled: Vec<NodeIndex>,