Consensus Engines Implementation: Aura (#911)

* Generalize BlockImport

 - move ImportBlock, BlockOrigin, ImportResult into shared sr-primitives
 - let Consensus provide  and  traits again
 - update consensus traits to latest development
 - implement traits on client::Client, test_client::TestClient
 - update RHD to use the new import_block API

* Move ImportBlock into consensus-common
* Send import notification in aura tests
* Integrating aura into service
* Make Signatures more generic
* Aura Block Production with the given key
* run aura on the thread pool
* start at exact step start in aura
* Add needed wasm blob, in leiu of better solutions.
* Make API ids consistent with traits and bring upstream for sharing.
* Add decrease_free_balance to Balances module
* Encode `Metadata` once instead of two times
* Bitops include xor
* Upgrade key module.
* Default pages to somewhat bigger.
* Introduce upgrade key into node
* Add `Created` event

substrate/core/consensus/aura/Cargo.toml

@@ -0,0 +1,42 @@
+[package]
+name = "substrate-consensus-aura"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+description = "Rhododendron Round-Based consensus-algorithm for substrate"
+
+[dependencies]
+futures = "0.1.17"
+parity-codec = { version = "2.1" }
+substrate-consensus-common = { path = "../common" }
+substrate-client = { path = "../../client" }
+substrate-primitives = { path = "../../primitives" }
+substrate-network = { path = "../../network" }
+srml-support = { path = "../../../srml/support" }
+sr-primitives = { path = "../../sr-primitives" }
+sr-version = { path = "../../sr-version" }
+sr-io = { path = "../../sr-io" }
+srml-consensus = { path = "../../../srml/consensus" }
+tokio = "0.1.7"
+parking_lot = "0.4"
+error-chain = "0.12"
+log = "0.3"
+
+[dev-dependencies]
+substrate-keyring = { path = "../../keyring" }
+substrate-executor = { path = "../../executor" }
+substrate-service = { path = "../../service" }
+substrate-test-client = { path = "../../test-client" }
+env_logger = { version = "0.4" }
+
+[target.'cfg(test)'.dependencies]
+substrate-network = { path = "../../network", features = ["test-helpers"] }
+
+[features]
+default = ["std"]
+std = [
+	"substrate-primitives/std",
+	"srml-support/std",
+	"sr-primitives/std",
+	"sr-version/std",
+]
+

substrate/core/consensus/aura/src/lib.rs

@@ -0,0 +1,560 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate 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.
+
+// Substrate 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 Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Aura (Authority-round) consensus in substrate.
+//!
+//! Aura works by having a list of authorities A who are expected to roughly
+//! agree on the current time. Time is divided up into discrete slots of t
+//! seconds each. For each slot s, the author of that slot is A[s % |A|].
+//!
+//! The author is allowed to issue one block but not more during that slot,
+//! and it will be built upon the longest valid chain that has been seen.
+//!
+//! Blocks from future steps will be either deferred or rejected depending on how
+//! far in the future they are.
+
+extern crate parity_codec as codec;
+extern crate substrate_consensus_common as consensus_common;
+extern crate substrate_client as client;
+extern crate substrate_primitives as primitives;
+extern crate substrate_network as network;
+extern crate srml_support as runtime_support;
+extern crate sr_primitives as runtime_primitives;
+extern crate sr_version as runtime_version;
+extern crate sr_io as runtime_io;
+extern crate tokio;
+
+#[cfg(test)]
+extern crate substrate_keyring as keyring;
+#[cfg(test)]
+extern crate substrate_service as service;
+#[cfg(test)]
+extern crate substrate_test_client as test_client;
+#[cfg(test)]
+extern crate env_logger;
+
+extern crate parking_lot;
+
+#[macro_use]
+extern crate log;
+
+extern crate futures;
+
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use codec::Encode;
+use consensus_common::{Authorities, BlockImport, Environment, Proposer};
+use client::ChainHead;
+use consensus_common::{ImportBlock, BlockOrigin};
+use runtime_primitives::{generic, generic::BlockId};
+use runtime_primitives::traits::{Block, Header, Digest, DigestItemFor};
+use network::import_queue::{Verifier, BasicQueue};
+use primitives::{AuthorityId, ed25519};
+
+use futures::{Stream, Future, IntoFuture, future::{self, Either}};
+use tokio::timer::Interval;
+
+pub use consensus_common::SyncOracle;
+
+/// A handle to the network. This is generally implemented by providing some
+/// handle to a gossip service or similar.
+///
+/// Intended to be a lightweight handle such as an `Arc`.
+pub trait Network: Clone {
+	/// A stream of input messages for a topic.
+	type In: Stream<Item=Vec<u8>,Error=()>;
+
+	/// Send a message at a specific round out.
+	fn send_message(&self, slot: u64, message: Vec<u8>);
+}
+
+/// Configuration for Aura consensus.
+#[derive(Clone)]
+pub struct Config {
+	/// The local authority keypair. Can be none if this is just an observer.
+	pub local_key: Option<Arc<ed25519::Pair>>,
+	/// The slot duration in seconds.
+	pub slot_duration: u64
+}
+
+/// Get slot author for given block along with authorities.
+fn slot_author(slot_num: u64, authorities: &[AuthorityId]) -> Option<AuthorityId> {
+	if authorities.is_empty() { return None }
+
+	let idx = slot_num % (authorities.len() as u64);
+	assert!(idx <= usize::max_value() as u64,
+		"It is impossible to have a vector with length beyond the address space; qed");
+
+	let current_author = *authorities.get(idx as usize)
+		.expect("authorities not empty; index constrained to list length;\
+				this is a valid index; qed");
+
+	Some(current_author)
+}
+
+fn duration_now() -> Option<Duration> {
+	use std::time::SystemTime;
+
+	let now = SystemTime::now();
+	now.duration_since(SystemTime::UNIX_EPOCH).map_err(|e| {
+			warn!("Current time {:?} is before unix epoch. Something is wrong: {:?}", now, e);
+	}).ok()
+}
+
+/// Get the slot for now.
+fn slot_now(slot_duration: u64) -> Option<u64> {
+	duration_now().map(|s| s.as_secs() / slot_duration)
+}
+
+/// A digest item which is usable with aura consensus.
+pub trait CompatibleDigestItem: Sized {
+	/// Construct a digest item which is a slot number and a signature on the
+	/// hash.
+	fn aura_seal(slot_number: u64, signature: ed25519::Signature) -> Self;
+
+	/// If this item is an Aura seal, return the slot number and signature.
+	fn as_aura_seal(&self) -> Option<(u64, &ed25519::Signature)>;
+}
+
+impl CompatibleDigestItem for generic::DigestItem<primitives::H256, u64> {
+	/// Construct a digest item which is a slot number and a signature on the
+	/// hash.
+	fn aura_seal(slot_number: u64, signature: ed25519::Signature) -> Self {
+		generic::DigestItem::Seal(slot_number, signature)
+	}
+	/// If this item is an Aura seal, return the slot number and signature.
+	fn as_aura_seal(&self) -> Option<(u64, &ed25519::Signature)> {
+		match self {
+			generic::DigestItem::Seal(slot, ref sign) => Some((*slot, sign)),
+			_ => None
+		}
+	}
+}
+
+impl CompatibleDigestItem for generic::DigestItem<primitives::H256, primitives::AuthorityId> {
+	/// Construct a digest item which is a slot number and a signature on the
+	/// hash.
+	fn aura_seal(slot_number: u64, signature: ed25519::Signature) -> Self {
+		generic::DigestItem::Seal(slot_number, signature)
+	}
+	/// If this item is an Aura seal, return the slot number and signature.
+	fn as_aura_seal(&self) -> Option<(u64, &ed25519::Signature)> {
+		match self {
+			generic::DigestItem::Seal(slot, ref sign) => Some((*slot, sign)),
+			_ => None
+		}
+	}
+}
+
+/// Start the aura worker. This should be run in a tokio runtime.
+pub fn start_aura<B, C, E, SO, Error>(
+	config: Config,
+	client: Arc<C>,
+	env: Arc<E>,
+	sync_oracle: SO,
+)
+	-> impl Future<Item=(),Error=()> where
+	B: Block,
+	C: Authorities<B, Error=Error> + BlockImport<B, Error=Error> + ChainHead<B>,
+	E: Environment<B, Error=Error>,
+	E::Proposer: Proposer<B, Error=Error>,
+	SO: SyncOracle + Send + Clone,
+	DigestItemFor<B>: CompatibleDigestItem,
+	Error: ::std::error::Error + Send + 'static + From<::consensus_common::Error>,
+{
+	let make_authorship = move || {
+		let config = config.clone();
+		let client = client.clone();
+		let env = env.clone();
+		let sync_oracle = sync_oracle.clone();
+
+		let local_keys = config.local_key.map(|pair| (pair.public(), pair));
+		let slot_duration = config.slot_duration;
+		let mut last_authored_slot = 0;
+		let next_slot_start = duration_now().map(|now| {
+			let remaining_full_secs = slot_duration - (now.as_secs() % slot_duration) - 1;
+			let remaining_nanos = 1_000_000_000 - now.subsec_nanos();
+			Instant::now() + Duration::new(remaining_full_secs, remaining_nanos)
+		}).unwrap_or_else(|| Instant::now());
+
+		Interval::new(next_slot_start, Duration::from_secs(slot_duration))
+			.filter(move |_| !sync_oracle.is_major_syncing()) // only propose when we are not syncing.
+			.filter_map(move |_| local_keys.clone()) // skip if not authoring.
+			.map_err(|e|  debug!(target: "aura", "Faulty timer: {:?}", e))
+			.for_each(move |(public_key, key)| {
+				use futures::future;
+
+				let slot_num = match slot_now(slot_duration) {
+					Some(n) => n,
+					None => return Either::B(future::err(())),
+				};
+
+				if last_authored_slot >= slot_num { return Either::B(future::ok(())) }
+				last_authored_slot = slot_num;
+
+				let chain_head = match client.best_block_header() {
+					Ok(x) => x,
+					Err(e) => {
+						warn!(target:"aura", "Unable to author block in slot {}. no best block header: {:?}", slot_num, e);
+						return Either::B(future::ok(()))
+					}
+				};
+
+				let authorities = match client.authorities(&BlockId::Hash(chain_head.hash())){
+					Ok(authorities) => authorities,
+					Err(e) => {
+						warn!("Unable to fetch authorities at block {:?}: {:?}", chain_head.hash(), e);
+						return Either::B(future::ok(()));
+					}
+				};
+
+				let proposal_work = match slot_author(slot_num, &authorities) {
+					None => return Either::B(future::ok(())),
+					Some(author) => if author.0 == public_key.0 {
+						// we are the slot author. make a block and sign it.
+						let proposer = match env.init(&chain_head, &authorities, key.clone()) {
+							Ok(p) => p,
+							Err(e) => {
+								warn!("Unable to author block in slot {:?}: {:?}", slot_num, e);
+								return Either::B(future::ok(()))
+							}
+						};
+
+						proposer.propose().into_future()
+					} else {
+						return Either::B(future::ok(()));
+					}
+				};
+
+				let block_import = client.clone();
+				Either::A(proposal_work
+					.map(move |b| {
+						let (header, body) = b.deconstruct();
+						let pre_hash = header.hash();
+						let parent_hash = header.parent_hash().clone();
+
+						// sign the pre-sealed hash of the block and then
+						// add it to a digest item.
+						let to_sign = (slot_num, pre_hash).encode();
+						let signature = key.sign(&to_sign[..]);
+						let item = <DigestItemFor<B> as CompatibleDigestItem>::aura_seal(slot_num, signature);
+						let import_block = ImportBlock {
+							origin: BlockOrigin::Own,
+							header,
+							external_justification: Vec::new(),
+							post_runtime_digests: vec![item],
+							body: Some(body),
+							finalized: false,
+							auxiliary: Vec::new(),
+						};
+
+						if let Err(e) = block_import.import_block(import_block, None) {
+							warn!(target: "aura", "Error with block built on {:?}: {:?}", parent_hash, e);
+						}
+					})
+					.map_err(|e| warn!("Failed to construct block: {:?}", e))
+				)
+			})
+	};
+
+	future::loop_fn((), move |()| {
+		let authorship_task = ::std::panic::AssertUnwindSafe(make_authorship());
+		authorship_task.catch_unwind().then(|res| {
+			match res {
+				Ok(Ok(())) => (),
+				Ok(Err(())) => warn!("Aura authorship task terminated unexpectedly. Restarting"),
+				Err(e) => {
+					if let Some(s) = e.downcast_ref::<&'static str>() {
+						warn!("Aura authorship task panicked at {:?}", s);
+					}
+
+					warn!("Restarting Aura authorship task");
+				}
+			}
+
+			Ok(future::Loop::Continue(()))
+		})
+	})
+}
+
+// a header which has been checked
+enum CheckedHeader<H> {
+	// a header which has slot in the future. this is the full header (not stripped)
+	// and the slot in which it should be processed.
+	Deferred(H, u64),
+	// a header which is fully checked, including signature. This is the pre-header
+	// accompanied by the seal components.
+	Checked(H, u64, ed25519::Signature),
+}
+
+
+/// check a header has been signed by the right key. If the slot is too far in the future, an error will be returned.
+/// if it's successful, returns the pre-header, the slot number, and the signat.
+//
+// FIXME: needs misbehavior types - https://github.com/paritytech/substrate/issues/1018
+fn check_header<B: Block>(slot_now: u64, mut header: B::Header, hash: B::Hash, authorities: &[AuthorityId])
+	-> Result<CheckedHeader<B::Header>, String>
+	where DigestItemFor<B>: CompatibleDigestItem
+{
+	let digest_item = match header.digest_mut().pop() {
+		Some(x) => x,
+		None => return Err(format!("Header {:?} is unsealed", hash)),
+	};
+	let (slot_num, &sig) = match digest_item.as_aura_seal() {
+		Some(x) => x,
+		None => return Err(format!("Header {:?} is unsealed", hash)),
+	};
+
+	if slot_num > slot_now {
+		header.digest_mut().push(digest_item);
+		Ok(CheckedHeader::Deferred(header, slot_num))
+	} else {
+		// check the signature is valid under the expected authority and
+		// chain state.
+
+		let expected_author = match slot_author(slot_num, &authorities) {
+			None => return Err("Slot Author not found".to_string()),
+			Some(author) => author
+		};
+
+		let pre_hash = header.hash();
+		let to_sign = (slot_num, pre_hash).encode();
+		let public = ed25519::Public(expected_author.0);
+
+		if ed25519::verify_strong(&sig, &to_sign[..], public) {
+			Ok(CheckedHeader::Checked(header, slot_num, sig))
+		} else {
+			Err(format!("Bad signature on {:?}", hash))
+		}
+	}
+}
+
+/// A verifier for Aura blocks.
+pub struct AuraVerifier<C> {
+	config: Config,
+	client: Arc<C>,
+}
+
+impl<B: Block, C> Verifier<B> for AuraVerifier<C> where
+	C: Authorities<B> + BlockImport<B> + Send + Sync,
+	DigestItemFor<B>: CompatibleDigestItem,
+{
+	fn verify(
+		&self,
+		origin: BlockOrigin,
+		header: B::Header,
+		_justification: Vec<u8>,
+		body: Option<Vec<B::Extrinsic>>
+	) -> Result<(ImportBlock<B>, Option<Vec<AuthorityId>>), String> {
+		let slot_now = slot_now(self.config.slot_duration)
+			.ok_or("System time is before UnixTime?".to_owned())?;
+		let hash = header.hash();
+		let parent_hash = *header.parent_hash();
+		let authorities = self.client.authorities(&BlockId::Hash(parent_hash))
+			.map_err(|e| format!("Could not fetch authorities at {:?}: {:?}", parent_hash, e))?;
+
+		// we add one to allow for some small drift.
+		// FIXME: in the future, alter this queue to allow deferring of headers
+		// https://github.com/paritytech/substrate/issues/1019
+		let checked_header = check_header::<B>(slot_now + 1, header, hash, &authorities[..])?;
+		match checked_header {
+			CheckedHeader::Checked(pre_header, slot_num, sig) => {
+				let item = <DigestItemFor<B>>::aura_seal(slot_num, sig);
+
+				debug!(target: "aura", "Checked {:?}; importing.", pre_header);
+
+				let import_block = ImportBlock {
+					origin,
+					header: pre_header,
+					external_justification: Vec::new(),
+					post_runtime_digests: vec![item],
+					body,
+					finalized: false,
+					auxiliary: Vec::new(),
+				};
+
+				// FIXME: extract authorities - https://github.com/paritytech/substrate/issues/1019
+				Ok((import_block, None))
+			}
+			CheckedHeader::Deferred(a, b) => {
+				debug!(target: "aura", "Checking {:?} failed; {:?}, {:?}.", hash, a, b);
+				Err(format!("Header {:?} rejected: too far in the future", hash))
+			}
+		}
+	}
+}
+
+/// The Aura import queue type.
+pub type AuraImportQueue<B, C> = BasicQueue<B, AuraVerifier<C>>;
+
+/// Start an import queue for the Aura consensus algorithm.
+pub fn import_queue<B, C>(config: Config, client: Arc<C>) -> AuraImportQueue<B, C> where
+	B: Block,
+	C: Authorities<B> + BlockImport<B> + Send + Sync,
+	DigestItemFor<B>: CompatibleDigestItem,
+{
+	let verifier = Arc::new(AuraVerifier { config, client });
+	BasicQueue::new(verifier)
+}
+
+
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use consensus_common::NoNetwork as DummyOracle;
+	use network::test::*;
+	use network::test::{Block as TestBlock, PeersClient};
+	use runtime_primitives::traits::Block as BlockT;
+	use network::ProtocolConfig;
+	use parking_lot::Mutex;
+	use tokio::runtime::current_thread;
+	use keyring::Keyring;
+	use client::BlockchainEvents;
+	use test_client;
+
+	type Error = client::error::Error;
+
+	type TestClient = client::Client<test_client::Backend, test_client::Executor, TestBlock>;
+
+	struct DummyFactory(Arc<TestClient>);
+	struct DummyProposer(u64, Arc<TestClient>);
+
+	impl Environment<TestBlock> for DummyFactory {
+		type Proposer = DummyProposer;
+		type Error = Error;
+
+		fn init(&self, parent_header: &<TestBlock as BlockT>::Header, _authorities: &[AuthorityId], _sign_with: Arc<ed25519::Pair>)
+			-> Result<DummyProposer, Error>
+		{
+			Ok(DummyProposer(parent_header.number + 1, self.0.clone()))
+		}
+	}
+
+	impl Proposer<TestBlock> for DummyProposer {
+		type Error = Error;
+		type Create = Result<TestBlock, Error>;
+
+		fn propose(&self) -> Result<TestBlock, Error> {
+			self.1.new_block().unwrap().bake().map_err(|e| e.into())
+		}
+	}
+
+	const SLOT_DURATION: u64 = 1;
+	const TEST_ROUTING_INTERVAL: Duration = Duration::from_millis(50);
+
+	pub struct AuraTestNet {
+		peers: Vec<Arc<Peer<AuraVerifier<PeersClient>>>>,
+		started: bool
+	}
+
+	impl TestNetFactory for AuraTestNet {
+		type Verifier = AuraVerifier<PeersClient>;
+
+		/// Create new test network with peers and given config.
+		fn from_config(_config: &ProtocolConfig) -> Self {
+			AuraTestNet {
+				peers: Vec::new(),
+				started: false
+			}
+		}
+
+		fn make_verifier(&self, client: Arc<PeersClient>, _cfg: &ProtocolConfig)
+			-> Arc<Self::Verifier>
+		{
+			let config = Config { local_key: None, slot_duration: SLOT_DURATION };
+			Arc::new(AuraVerifier { client, config })
+		}
+
+		fn peer(&self, i: usize) -> &Peer<Self::Verifier> {
+			&self.peers[i]
+		}
+
+		fn peers(&self) -> &Vec<Arc<Peer<Self::Verifier>>> {
+			&self.peers
+		}
+
+		fn mut_peers<F: Fn(&mut Vec<Arc<Peer<Self::Verifier>>>)>(&mut self, closure: F ) {
+			closure(&mut self.peers);
+		}
+
+		fn started(&self) -> bool {
+			self.started
+		}
+
+		fn set_started(&mut self, new: bool) {
+			self.started = new;
+		}
+	}
+
+	#[test]
+	fn authoring_blocks() {
+		::env_logger::init().ok();
+		let mut net = AuraTestNet::new(3);
+
+		net.start();
+
+		let peers = &[
+			(0, Keyring::Alice),
+			(1, Keyring::Bob),
+			(2, Keyring::Charlie),
+		];
+
+		let net = Arc::new(Mutex::new(net));
+		let mut import_notifications = Vec::new();
+
+		let mut runtime = current_thread::Runtime::new().unwrap();
+		for (peer_id, key) in peers {
+			let mut client = net.lock().peer(*peer_id).client().clone();
+			let environ = Arc::new(DummyFactory(client.clone()));
+			import_notifications.push(
+				client.import_notification_stream()
+					.take_while(|n| {
+						Ok(!(n.origin != BlockOrigin::Own && n.header.number() < &5))
+					})
+					.for_each(move |_| Ok(()))
+			);
+			let aura = start_aura(
+				Config {
+					local_key: Some(Arc::new(key.clone().into())),
+					slot_duration: SLOT_DURATION
+				},
+				client,
+				environ.clone(),
+				DummyOracle,
+			);
+
+			runtime.spawn(aura);
+		}
+
+		// wait for all finalized on each.
+		let wait_for = ::futures::future::join_all(import_notifications)
+			.map(|_| ())
+			.map_err(|_| ());
+
+		let drive_to_completion = ::tokio::timer::Interval::new_interval(TEST_ROUTING_INTERVAL)
+			.for_each(move |_| {
+				net.lock().send_import_notifications();
+				net.lock().sync();
+				Ok(())
+			})
+			.map(|_| ())
+			.map_err(|_| ());
+
+		runtime.block_on(wait_for.select(drive_to_completion).map_err(|_| ())).unwrap();
+	}
+}

substrate/core/consensus/common/Cargo.toml

@@ -4,5 +4,12 @@ version = "0.1.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 description = "Common utilities for substrate consensus"
 
-[dev-dependencies]
-substrate-primitives = { path= "../../primitives"}
+[dependencies]
+substrate-primitives = { path= "../../primitives" }
+error-chain = "0.12"
+futures = "0.1"
+sr-version = { path = "../../sr-version" }
+sr-primitives = { path = "../../sr-primitives" }
+tokio = "0.1.7"
+parity-codec = "2.1"
+parity-codec-derive = "2.0"

substrate/core/consensus/common/src/block_import.rs

@@ -0,0 +1,104 @@
+
+use primitives::AuthorityId;
+use runtime_primitives::traits::{Block as BlockT, DigestItemFor};
+use runtime_primitives::Justification;
+
+/// Block import result.
+#[derive(Debug)]
+pub enum ImportResult {
+	/// Added to the import queue.
+	Queued,
+	/// Already in the import queue.
+	AlreadyQueued,
+	/// Already in the blockchain.
+	AlreadyInChain,
+	/// Block or parent is known to be bad.
+	KnownBad,
+	/// Block parent is not in the chain.
+	UnknownParent,
+}
+
+/// Block data origin.
+#[derive(Debug, PartialEq, Eq, Clone, Copy)]
+pub enum BlockOrigin {
+	/// Genesis block built into the client.
+	Genesis,
+	/// Block is part of the initial sync with the network.
+	NetworkInitialSync,
+	/// Block was broadcasted on the network.
+	NetworkBroadcast,
+	/// Block that was received from the network and validated in the consensus process.
+	ConsensusBroadcast,
+	/// Block that was collated by this node.
+	Own,
+	/// Block was imported from a file.
+	File,
+}
+
+/// Data required to import a Block
+pub struct ImportBlock<Block: BlockT> {
+	/// Origin of the Block
+	pub origin: BlockOrigin,
+	/// The header, without consensus post-digests applied. This should be in the same
+	/// state as it comes out of the runtime.
+	///
+	/// Consensus engines which alter the header (by adding post-runtime digests)
+	/// should strip those off in the initial verification process and pass them
+	/// via the `post_runtime_digests` field. During block authorship, they should
+	/// not be pushed to the header directly.
+	///
+	/// The reason for this distinction is so the header can be directly
+	/// re-executed in a runtime that checks digest equivalence -- the
+	/// post-runtime digests are pushed back on after.
+	pub header: Block::Header,
+	/// Justification provided for this block from the outside:.
+	pub external_justification: Justification,
+	/// Digest items that have been added after the runtime for external
+	/// work, like a consensus signature.
+	pub post_runtime_digests: Vec<DigestItemFor<Block>>,
+	/// Block's body
+	pub body: Option<Vec<Block::Extrinsic>>,
+	/// Is this block finalized already?
+	/// `true` implies instant finality.
+	pub finalized: bool,
+	/// Auxiliary consensus data produced by the block.
+	/// Contains a list of key-value pairs. If values are `None`, the keys
+	/// will be deleted.
+	pub auxiliary: Vec<(Vec<u8>, Option<Vec<u8>>)>,
+}
+
+impl<Block: BlockT> ImportBlock<Block> {
+	/// Deconstruct the justified header into parts.
+	pub fn into_inner(self)
+		-> (
+			BlockOrigin,
+			<Block as BlockT>::Header,
+			Justification,
+			Vec<DigestItemFor<Block>>,
+			Option<Vec<<Block as BlockT>::Extrinsic>>,
+			bool,
+			Vec<(Vec<u8>, Option<Vec<u8>>)>,
+		) {
+		(
+			self.origin,
+			self.header,
+			self.external_justification,
+			self.post_runtime_digests,
+			self.body,
+			self.finalized,
+			self.auxiliary,
+		)
+	}
+}
+
+
+
+/// Block import trait.
+pub trait BlockImport<B: BlockT> {
+	type Error: ::std::error::Error + Send + 'static;
+	/// Import a Block alongside the new authorities valid form this block forward
+	fn import_block(&self,
+		block: ImportBlock<B>,
+		new_authorities: Option<Vec<AuthorityId>>
+	) -> Result<ImportResult, Self::Error>;
+}

substrate/core/consensus/common/src/error.rs

@@ -0,0 +1,88 @@
+// Copyright 2017-2018 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate 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.
+
+// Substrate 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 Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Error types in Consensus
+use runtime_version::RuntimeVersion;
+
+error_chain! {
+	errors {
+		/// Missing state at block with given descriptor.
+		StateUnavailable(b: String) {
+			description("State missing at given block."),
+			display("State unavailable at block {}", b),
+		}
+
+		/// I/O terminated unexpectedly
+		IoTerminated {
+			description("I/O terminated unexpectedly."),
+			display("I/O terminated unexpectedly."),
+		}
+
+		/// Unable to schedule wakeup.
+		FaultyTimer(e: ::tokio::timer::Error) {
+			description("Timer error"),
+			display("Timer error: {}", e),
+		}
+
+		/// Unable to propose a block.
+		CannotPropose {
+			description("Unable to create block proposal."),
+			display("Unable to create block proposal."),
+		}
+
+		/// Error checking signature
+		InvalidSignature(s: ::primitives::ed25519::Signature, a: ::primitives::AuthorityId) {
+			description("Message signature is invalid"),
+			display("Message signature {:?} by {:?} is invalid.", s, a),
+		}
+
+		/// Account is not an authority.
+		InvalidAuthority(a: ::primitives::AuthorityId) {
+			description("Message sender is not a valid authority"),
+			display("Message sender {:?} is not a valid authority.", a),
+		}
+
+		/// Authoring interface does not match the runtime.
+		IncompatibleAuthoringRuntime(native: RuntimeVersion, on_chain: RuntimeVersion) {
+			description("Authoring for current runtime is not supported"),
+			display("Authoring for current runtime is not supported. Native ({}) cannot author for on-chain ({}).", native, on_chain),
+		}
+
+		/// Authoring interface does not match the runtime.
+		RuntimeVersionMissing {
+			description("Current runtime has no version"),
+			display("Authoring for current runtime is not supported since it has no version."),
+		}
+
+		/// Authoring interface does not match the runtime.
+		NativeRuntimeMissing {
+			description("This build has no native runtime"),
+			display("Authoring in current build is not supported since it has no runtime."),
+		}
+
+		/// Justification requirements not met.
+		InvalidJustification {
+			description("Invalid justification"),
+			display("Invalid justification."),
+		}
+
+		/// Some other error.
+		Other(e: Box<::std::error::Error + Send>) {
+			description("Other error")
+			display("Other error: {}", e.description())
+		}
+	}
+}

substrate/core/consensus/common/src/evaluation.rs

@@ -0,0 +1,82 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate 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.
+
+// Substrate 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 Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Block evaluation and evaluation errors.
+
+use super::MAX_TRANSACTIONS_SIZE;
+
+use codec::Encode;
+use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, As};
+
+type BlockNumber = u64;
+
+error_chain! {
+	errors {
+		BadProposalFormat {
+			description("Proposal provided not a block."),
+			display("Proposal provided not a block."),
+		}
+		WrongParentHash(expected: String, got: String) {
+			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, size.saturating_sub(MAX_TRANSACTIONS_SIZE)
+			),
+		}
+	}
+}
+
+/// Attempt to evaluate a substrate block as a node block, returning error
+/// upon any initial validity checks failing.
+pub fn evaluate_initial<Block: BlockT>(
+	proposal: &Block,
+	parent_hash: &<Block as BlockT>::Hash,
+	parent_number: <<Block as BlockT>::Header as HeaderT>::Number,
+) -> Result<()> {
+
+	let encoded = Encode::encode(proposal);
+	let proposal = Block::decode(&mut &encoded[..])
+		.ok_or_else(|| ErrorKind::BadProposalFormat)?;
+
+	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 *parent_hash != *proposal.header().parent_hash() {
+		bail!(ErrorKind::WrongParentHash(
+			format!("{:?}", *parent_hash),
+			format!("{:?}", proposal.header().parent_hash())
+		));
+	}
+
+	if parent_number.as_() + 1 != proposal.header().number().as_() {
+		bail!(ErrorKind::WrongNumber(parent_number.as_() + 1, proposal.header().number().as_()));
+	}
+
+	Ok(())
+}

substrate/core/consensus/common/src/lib.rs

@@ -15,22 +15,118 @@
 // along with Substrate Consensus Common.  If not, see <http://www.gnu.org/licenses/>.
 
 // tag::description[]
-//! Tracks offline validators.
+//! Consensus basics and common features
 // end::description[]
 
+// This provides "unused" building blocks to other crates
 #![allow(dead_code)]
 
-#![cfg(feature="rhd")]
+// our error-chain could potentially blow up otherwise
+#![recursion_limit="128"]
 
 extern crate substrate_primitives as primitives;
+extern crate futures;
+extern crate sr_version as runtime_version;
+extern crate sr_primitives as runtime_primitives;
+extern crate tokio;
 
-use primitives::{generic::BlockId, Justification};
-use primitives::traits::{Block, Header};
+extern crate parity_codec as codec;
+#[macro_use]
+extern crate parity_codec_derive;
 
-/// Block import trait.
-pub trait BlockImport<B: Block> {
-	/// Import a block alongside its corresponding justification.
-	fn import_block(&self, block: B, justification: Justification, authorities: &[AuthorityId]) -> bool;
+#[macro_use]
+extern crate error_chain;
+
+use std::sync::Arc;
+
+use primitives::{ed25519, AuthorityId};
+use runtime_primitives::generic::BlockId;
+use runtime_primitives::traits::Block;
+use futures::prelude::*;
+
+pub mod offline_tracker;
+pub mod error;
+mod block_import;
+pub mod evaluation;
+
+// block size limit.
+const MAX_TRANSACTIONS_SIZE: usize = 4 * 1024 * 1024;
+
+pub use self::error::{Error, ErrorKind};
+pub use block_import::{BlockImport, ImportBlock, BlockOrigin, ImportResult};
+
+/// Trait for getting the authorities at a given block.
+pub trait Authorities<B: Block> {
+	type Error: ::std::error::Error + Send + 'static;	/// Get the authorities at the given block.
+	fn authorities(&self, at: &BlockId<B>) -> Result<Vec<AuthorityId>, Self::Error>;
 }
 
-pub mod offline_tracker;
+/// Environment producer for a Consensus instance. Creates proposer instance and communication streams.
+pub trait Environment<B: Block> {
+	/// The proposer type this creates.
+	type Proposer: Proposer<B>;
+	/// Error which can occur upon creation.
+	type Error: From<Error>;
+
+	/// Initialize the proposal logic on top of a specific header. Provide
+	/// the authorities at that header, and a local key to sign any additional
+	/// consensus messages with as well.
+	fn init(&self, parent_header: &B::Header, authorities: &[AuthorityId], sign_with: Arc<ed25519::Pair>)
+		-> Result<Self::Proposer, Self::Error>;
+}
+
+/// Logic for a proposer.
+///
+/// This will encapsulate creation and evaluation of proposals at a specific
+/// block.
+pub trait Proposer<B: Block> {
+	/// Error type which can occur when proposing or evaluating.
+	type Error: From<Error> + ::std::fmt::Debug + 'static;
+	/// Future that resolves to a committed proposal.
+	type Create: IntoFuture<Item=B,Error=Self::Error>;
+	/// Create a proposal.
+	fn propose(&self) -> Self::Create;
+}
+
+/// Inherent data to include in a block.
+#[derive(Encode, Decode)]
+pub struct InherentData {
+	/// Current timestamp.
+	pub timestamp: u64,
+	/// Indices of offline validators.
+	pub offline_indices: Vec<u32>,
+}
+
+impl InherentData {
+	/// Create a new `InherentData` instance.
+	pub fn new(timestamp: u64, offline_indices: Vec<u32>) -> Self {
+		Self {
+			timestamp,
+			offline_indices
+		}
+	}
+}
+
+/// An oracle for when major synchronization work is being undertaken.
+///
+/// Generally, consensus authoring work isn't undertaken while well behind
+/// the head of the chain.
+pub trait SyncOracle {
+	/// Whether the synchronization service is undergoing major sync.
+	/// Returns true if so.
+	fn is_major_syncing(&self) -> bool;
+}
+
+/// A synchronization oracle for when there is no network.
+#[derive(Clone, Copy, Debug)]
+pub struct NoNetwork;
+
+impl SyncOracle for NoNetwork {
+	fn is_major_syncing(&self) -> bool { false }
+}
+
+impl<T: SyncOracle> SyncOracle for Arc<T> {
+	fn is_major_syncing(&self) -> bool {
+		T::is_major_syncing(&*self)
+	}
+}

substrate/core/consensus/common/src/offline_tracker.rs

@@ -16,7 +16,7 @@
 
 //! Tracks offline validators.
 
-use node_primitives::AccountId;
+use primitives::AuthorityId;
 
 use std::collections::HashMap;
 use std::time::{Instant, Duration};
@@ -56,7 +56,7 @@ impl Observed {
 
 /// Tracks offline validators and can issue a report for those offline.
 pub struct OfflineTracker {
-	observed: HashMap<AccountId, Observed>,
+	observed: HashMap<AuthorityId, Observed>,
 }
 
 impl OfflineTracker {
@@ -66,7 +66,7 @@ impl OfflineTracker {
 	}
 
 	/// Note new consensus is starting with the given set of validators.
-	pub fn note_new_block(&mut self, validators: &[AccountId]) {
+	pub fn note_new_block(&mut self, validators: &[AuthorityId]) {
 		use std::collections::HashSet;
 
 		let set: HashSet<_> = validators.iter().cloned().collect();
@@ -74,14 +74,14 @@ impl OfflineTracker {
 	}
 
 	/// Note that a round has ended.
-	pub fn note_round_end(&mut self, validator: AccountId, was_online: bool) {
+	pub fn note_round_end(&mut self, validator: AuthorityId, was_online: bool) {
 		self.observed.entry(validator)
 			.or_insert_with(Observed::new)
 			.note_round_end(was_online);
 	}
 
 	/// Generate a vector of indices for offline account IDs.
-	pub fn reports(&self, validators: &[AccountId]) -> Vec<u32> {
+	pub fn reports(&self, validators: &[AuthorityId]) -> Vec<u32> {
 		validators.iter()
 			.enumerate()
 			.filter_map(|(i, v)| if self.is_online(v) {
@@ -93,7 +93,7 @@ impl OfflineTracker {
 	}
 
 	/// Whether reports on a validator set are consistent with our view of things.
-	pub fn check_consistency(&self, validators: &[AccountId], reports: &[u32]) -> bool {
+	pub fn check_consistency(&self, validators: &[AuthorityId], reports: &[u32]) -> bool {
 		reports.iter().cloned().all(|r| {
 			let v = match validators.get(r as usize) {
 				Some(v) => v,
@@ -106,7 +106,7 @@ impl OfflineTracker {
 		})
 	}
 
-	fn is_online(&self, v: &AccountId) -> bool {
+	fn is_online(&self, v: &AuthorityId) -> bool {
 		self.observed.get(v).map(Observed::is_active).unwrap_or(true)
 	}
 }

substrate/core/consensus/rhd/Cargo.toml

@@ -6,30 +6,33 @@ description = "Rhododendron Round-Based consensus-algorithm for substrate"
 
 [dependencies]
 futures = "0.1.17"
-parity-codec = { version = "1.1" }
+parity-codec = { version = "2.1" }
 parity-codec-derive = { version = "2.0" }
 substrate-primitives = { path = "../../primitives" }
+substrate-consensus-common = { path = "../common" }
+substrate-client = { path = "../../client" }
+substrate-transaction-pool = { path = "../../transaction-pool" }
 srml-support = { path = "../../../srml/support" }
+srml-system = { path = "../../../srml/system" }
+srml-consensus = { path = "../../../srml/consensus" }
 sr-primitives = { path = "../../sr-primitives" }
 sr-version = { path = "../../sr-version" }
 sr-io = { path = "../../sr-io" }
-srml-consensus = { path = "../../../srml/consensus" }
 tokio = "0.1.7"
 parking_lot = "0.4"
 error-chain = "0.12"
-log = "0.3"
-rhododendron = { git = "https://github.com/paritytech/rhododendron.git", features = ["codec"] }
-serde = { version = "1.0", features = ["derive"] }
+log = "0.4"
+rhododendron = { version = "0.4.0", features = ["codec"] }
+exit-future = "0.1"
+
 
 [dev-dependencies]
 substrate-keyring = { path = "../../keyring" }
 substrate-executor = { path = "../../executor" }
 
-
 [features]
 default = ["std"]
 std = [
-	"serde/std",
 	"substrate-primitives/std",
 	"srml-support/std",
 	"sr-primitives/std",

substrate/core/consensus/rhd/src/error.rs

@@ -1,4 +1,4 @@
-// Copyright 2017-2018 Parity Technologies (UK) Ltd.
+// Copyright 2018 Parity Technologies (UK) Ltd.
 // This file is part of Substrate.
 
 // Substrate is free software: you can redistribute it and/or modify
@@ -14,81 +14,44 @@
 // You should have received a copy of the GNU General Public License
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
-//! Error types in the BFT service.
-use runtime_version::RuntimeVersion;
+//! Error types in the rhododendron Consensus service.
+use consensus::error::{Error as CommonError, ErrorKind as CommonErrorKind};
+use primitives::AuthorityId;
+use client;
 
 error_chain! {
+	links {
+		Client(client::error::Error, client::error::ErrorKind);
+		Common(CommonError, CommonErrorKind);
+	}
 	errors {
-		/// Missing state at block with given descriptor.
-		StateUnavailable(b: String) {
-			description("State missing at given block."),
-			display("State unavailable at block {}", b),
+		NotValidator(id: AuthorityId) {
+			description("Local account ID not a validator at this block."),
+			display("Local account ID ({:?}) not a validator at this block.", id),
 		}
-
-		/// I/O terminated unexpectedly
-		IoTerminated {
-			description("I/O terminated unexpectedly."),
-			display("I/O terminated unexpectedly."),
+		PrematureDestruction {
+			description("Proposer destroyed before finishing proposing or evaluating"),
+			display("Proposer destroyed before finishing proposing or evaluating"),
 		}
-
-		/// Unable to schedule wakeup.
-		FaultyTimer(e: ::tokio::timer::Error) {
-			description("Timer error"),
-			display("Timer error: {}", e),
+		Timer(e: ::tokio::timer::Error) {
+			description("Failed to register or resolve async timer."),
+			display("Timer failed: {}", e),
 		}
-
-		/// Unable to propose a block.
-		CannotPropose {
-			description("Unable to create block proposal."),
-			display("Unable to create block proposal."),
-		}
-
-		/// Error checking signature
-		InvalidSignature(s: ::primitives::ed25519::Signature, a: ::primitives::AuthorityId) {
-			description("Message signature is invalid"),
-			display("Message signature {:?} by {:?} is invalid.", s, a),
-		}
-
-		/// Account is not an authority.
-		InvalidAuthority(a: ::primitives::AuthorityId) {
-			description("Message sender is not a valid authority"),
-			display("Message sender {:?} is not a valid authority.", a),
-		}
-
-		/// Authoring interface does not match the runtime.
-		IncompatibleAuthoringRuntime(native: RuntimeVersion, on_chain: RuntimeVersion) {
-			description("Authoring for current runtime is not supported"),
-			display("Authoring for current runtime is not supported. Native ({}) cannot author for on-chain ({}).", native, on_chain),
-		}
-
-		/// Authoring interface does not match the runtime.
-		RuntimeVersionMissing {
-			description("Current runtime has no version"),
-			display("Authoring for current runtime is not supported since it has no version."),
-		}
-
-		/// Authoring interface does not match the runtime.
-		NativeRuntimeMissing {
-			description("This build has no native runtime"),
-			display("Authoring in current build is not supported since it has no runtime."),
-		}
-
-		/// Justification requirements not met.
-		InvalidJustification {
-			description("Invalid justification"),
-			display("Invalid justification."),
-		}
-
-		/// Some other error.
-		Other(e: Box<::std::error::Error + Send>) {
-			description("Other error")
-			display("Other error: {}", e.description())
+		Executor(e: ::futures::future::ExecuteErrorKind) {
+			description("Unable to dispatch agreement future"),
+			display("Unable to dispatch agreement future: {:?}", e),
 		}
 	}
 }
 
 impl From<::rhododendron::InputStreamConcluded> for Error {
-	fn from(_: ::rhododendron::InputStreamConcluded) -> Error {
-		ErrorKind::IoTerminated.into()
+	fn from(_: ::rhododendron::InputStreamConcluded) -> Self {
+		CommonErrorKind::IoTerminated.into()
 	}
 }
+
+impl From<CommonErrorKind> for Error {
+	fn from(e: CommonErrorKind) -> Self {
+		CommonError::from(e).into()
+	}
+}

substrate/core/consensus/rhd/src/service.rs

@@ -0,0 +1,181 @@
+// Copyright 2018 Parity Technologies (UK) Ltd.
+// This file is part of Substrate.
+
+// Substrate 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.
+
+// Substrate 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 Substrate.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Consensus service.
+
+/// Consensus service. A long running service that manages BFT agreement
+/// the network.
+use std::thread;
+use std::time::{Duration, Instant};
+use std::sync::Arc;
+
+use client::{BlockchainEvents, ChainHead, BlockBody};
+use futures::prelude::*;
+use transaction_pool::txpool::{Pool as TransactionPool, ChainApi as PoolChainApi};
+use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, BlockNumberToHash};
+
+use tokio::executor::current_thread::TaskExecutor as LocalThreadHandle;
+use tokio::runtime::TaskExecutor as ThreadPoolHandle;
+use tokio::runtime::current_thread::Runtime as LocalRuntime;
+use tokio::timer::Interval;
+
+use parking_lot::RwLock;
+use consensus::offline_tracker::OfflineTracker;
+
+use super::{Network, ProposerFactory, AuthoringApi};
+use {consensus, primitives, ed25519, error, BftService, LocalProposer};
+
+const TIMER_DELAY_MS: u64 = 5000;
+const TIMER_INTERVAL_MS: u64 = 500;
+
+// spin up an instance of BFT agreement on the current thread's executor.
+// panics if there is no current thread executor.
+fn start_bft<F, C, Block>(
+	header: <Block as BlockT>::Header,
+	bft_service: Arc<BftService<Block, F, C>>,
+) where
+	F: consensus::Environment<Block> + 'static,
+	C: consensus::BlockImport<Block> + consensus::Authorities<Block> + 'static,
+	F::Error: ::std::fmt::Debug,
+	<F::Proposer as consensus::Proposer<Block>>::Error: ::std::fmt::Display + Into<error::Error>,
+	<F as consensus::Environment<Block>>::Proposer : LocalProposer<Block>,
+	<F as consensus::Environment<Block>>::Error: ::std::fmt::Display,
+	Block: BlockT,
+{
+	let mut handle = LocalThreadHandle::current();
+	match bft_service.build_upon(&header) {
+		Ok(Some(bft_work)) => if let Err(e) = handle.spawn_local(Box::new(bft_work)) {
+		    warn!(target: "bft", "Couldn't initialize BFT agreement: {:?}", e);
+		}
+		Ok(None) => trace!(target: "bft", "Could not start agreement on top of {}", header.hash()),
+		Err(e) => warn!(target: "bft", "BFT agreement error: {}", e),
+ 	}
+}
+
+/// Consensus service. Starts working when created.
+pub struct Service {
+	thread: Option<thread::JoinHandle<()>>,
+	exit_signal: Option<::exit_future::Signal>,
+}
+
+impl Service {
+	/// Create and start a new instance.
+	pub fn new<A, P, C, N>(
+		client: Arc<C>,
+		api: Arc<A>,
+		network: N,
+		transaction_pool: Arc<TransactionPool<P>>,
+		thread_pool: ThreadPoolHandle,
+		key: ed25519::Pair,
+		block_delay: u64,
+	) -> Service
+		where
+			error::Error: From<<A as AuthoringApi>::Error>,
+			A: AuthoringApi + BlockNumberToHash + 'static,
+			P: PoolChainApi<Block = <A as AuthoringApi>::Block> + 'static,
+			C: BlockchainEvents<<A as AuthoringApi>::Block>
+				+ ChainHead<<A as AuthoringApi>::Block>
+				+ BlockBody<<A as AuthoringApi>::Block>,
+			C: consensus::BlockImport<<A as AuthoringApi>::Block>
+				+ consensus::Authorities<<A as AuthoringApi>::Block> + Send + Sync + 'static,
+			primitives::H256: From<<<A as AuthoringApi>::Block as BlockT>::Hash>,
+			<<A as AuthoringApi>::Block as BlockT>::Hash: PartialEq<primitives::H256> + PartialEq,
+			N: Network<Block = <A as AuthoringApi>::Block> + Send + 'static,
+	{
+
+		let (signal, exit) = ::exit_future::signal();
+		let thread = thread::spawn(move || {
+			let mut runtime = LocalRuntime::new().expect("Could not create local runtime");
+			let key = Arc::new(key);
+
+			let factory = ProposerFactory {
+				client: api.clone(),
+				transaction_pool: transaction_pool.clone(),
+				network,
+				handle: thread_pool.clone(),
+				offline: Arc::new(RwLock::new(OfflineTracker::new())),
+				force_delay: block_delay,
+			};
+			let bft_service = Arc::new(BftService::new(client.clone(), key, factory));
+
+			let notifications = {
+				let client = client.clone();
+				let bft_service = bft_service.clone();
+
+				client.import_notification_stream().for_each(move |notification| {
+					if notification.is_new_best {
+						start_bft(notification.header, bft_service.clone());
+					}
+					Ok(())
+				})
+			};
+
+			let interval = Interval::new(
+				Instant::now() + Duration::from_millis(TIMER_DELAY_MS),
+				Duration::from_millis(TIMER_INTERVAL_MS),
+			);
+
+			let mut prev_best = match client.best_block_header() {
+				Ok(header) => header.hash(),
+				Err(e) => {
+					warn!("Cant's start consensus service. Error reading best block header: {:?}", e);
+					return;
+				}
+			};
+
+			let timed = {
+				let c = client.clone();
+				let s = bft_service.clone();
+
+				interval.map_err(|e| debug!(target: "bft", "Timer error: {:?}", e)).for_each(move |_| {
+					if let Ok(best_block) = c.best_block_header() {
+						let hash = best_block.hash();
+
+						if hash == prev_best {
+							debug!(target: "bft", "Starting consensus round after a timeout");
+							start_bft(best_block, s.clone());
+						}
+						prev_best = hash;
+					}
+					Ok(())
+				})
+			};
+
+			runtime.spawn(notifications);
+			runtime.spawn(timed);
+
+			if let Err(e) = runtime.block_on(exit) {
+				debug!("BFT event loop error {:?}", e);
+			}
+		});
+		Service {
+			thread: Some(thread),
+			exit_signal: Some(signal),
+		}
+	}
+}
+
+impl Drop for Service {
+	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");
+		}
+	}
+}