Rename polkadot-consensus -> polkadot-validation (#151)

* Initial rename of consensus -> validation

* Rename crate imports

* network: rename consensus to validation

* network: rename consensus in comments and logs

* Grumbles

* Rename tests consensus -> validation

polkadot/validation/Cargo.toml

@@ -0,0 +1,31 @@
+[package]
+name = "polkadot-validation"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+
+[dependencies]
+futures = "0.1.17"
+parking_lot = "0.7.1"
+tokio = "0.1.7"
+error-chain = "0.12"
+log = "0.4.6"
+exit-future = "0.1"
+parity-codec = "3.0"
+polkadot-availability-store = { path = "../availability-store" }
+polkadot-parachain = { path = "../parachain" }
+polkadot-primitives = { path = "../primitives" }
+polkadot-runtime = { path = "../runtime" }
+polkadot-statement-table = { path = "../statement-table" }
+substrate-consensus-aura = { git = "https://github.com/paritytech/substrate" }
+substrate-finality-grandpa = { git = "https://github.com/paritytech/substrate" }
+substrate-consensus-common = { git = "https://github.com/paritytech/substrate" }
+substrate-primitives = { git = "https://github.com/paritytech/substrate" }
+substrate-inherents = { git = "https://github.com/paritytech/substrate" }
+substrate-transaction-pool = { git = "https://github.com/paritytech/substrate" }
+srml-aura = { git = "https://github.com/paritytech/substrate" }
+substrate-client = { git = "https://github.com/paritytech/substrate" }
+substrate-trie = { git = "https://github.com/paritytech/substrate" }
+sr-primitives = { git = "https://github.com/paritytech/substrate" }
+
+[dev-dependencies]
+substrate-keyring = { git = "https://github.com/paritytech/substrate" }

polkadot/validation/README.adoc

@@ -0,0 +1,5 @@
+
+= Polkadot Validation
+
+placeholder
+//TODO Write content :)

polkadot/validation/src/attestation_service.rs

@@ -0,0 +1,215 @@
+// Copyright 2017 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/>.
+
+//! Attestation service.
+
+/// Attestation service. A long running service that creates and manages parachain attestation
+/// instances.
+///
+/// This uses a handle to an underlying thread pool to dispatch heavy work
+/// such as candidate verification while performing event-driven work
+/// on a local event loop.
+
+use std::thread;
+use std::time::{Duration, Instant};
+use std::sync::Arc;
+
+use client::{error::Result as ClientResult, BlockchainEvents, ChainHead, BlockBody};
+use client::block_builder::api::BlockBuilder;
+use client::blockchain::HeaderBackend;
+use client::runtime_api::Core;
+use primitives::ed25519;
+use futures::prelude::*;
+use polkadot_primitives::{Block, BlockId};
+use polkadot_primitives::parachain::{CandidateReceipt, ParachainHost};
+use extrinsic_store::Store as ExtrinsicStore;
+use runtime_primitives::traits::{ProvideRuntimeApi, Header as HeaderT};
+
+use tokio::runtime::TaskExecutor;
+use tokio::runtime::current_thread::Runtime as LocalRuntime;
+use tokio::timer::Interval;
+
+use super::{Network, Collators, TableRouter};
+
+/// Gets a list of the candidates in a block.
+pub(crate) fn fetch_candidates<P: BlockBody<Block>>(client: &P, block: &BlockId)
+	-> ClientResult<Option<impl Iterator<Item=CandidateReceipt>>>
+{
+	use codec::{Encode, Decode};
+	use polkadot_runtime::{Call, ParachainsCall, UncheckedExtrinsic as RuntimeExtrinsic};
+
+	let extrinsics = client.block_body(block)?;
+	Ok(extrinsics
+		.into_iter()
+		.filter_map(|ex| RuntimeExtrinsic::decode(&mut ex.encode().as_slice()))
+		.filter_map(|ex| match ex.function {
+			Call::Parachains(ParachainsCall::set_heads(heads)) =>
+				Some(heads.into_iter().map(|c| c.candidate)),
+			_ => None,
+		})
+		.next())
+}
+
+// creates a task to prune redundant entries in availability store upon block finalization
+//
+// NOTE: this will need to be changed to finality notification rather than
+// block import notifications when the consensus switches to non-instant finality.
+fn prune_unneeded_availability<P>(client: Arc<P>, extrinsic_store: ExtrinsicStore)
+	-> impl Future<Item=(),Error=()> + Send
+	where P: Send + Sync + BlockchainEvents<Block> + BlockBody<Block> + 'static
+{
+	client.finality_notification_stream()
+		.for_each(move |notification| {
+			let hash = notification.hash;
+			let parent_hash = notification.header.parent_hash;
+			let candidate_hashes = match fetch_candidates(&*client, &BlockId::hash(hash)) {
+				Ok(Some(candidates)) => candidates.map(|c| c.hash()).collect(),
+				Ok(None) => {
+					warn!("Could not extract candidates from block body of imported block {:?}", hash);
+					return Ok(())
+				}
+				Err(e) => {
+					warn!("Failed to fetch block body for imported block {:?}: {:?}", hash, e);
+					return Ok(())
+				}
+			};
+
+			if let Err(e) = extrinsic_store.candidates_finalized(parent_hash, candidate_hashes) {
+				warn!(target: "validation", "Failed to prune unneeded available data: {:?}", e);
+			}
+
+			Ok(())
+		})
+}
+
+/// Parachain candidate attestation service handle.
+pub(crate) struct ServiceHandle {
+	thread: Option<thread::JoinHandle<()>>,
+	exit_signal: Option<::exit_future::Signal>,
+}
+
+/// Create and start a new instance of the attestation service.
+pub(crate) fn start<C, N, P>(
+	client: Arc<P>,
+	parachain_validation: Arc<::ParachainValidation<C, N, P>>,
+	thread_pool: TaskExecutor,
+	key: Arc<ed25519::Pair>,
+	extrinsic_store: ExtrinsicStore,
+) -> ServiceHandle
+	where
+		C: Collators + Send + Sync + 'static,
+		<C::Collation as IntoFuture>::Future: Send + 'static,
+		P: BlockchainEvents<Block> + ChainHead<Block> + BlockBody<Block>,
+		P: ProvideRuntimeApi + HeaderBackend<Block> + Send + Sync + 'static,
+		P::Api: ParachainHost<Block> + Core<Block> + BlockBuilder<Block>,
+		N: Network + Send + Sync + 'static,
+		N::TableRouter: Send + 'static,
+		<<N::TableRouter as TableRouter>::FetchIncoming as IntoFuture>::Future: Send + 'static,
+{
+	const TIMER_DELAY: Duration = Duration::from_secs(5);
+	const TIMER_INTERVAL: Duration = Duration::from_secs(30);
+
+	let (signal, exit) = ::exit_future::signal();
+	let thread = thread::spawn(move || {
+		let mut runtime = LocalRuntime::new().expect("Could not create local runtime");
+		let notifications = {
+			let client = client.clone();
+			let validation = parachain_validation.clone();
+			let key = key.clone();
+
+			client.import_notification_stream()
+				.for_each(move |notification| {
+					let parent_hash = notification.hash;
+					if notification.is_new_best {
+						let res = client
+							.runtime_api()
+							.authorities(&BlockId::hash(parent_hash))
+							.map_err(Into::into)
+							.and_then(|authorities| {
+								validation.get_or_instantiate(
+									parent_hash,
+									notification.header.parent_hash().clone(),
+									&authorities,
+									key.clone(),
+								)
+							});
+
+						if let Err(e) = res {
+							warn!("Unable to start parachain validation on top of {:?}: {}",
+								parent_hash, e);
+						}
+					}
+					Ok(())
+				})
+				.select(exit.clone())
+				.then(|_| Ok(()))
+		};
+
+		let prune_old_sessions = {
+			let client = client.clone();
+			let interval = Interval::new(
+				Instant::now() + TIMER_DELAY,
+				TIMER_INTERVAL,
+			);
+
+			interval
+				.for_each(move |_| match client.leaves() {
+					Ok(leaves) => {
+						parachain_validation.retain(|h| leaves.contains(h));
+						Ok(())
+					}
+					Err(e) => {
+						warn!("Error fetching leaves from client: {:?}", e);
+						Ok(())
+					}
+				})
+				.map_err(|e| warn!("Timer error {:?}", e))
+				.select(exit.clone())
+				.then(|_| Ok(()))
+		};
+
+		runtime.spawn(notifications);
+		thread_pool.spawn(prune_old_sessions);
+
+		let prune_available = prune_unneeded_availability(client, extrinsic_store)
+			.select(exit.clone())
+			.then(|_| Ok(()));
+
+		// spawn this on the tokio executor since it's fine on a thread pool.
+		thread_pool.spawn(prune_available);
+
+		if let Err(e) = runtime.block_on(exit) {
+			debug!("BFT event loop error {:?}", e);
+		}
+	});
+
+	ServiceHandle {
+		thread: Some(thread),
+		exit_signal: Some(signal),
+	}
+}
+
+impl Drop for ServiceHandle {
+	fn drop(&mut self) {
+		if let Some(signal) = self.exit_signal.take() {
+			signal.fire();
+		}
+
+		if let Some(thread) = self.thread.take() {
+			thread.join().expect("The service thread has panicked");
+		}
+	}
+}

polkadot/validation/src/collation.rs

@@ -0,0 +1,396 @@
+// Copyright 2017 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/>.
+
+//! Validator-side view of collation.
+//!
+//! This module contains type definitions, a trait for a batch of collators, and a trait for
+//! attempting to fetch a collation repeatedly until a valid one is obtained.
+
+use std::sync::Arc;
+
+use polkadot_primitives::{Block, Hash, AccountId, BlockId};
+use polkadot_primitives::parachain::{Id as ParaId, Collation, Extrinsic, OutgoingMessage};
+use polkadot_primitives::parachain::{CandidateReceipt, ParachainHost};
+use runtime_primitives::traits::ProvideRuntimeApi;
+use parachain::{wasm_executor::{self, ExternalitiesError}, MessageRef};
+use super::Incoming;
+
+use futures::prelude::*;
+
+/// Encapsulates connections to collators and allows collation on any parachain.
+///
+/// This is expected to be a lightweight, shared type like an `Arc`.
+pub trait Collators: Clone {
+	/// Errors when producing collations.
+	type Error: std::fmt::Debug;
+	/// A full collation.
+	type Collation: IntoFuture<Item=Collation,Error=Self::Error>;
+
+	/// Collate on a specific parachain, building on a given relay chain parent hash.
+	///
+	/// The returned collation should be checked for basic validity in the signature
+	/// and will be checked for state-transition validity by the consumer of this trait.
+	///
+	/// This does not have to guarantee local availability, as a valid collation
+	/// will be passed to the `TableRouter` instance.
+	fn collate(&self, parachain: ParaId, relay_parent: Hash) -> Self::Collation;
+
+	/// Note a bad collator. TODO: take proof
+	fn note_bad_collator(&self, collator: AccountId);
+}
+
+/// A future which resolves when a collation is available.
+///
+/// This future is fused.
+pub struct CollationFetch<C: Collators, P> {
+	parachain: ParaId,
+	relay_parent_hash: Hash,
+	relay_parent: BlockId,
+	collators: C,
+	incoming: Incoming,
+	live_fetch: Option<<C::Collation as IntoFuture>::Future>,
+	client: Arc<P>,
+}
+
+impl<C: Collators, P> CollationFetch<C, P> {
+	/// Create a new collation fetcher for the given chain.
+	pub fn new(
+		parachain: ParaId,
+		relay_parent_hash: Hash,
+		collators: C,
+		client: Arc<P>,
+		incoming: Incoming,
+	) -> Self {
+		CollationFetch {
+			relay_parent: BlockId::hash(relay_parent_hash),
+			relay_parent_hash,
+			collators,
+			client,
+			parachain,
+			live_fetch: None,
+			incoming,
+		}
+	}
+
+	/// Access the underlying relay parent hash.
+	pub fn relay_parent(&self) -> Hash {
+		self.relay_parent_hash
+	}
+
+	/// Access the local parachain ID.
+	pub fn parachain(&self) -> ParaId {
+		self.parachain
+	}
+}
+
+impl<C: Collators, P: ProvideRuntimeApi> Future for CollationFetch<C, P>
+	where P::Api: ParachainHost<Block>,
+{
+	type Item = (Collation, Extrinsic);
+	type Error = C::Error;
+
+	fn poll(&mut self) -> Poll<(Collation, Extrinsic), C::Error> {
+		loop {
+			let x = {
+				let parachain = self.parachain.clone();
+				let (r, c)  = (self.relay_parent_hash, &self.collators);
+				let poll = self.live_fetch
+					.get_or_insert_with(move || c.collate(parachain, r).into_future())
+					.poll();
+
+				try_ready!(poll)
+			};
+
+			match validate_collation(&*self.client, &self.relay_parent, &x, &self.incoming) {
+				Ok(e) => {
+					return Ok(Async::Ready((x, e)))
+				}
+				Err(e) => {
+					debug!("Failed to validate parachain due to API error: {}", e);
+
+					// just continue if we got a bad collation or failed to validate
+					self.live_fetch = None;
+					self.collators.note_bad_collator(x.receipt.collator)
+				}
+			}
+		}
+	}
+}
+
+// Errors that can occur when validating a parachain.
+error_chain! {
+	types { Error, ErrorKind, ResultExt; }
+
+	links {
+		Client(::client::error::Error, ::client::error::ErrorKind);
+		WasmValidation(wasm_executor::Error, wasm_executor::ErrorKind);
+	}
+
+	errors {
+		InactiveParachain(id: ParaId) {
+			description("Collated for inactive parachain"),
+			display("Collated for inactive parachain: {:?}", id),
+		}
+		EgressRootMismatch(id: ParaId, expected: Hash, got: Hash) {
+			description("Got unexpected egress route."),
+			display(
+				"Got unexpected egress route to {:?}. (expected: {:?}, got {:?})",
+				id, expected, got
+			),
+		}
+		MissingEgressRoute(expected: Option<ParaId>, got: Option<ParaId>) {
+			description("Missing or extra egress route."),
+			display("Missing or extra egress route. (expected: {:?}, got {:?})", expected, got),
+		}
+		WrongHeadData(expected: Vec<u8>, got: Vec<u8>) {
+			description("Parachain validation produced wrong head data."),
+			display("Parachain validation produced wrong head data (expected: {:?}, got {:?}", expected, got),
+		}
+	}
+}
+
+/// Compute a trie root for a set of messages.
+pub fn message_queue_root<A, I: IntoIterator<Item=A>>(messages: I) -> Hash
+	where A: AsRef<[u8]>
+{
+	::trie::ordered_trie_root::<primitives::Blake2Hasher, _, _>(messages)
+}
+
+/// Compute the set of egress roots for all given outgoing messages.
+pub fn egress_roots(mut outgoing: Vec<OutgoingMessage>) -> Vec<(ParaId, Hash)> {
+	// stable sort messages by parachain ID.
+	outgoing.sort_by_key(|msg| ParaId::from(msg.target));
+
+	let mut egress_roots = Vec::new();
+	{
+		let mut messages_iter = outgoing.iter().peekable();
+		while let Some(batch_target) = messages_iter.peek().map(|o| o.target) {
+ 			// we borrow the iterator mutably to ensure it advances so the
+			// next iteration of the loop starts with `messages_iter` pointing to
+			// the next batch.
+			let messages_to = messages_iter
+				.clone()
+				.take_while(|o| o.target == batch_target)
+				.map(|o| { let _ = messages_iter.next(); &o.data[..] });
+
+			let computed_root = message_queue_root(messages_to);
+			egress_roots.push((batch_target, computed_root));
+		}
+	}
+
+	egress_roots
+}
+
+fn check_extrinsic(
+	mut outgoing: Vec<OutgoingMessage>,
+	expected_egress_roots: &[(ParaId, Hash)],
+) -> Result<Extrinsic, Error> {
+	// stable sort messages by parachain ID.
+	outgoing.sort_by_key(|msg| ParaId::from(msg.target));
+
+	{
+		let mut messages_iter = outgoing.iter().peekable();
+		let mut expected_egress_roots = expected_egress_roots.iter();
+		while let Some(batch_target) = messages_iter.peek().map(|o| o.target) {
+			let expected_root = match expected_egress_roots.next() {
+				None => return Err(ErrorKind::MissingEgressRoute(Some(batch_target), None).into()),
+				Some(&(id, ref root)) => if id == batch_target {
+					root
+				} else {
+					return Err(ErrorKind::MissingEgressRoute(Some(batch_target), Some(id)).into());
+				}
+			};
+
+ 			// we borrow the iterator mutably to ensure it advances so the
+			// next iteration of the loop starts with `messages_iter` pointing to
+			// the next batch.
+			let messages_to = messages_iter
+				.clone()
+				.take_while(|o| o.target == batch_target)
+				.map(|o| { let _ = messages_iter.next(); &o.data[..] });
+
+			let computed_root = message_queue_root(messages_to);
+			if &computed_root != expected_root {
+				return Err(ErrorKind::EgressRootMismatch(
+					batch_target,
+					expected_root.clone(),
+					computed_root,
+				).into());
+			}
+		}
+
+		// also check that there are no more additional expected roots.
+		if let Some((next_target, _)) = expected_egress_roots.next() {
+			return Err(ErrorKind::MissingEgressRoute(None, Some(*next_target)).into());
+		}
+	}
+
+	Ok(Extrinsic { outgoing_messages: outgoing })
+}
+
+struct Externalities {
+	parachain_index: ParaId,
+	outgoing: Vec<OutgoingMessage>,
+}
+
+impl wasm_executor::Externalities for Externalities {
+	fn post_message(&mut self, message: MessageRef) -> Result<(), ExternalitiesError> {
+		// TODO: https://github.com/paritytech/polkadot/issues/92
+		// check per-message and per-byte fees for the parachain.
+		let target: ParaId = message.target.into();
+		if target == self.parachain_index {
+			return Err(ExternalitiesError::CannotPostMessage("posted message to self"));
+		}
+
+		self.outgoing.push(OutgoingMessage {
+			target,
+			data: message.data.to_vec(),
+		});
+
+		Ok(())
+	}
+}
+
+impl Externalities {
+	// Performs final checks of validity, producing the extrinsic data.
+	fn final_checks(
+		self,
+		candidate: &CandidateReceipt,
+	) -> Result<Extrinsic, Error> {
+		check_extrinsic(
+			self.outgoing,
+			&candidate.egress_queue_roots[..],
+		)
+	}
+}
+
+/// Check whether a given collation is valid. Returns `Ok` on success, error otherwise.
+///
+/// This assumes that basic validity checks have been done:
+///   - Block data hash is the same as linked in candidate receipt.
+///   - incoming messages have been validated against canonical ingress roots
+pub fn validate_collation<P>(
+	client: &P,
+	relay_parent: &BlockId,
+	collation: &Collation,
+	incoming: &Incoming,
+) -> Result<Extrinsic, Error> where
+	P: ProvideRuntimeApi,
+	P::Api: ParachainHost<Block>,
+{
+	use parachain::{IncomingMessage, ValidationParams};
+
+	let api = client.runtime_api();
+	let para_id = collation.receipt.parachain_index;
+	let validation_code = api.parachain_code(relay_parent, para_id)?
+		.ok_or_else(|| ErrorKind::InactiveParachain(para_id))?;
+
+	let chain_head = api.parachain_head(relay_parent, para_id)?
+		.ok_or_else(|| ErrorKind::InactiveParachain(para_id))?;
+
+	let params = ValidationParams {
+		parent_head: chain_head,
+		block_data: collation.block_data.0.clone(),
+		ingress: incoming.iter()
+			.flat_map(|&(source, ref messages)| {
+				messages.iter().map(move |msg| IncomingMessage {
+					source,
+					data: msg.0.clone(),
+				})
+			})
+			.collect()
+	};
+
+	let mut ext = Externalities {
+		parachain_index: collation.receipt.parachain_index.clone(),
+		outgoing: Vec::new(),
+	};
+
+	match wasm_executor::validate_candidate(&validation_code, params, &mut ext) {
+		Ok(result) => {
+			if result.head_data == collation.receipt.head_data.0 {
+				ext.final_checks(&collation.receipt)
+			} else {
+				Err(ErrorKind::WrongHeadData(
+					collation.receipt.head_data.0.clone(),
+					result.head_data
+				).into())
+			}
+		}
+		Err(e) => Err(e.into())
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use parachain::wasm_executor::Externalities as ExternalitiesTrait;
+
+	#[test]
+	fn compute_and_check_egress() {
+		let messages = vec![
+			OutgoingMessage { target: 3.into(), data: vec![1, 1, 1] },
+			OutgoingMessage { target: 1.into(), data: vec![1, 2, 3] },
+			OutgoingMessage { target: 2.into(), data: vec![4, 5, 6] },
+			OutgoingMessage { target: 1.into(), data: vec![7, 8, 9] },
+		];
+
+		let root_1 = message_queue_root(&[vec![1, 2, 3], vec![7, 8, 9]]);
+		let root_2 = message_queue_root(&[vec![4, 5, 6]]);
+		let root_3 = message_queue_root(&[vec![1, 1, 1]]);
+
+		assert!(check_extrinsic(
+			messages.clone(),
+			&[(1.into(), root_1), (2.into(), root_2), (3.into(), root_3)],
+		).is_ok());
+
+		let egress_roots = egress_roots(messages.clone());
+
+		assert!(check_extrinsic(
+			messages.clone(),
+			&egress_roots[..],
+		).is_ok());
+
+		// missing root.
+		assert!(check_extrinsic(
+			messages.clone(),
+			&[(1.into(), root_1), (3.into(), root_3)],
+		).is_err());
+
+		// extra root.
+		assert!(check_extrinsic(
+			messages.clone(),
+			&[(1.into(), root_1), (2.into(), root_2), (3.into(), root_3), (4.into(), Default::default())],
+		).is_err());
+
+		// root mismatch.
+		assert!(check_extrinsic(
+			messages.clone(),
+			&[(1.into(), root_2), (2.into(), root_1), (3.into(), root_3)],
+		).is_err());
+	}
+
+	#[test]
+	fn ext_rejects_local_message() {
+		let mut ext = Externalities {
+			parachain_index: 5.into(),
+			outgoing: Vec::new(),
+		};
+
+		assert!(ext.post_message(MessageRef { target: 1.into(), data: &[] }).is_ok());
+		assert!(ext.post_message(MessageRef { target: 5.into(), data: &[] }).is_err());
+	}
+}

polkadot/validation/src/dynamic_inclusion.rs

@@ -0,0 +1,131 @@
+// Copyright 2017 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/>.
+
+//! Dynamic inclusion threshold over time.
+
+use std::time::{Duration, Instant};
+
+fn duration_to_micros(duration: &Duration) -> u64 {
+	duration.as_secs() * 1_000_000 + (duration.subsec_nanos() / 1000) as u64
+}
+
+/// Dynamic inclusion threshold over time.
+///
+/// The acceptable proportion of parachains which must have parachain candidates
+/// reduces over time (eventually going to zero).
+#[derive(Debug, Clone)]
+pub struct DynamicInclusion {
+	start: Instant,
+	y: u64,
+	m: u64,
+}
+
+impl DynamicInclusion {
+	/// Constructs a new dynamic inclusion threshold calculator based on the time now,
+	/// how many parachain candidates are required at the beginning, and when an empty
+	/// block will be allowed.
+	pub fn new(initial: usize, start: Instant, allow_empty: Duration) -> Self {
+		// linear function f(n_candidates) -> valid after microseconds
+		// f(0) = allow_empty
+		// f(initial) = 0
+		// m is actually the negative slope to avoid using signed arithmetic.
+		let (y, m) = if initial != 0 {
+			let y = duration_to_micros(&allow_empty);
+
+			(y, y / initial as u64)
+		} else {
+			(0, 0)
+		};
+
+		DynamicInclusion {
+			start,
+			y,
+			m,
+		}
+	}
+
+	/// Returns the duration from `now` after which the amount of included parachain candidates
+	/// would be enough, or `None` if it is sufficient now.
+	///
+	/// Panics if `now` is earlier than the `start`.
+	pub fn acceptable_in(&self, now: Instant, included: usize) -> Option<Instant> {
+		let elapsed = now.duration_since(self.start);
+		let elapsed = duration_to_micros(&elapsed);
+
+		let valid_after = self.y.saturating_sub(self.m * included as u64);
+
+		if elapsed >= valid_after {
+			None
+		} else {
+			let until = Duration::from_millis((valid_after - elapsed) as u64 / 1000);
+			Some(now + until)
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn full_immediately_allowed() {
+		let now = Instant::now();
+
+		let dynamic = DynamicInclusion::new(
+			10,
+			now,
+			Duration::from_millis(4000),
+		);
+
+		assert!(dynamic.acceptable_in(now, 10).is_none());
+		assert!(dynamic.acceptable_in(now, 11).is_none());
+		assert!(dynamic.acceptable_in(now + Duration::from_millis(2000), 10).is_none());
+	}
+
+	#[test]
+	fn half_allowed_halfway() {
+		let now = Instant::now();
+
+		let dynamic = DynamicInclusion::new(
+			10,
+			now,
+			Duration::from_millis(4000),
+		);
+
+		assert_eq!(dynamic.acceptable_in(now, 5), Some(now + Duration::from_millis(2000)));
+		assert!(dynamic.acceptable_in(now + Duration::from_millis(2000), 5).is_none());
+		assert!(dynamic.acceptable_in(now + Duration::from_millis(3000), 5).is_none());
+		assert!(dynamic.acceptable_in(now + Duration::from_millis(4000), 5).is_none());
+	}
+
+	#[test]
+	fn zero_initial_is_flat() {
+		let now = Instant::now();
+
+		let dynamic = DynamicInclusion::new(
+			0,
+			now,
+			Duration::from_secs(10_000),
+		);
+
+		for i in 0..10_001 {
+			let now = now + Duration::from_secs(i);
+			assert!(dynamic.acceptable_in(now, 0).is_none());
+			assert!(dynamic.acceptable_in(now, 1).is_none());
+			assert!(dynamic.acceptable_in(now, 10).is_none());
+		}
+	}
+}

polkadot/validation/src/error.rs

@@ -0,0 +1,60 @@
+// Copyright 2017 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/>.
+
+//! Errors that can occur during the validation process.
+
+use primitives::Ed25519AuthorityId as AuthorityId;
+use runtime_primitives::RuntimeString;
+
+error_chain! {
+	links {
+		Client(::client::error::Error, ::client::error::ErrorKind);
+		Consensus(::consensus::error::Error, ::consensus::error::ErrorKind);
+	}
+
+	errors {
+		InvalidDutyRosterLength(expected: usize, got: usize) {
+			description("Duty Roster had invalid length"),
+			display("Invalid duty roster length: expected {}, got {}", expected, got),
+		}
+		NotValidator(id: AuthorityId) {
+			description("Local account ID not a validator at this block."),
+			display("Local account ID ({:?}) not a validator at this block.", id),
+		}
+		InherentError(reason: RuntimeString) {
+			description("Unexpected error while checking inherents"),
+			display("Unexpected error while checking inherents: {}", reason),
+		}
+		PrematureDestruction {
+			description("Proposer destroyed before finishing proposing or evaluating"),
+			display("Proposer destroyed before finishing proposing or evaluating"),
+		}
+		Timer(e: ::tokio::timer::Error) {
+			description("Failed to register or resolve async timer."),
+			display("Timer failed: {}", e),
+		}
+		Executor(e: ::futures::future::ExecuteErrorKind) {
+			description("Unable to dispatch agreement future"),
+			display("Unable to dispatch agreement future: {:?}", e),
+		}
+	}
+}
+
+// impl From<::bft::InputStreamConcluded> for Error {
+// 	fn from(err: ::bft::InputStreamConcluded) -> Self {
+// 		::bft::Error::from(err).into()
+// 	}
+// }

polkadot/validation/src/evaluation.rs

@@ -0,0 +1,91 @@
+// Copyright 2017 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/>.
+
+//! Polkadot block evaluation and evaluation errors.
+
+use super::MAX_TRANSACTIONS_SIZE;
+
+use codec::Encode;
+use polkadot_primitives::{Block, Hash, BlockNumber};
+use polkadot_primitives::parachain::Id as ParaId;
+
+error_chain! {
+	links {
+		Client(::client::error::Error, ::client::error::ErrorKind);
+	}
+
+	errors {
+		ProposalNotForPolkadot {
+			description("Proposal provided not a Polkadot block."),
+			display("Proposal provided not a Polkadot block."),
+		}
+		TooManyCandidates(expected: usize, got: usize) {
+			description("Proposal included more candidates than is possible."),
+			display("Proposal included {} candidates for {} parachains", got, expected),
+		}
+		ParachainOutOfOrder {
+			description("Proposal included parachains out of order."),
+			display("Proposal included parachains out of order."),
+		}
+		UnknownParachain(id: ParaId) {
+			description("Proposal included unregistered parachain."),
+			display("Proposal included unregistered parachain {:?}", id),
+		}
+		WrongParentHash(expected: Hash, got: Hash) {
+			description("Proposal had wrong parent hash."),
+			display("Proposal had wrong parent hash. Expected {:?}, got {:?}", expected, got),
+		}
+		WrongNumber(expected: BlockNumber, got: BlockNumber) {
+			description("Proposal had wrong number."),
+			display("Proposal had wrong number. Expected {:?}, got {:?}", expected, got),
+		}
+		ProposalTooLarge(size: usize) {
+			description("Proposal exceeded the maximum size."),
+			display(
+				"Proposal exceeded the maximum size of {} by {} bytes.",
+				MAX_TRANSACTIONS_SIZE, MAX_TRANSACTIONS_SIZE.saturating_sub(*size)
+			),
+		}
+	}
+}
+
+/// Attempt to evaluate a substrate block as a polkadot block, returning error
+/// upon any initial validity checks failing.
+pub fn evaluate_initial(
+	proposal: &Block,
+	_now: u64,
+	parent_hash: &Hash,
+	parent_number: BlockNumber,
+	_active_parachains: &[ParaId],
+) -> Result<()> {
+	let transactions_size = proposal.extrinsics.iter().fold(0, |a, tx| {
+		a + Encode::encode(tx).len()
+	});
+
+	if transactions_size > MAX_TRANSACTIONS_SIZE {
+		bail!(ErrorKind::ProposalTooLarge(transactions_size))
+	}
+
+	if proposal.header.parent_hash != *parent_hash {
+		bail!(ErrorKind::WrongParentHash(*parent_hash, proposal.header.parent_hash));
+	}
+
+	if proposal.header.number != parent_number + 1 {
+		bail!(ErrorKind::WrongNumber(parent_number + 1, proposal.header.number));
+	}
+
+	Ok(())
+}

polkadot/validation/src/lib.rs

@@ -0,0 +1,813 @@
+// Copyright 2017 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/>.
+
+//! Propagation and agreement of candidates.
+//!
+//! Authorities are split into groups by parachain, and each authority might come
+//! up its own candidate for their parachain. Within groups, authorities pass around
+//! their candidates and produce statements of validity.
+//!
+//! Any candidate that receives majority approval by the authorities in a group
+//! may be subject to inclusion, unless any authorities flag that candidate as invalid.
+//!
+//! Wrongly flagging as invalid should be strongly disincentivized, so that in the
+//! equilibrium state it is not expected to happen. Likewise with the submission
+//! of invalid blocks.
+//!
+//! Groups themselves may be compromised by malicious authorities.
+
+extern crate parking_lot;
+extern crate polkadot_availability_store as extrinsic_store;
+extern crate polkadot_statement_table as table;
+extern crate polkadot_parachain as parachain;
+extern crate polkadot_runtime;
+extern crate polkadot_primitives;
+
+extern crate parity_codec as codec;
+extern crate substrate_primitives as primitives;
+extern crate sr_primitives as runtime_primitives;
+extern crate substrate_client as client;
+extern crate substrate_trie as trie;
+
+extern crate exit_future;
+extern crate tokio;
+extern crate substrate_consensus_common as consensus;
+extern crate substrate_consensus_aura as aura;
+extern crate substrate_finality_grandpa as grandpa;
+extern crate substrate_transaction_pool as transaction_pool;
+extern crate substrate_inherents as inherents;
+extern crate srml_aura as runtime_aura;
+
+#[macro_use]
+extern crate error_chain;
+
+#[macro_use]
+extern crate futures;
+
+#[macro_use]
+extern crate log;
+
+#[cfg(test)]
+extern crate substrate_keyring;
+
+use std::collections::{HashMap, HashSet};
+use std::sync::Arc;
+use std::time::{self, Duration, Instant};
+
+use client::{BlockchainEvents, ChainHead, BlockBody};
+use client::blockchain::HeaderBackend;
+use client::block_builder::api::BlockBuilder as BlockBuilderApi;
+use client::runtime_api::Core;
+use codec::Encode;
+use extrinsic_store::Store as ExtrinsicStore;
+use parking_lot::Mutex;
+use polkadot_primitives::{Hash, Block, BlockId, BlockNumber, Header, SessionKey};
+use polkadot_primitives::parachain::{
+	Id as ParaId, Chain, DutyRoster, BlockData, Extrinsic as ParachainExtrinsic, CandidateReceipt,
+	CandidateSignature, ParachainHost, AttestedCandidate, Statement as PrimitiveStatement, Message,
+	OutgoingMessage,
+};
+use primitives::{Ed25519AuthorityId as AuthorityId, ed25519};
+use runtime_primitives::{traits::{ProvideRuntimeApi, Header as HeaderT}, ApplyError};
+use tokio::runtime::TaskExecutor;
+use tokio::timer::{Delay, Interval};
+use transaction_pool::txpool::{Pool, ChainApi as PoolChainApi};
+
+use attestation_service::ServiceHandle;
+use futures::prelude::*;
+use futures::future::{self, Either};
+use collation::CollationFetch;
+use dynamic_inclusion::DynamicInclusion;
+use inherents::InherentData;
+use runtime_aura::timestamp::TimestampInherentData;
+use aura::SlotDuration;
+
+pub use self::collation::{validate_collation, message_queue_root, egress_roots, Collators};
+pub use self::error::{ErrorKind, Error};
+pub use self::shared_table::{
+	SharedTable, ParachainWork, PrimedParachainWork, Validated, Statement, SignedStatement,
+	GenericStatement,
+};
+
+mod attestation_service;
+mod dynamic_inclusion;
+mod evaluation;
+mod error;
+mod shared_table;
+
+pub mod collation;
+
+// block size limit.
+const MAX_TRANSACTIONS_SIZE: usize = 4 * 1024 * 1024;
+
+/// Incoming messages; a series of sorted (ParaId, Message) pairs.
+pub type Incoming = Vec<(ParaId, Vec<Message>)>;
+
+/// Outgoing messages from various candidates.
+pub type Outgoing = Vec<MessagesFrom>;
+
+/// Some messages from a parachain.
+pub struct MessagesFrom {
+	/// The parachain originating the messages.
+	pub from: ParaId,
+	/// The messages themselves.
+	pub messages: ParachainExtrinsic,
+}
+
+impl MessagesFrom {
+	/// Construct from the raw messages.
+	pub fn from_messages(from: ParaId, messages: Vec<OutgoingMessage>) -> Self {
+		MessagesFrom {
+			from,
+			messages: ParachainExtrinsic { outgoing_messages: messages },
+		}
+	}
+}
+
+/// A handle to a statement table router.
+///
+/// This is expected to be a lightweight, shared type like an `Arc`.
+pub trait TableRouter: Clone {
+	/// Errors when fetching data from the network.
+	type Error: std::fmt::Debug;
+	/// Future that resolves when candidate data is fetched.
+	type FetchCandidate: IntoFuture<Item=BlockData,Error=Self::Error>;
+	/// Fetch incoming messages for a candidate.
+	type FetchIncoming: IntoFuture<Item=Incoming,Error=Self::Error>;
+
+	/// Call with local candidate data. This will make the data available on the network,
+	/// and sign, import, and broadcast a statement about the candidate.
+	fn local_candidate(&self, candidate: CandidateReceipt, block_data: BlockData, extrinsic: ParachainExtrinsic);
+
+	/// Fetch block data for a specific candidate.
+	fn fetch_block_data(&self, candidate: &CandidateReceipt) -> Self::FetchCandidate;
+
+	/// Fetches the incoming message data to a parachain from the network. Incoming data should be
+	/// checked.
+	///
+	/// The `ParachainHost::ingress` function can be used to fetch incoming roots,
+	/// and the `message_queue_root` function can be used to check that messages actually have
+	/// expected root.
+	fn fetch_incoming(&self, id: ParaId) -> Self::FetchIncoming;
+}
+
+/// A long-lived network which can create parachain statement and BFT message routing processes on demand.
+pub trait Network {
+	/// The table router type. This should handle importing of any statements,
+	/// routing statements to peers, and driving completion of any `StatementProducers`.
+	type TableRouter: TableRouter;
+
+	/// Instantiate a table router using the given shared table.
+	/// Also pass through any outgoing messages to be broadcast to peers.
+	fn communication_for(
+		&self,
+		table: Arc<SharedTable>,
+		outgoing: Outgoing,
+	) -> Self::TableRouter;
+}
+
+/// Information about a specific group.
+#[derive(Debug, Clone, Default)]
+pub struct GroupInfo {
+	/// Authorities meant to check validity of candidates.
+	pub validity_guarantors: HashSet<SessionKey>,
+	/// Number of votes needed for validity.
+	pub needed_validity: usize,
+}
+
+/// Sign a table statement against a parent hash.
+/// The actual message signed is the encoded statement concatenated with the
+/// parent hash.
+pub fn sign_table_statement(statement: &Statement, key: &ed25519::Pair, parent_hash: &Hash) -> CandidateSignature {
+	// we sign using the primitive statement type because that's what the runtime
+	// expects. These types probably encode the same way so this clone could be optimized
+	// out in the future.
+	let mut encoded = PrimitiveStatement::from(statement.clone()).encode();
+	encoded.extend(parent_hash.as_ref());
+
+	key.sign(&encoded).into()
+}
+
+/// Check signature on table statement.
+pub fn check_statement(statement: &Statement, signature: &CandidateSignature, signer: SessionKey, parent_hash: &Hash) -> bool {
+	use runtime_primitives::traits::Verify;
+
+	let mut encoded = PrimitiveStatement::from(statement.clone()).encode();
+	encoded.extend(parent_hash.as_ref());
+
+	signature.verify(&encoded[..], &signer.into())
+}
+
+/// Compute group info out of a duty roster and a local authority set.
+pub fn make_group_info(
+	roster: DutyRoster,
+	authorities: &[AuthorityId],
+	local_id: AuthorityId,
+) -> Result<(HashMap<ParaId, GroupInfo>, LocalDuty), Error> {
+	if roster.validator_duty.len() != authorities.len() {
+		bail!(ErrorKind::InvalidDutyRosterLength(authorities.len(), roster.validator_duty.len()))
+	}
+
+	let mut local_validation = None;
+	let mut map = HashMap::new();
+
+	let duty_iter = authorities.iter().zip(&roster.validator_duty);
+	for (authority, v_duty) in duty_iter {
+		if authority == &local_id {
+			local_validation = Some(v_duty.clone());
+		}
+
+		match *v_duty {
+			Chain::Relay => {}, // does nothing for now.
+			Chain::Parachain(ref id) => {
+				map.entry(id.clone()).or_insert_with(GroupInfo::default)
+					.validity_guarantors
+					.insert(authority.clone());
+			}
+		}
+	}
+
+	for live_group in map.values_mut() {
+		let validity_len = live_group.validity_guarantors.len();
+		live_group.needed_validity = validity_len / 2 + validity_len % 2;
+	}
+
+	match local_validation {
+		Some(local_validation) => {
+			let local_duty = LocalDuty {
+				validation: local_validation,
+			};
+
+			Ok((map, local_duty))
+		}
+		None => bail!(ErrorKind::NotValidator(local_id)),
+	}
+}
+
+/// Constructs parachain-agreement instances.
+struct ParachainValidation<C, N, P> {
+	/// The client instance.
+	client: Arc<P>,
+	/// The backing network handle.
+	network: N,
+	/// Parachain collators.
+	collators: C,
+	/// handle to remote task executor
+	handle: TaskExecutor,
+	/// Store for extrinsic data.
+	extrinsic_store: ExtrinsicStore,
+	/// Live agreements. Maps relay chain parent hashes to attestation
+	/// instances.
+	live_instances: Mutex<HashMap<Hash, Arc<AttestationTracker>>>,
+}
+
+impl<C, N, P> ParachainValidation<C, N, P> where
+	C: Collators + Send + 'static,
+	N: Network,
+	P: ProvideRuntimeApi + HeaderBackend<Block> + BlockBody<Block> + Send + Sync + 'static,
+	P::Api: ParachainHost<Block> + BlockBuilderApi<Block>,
+	<C::Collation as IntoFuture>::Future: Send + 'static,
+	N::TableRouter: Send + 'static,
+	<<N::TableRouter as TableRouter>::FetchIncoming as IntoFuture>::Future: Send + 'static,
+{
+	/// Get an attestation table for given parent hash.
+	///
+	/// This starts a parachain agreement process on top of the parent hash if
+	/// one has not already started.
+	///
+	/// Additionally, this will trigger broadcast of data to the new block's duty
+	/// roster.
+	fn get_or_instantiate(
+		&self,
+		parent_hash: Hash,
+		grandparent_hash: Hash,
+		authorities: &[AuthorityId],
+		sign_with: Arc<ed25519::Pair>,
+	)
+		-> Result<Arc<AttestationTracker>, Error>
+	{
+		let mut live_instances = self.live_instances.lock();
+		if let Some(tracker) = live_instances.get(&parent_hash) {
+			return Ok(tracker.clone());
+		}
+
+		let id = BlockId::hash(parent_hash);
+
+		// compute the parent candidates, if we know of them.
+		// this will allow us to circulate outgoing messages to other peers as necessary.
+		let parent_candidates: Vec<_> = ::attestation_service::fetch_candidates(&*self.client, &id)
+			.ok()
+			.and_then(|x| x)
+			.map(|x| x.collect())
+			.unwrap_or_default();
+
+		let outgoing: Vec<_> = {
+			// extract all extrinsic data that we have and propagate to peers.
+			live_instances.get(&grandparent_hash).map(|parent_validation| {
+				parent_candidates.iter().filter_map(|c| {
+					let para_id = c.parachain_index;
+					let hash = c.hash();
+					parent_validation.table.extrinsic_data(&hash).map(|ex| MessagesFrom {
+						from: para_id,
+						messages: ex,
+					})
+				}).collect()
+			}).unwrap_or_default()
+		};
+
+		let duty_roster = self.client.runtime_api().duty_roster(&id)?;
+
+		let (group_info, local_duty) = make_group_info(
+			duty_roster,
+			authorities,
+			sign_with.public().into(),
+		)?;
+
+		info!("Starting parachain attestation session on top of parent {:?}. Local parachain duty is {:?}",
+			parent_hash, local_duty.validation);
+
+		let active_parachains = self.client.runtime_api().active_parachains(&id)?;
+
+		debug!(target: "validation", "Active parachains: {:?}", active_parachains);
+
+		let table = Arc::new(SharedTable::new(group_info, sign_with.clone(), parent_hash, self.extrinsic_store.clone()));
+		let router = self.network.communication_for(
+			table.clone(),
+			outgoing,
+		);
+
+		let drop_signal = match local_duty.validation {
+			Chain::Parachain(id) => Some(self.launch_work(
+				parent_hash,
+				id,
+				router,
+			)),
+			Chain::Relay => None,
+		};
+
+		let tracker = Arc::new(AttestationTracker {
+			table,
+			started: Instant::now(),
+			_drop_signal: drop_signal
+		});
+
+		live_instances.insert(parent_hash, tracker.clone());
+
+		Ok(tracker)
+	}
+
+	/// Retain validation sessions matching predicate.
+	fn retain<F: FnMut(&Hash) -> bool>(&self, mut pred: F) {
+		self.live_instances.lock().retain(|k, _| pred(k))
+	}
+
+	// launch parachain work asynchronously.
+	fn launch_work(
+		&self,
+		relay_parent: Hash,
+		validation_para: ParaId,
+		router: N::TableRouter,
+	) -> exit_future::Signal {
+		use extrinsic_store::Data;
+
+		let (signal, exit) = exit_future::signal();
+
+		let fetch_incoming = router.fetch_incoming(validation_para)
+			.into_future()
+			.map_err(|e| format!("{:?}", e));
+
+		// fetch incoming messages to our parachain from network and
+		// then fetch a local collation.
+		let (collators, client) = (self.collators.clone(), self.client.clone());
+		let collation_work = fetch_incoming
+			.map_err(|e| String::clone(&e))
+			.and_then(move |incoming| {
+				CollationFetch::new(
+					validation_para,
+					relay_parent,
+					collators,
+					client,
+					incoming,
+				).map_err(|e| format!("{:?}", e))
+			});
+
+		let extrinsic_store = self.extrinsic_store.clone();
+		let handled_work = collation_work.then(move |result| match result {
+			Ok((collation, extrinsic)) => {
+				let res = extrinsic_store.make_available(Data {
+					relay_parent,
+					parachain_id: collation.receipt.parachain_index,
+					candidate_hash: collation.receipt.hash(),
+					block_data: collation.block_data.clone(),
+					extrinsic: Some(extrinsic.clone()),
+				});
+
+				match res {
+					Ok(()) => {
+						// TODO: https://github.com/paritytech/polkadot/issues/51
+						// Erasure-code and provide merkle branches.
+						router.local_candidate(collation.receipt, collation.block_data, extrinsic)
+					}
+					Err(e) => warn!(
+						target: "validation",
+						"Failed to make collation data available: {:?}",
+						e,
+					),
+				}
+
+				Ok(())
+			}
+			Err(e) => {
+				warn!(target: "validation", "Failed to collate candidate: {}", e);
+				Ok(())
+			}
+		});
+
+		let cancellable_work = handled_work.select(exit).then(|_| Ok(()));
+
+		// spawn onto thread pool.
+		self.handle.spawn(cancellable_work);
+		signal
+	}
+}
+
+/// Parachain validation for a single block.
+struct AttestationTracker {
+	_drop_signal: Option<exit_future::Signal>,
+	table: Arc<SharedTable>,
+	started: Instant,
+}
+
+/// Polkadot proposer factory.
+pub struct ProposerFactory<C, N, P, TxApi: PoolChainApi> {
+	parachain_validation: Arc<ParachainValidation<C, N, P>>,
+	transaction_pool: Arc<Pool<TxApi>>,
+	key: Arc<ed25519::Pair>,
+	_service_handle: ServiceHandle,
+	aura_slot_duration: SlotDuration,
+}
+
+impl<C, N, P, TxApi> ProposerFactory<C, N, P, TxApi> where
+	C: Collators + Send + Sync + 'static,
+	<C::Collation as IntoFuture>::Future: Send + 'static,
+	P: BlockchainEvents<Block> + ChainHead<Block> + BlockBody<Block>,
+	P: ProvideRuntimeApi + HeaderBackend<Block> + Send + Sync + 'static,
+	P::Api: ParachainHost<Block> + Core<Block> + BlockBuilderApi<Block>,
+	N: Network + Send + Sync + 'static,
+	N::TableRouter: Send + 'static,
+	<<N::TableRouter as TableRouter>::FetchIncoming as IntoFuture>::Future: Send + 'static,
+	TxApi: PoolChainApi,
+{
+	/// Create a new proposer factory.
+	pub fn new(
+		client: Arc<P>,
+		network: N,
+		collators: C,
+		transaction_pool: Arc<Pool<TxApi>>,
+		thread_pool: TaskExecutor,
+		key: Arc<ed25519::Pair>,
+		extrinsic_store: ExtrinsicStore,
+		aura_slot_duration: SlotDuration,
+	) -> Self {
+		let parachain_validation = Arc::new(ParachainValidation {
+			client: client.clone(),
+			network,
+			collators,
+			handle: thread_pool.clone(),
+			extrinsic_store: extrinsic_store.clone(),
+			live_instances: Mutex::new(HashMap::new()),
+		});
+
+		let service_handle = ::attestation_service::start(
+			client,
+			parachain_validation.clone(),
+			thread_pool,
+			key.clone(),
+			extrinsic_store,
+		);
+
+		ProposerFactory {
+			parachain_validation,
+			transaction_pool,
+			key,
+			_service_handle: service_handle,
+			aura_slot_duration,
+		}
+	}
+}
+
+impl<C, N, P, TxApi> consensus::Environment<Block> for ProposerFactory<C, N, P, TxApi> where
+	C: Collators + Send + 'static,
+	N: Network,
+	TxApi: PoolChainApi<Block=Block>,
+	P: ProvideRuntimeApi + HeaderBackend<Block> + BlockBody<Block> + Send + Sync + 'static,
+	P::Api: ParachainHost<Block> + BlockBuilderApi<Block>,
+	<C::Collation as IntoFuture>::Future: Send + 'static,
+	N::TableRouter: Send + 'static,
+	<<N::TableRouter as TableRouter>::FetchIncoming as IntoFuture>::Future: Send + 'static,
+{
+	type Proposer = Proposer<P, TxApi>;
+	type Error = Error;
+
+	fn init(
+		&self,
+		parent_header: &Header,
+		authorities: &[AuthorityId],
+	) -> Result<Self::Proposer, Error> {
+		let parent_hash = parent_header.hash();
+		let parent_id = BlockId::hash(parent_hash);
+		let sign_with = self.key.clone();
+		let tracker = self.parachain_validation.get_or_instantiate(
+			parent_hash,
+			parent_header.parent_hash().clone(),
+			authorities,
+			sign_with,
+		)?;
+
+		Ok(Proposer {
+			client: self.parachain_validation.client.clone(),
+			tracker,
+			parent_hash,
+			parent_id,
+			parent_number: parent_header.number,
+			transaction_pool: self.transaction_pool.clone(),
+			slot_duration: self.aura_slot_duration,
+		})
+	}
+}
+
+/// The local duty of a validator.
+pub struct LocalDuty {
+	validation: Chain,
+}
+
+/// The Polkadot proposer logic.
+pub struct Proposer<C: Send + Sync, TxApi: PoolChainApi> where
+	C: ProvideRuntimeApi + HeaderBackend<Block>,
+{
+	client: Arc<C>,
+	parent_hash: Hash,
+	parent_id: BlockId,
+	parent_number: BlockNumber,
+	tracker: Arc<AttestationTracker>,
+	transaction_pool: Arc<Pool<TxApi>>,
+	slot_duration: SlotDuration,
+}
+
+impl<C, TxApi> consensus::Proposer<Block> for Proposer<C, TxApi> where
+	TxApi: PoolChainApi<Block=Block>,
+	C: ProvideRuntimeApi + HeaderBackend<Block> + Send + Sync,
+	C::Api: ParachainHost<Block> + BlockBuilderApi<Block>,
+{
+	type Error = Error;
+	type Create = Either<CreateProposal<C, TxApi>, future::FutureResult<Block, Error>>;
+
+	fn propose(&self, inherent_data: InherentData, max_duration: Duration) -> Self::Create {
+		const ATTEMPT_PROPOSE_EVERY: Duration = Duration::from_millis(100);
+		const SLOT_DURATION_DENOMINATOR: u64 = 3; // wait up to 1/3 of the slot for candidates.
+
+		let initial_included = self.tracker.table.includable_count();
+		let now = Instant::now();
+
+		let dynamic_inclusion = DynamicInclusion::new(
+			self.tracker.table.num_parachains(),
+			self.tracker.started,
+			Duration::from_secs(self.slot_duration.get() / SLOT_DURATION_DENOMINATOR),
+		);
+
+		let enough_candidates = dynamic_inclusion.acceptable_in(
+			now,
+			initial_included,
+		).unwrap_or_else(|| now + Duration::from_millis(1));
+
+		let believed_timestamp = match inherent_data.timestamp_inherent_data() {
+			Ok(timestamp) => timestamp,
+			Err(e) => return Either::B(future::err(ErrorKind::InherentError(e).into())),
+		};
+
+		// set up delay until next allowed timestamp.
+		let current_timestamp = current_timestamp();
+		let delay_future = if current_timestamp >= believed_timestamp {
+			None
+		} else {
+			Some(Delay::new(
+				Instant::now() + Duration::from_secs(current_timestamp - believed_timestamp)
+			))
+		};
+
+		let timing = ProposalTiming {
+			minimum: delay_future,
+			attempt_propose: Interval::new(now + ATTEMPT_PROPOSE_EVERY, ATTEMPT_PROPOSE_EVERY),
+			enough_candidates: Delay::new(enough_candidates),
+			dynamic_inclusion,
+			last_included: initial_included,
+		};
+
+		Either::A(CreateProposal {
+			parent_hash: self.parent_hash.clone(),
+			parent_number: self.parent_number.clone(),
+			parent_id: self.parent_id.clone(),
+			client: self.client.clone(),
+			transaction_pool: self.transaction_pool.clone(),
+			table: self.tracker.table.clone(),
+			believed_minimum_timestamp: believed_timestamp,
+			timing,
+			inherent_data: Some(inherent_data),
+			// leave some time for the proposal finalisation
+			deadline: Instant::now() + max_duration - max_duration / 3,
+		})
+	}
+}
+
+fn current_timestamp() -> u64 {
+	time::SystemTime::now().duration_since(time::UNIX_EPOCH)
+		.expect("now always later than unix epoch; qed")
+		.as_secs()
+		.into()
+}
+
+struct ProposalTiming {
+	minimum: Option<Delay>,
+	attempt_propose: Interval,
+	dynamic_inclusion: DynamicInclusion,
+	enough_candidates: Delay,
+	last_included: usize,
+}
+
+impl ProposalTiming {
+	// whether it's time to attempt a proposal.
+	// shouldn't be called outside of the context of a task.
+	fn poll(&mut self, included: usize) -> Poll<(), ErrorKind> {
+		// first drain from the interval so when the minimum delay is up
+		// we don't have any notifications built up.
+		//
+		// this interval is just meant to produce periodic task wakeups
+		// that lead to the `dynamic_inclusion` getting updated as necessary.
+		while let Async::Ready(x) = self.attempt_propose.poll().map_err(ErrorKind::Timer)? {
+			x.expect("timer still alive; intervals never end; qed");
+		}
+
+		// wait until the minimum time has passed.
+		if let Some(mut minimum) = self.minimum.take() {
+			if let Async::NotReady = minimum.poll().map_err(ErrorKind::Timer)? {
+				self.minimum = Some(minimum);
+				return Ok(Async::NotReady);
+			}
+		}
+
+		if included == self.last_included {
+			return self.enough_candidates.poll().map_err(ErrorKind::Timer);
+		}
+
+		// the amount of includable candidates has changed. schedule a wakeup
+		// if it's not sufficient anymore.
+		match self.dynamic_inclusion.acceptable_in(Instant::now(), included) {
+			Some(instant) => {
+				self.last_included = included;
+				self.enough_candidates.reset(instant);
+				self.enough_candidates.poll().map_err(ErrorKind::Timer)
+			}
+			None => Ok(Async::Ready(())),
+		}
+	}
+}
+
+/// Future which resolves upon the creation of a proposal.
+pub struct CreateProposal<C: Send + Sync, TxApi: PoolChainApi> {
+	parent_hash: Hash,
+	parent_number: BlockNumber,
+	parent_id: BlockId,
+	client: Arc<C>,
+	transaction_pool: Arc<Pool<TxApi>>,
+	table: Arc<SharedTable>,
+	timing: ProposalTiming,
+	believed_minimum_timestamp: u64,
+	inherent_data: Option<InherentData>,
+	deadline: Instant,
+}
+
+impl<C, TxApi> CreateProposal<C, TxApi> where
+	TxApi: PoolChainApi<Block=Block>,
+	C: ProvideRuntimeApi + HeaderBackend<Block> + Send + Sync,
+	C::Api: ParachainHost<Block> + BlockBuilderApi<Block>,
+{
+	fn propose_with(&mut self, candidates: Vec<AttestedCandidate>) -> Result<Block, Error> {
+		use client::block_builder::BlockBuilder;
+		use runtime_primitives::traits::{Hash as HashT, BlakeTwo256};
+
+		let mut inherent_data = self.inherent_data.take().expect("CreateProposal is not polled after finishing; qed");
+		inherent_data.put_data(polkadot_runtime::PARACHAIN_INHERENT_IDENTIFIER, &candidates).map_err(ErrorKind::InherentError)?;
+
+		let runtime_api = self.client.runtime_api();
+
+		let mut block_builder = BlockBuilder::at_block(&self.parent_id, &*self.client)?;
+
+		{
+			let inherents = runtime_api.inherent_extrinsics(&self.parent_id, inherent_data)?;
+			for inherent in inherents {
+				block_builder.push(inherent)?;
+			}
+
+			let mut unqueue_invalid = Vec::new();
+
+			for ready in self.transaction_pool.ready() {
+				if Instant::now() > self.deadline {
+					debug!("Validation deadline reached when pushing block transactions, proceeding with proposing.");
+					break;
+				}
+
+				match block_builder.push(ready.data.clone()) {
+					Ok(()) => {
+						debug!("[{:?}] Pushed to the block.", ready.hash);
+					}
+					Err(client::error::Error(client::error::ErrorKind::ApplyExtrinsicFailed(ApplyError::FullBlock), _)) => {
+						debug!("Block is full, proceed with proposing.");
+						break;
+					}
+					Err(e) => {
+						trace!(target: "transaction-pool", "Invalid transaction: {}", e);
+						unqueue_invalid.push(ready.hash.clone());
+					}
+				}
+			}
+
+			self.transaction_pool.remove_invalid(&unqueue_invalid);
+		}
+
+		let new_block = block_builder.bake()?;
+
+		info!("Prepared block for proposing at {} [hash: {:?}; parent_hash: {}; extrinsics: [{}]]",
+			new_block.header.number,
+			Hash::from(new_block.header.hash()),
+			new_block.header.parent_hash,
+			new_block.extrinsics.iter()
+				.map(|xt| format!("{}", BlakeTwo256::hash_of(xt)))
+				.collect::<Vec<_>>()
+				.join(", ")
+		);
+
+		// TODO: full re-evaluation
+		let active_parachains = runtime_api.active_parachains(&self.parent_id)?;
+		assert!(evaluation::evaluate_initial(
+			&new_block,
+			self.believed_minimum_timestamp,
+			&self.parent_hash,
+			self.parent_number,
+			&active_parachains,
+		).is_ok());
+
+		Ok(new_block)
+	}
+}
+
+impl<C, TxApi> Future for CreateProposal<C, TxApi> where
+	TxApi: PoolChainApi<Block=Block>,
+	C: ProvideRuntimeApi + HeaderBackend<Block> + Send + Sync,
+	C::Api: ParachainHost<Block> + BlockBuilderApi<Block>,
+{
+	type Item = Block;
+	type Error = Error;
+
+	fn poll(&mut self) -> Poll<Block, Error> {
+		// 1. try to propose if we have enough includable candidates and other
+		// delays have concluded.
+		let included = self.table.includable_count();
+		try_ready!(self.timing.poll(included));
+
+		// 2. propose
+		let proposed_candidates = self.table.proposed_set();
+
+		self.propose_with(proposed_candidates).map(Async::Ready)
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use substrate_keyring::Keyring;
+
+	#[test]
+	fn sign_and_check_statement() {
+		let statement: Statement = GenericStatement::Valid([1; 32].into());
+		let parent_hash = [2; 32].into();
+
+		let sig = sign_table_statement(&statement, &Keyring::Alice.pair(), &parent_hash);
+
+		assert!(check_statement(&statement, &sig, Keyring::Alice.to_raw_public().into(), &parent_hash));
+		assert!(!check_statement(&statement, &sig, Keyring::Alice.to_raw_public().into(), &[0xff; 32].into()));
+		assert!(!check_statement(&statement, &sig, Keyring::Bob.to_raw_public().into(), &parent_hash));
+	}
+}

polkadot/validation/src/shared_table/includable.rs

@@ -0,0 +1,137 @@
+// Copyright 2017 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/>.
+
+//! Implements a future which resolves when all of the candidates referenced are includable.
+
+use std::collections::HashMap;
+
+use futures::prelude::*;
+use futures::sync::oneshot;
+
+use polkadot_primitives::Hash;
+
+/// Track includability of a set of candidates,
+pub(super) fn track<I: IntoIterator<Item=(Hash, bool)>>(candidates: I) -> (IncludabilitySender, Includable) {
+	let (tx, rx) = oneshot::channel();
+	let tracking: HashMap<_, _> = candidates.into_iter().collect();
+	let includable_count = tracking.values().filter(|x| **x).count();
+
+	let mut sender = IncludabilitySender {
+		tracking,
+		includable_count,
+		sender: Some(tx),
+	};
+
+	sender.try_complete();
+
+	(
+		sender,
+		Includable(rx),
+	)
+}
+
+/// The sending end of the includability sender.
+pub(super) struct IncludabilitySender {
+	tracking: HashMap<Hash, bool>,
+	includable_count: usize,
+	sender: Option<oneshot::Sender<()>>,
+}
+
+impl IncludabilitySender {
+	/// update the inner candidate. wakes up the task as necessary.
+	/// returns `Err(Canceled)` if the other end has hung up.
+	///
+	/// returns `true` when this is completed and should be destroyed.
+	pub fn update_candidate(&mut self, candidate: Hash, includable: bool) -> bool {
+		use std::collections::hash_map::Entry;
+
+		match self.tracking.entry(candidate) {
+			Entry::Vacant(_) => {}
+			Entry::Occupied(mut entry) => {
+				let old = entry.insert(includable);
+				if !old && includable {
+					self.includable_count += 1;
+				} else if old && !includable {
+					self.includable_count -= 1;
+				}
+			}
+		}
+
+		self.try_complete()
+	}
+
+	/// whether the sender is completed.
+	pub fn is_complete(&self) -> bool {
+		self.sender.is_none()
+	}
+
+	fn try_complete(&mut self) -> bool {
+		if self.includable_count == self.tracking.len() {
+			if let Some(sender) = self.sender.take() {
+				let _ = sender.send(());
+			}
+
+			true
+		} else {
+			false
+		}
+	}
+}
+
+/// Future that resolves when all the candidates within are includable.
+pub struct Includable(oneshot::Receiver<()>);
+
+impl Future for Includable {
+	type Item = ();
+	type Error = oneshot::Canceled;
+
+	fn poll(&mut self) -> Poll<(), oneshot::Canceled> {
+		self.0.poll()
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[test]
+	fn it_works() {
+		let hash1 = [1; 32].into();
+		let hash2 = [2; 32].into();
+		let hash3 = [3; 32].into();
+
+		let (mut sender, recv) = track([
+			(hash1, true),
+			(hash2, true),
+			(hash2, false), // overwrite should favor latter.
+			(hash3, true),
+		].iter().cloned());
+
+		assert!(!sender.is_complete());
+
+		// true -> false transition is possible and should be handled.
+		sender.update_candidate(hash1, false);
+		assert!(!sender.is_complete());
+
+		sender.update_candidate(hash2, true);
+		assert!(!sender.is_complete());
+
+		sender.update_candidate(hash1, true);
+		assert!(sender.is_complete());
+
+		recv.wait().unwrap();
+	}
+}

polkadot/validation/src/shared_table/mod.rs

@@ -0,0 +1,861 @@
+// Copyright 2017 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/>.
+
+//! Parachain statement table meant to be shared with a message router
+//! and a consensus proposer.
+
+use std::collections::hash_map::{HashMap, Entry};
+use std::sync::Arc;
+
+use extrinsic_store::{Data, Store as ExtrinsicStore};
+use table::{self, Table, Context as TableContextTrait};
+use polkadot_primitives::{Block, BlockId, Hash, SessionKey};
+use polkadot_primitives::parachain::{
+	Id as ParaId, BlockData, Collation, Extrinsic, CandidateReceipt,
+	AttestedCandidate, ParachainHost
+};
+
+use parking_lot::Mutex;
+use futures::{future, prelude::*};
+
+use super::{GroupInfo, Incoming, TableRouter};
+use self::includable::IncludabilitySender;
+use primitives::ed25519;
+use runtime_primitives::{traits::ProvideRuntimeApi};
+
+mod includable;
+
+pub use self::includable::Includable;
+pub use table::{SignedStatement, Statement};
+pub use table::generic::Statement as GenericStatement;
+
+struct TableContext {
+	parent_hash: Hash,
+	key: Arc<ed25519::Pair>,
+	groups: HashMap<ParaId, GroupInfo>,
+}
+
+impl table::Context for TableContext {
+	fn is_member_of(&self, authority: &SessionKey, group: &ParaId) -> bool {
+		self.groups.get(group).map_or(false, |g| g.validity_guarantors.contains(authority))
+	}
+
+	fn requisite_votes(&self, group: &ParaId) -> usize {
+		self.groups.get(group).map_or(usize::max_value(), |g| g.needed_validity)
+	}
+}
+
+impl TableContext {
+	fn local_id(&self) -> SessionKey {
+		self.key.public().into()
+	}
+
+	fn sign_statement(&self, statement: table::Statement) -> table::SignedStatement {
+		let signature = ::sign_table_statement(&statement, &self.key, &self.parent_hash).into();
+
+		table::SignedStatement {
+			statement,
+			signature,
+			sender: self.local_id(),
+		}
+	}
+}
+
+pub(crate) enum Validation {
+	Valid(BlockData, Extrinsic),
+	Invalid(BlockData), // should take proof.
+}
+
+enum ValidationWork {
+	Done(Validation),
+	InProgress,
+	Error(String),
+}
+
+#[cfg(test)]
+impl ValidationWork {
+	fn is_in_progress(&self) -> bool {
+		match *self {
+			ValidationWork::InProgress => true,
+			_ => false,
+		}
+	}
+
+	fn is_done(&self) -> bool {
+		match *self {
+			ValidationWork::Done(_) => true,
+			_ => false,
+		}
+	}
+}
+
+// A shared table object.
+struct SharedTableInner {
+	table: Table<TableContext>,
+	trackers: Vec<IncludabilitySender>,
+	extrinsic_store: ExtrinsicStore,
+	validated: HashMap<Hash, ValidationWork>,
+}
+
+impl SharedTableInner {
+	// Import a single statement. Provide a handle to a table router and a function
+	// used to determine if a referenced candidate is valid.
+	//
+	// the statement producer, if any, will produce only statements concerning the same candidate
+	// as the one just imported
+	fn import_remote_statement<R: TableRouter>(
+		&mut self,
+		context: &TableContext,
+		router: &R,
+		statement: table::SignedStatement,
+	) -> Option<ParachainWork<future::Join<
+		<R::FetchCandidate as IntoFuture>::Future,
+		<R::FetchIncoming as IntoFuture>::Future,
+	>>> {
+		let summary = match self.table.import_statement(context, statement) {
+			Some(summary) => summary,
+			None => return None,
+		};
+
+		self.update_trackers(&summary.candidate, context);
+
+		let local_id = context.local_id();
+
+		let para_member = context.is_member_of(&local_id, &summary.group_id);
+
+		let digest = &summary.candidate;
+
+		// TODO: consider a strategy based on the number of candidate votes as well.
+		let do_validation = para_member && match self.validated.entry(digest.clone()) {
+			Entry::Occupied(_) => false,
+			Entry::Vacant(entry) => {
+				entry.insert(ValidationWork::InProgress);
+				true
+			}
+		};
+
+		let work = if do_validation {
+			let fetch_incoming = router.fetch_incoming(summary.group_id);
+			match self.table.get_candidate(&digest) {
+				None => {
+					let message = format!(
+						"Table inconsistency detected. Summary returned for candidate {} \
+						but receipt not present in table.",
+						digest,
+					);
+
+					warn!(target: "validation", "{}", message);
+					self.validated.insert(digest.clone(), ValidationWork::Error(message));
+					None
+				}
+				Some(candidate) => {
+					let fetch_block_data = router.fetch_block_data(candidate).into_future();
+
+					Some(Work {
+						candidate_receipt: candidate.clone(),
+						fetch: fetch_block_data.join(fetch_incoming),
+					})
+				}
+			}
+		} else {
+			None
+		};
+
+		work.map(|work| ParachainWork {
+			extrinsic_store: self.extrinsic_store.clone(),
+			relay_parent: context.parent_hash.clone(),
+			work
+		})
+	}
+
+	fn update_trackers(&mut self, candidate: &Hash, context: &TableContext) {
+		let includable = self.table.candidate_includable(candidate, context);
+		for i in (0..self.trackers.len()).rev() {
+			if self.trackers[i].update_candidate(candidate.clone(), includable) {
+				self.trackers.swap_remove(i);
+			}
+		}
+	}
+}
+
+/// Produced after validating a candidate.
+pub struct Validated {
+	/// A statement about the validity of the candidate.
+	statement: table::Statement,
+	/// The result of validation.
+	result: Validation,
+}
+
+impl Validated {
+	/// Note that we've validated a candidate with given hash and it is bad.
+	pub fn known_bad(hash: Hash, block_data: BlockData) -> Self {
+		Validated {
+			statement: GenericStatement::Invalid(hash),
+			result: Validation::Invalid(block_data),
+		}
+	}
+
+	/// Note that we've validated a candidate with given hash and it is good.
+	/// Extrinsic data required.
+	pub fn known_good(hash: Hash, block_data: BlockData, extrinsic: Extrinsic) -> Self {
+		Validated {
+			statement: GenericStatement::Valid(hash),
+			result: Validation::Valid(block_data, extrinsic),
+		}
+	}
+
+	/// Note that we've collated a candidate.
+	/// Extrinsic data required.
+	pub fn collated_local(
+		receipt: CandidateReceipt,
+		block_data: BlockData,
+		extrinsic: Extrinsic,
+	) -> Self {
+		Validated {
+			statement: GenericStatement::Candidate(receipt),
+			result: Validation::Valid(block_data, extrinsic),
+		}
+	}
+
+	/// Get a reference to the block data.
+	pub fn block_data(&self) -> &BlockData {
+		match self.result {
+			Validation::Valid(ref b, _) | Validation::Invalid(ref b) => b,
+		}
+	}
+
+	/// Get a reference to the extrinsic data, if any.
+	pub fn extrinsic(&self) -> Option<&Extrinsic> {
+		match self.result {
+			Validation::Valid(_, ref ex) => Some(ex),
+			Validation::Invalid(_) => None,
+		}
+	}
+}
+
+/// Future that performs parachain validation work.
+pub struct ParachainWork<Fetch> {
+	work: Work<Fetch>,
+	relay_parent: Hash,
+	extrinsic_store: ExtrinsicStore,
+}
+
+impl<Fetch: Future> ParachainWork<Fetch> {
+	/// Prime the parachain work with an API reference for extracting
+	/// chain information.
+	pub fn prime<P: ProvideRuntimeApi>(self, api: Arc<P>)
+		-> PrimedParachainWork<
+			Fetch,
+			impl Send + FnMut(&BlockId, &Collation, &Incoming) -> Result<Extrinsic, ()>,
+		>
+		where
+			P: Send + Sync + 'static,
+			P::Api: ParachainHost<Block>,
+	{
+		let validate = move |id: &_, collation: &_, incoming: &_| {
+			let res = ::collation::validate_collation(
+				&*api,
+				id,
+				collation,
+				incoming,
+			);
+
+			match res {
+				Ok(e) => Ok(e),
+				Err(e) => {
+					debug!(target: "validation", "Encountered bad collation: {}", e);
+					Err(())
+				}
+			}
+		};
+
+		PrimedParachainWork { inner: self, validate }
+	}
+
+	/// Prime the parachain work with a custom validation function.
+	pub fn prime_with<F>(self, validate: F) -> PrimedParachainWork<Fetch, F>
+		where F: FnMut(&BlockId, &Collation, &Incoming) -> Result<Extrinsic, ()>
+	{
+		PrimedParachainWork { inner: self, validate }
+	}
+}
+
+struct Work<Fetch> {
+	candidate_receipt: CandidateReceipt,
+	fetch: Fetch
+}
+
+/// Primed statement producer.
+pub struct PrimedParachainWork<Fetch, F> {
+	inner: ParachainWork<Fetch>,
+	validate: F,
+}
+
+impl<Fetch, F, Err> Future for PrimedParachainWork<Fetch, F>
+	where
+		Fetch: Future<Item=(BlockData, Incoming),Error=Err>,
+		F: FnMut(&BlockId, &Collation, &Incoming) -> Result<Extrinsic, ()>,
+		Err: From<::std::io::Error>,
+{
+	type Item = Validated;
+	type Error = Err;
+
+	fn poll(&mut self) -> Poll<Validated, Err> {
+		let work = &mut self.inner.work;
+		let candidate = &work.candidate_receipt;
+
+		let (block, incoming) = try_ready!(work.fetch.poll());
+		let validation_res = (self.validate)(
+			&BlockId::hash(self.inner.relay_parent),
+			&Collation { block_data: block.clone(), receipt: candidate.clone() },
+			&incoming,
+		);
+
+		let candidate_hash = candidate.hash();
+
+		debug!(target: "validation", "Making validity statement about candidate {}: is_good? {:?}",
+			candidate_hash, validation_res.is_ok());
+
+		let (validity_statement, result) = match validation_res {
+			Err(()) => (
+				GenericStatement::Invalid(candidate_hash),
+				Validation::Invalid(block),
+			),
+			Ok(extrinsic) => {
+				self.inner.extrinsic_store.make_available(Data {
+					relay_parent: self.inner.relay_parent,
+					parachain_id: work.candidate_receipt.parachain_index,
+					candidate_hash,
+					block_data: block.clone(),
+					extrinsic: Some(extrinsic.clone()),
+				})?;
+
+				(
+					GenericStatement::Valid(candidate_hash),
+					Validation::Valid(block, extrinsic)
+				)
+			}
+		};
+
+		Ok(Async::Ready(Validated {
+			statement: validity_statement,
+			result,
+		}))
+	}
+}
+
+/// A shared table object.
+pub struct SharedTable {
+	context: Arc<TableContext>,
+	inner: Arc<Mutex<SharedTableInner>>,
+}
+
+impl Clone for SharedTable {
+	fn clone(&self) -> Self {
+		SharedTable {
+			context: self.context.clone(),
+			inner: self.inner.clone(),
+		}
+	}
+}
+
+impl SharedTable {
+	/// Create a new shared table.
+	///
+	/// Provide the key to sign with, and the parent hash of the relay chain
+	/// block being built.
+	pub fn new(
+		groups: HashMap<ParaId, GroupInfo>,
+		key: Arc<ed25519::Pair>,
+		parent_hash: Hash,
+		extrinsic_store: ExtrinsicStore,
+	) -> Self {
+		SharedTable {
+			context: Arc::new(TableContext { groups, key, parent_hash }),
+			inner: Arc::new(Mutex::new(SharedTableInner {
+				table: Table::default(),
+				validated: HashMap::new(),
+				trackers: Vec::new(),
+				extrinsic_store,
+			}))
+		}
+	}
+
+	/// Get the parent hash this table should hold statements localized to.
+	pub fn consensus_parent_hash(&self) -> &Hash {
+		&self.context.parent_hash
+	}
+
+	/// Get the local validator session key.
+	pub fn session_key(&self) -> SessionKey {
+		self.context.local_id()
+	}
+
+	/// Get group info.
+	pub fn group_info(&self) -> &HashMap<ParaId, GroupInfo> {
+		&self.context.groups
+	}
+
+	/// Get extrinsic data for candidate with given hash, if any.
+	///
+	/// This will return `Some` for any candidates that have been validated
+	/// locally.
+	pub(crate) fn extrinsic_data(&self, hash: &Hash) -> Option<Extrinsic> {
+		self.inner.lock().validated.get(hash).and_then(|x| match *x {
+			ValidationWork::Error(_) => None,
+			ValidationWork::InProgress => None,
+			ValidationWork::Done(Validation::Invalid(_)) => None,
+			ValidationWork::Done(Validation::Valid(_, ref ex)) => Some(ex.clone()),
+		})
+	}
+
+	/// Import a single statement with remote source, whose signature has already been checked.
+	///
+	/// The statement producer, if any, will produce only statements concerning the same candidate
+	/// as the one just imported
+	pub fn import_remote_statement<R: TableRouter>(
+		&self,
+		router: &R,
+		statement: table::SignedStatement,
+	) -> Option<ParachainWork<future::Join<
+		<R::FetchCandidate as IntoFuture>::Future,
+		<R::FetchIncoming as IntoFuture>::Future,
+	>>> {
+		self.inner.lock().import_remote_statement(&*self.context, router, statement)
+	}
+
+	/// Import many statements at once.
+	///
+	/// Provide an iterator yielding remote, pre-checked statements.
+	///
+	/// The statement producer, if any, will produce only statements concerning the same candidate
+	/// as the one just imported
+	pub fn import_remote_statements<R, I, U>(&self, router: &R, iterable: I) -> U
+		where
+			R: TableRouter,
+			I: IntoIterator<Item=table::SignedStatement>,
+			U: ::std::iter::FromIterator<Option<ParachainWork<future::Join<
+				<R::FetchCandidate as IntoFuture>::Future,
+				<R::FetchIncoming as IntoFuture>::Future,
+			>>>>,
+	{
+		let mut inner = self.inner.lock();
+
+		iterable.into_iter().map(move |statement| {
+			inner.import_remote_statement(&*self.context, router, statement)
+		}).collect()
+	}
+
+	/// Sign and import the result of candidate validation.
+	pub fn import_validated(&self, validated: Validated)
+		-> SignedStatement
+	{
+		let digest = match validated.statement {
+			GenericStatement::Candidate(ref c) => c.hash(),
+			GenericStatement::Valid(h) | GenericStatement::Invalid(h) => h,
+		};
+
+		let signed_statement = self.context.sign_statement(validated.statement);
+
+		let mut inner = self.inner.lock();
+		inner.table.import_statement(&*self.context, signed_statement.clone());
+		inner.validated.insert(digest, ValidationWork::Done(validated.result));
+
+		signed_statement
+	}
+
+	/// Execute a closure using a specific candidate.
+	///
+	/// Deadlocks if called recursively.
+	pub fn with_candidate<F, U>(&self, digest: &Hash, f: F) -> U
+		where F: FnOnce(Option<&CandidateReceipt>) -> U
+	{
+		let inner = self.inner.lock();
+		f(inner.table.get_candidate(digest))
+	}
+
+	/// Get a set of candidates that can be proposed.
+	pub fn proposed_set(&self) -> Vec<AttestedCandidate> {
+		use table::generic::{ValidityAttestation as GAttestation};
+		use polkadot_primitives::parachain::ValidityAttestation;
+
+		// we transform the types of the attestations gathered from the table
+		// into the type expected by the runtime. This may do signature
+		// aggregation in the future.
+		let table_attestations = self.inner.lock().table.proposed_candidates(&*self.context);
+		table_attestations.into_iter()
+			.map(|attested| AttestedCandidate {
+				candidate: attested.candidate,
+				validity_votes: attested.validity_votes.into_iter().map(|(a, v)| match v {
+					GAttestation::Implicit(s) => (a, ValidityAttestation::Implicit(s)),
+					GAttestation::Explicit(s) => (a, ValidityAttestation::Explicit(s)),
+				}).collect(),
+			})
+			.collect()
+	}
+
+	/// Get the number of total parachains.
+	pub fn num_parachains(&self) -> usize {
+		self.group_info().len()
+	}
+
+	/// Get the number of parachains whose candidates may be included.
+	pub fn includable_count(&self) -> usize {
+		self.inner.lock().table.includable_count()
+	}
+
+	/// Get all witnessed misbehavior.
+	pub fn get_misbehavior(&self) -> HashMap<SessionKey, table::Misbehavior> {
+		self.inner.lock().table.get_misbehavior().clone()
+	}
+
+	/// Track includability  of a given set of candidate hashes.
+	pub fn track_includability<I>(&self, iterable: I) -> Includable
+		where I: IntoIterator<Item=Hash>
+	{
+		let mut inner = self.inner.lock();
+
+		let (tx, rx) = includable::track(iterable.into_iter().map(|x| {
+			let includable = inner.table.candidate_includable(&x, &*self.context);
+			(x, includable)
+		}));
+
+		if !tx.is_complete() {
+			inner.trackers.push(tx);
+		}
+
+		rx
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use substrate_keyring::Keyring;
+	use futures::future;
+
+	#[derive(Clone)]
+	struct DummyRouter;
+	impl TableRouter for DummyRouter {
+		type Error = ::std::io::Error;
+		type FetchCandidate = ::futures::future::FutureResult<BlockData,Self::Error>;
+		type FetchIncoming = ::futures::future::FutureResult<Incoming,Self::Error>;
+
+		fn local_candidate(&self, _candidate: CandidateReceipt, _block_data: BlockData, _extrinsic: Extrinsic) {
+
+		}
+
+		fn fetch_block_data(&self, _candidate: &CandidateReceipt) -> Self::FetchCandidate {
+			future::ok(BlockData(vec![1, 2, 3, 4, 5]))
+		}
+
+		fn fetch_incoming(&self, _para_id: ParaId) -> Self::FetchIncoming {
+			future::ok(Vec::new())
+		}
+	}
+
+	#[test]
+	fn statement_triggers_fetch_and_evaluate() {
+		let mut groups = HashMap::new();
+
+		let para_id = ParaId::from(1);
+		let local_id = Keyring::Alice.to_raw_public().into();
+		let local_key = Arc::new(Keyring::Alice.pair());
+
+		let validity_other = Keyring::Bob.to_raw_public().into();
+		let validity_other_key = Keyring::Bob.pair();
+		let parent_hash = Default::default();
+
+		groups.insert(para_id, GroupInfo {
+			validity_guarantors: [local_id, validity_other].iter().cloned().collect(),
+			needed_validity: 2,
+		});
+
+		let shared_table = SharedTable::new(
+			groups,
+			local_key.clone(),
+			parent_hash,
+			ExtrinsicStore::new_in_memory(),
+		);
+
+		let candidate = CandidateReceipt {
+			parachain_index: para_id,
+			collator: [1; 32].into(),
+			signature: Default::default(),
+			head_data: ::polkadot_primitives::parachain::HeadData(vec![1, 2, 3, 4]),
+			balance_uploads: Vec::new(),
+			egress_queue_roots: Vec::new(),
+			fees: 1_000_000,
+			block_data_hash: [2; 32].into(),
+		};
+
+		let candidate_statement = GenericStatement::Candidate(candidate);
+
+		let signature = ::sign_table_statement(&candidate_statement, &validity_other_key, &parent_hash);
+		let signed_statement = ::table::generic::SignedStatement {
+			statement: candidate_statement,
+			signature: signature.into(),
+			sender: validity_other,
+		};
+
+		shared_table.import_remote_statement(
+			&DummyRouter,
+			signed_statement,
+		).expect("candidate and local validity group are same");
+	}
+
+	#[test]
+	fn statement_triggers_fetch_and_validity() {
+		let mut groups = HashMap::new();
+
+		let para_id = ParaId::from(1);
+		let local_id = Keyring::Alice.to_raw_public().into();
+		let local_key = Arc::new(Keyring::Alice.pair());
+
+		let validity_other = Keyring::Bob.to_raw_public().into();
+		let validity_other_key = Keyring::Bob.pair();
+		let parent_hash = Default::default();
+
+		groups.insert(para_id, GroupInfo {
+			validity_guarantors: [local_id, validity_other].iter().cloned().collect(),
+			needed_validity: 1,
+		});
+
+		let shared_table = SharedTable::new(
+			groups,
+			local_key.clone(),
+			parent_hash,
+			ExtrinsicStore::new_in_memory(),
+		);
+
+		let candidate = CandidateReceipt {
+			parachain_index: para_id,
+			collator: [1; 32].into(),
+			signature: Default::default(),
+			head_data: ::polkadot_primitives::parachain::HeadData(vec![1, 2, 3, 4]),
+			balance_uploads: Vec::new(),
+			egress_queue_roots: Vec::new(),
+			fees: 1_000_000,
+			block_data_hash: [2; 32].into(),
+		};
+
+		let candidate_statement = GenericStatement::Candidate(candidate);
+
+		let signature = ::sign_table_statement(&candidate_statement, &validity_other_key, &parent_hash);
+		let signed_statement = ::table::generic::SignedStatement {
+			statement: candidate_statement,
+			signature: signature.into(),
+			sender: validity_other,
+		};
+
+		shared_table.import_remote_statement(
+			&DummyRouter,
+			signed_statement,
+		).expect("should produce work");
+	}
+
+	#[test]
+	fn evaluate_makes_block_data_available() {
+		let store = ExtrinsicStore::new_in_memory();
+		let relay_parent = [0; 32].into();
+		let para_id = 5.into();
+		let block_data = BlockData(vec![1, 2, 3]);
+
+		let candidate = CandidateReceipt {
+			parachain_index: para_id,
+			collator: [1; 32].into(),
+			signature: Default::default(),
+			head_data: ::polkadot_primitives::parachain::HeadData(vec![1, 2, 3, 4]),
+			balance_uploads: Vec::new(),
+			egress_queue_roots: Vec::new(),
+			fees: 1_000_000,
+			block_data_hash: [2; 32].into(),
+		};
+
+		let hash = candidate.hash();
+
+		let producer: ParachainWork<future::FutureResult<_, ::std::io::Error>> = ParachainWork {
+			work: Work {
+				candidate_receipt: candidate,
+				fetch: future::ok((block_data.clone(), Vec::new())),
+			},
+			relay_parent,
+			extrinsic_store: store.clone(),
+		};
+
+		let validated = producer.prime_with(|_, _, _| Ok(Extrinsic { outgoing_messages: Vec::new() }))
+			.wait()
+			.unwrap();
+
+		assert_eq!(validated.block_data(), &block_data);
+		assert_eq!(validated.statement, GenericStatement::Valid(hash));
+
+		assert_eq!(store.block_data(relay_parent, hash).unwrap(), block_data);
+		assert!(store.extrinsic(relay_parent, hash).is_some());
+	}
+
+	#[test]
+	fn full_availability() {
+		let store = ExtrinsicStore::new_in_memory();
+		let relay_parent = [0; 32].into();
+		let para_id = 5.into();
+		let block_data = BlockData(vec![1, 2, 3]);
+
+		let candidate = CandidateReceipt {
+			parachain_index: para_id,
+			collator: [1; 32].into(),
+			signature: Default::default(),
+			head_data: ::polkadot_primitives::parachain::HeadData(vec![1, 2, 3, 4]),
+			balance_uploads: Vec::new(),
+			egress_queue_roots: Vec::new(),
+			fees: 1_000_000,
+			block_data_hash: [2; 32].into(),
+		};
+
+		let hash = candidate.hash();
+
+		let producer = ParachainWork {
+			work: Work {
+				candidate_receipt: candidate,
+				fetch: future::ok::<_, ::std::io::Error>((block_data.clone(), Vec::new())),
+			},
+			relay_parent,
+			extrinsic_store: store.clone(),
+		};
+
+		let validated = producer.prime_with(|_, _, _| Ok(Extrinsic { outgoing_messages: Vec::new() }))
+			.wait()
+			.unwrap();
+
+		assert_eq!(validated.block_data(), &block_data);
+
+		assert_eq!(store.block_data(relay_parent, hash).unwrap(), block_data);
+		assert!(store.extrinsic(relay_parent, hash).is_some());
+	}
+
+	#[test]
+	fn does_not_dispatch_work_after_starting_validation() {
+		let mut groups = HashMap::new();
+
+		let para_id = ParaId::from(1);
+		let local_id = Keyring::Alice.to_raw_public().into();
+		let local_key = Arc::new(Keyring::Alice.pair());
+
+		let validity_other = Keyring::Bob.to_raw_public().into();
+		let validity_other_key = Keyring::Bob.pair();
+		let parent_hash = Default::default();
+
+		groups.insert(para_id, GroupInfo {
+			validity_guarantors: [local_id, validity_other].iter().cloned().collect(),
+			needed_validity: 1,
+		});
+
+		let shared_table = SharedTable::new(
+			groups,
+			local_key.clone(),
+			parent_hash,
+			ExtrinsicStore::new_in_memory(),
+		);
+
+		let candidate = CandidateReceipt {
+			parachain_index: para_id,
+			collator: [1; 32].into(),
+			signature: Default::default(),
+			head_data: ::polkadot_primitives::parachain::HeadData(vec![1, 2, 3, 4]),
+			balance_uploads: Vec::new(),
+			egress_queue_roots: Vec::new(),
+			fees: 1_000_000,
+			block_data_hash: [2; 32].into(),
+		};
+
+		let hash = candidate.hash();
+		let candidate_statement = GenericStatement::Candidate(candidate);
+
+		let signature = ::sign_table_statement(&candidate_statement, &validity_other_key, &parent_hash);
+		let signed_statement = ::table::generic::SignedStatement {
+			statement: candidate_statement,
+			signature: signature.into(),
+			sender: validity_other,
+		};
+
+		let _a = shared_table.import_remote_statement(
+			&DummyRouter,
+			signed_statement.clone(),
+		).expect("should produce work");
+
+		assert!(shared_table.inner.lock().validated.get(&hash).expect("validation has started").is_in_progress());
+
+		let b = shared_table.import_remote_statement(
+			&DummyRouter,
+			signed_statement.clone(),
+		);
+
+		assert!(b.is_none(), "cannot work when validation has started");
+	}
+
+	#[test]
+	fn does_not_dispatch_after_local_candidate() {
+		let mut groups = HashMap::new();
+
+		let para_id = ParaId::from(1);
+		let local_id = Keyring::Alice.to_raw_public().into();
+		let local_key = Arc::new(Keyring::Alice.pair());
+		let block_data = BlockData(vec![1, 2, 3]);
+		let extrinsic = Extrinsic { outgoing_messages: Vec::new() };
+
+		let validity_other = Keyring::Bob.to_raw_public().into();
+		let parent_hash = Default::default();
+
+		groups.insert(para_id, GroupInfo {
+			validity_guarantors: [local_id, validity_other].iter().cloned().collect(),
+			needed_validity: 1,
+		});
+
+		let shared_table = SharedTable::new(
+			groups,
+			local_key.clone(),
+			parent_hash,
+			ExtrinsicStore::new_in_memory(),
+		);
+
+		let candidate = CandidateReceipt {
+			parachain_index: para_id,
+			collator: [1; 32].into(),
+			signature: Default::default(),
+			head_data: ::polkadot_primitives::parachain::HeadData(vec![1, 2, 3, 4]),
+			balance_uploads: Vec::new(),
+			egress_queue_roots: Vec::new(),
+			fees: 1_000_000,
+			block_data_hash: [2; 32].into(),
+		};
+
+		let hash = candidate.hash();
+		let signed_statement = shared_table.import_validated(Validated::collated_local(
+			candidate,
+			block_data,
+			extrinsic,
+		));
+
+		assert!(shared_table.inner.lock().validated.get(&hash).expect("validation has started").is_done());
+
+		let a = shared_table.import_remote_statement(
+			&DummyRouter,
+			signed_statement,
+		);
+
+		assert!(a.is_none());
+	}
+}