Split BFT into substrate-bft and runtime-specific proposer logic (#72)

* reshuffle consensus libraries

* polkadot-useful type definitions for statement table

* begin BftService

* primary selection logic

* bft service implementation without I/O

* extract out `BlockImport` trait

* allow bft primitives to compile on wasm

* take polkadot-consensus down to the core.

* test for preemption

* fix test build

substrate/substrate/bft/Cargo.toml

@@ -0,0 +1,20 @@
+[package]
+name = "substrate-bft"
+version = "0.1.0"
+authors = ["Parity Technologies <admin@parity.io>"]
+
+[dependencies]
+futures = "0.1.17"
+substrate-client = { path = "../client" }
+substrate-codec = { path = "../codec" }
+substrate-primitives = { path = "../primitives" }
+substrate-state-machine = { path = "../state-machine" }
+ed25519 = { path = "../ed25519" }
+tokio-timer = "0.1.2"
+parking_lot = "0.4"
+error-chain = "0.11"
+
+[dev-dependencies]
+substrate-keyring = { path = "../keyring" }
+substrate-executor = { path = "../executor" }
+tokio-core = "0.1.12"

substrate/substrate/bft/src/error.rs

@@ -0,0 +1,57 @@
+// Copyright 2017 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 the BFT service.
+
+error_chain! {
+	errors {
+		/// Missing state at block with given Id.
+		StateUnavailable(b: ::client::BlockId) {
+			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 {
+			description("Faulty timer: unable to schedule wakeup"),
+			display("Faulty timer: unable to schedule wakeup"),
+		}
+
+		/// Unable to propose a block.
+		CannotPropose {
+			description("Unable to create block proposal."),
+			display("Unable to create block proposal."),
+		}
+
+		/// Error dispatching the agreement future onto the executor.
+		Executor(e: ::futures::future::ExecuteErrorKind) {
+			description("Unable to dispatch agreement future"),
+			display("Unable to dispatch agreement future: {:?}", e),
+		}
+	}
+}
+
+impl From<::generic::InputStreamConcluded> for Error {
+	fn from(_: ::generic::InputStreamConcluded) -> Error {
+		ErrorKind::IoTerminated.into()
+	}
+}

substrate/substrate/bft/src/generic/accumulator.rs

@@ -0,0 +1,602 @@
+// Copyright 2017 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/>.
+
+//! Message accumulator for each round of BFT consensus.
+
+use std::collections::{HashMap, HashSet};
+use std::collections::hash_map::Entry;
+use std::hash::Hash;
+
+use generic::{Message, LocalizedMessage};
+
+/// Justification for some state at a given round.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct UncheckedJustification<D, S> {
+	/// The round.
+	pub round_number: usize,
+	/// The digest prepared for.
+	pub digest: D,
+	/// Signatures for the prepare messages.
+	pub signatures: Vec<S>,
+}
+
+impl<D, S> UncheckedJustification<D, S> {
+	/// Fails if there are duplicate signatures or invalid.
+	///
+	/// Provide a closure for checking whether the signature is valid on a
+	/// digest.
+	///
+	/// The closure should returns a checked justification iff the round number, digest, and signature
+	/// represent a valid message and the signer was authorized to issue
+	/// it.
+	///
+	/// The `check_message` closure may vary based on context.
+	pub fn check<F, V>(self, threshold: usize, mut check_message: F)
+		-> Result<Justification<D, S>, Self>
+		where
+			F: FnMut(usize, &D, &S) -> Option<V>,
+			V: Hash + Eq,
+	{
+		let checks_out = {
+			let mut checks_out = || {
+				let mut voted = HashSet::new();
+
+				for signature in &self.signatures {
+					match check_message(self.round_number, &self.digest, signature) {
+						None => return false,
+						Some(v) => {
+							if !voted.insert(v) {
+								return false;
+							}
+						}
+					}
+				}
+
+				voted.len() >= threshold
+			};
+
+			checks_out()
+		};
+
+		if checks_out {
+			Ok(Justification(self))
+		} else {
+			Err(self)
+		}
+	}
+}
+
+/// A checked justification.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Justification<D,S>(UncheckedJustification<D,S>);
+
+impl<D, S> ::std::ops::Deref for Justification<D, S> {
+	type Target = UncheckedJustification<D, S>;
+
+	fn deref(&self) -> &Self::Target {
+		&self.0
+	}
+}
+
+/// Type alias to represent a justification specifically for a prepare.
+pub type PrepareJustification<D, S> = Justification<D, S>;
+
+/// The round's state, based on imported messages.
+#[derive(PartialEq, Eq, Debug)]
+pub enum State<Candidate, Digest, Signature> {
+	/// No proposal yet.
+	Begin,
+	/// Proposal received.
+	Proposed(Candidate),
+	/// Seen n - f prepares for this digest.
+	Prepared(PrepareJustification<Digest, Signature>),
+	/// Seen n - f commits for a digest.
+	Committed(Justification<Digest, Signature>),
+	/// Seen n - f round-advancement messages.
+	Advanced(Option<PrepareJustification<Digest, Signature>>),
+}
+
+#[derive(Debug, Default)]
+struct VoteCounts {
+	prepared: usize,
+	committed: usize,
+}
+
+/// Accumulates messages for a given round of BFT consensus.
+///
+/// This isn't tied to the "view" of a single authority. It
+/// keeps accurate track of the state of the BFT consensus based
+/// on all messages imported.
+#[derive(Debug)]
+pub struct Accumulator<Candidate, Digest, AuthorityId, Signature>
+	where
+	Candidate: Eq + Clone,
+	Digest: Hash + Eq + Clone,
+	AuthorityId: Hash + Eq,
+	Signature: Eq + Clone,
+{
+	round_number: usize,
+	threshold: usize,
+	round_proposer: AuthorityId,
+	proposal: Option<Candidate>,
+	prepares: HashMap<AuthorityId, (Digest, Signature)>,
+	commits: HashMap<AuthorityId, (Digest, Signature)>,
+	vote_counts: HashMap<Digest, VoteCounts>,
+	advance_round: HashSet<AuthorityId>,
+	state: State<Candidate, Digest, Signature>,
+}
+
+impl<Candidate, Digest, AuthorityId, Signature> Accumulator<Candidate, Digest, AuthorityId, Signature>
+	where
+	Candidate: Eq + Clone,
+	Digest: Hash + Eq + Clone,
+	AuthorityId: Hash + Eq,
+	Signature: Eq + Clone,
+{
+	/// Create a new state accumulator.
+	pub fn new(round_number: usize, threshold: usize, round_proposer: AuthorityId) -> Self {
+		Accumulator {
+			round_number,
+			threshold,
+			round_proposer,
+			proposal: None,
+			prepares: HashMap::new(),
+			commits: HashMap::new(),
+			vote_counts: HashMap::new(),
+			advance_round: HashSet::new(),
+			state: State::Begin,
+		}
+	}
+
+	/// How advance votes we have seen.
+	pub fn advance_votes(&self) -> usize {
+		self.advance_round.len()
+	}
+
+	/// Get the round number.
+	pub fn round_number(&self) -> usize {
+		self.round_number.clone()
+	}
+
+	pub fn proposal(&self) -> Option<&Candidate> {
+		self.proposal.as_ref()
+	}
+
+	/// Inspect the current consensus state.
+	pub fn state(&self) -> &State<Candidate, Digest, Signature> {
+		&self.state
+	}
+
+	/// Import a message. Importing duplicates is fine, but the signature
+	/// and authorization should have already been checked.
+	pub fn import_message(
+		&mut self,
+		message: LocalizedMessage<Candidate, Digest, AuthorityId, Signature>,
+	)
+	{
+		// message from different round.
+		if message.message.round_number() != self.round_number {
+			return;
+		}
+
+		let (sender, signature) = (message.sender, message.signature);
+
+		match message.message {
+			Message::Propose(_, p) => self.import_proposal(p, sender),
+			Message::Prepare(_, d) => self.import_prepare(d, sender, signature),
+			Message::Commit(_, d) => self.import_commit(d, sender, signature),
+			Message::AdvanceRound(_) => self.import_advance_round(sender),
+		}
+	}
+
+	fn import_proposal(
+		&mut self,
+		proposal: Candidate,
+		sender: AuthorityId,
+	) {
+		if sender != self.round_proposer || self.proposal.is_some() { return }
+
+		self.proposal = Some(proposal.clone());
+		self.state = State::Proposed(proposal);
+	}
+
+	fn import_prepare(
+		&mut self,
+		digest: Digest,
+		sender: AuthorityId,
+		signature: Signature,
+	) {
+		// ignore any subsequent prepares by the same sender.
+		// TODO: if digest is different, that's misbehavior.
+		let threshold_prepared = if let Entry::Vacant(vacant) = self.prepares.entry(sender) {
+			vacant.insert((digest.clone(), signature));
+			let count = self.vote_counts.entry(digest.clone()).or_insert_with(Default::default);
+			count.prepared += 1;
+
+			if count.prepared >= self.threshold {
+				Some(digest)
+			} else {
+				None
+			}
+		} else {
+			None
+		};
+
+		// only allow transition to prepare from begin or proposed state.
+		let valid_transition = match self.state {
+			State::Begin | State::Proposed(_) => true,
+			_ => false,
+		};
+
+		if let (true, Some(threshold_prepared)) = (valid_transition, threshold_prepared) {
+			let signatures = self.prepares
+				.values()
+				.filter(|&&(ref d, _)| d == &threshold_prepared)
+				.map(|&(_, ref s)| s.clone())
+				.collect();
+
+			self.state = State::Prepared(Justification(UncheckedJustification {
+				round_number: self.round_number,
+				digest: threshold_prepared,
+				signatures: signatures,
+			}));
+		}
+	}
+
+	fn import_commit(
+		&mut self,
+		digest: Digest,
+		sender: AuthorityId,
+		signature: Signature,
+	) {
+		// ignore any subsequent commits by the same sender.
+		// TODO: if digest is different, that's misbehavior.
+		let threshold_committed = if let Entry::Vacant(vacant) = self.commits.entry(sender) {
+			vacant.insert((digest.clone(), signature));
+			let count = self.vote_counts.entry(digest.clone()).or_insert_with(Default::default);
+			count.committed += 1;
+
+			if count.committed >= self.threshold {
+				Some(digest)
+			} else {
+				None
+			}
+		} else {
+			None
+		};
+
+		// transition to concluded state always valid.
+		// only weird case is if the prior state was "advanced",
+		// but technically it's the same behavior as if the order of receiving
+		// the last "advance round" and "commit" messages were reversed.
+		if let Some(threshold_committed) = threshold_committed {
+			let signatures = self.commits
+				.values()
+				.filter(|&&(ref d, _)| d == &threshold_committed)
+				.map(|&(_, ref s)| s.clone())
+				.collect();
+
+			self.state = State::Committed(Justification(UncheckedJustification {
+				round_number: self.round_number,
+				digest: threshold_committed,
+				signatures: signatures,
+			}));
+		}
+	}
+
+	fn import_advance_round(
+		&mut self,
+		sender: AuthorityId,
+	) {
+		self.advance_round.insert(sender);
+
+		if self.advance_round.len() < self.threshold { return }
+
+		// allow transition to new round only if we haven't produced a justification
+		// yet.
+		self.state = match ::std::mem::replace(&mut self.state, State::Begin) {
+			State::Committed(j) => State::Committed(j),
+			State::Prepared(j) => State::Advanced(Some(j)),
+			State::Advanced(j) => State::Advanced(j),
+			State::Begin | State::Proposed(_) => State::Advanced(None),
+		}
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+
+	#[derive(Clone, PartialEq, Eq, Debug)]
+	pub struct Candidate(usize);
+
+	#[derive(Hash, PartialEq, Eq, Clone, Debug)]
+	pub struct Digest(usize);
+
+	#[derive(Hash, PartialEq, Eq, Debug)]
+	pub struct AuthorityId(usize);
+
+	#[derive(PartialEq, Eq, Clone, Debug)]
+	pub struct Signature(usize, usize);
+
+	#[test]
+	fn justification_checks_out() {
+		let mut justification = UncheckedJustification {
+			round_number: 2,
+			digest: Digest(600),
+			signatures: (0..10).map(|i| Signature(600, i)).collect(),
+		};
+
+		let check_message = |r, d: &Digest, s: &Signature| {
+			if r == 2 && d.0 == 600 && s.0 == 600 {
+				Some(AuthorityId(s.1))
+			} else {
+				None
+			}
+		};
+
+		assert!(justification.clone().check(7, &check_message).is_ok());
+		assert!(justification.clone().check(11, &check_message).is_err());
+
+		{
+			// one bad signature is enough to spoil it.
+			justification.signatures.push(Signature(1001, 255));
+			assert!(justification.clone().check(7, &check_message).is_err());
+
+			justification.signatures.pop();
+		}
+		// duplicates not allowed.
+		justification.signatures.extend((0..10).map(|i| Signature(600, i)));
+		assert!(justification.clone().check(11, &check_message).is_err());
+	}
+
+	#[test]
+	fn accepts_proposal_from_proposer_only() {
+		let mut accumulator = Accumulator::<_, Digest, _, _>::new(1, 7, AuthorityId(8));
+		assert_eq!(accumulator.state(), &State::Begin);
+
+		accumulator.import_message(LocalizedMessage {
+			sender: AuthorityId(5),
+			signature: Signature(999, 5),
+			message: Message::Propose(1, Candidate(999)),
+		});
+
+		assert_eq!(accumulator.state(), &State::Begin);
+
+		accumulator.import_message(LocalizedMessage {
+			sender: AuthorityId(8),
+			signature: Signature(999, 8),
+			message: Message::Propose(1, Candidate(999)),
+		});
+
+		assert_eq!(accumulator.state(), &State::Proposed(Candidate(999)));
+	}
+
+	#[test]
+	fn reaches_prepare_phase() {
+		let mut accumulator = Accumulator::new(1, 7, AuthorityId(8));
+		assert_eq!(accumulator.state(), &State::Begin);
+
+		accumulator.import_message(LocalizedMessage {
+			sender: AuthorityId(8),
+			signature: Signature(999, 8),
+			message: Message::Propose(1, Candidate(999)),
+		});
+
+		assert_eq!(accumulator.state(), &State::Proposed(Candidate(999)));
+
+		for i in 0..6 {
+			accumulator.import_message(LocalizedMessage {
+				sender: AuthorityId(i),
+				signature: Signature(999, i),
+				message: Message::Prepare(1, Digest(999)),
+			});
+
+			assert_eq!(accumulator.state(), &State::Proposed(Candidate(999)));
+		}
+
+		accumulator.import_message(LocalizedMessage {
+			sender: AuthorityId(7),
+			signature: Signature(999, 7),
+			message: Message::Prepare(1, Digest(999)),
+		});
+
+		match accumulator.state() {
+			&State::Prepared(ref j) => assert_eq!(j.digest, Digest(999)),
+			s => panic!("wrong state: {:?}", s),
+		}
+	}
+
+	#[test]
+	fn prepare_to_commit() {
+		let mut accumulator = Accumulator::new(1, 7, AuthorityId(8));
+		assert_eq!(accumulator.state(), &State::Begin);
+
+		accumulator.import_message(LocalizedMessage {
+			sender: AuthorityId(8),
+			signature: Signature(999, 8),
+			message: Message::Propose(1, Candidate(999)),
+		});
+
+		assert_eq!(accumulator.state(), &State::Proposed(Candidate(999)));
+
+		for i in 0..6 {
+			accumulator.import_message(LocalizedMessage {
+				sender: AuthorityId(i),
+				signature: Signature(999, i),
+				message: Message::Prepare(1, Digest(999)),
+			});
+
+			assert_eq!(accumulator.state(), &State::Proposed(Candidate(999)));
+		}
+
+		accumulator.import_message(LocalizedMessage {
+			sender: AuthorityId(7),
+			signature: Signature(999, 7),
+			message: Message::Prepare(1, Digest(999)),
+		});
+
+		match accumulator.state() {
+			&State::Prepared(ref j) => assert_eq!(j.digest, Digest(999)),
+			s => panic!("wrong state: {:?}", s),
+		}
+
+		for i in 0..6 {
+			accumulator.import_message(LocalizedMessage {
+				sender: AuthorityId(i),
+				signature: Signature(999, i),
+				message: Message::Commit(1, Digest(999)),
+			});
+
+			match accumulator.state() {
+				&State::Prepared(_) => {},
+				s => panic!("wrong state: {:?}", s),
+			}
+		}
+
+		accumulator.import_message(LocalizedMessage {
+			sender: AuthorityId(7),
+			signature: Signature(999, 7),
+			message: Message::Commit(1, Digest(999)),
+		});
+
+		match accumulator.state() {
+			&State::Committed(ref j) => assert_eq!(j.digest, Digest(999)),
+			s => panic!("wrong state: {:?}", s),
+		}
+	}
+
+	#[test]
+	fn prepare_to_advance() {
+		let mut accumulator = Accumulator::new(1, 7, AuthorityId(8));
+		assert_eq!(accumulator.state(), &State::Begin);
+
+		accumulator.import_message(LocalizedMessage {
+			sender: AuthorityId(8),
+			signature: Signature(999, 8),
+			message: Message::Propose(1, Candidate(999)),
+		});
+
+		assert_eq!(accumulator.state(), &State::Proposed(Candidate(999)));
+
+		for i in 0..7 {
+			accumulator.import_message(LocalizedMessage {
+				sender: AuthorityId(i),
+				signature: Signature(999, i),
+				message: Message::Prepare(1, Digest(999)),
+			});
+		}
+
+		match accumulator.state() {
+			&State::Prepared(ref j) => assert_eq!(j.digest, Digest(999)),
+			s => panic!("wrong state: {:?}", s),
+		}
+
+		for i in 0..6 {
+			accumulator.import_message(LocalizedMessage {
+				sender: AuthorityId(i),
+				signature: Signature(999, i),
+				message: Message::AdvanceRound(1),
+			});
+
+			match accumulator.state() {
+				&State::Prepared(_) => {},
+				s => panic!("wrong state: {:?}", s),
+			}
+		}
+
+		accumulator.import_message(LocalizedMessage {
+			sender: AuthorityId(7),
+			signature: Signature(999, 7),
+			message: Message::AdvanceRound(1),
+		});
+
+		match accumulator.state() {
+			&State::Advanced(Some(_)) => {},
+			s => panic!("wrong state: {:?}", s),
+		}
+	}
+
+	#[test]
+	fn conclude_different_than_proposed() {
+		let mut accumulator = Accumulator::<Candidate, _, _, _>::new(1, 7, AuthorityId(8));
+		assert_eq!(accumulator.state(), &State::Begin);
+
+		for i in 0..7 {
+			accumulator.import_message(LocalizedMessage {
+				sender: AuthorityId(i),
+				signature: Signature(999, i),
+				message: Message::Prepare(1, Digest(999)),
+			});
+		}
+
+		match accumulator.state() {
+			&State::Prepared(ref j) => assert_eq!(j.digest, Digest(999)),
+			s => panic!("wrong state: {:?}", s),
+		}
+
+		for i in 0..7 {
+			accumulator.import_message(LocalizedMessage {
+				sender: AuthorityId(i),
+				signature: Signature(999, i),
+				message: Message::Commit(1, Digest(999)),
+			});
+		}
+
+		match accumulator.state() {
+			&State::Committed(ref j) => assert_eq!(j.digest, Digest(999)),
+			s => panic!("wrong state: {:?}", s),
+		}
+	}
+
+	#[test]
+	fn begin_to_advance() {
+		let mut accumulator = Accumulator::<Candidate, Digest, _, _>::new(1, 7, AuthorityId(8));
+		assert_eq!(accumulator.state(), &State::Begin);
+
+		for i in 0..7 {
+			accumulator.import_message(LocalizedMessage {
+				sender: AuthorityId(i),
+				signature: Signature(1, i),
+				message: Message::AdvanceRound(1),
+			});
+		}
+
+		match accumulator.state() {
+			&State::Advanced(ref j) => assert!(j.is_none()),
+			s => panic!("wrong state: {:?}", s),
+		}
+	}
+
+	#[test]
+	fn conclude_without_prepare() {
+		let mut accumulator = Accumulator::<Candidate, _, _, _>::new(1, 7, AuthorityId(8));
+		assert_eq!(accumulator.state(), &State::Begin);
+
+		for i in 0..7 {
+			accumulator.import_message(LocalizedMessage {
+				sender: AuthorityId(i),
+				signature: Signature(999, i),
+				message: Message::Commit(1, Digest(999)),
+			});
+		}
+
+		match accumulator.state() {
+			&State::Committed(ref j) => assert_eq!(j.digest, Digest(999)),
+			s => panic!("wrong state: {:?}", s),
+		}
+	}
+}

substrate/substrate/bft/src/generic/mod.rs

@@ -0,0 +1,743 @@
+// Copyright 2017 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/>.
+
+//! BFT Agreement based on a rotating proposer in different rounds.
+//! Very general implementation.
+
+use std::collections::{HashMap, VecDeque};
+use std::fmt::Debug;
+use std::hash::Hash;
+
+use futures::{future, Future, Stream, Sink, Poll, Async, AsyncSink};
+
+use self::accumulator::State;
+
+pub use self::accumulator::{Accumulator, Justification, PrepareJustification, UncheckedJustification};
+
+mod accumulator;
+
+#[cfg(test)]
+mod tests;
+
+/// Messages over the proposal.
+/// Each message carries an associated round number.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Message<C, D> {
+	/// Send a full proposal.
+	Propose(usize, C),
+	/// Prepare to vote for proposal with digest D.
+	Prepare(usize, D),
+	/// Commit to proposal with digest D..
+	Commit(usize, D),
+	/// Propose advancement to a new round.
+	AdvanceRound(usize),
+}
+
+impl<C, D> Message<C, D> {
+	/// Extract the round number.
+	pub fn round_number(&self) -> usize {
+		match *self {
+			Message::Propose(round, _) => round,
+			Message::Prepare(round, _) => round,
+			Message::Commit(round, _) => round,
+			Message::AdvanceRound(round) => round,
+		}
+	}
+}
+
+/// A localized message, including the sender.
+#[derive(Debug, Clone)]
+pub struct LocalizedMessage<C, D, V, S> {
+	/// The message received.
+	pub message: Message<C, D>,
+	/// The sender of the message
+	pub sender: V,
+	/// The signature of the message.
+	pub signature: S,
+}
+
+/// Context necessary for agreement.
+///
+/// Provides necessary types for protocol messages, and functions necessary for a
+/// participant to evaluate and create those messages.
+pub trait Context {
+	/// Candidate proposed.
+	type Candidate: Debug + Eq + Clone;
+	/// Candidate digest.
+	type Digest: Debug + Hash + Eq + Clone;
+	/// Authority ID.
+	type AuthorityId: Debug + Hash + Eq + Clone;
+	/// Signature.
+	type Signature: Debug + Eq + Clone;
+	/// A future that resolves when a round timeout is concluded.
+	type RoundTimeout: Future<Item=()>;
+	/// A future that resolves when a proposal is ready.
+	type CreateProposal: Future<Item=Self::Candidate>;
+
+	/// Get the local authority ID.
+	fn local_id(&self) -> Self::AuthorityId;
+
+	/// Get the best proposal.
+	fn proposal(&self) -> Self::CreateProposal;
+
+	/// Get the digest of a candidate.
+	fn candidate_digest(&self, candidate: &Self::Candidate) -> Self::Digest;
+
+	/// Sign a message using the local authority ID.
+	fn sign_local(&self, message: Message<Self::Candidate, Self::Digest>)
+		-> LocalizedMessage<Self::Candidate, Self::Digest, Self::AuthorityId, Self::Signature>;
+
+	/// Get the proposer for a given round of consensus.
+	fn round_proposer(&self, round: usize) -> Self::AuthorityId;
+
+	/// Whether the candidate is valid.
+	fn candidate_valid(&self, candidate: &Self::Candidate) -> bool;
+
+	/// Create a round timeout. The context will determine the correct timeout
+	/// length, and create a future that will resolve when the timeout is
+	/// concluded.
+	fn begin_round_timeout(&self, round: usize) -> Self::RoundTimeout;
+}
+
+/// Communication that can occur between participants in consensus.
+#[derive(Debug, Clone)]
+pub enum Communication<C, D, V, S> {
+	/// A consensus message (proposal or vote)
+	Consensus(LocalizedMessage<C, D, V, S>),
+	/// Auxiliary communication (just proof-of-lock for now).
+	Auxiliary(PrepareJustification<D, S>),
+}
+
+/// Hack to get around type alias warning.
+pub trait TypeResolve {
+	/// Communication type.
+	type Communication;
+}
+
+impl<C: Context> TypeResolve for C {
+	type Communication = Communication<C::Candidate, C::Digest, C::AuthorityId, C::Signature>;
+}
+
+#[derive(Debug)]
+struct Sending<T> {
+	items: VecDeque<T>,
+	flushing: bool,
+}
+
+impl<T> Sending<T> {
+	fn with_capacity(n: usize) -> Self {
+		Sending {
+			items: VecDeque::with_capacity(n),
+			flushing: false,
+		}
+	}
+
+	fn push(&mut self, item: T) {
+		self.items.push_back(item);
+		self.flushing = false;
+	}
+
+	// process all the sends into the sink.
+	fn process_all<S: Sink<SinkItem=T>>(&mut self, sink: &mut S) -> Poll<(), S::SinkError> {
+		while let Some(item) = self.items.pop_front() {
+			match sink.start_send(item) {
+				Err(e) => return Err(e),
+				Ok(AsyncSink::NotReady(item)) => {
+					self.items.push_front(item);
+					return Ok(Async::NotReady);
+				}
+				Ok(AsyncSink::Ready) => { self.flushing = true; }
+			}
+		}
+
+		if self.flushing {
+			match sink.poll_complete() {
+				Err(e) => return Err(e),
+				Ok(Async::NotReady) => return Ok(Async::NotReady),
+				Ok(Async::Ready(())) => { self.flushing = false; }
+			}
+		}
+
+		Ok(Async::Ready(()))
+	}
+}
+
+/// Error returned when the input stream concludes.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct InputStreamConcluded;
+
+impl ::std::fmt::Display for InputStreamConcluded {
+	fn fmt(&self, f: &mut ::std::fmt::Formatter) -> ::std::fmt::Result {
+		write!(f, "{}", ::std::error::Error::description(self))
+	}
+}
+
+impl ::std::error::Error for InputStreamConcluded {
+	fn description(&self) -> &str {
+		"input stream of messages concluded prematurely"
+	}
+}
+
+// get the "full BFT" threshold based on an amount of nodes and
+// a maximum faulty. if nodes == 3f + 1, then threshold == 2f + 1.
+fn bft_threshold(nodes: usize, max_faulty: usize) -> usize {
+	nodes - max_faulty
+}
+
+/// Committed successfully.
+#[derive(Debug, Clone)]
+pub struct Committed<C, D, S> {
+	/// The candidate committed for. This will be unknown if
+	/// we never witnessed the proposal of the last round.
+	pub candidate: Option<C>,
+	/// A justification for the candidate.
+	pub justification: Justification<D, S>,
+}
+
+struct Locked<D, S> {
+	justification: PrepareJustification<D, S>,
+}
+
+impl<D, S> Locked<D, S> {
+	fn digest(&self) -> &D {
+		&self.justification.digest
+	}
+}
+
+// the state of the local node during the current state of consensus.
+//
+// behavior is different when locked on a proposal.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum LocalState {
+	Start,
+	Proposed,
+	Prepared,
+	Committed,
+	VoteAdvance,
+}
+
+// This structure manages a single "view" of consensus.
+//
+// We maintain two message accumulators: one for the round we are currently in,
+// and one for a future round.
+//
+// We advance the round accumulators when one of two conditions is met:
+//   - we witness consensus of advancement in the current round. in this case we
+//     advance by one.
+//   - a higher threshold-prepare is broadcast to us. in this case we can
+//     advance to the round of the threshold-prepare. this is an indication
+//     that we have experienced severe asynchrony/clock drift with the remainder
+//     of the other authorities, and it is unlikely that we can assist in
+//     consensus meaningfully. nevertheless we make an attempt.
+struct Strategy<C: Context> {
+	nodes: usize,
+	max_faulty: usize,
+	fetching_proposal: Option<C::CreateProposal>,
+	round_timeout: future::Fuse<C::RoundTimeout>,
+	local_state: LocalState,
+	locked: Option<Locked<C::Digest, C::Signature>>,
+	notable_candidates: HashMap<C::Digest, C::Candidate>,
+	current_accumulator: Accumulator<C::Candidate, C::Digest, C::AuthorityId, C::Signature>,
+	future_accumulator: Accumulator<C::Candidate, C::Digest, C::AuthorityId, C::Signature>,
+	local_id: C::AuthorityId,
+}
+
+impl<C: Context> Strategy<C> {
+	fn create(context: &C, nodes: usize, max_faulty: usize) -> Self {
+		let timeout = context.begin_round_timeout(0);
+		let threshold = bft_threshold(nodes, max_faulty);
+
+		let current_accumulator = Accumulator::new(
+			0,
+			threshold,
+			context.round_proposer(0),
+		);
+
+		let future_accumulator = Accumulator::new(
+			1,
+			threshold,
+			context.round_proposer(1),
+		);
+
+		Strategy {
+			nodes,
+			max_faulty,
+			current_accumulator,
+			future_accumulator,
+			fetching_proposal: None,
+			local_state: LocalState::Start,
+			locked: None,
+			notable_candidates: HashMap::new(),
+			round_timeout: timeout.fuse(),
+			local_id: context.local_id(),
+		}
+	}
+
+	fn import_message(
+		&mut self,
+		msg: LocalizedMessage<C::Candidate, C::Digest, C::AuthorityId, C::Signature>
+	) {
+		let round_number = msg.message.round_number();
+
+		if round_number == self.current_accumulator.round_number() {
+			self.current_accumulator.import_message(msg);
+		} else if round_number == self.future_accumulator.round_number() {
+			self.future_accumulator.import_message(msg);
+		}
+	}
+
+	fn import_lock_proof(
+		&mut self,
+		context: &C,
+		justification: PrepareJustification<C::Digest, C::Signature>,
+	) {
+		// TODO: find a way to avoid processing of the signatures if the sender is
+		// not the primary or the round number is low.
+		if justification.round_number > self.current_accumulator.round_number() {
+			// jump ahead to the prior round as this is an indication of a supermajority
+			// good nodes being at least on that round.
+			self.advance_to_round(context, justification.round_number);
+		}
+
+		let lock_to_new = self.locked.as_ref()
+			.map_or(true, |l| l.justification.round_number < justification.round_number);
+
+		if lock_to_new {
+			self.locked = Some(Locked { justification })
+		}
+	}
+
+	// poll the strategy: this will queue messages to be sent and advance
+	// rounds if necessary.
+	//
+	// only call within the context of a `Task`.
+	fn poll<E>(
+		&mut self,
+		context: &C,
+		sending: &mut Sending<<C as TypeResolve>::Communication>
+	)
+		-> Poll<Committed<C::Candidate, C::Digest, C::Signature>, E>
+		where
+			C::RoundTimeout: Future<Error=E>,
+			C::CreateProposal: Future<Error=E>,
+	{
+		let mut last_watermark = (
+			self.current_accumulator.round_number(),
+			self.local_state
+		);
+
+		// poll until either completion or state doesn't change.
+		loop {
+			match self.poll_once(context, sending)? {
+				Async::Ready(x) => return Ok(Async::Ready(x)),
+				Async::NotReady => {
+					let new_watermark = (
+						self.current_accumulator.round_number(),
+						self.local_state
+					);
+
+					if new_watermark == last_watermark {
+						return Ok(Async::NotReady)
+					} else {
+						last_watermark = new_watermark;
+					}
+				}
+			}
+		}
+	}
+
+	// perform one round of polling: attempt to broadcast messages and change the state.
+	// if the round or internal round-state changes, this should be called again.
+	fn poll_once<E>(
+		&mut self,
+		context: &C,
+		sending: &mut Sending<<C as TypeResolve>::Communication>
+	)
+		-> Poll<Committed<C::Candidate, C::Digest, C::Signature>, E>
+		where
+			C::RoundTimeout: Future<Error=E>,
+			C::CreateProposal: Future<Error=E>,
+	{
+		self.propose(context, sending)?;
+		self.prepare(context, sending);
+		self.commit(context, sending);
+		self.vote_advance(context, sending)?;
+
+		let advance = match self.current_accumulator.state() {
+			&State::Advanced(ref p_just) => {
+				// lock to any witnessed prepare justification.
+				if let Some(p_just) = p_just.as_ref() {
+					self.locked = Some(Locked { justification: p_just.clone() });
+				}
+
+				let round_number = self.current_accumulator.round_number();
+				Some(round_number + 1)
+			}
+			&State::Committed(ref just) => {
+				// fetch the agreed-upon candidate:
+				//   - we may not have received the proposal in the first place
+				//   - there is no guarantee that the proposal we got was agreed upon
+				//     (can happen if faulty primary)
+				//   - look in the candidates of prior rounds just in case.
+				let candidate = self.current_accumulator
+					.proposal()
+					.and_then(|c| if context.candidate_digest(c) == just.digest {
+						Some(c.clone())
+					} else {
+						None
+					})
+					.or_else(|| self.notable_candidates.get(&just.digest).cloned());
+
+				let committed = Committed {
+					candidate,
+					justification: just.clone()
+				};
+
+				return Ok(Async::Ready(committed))
+			}
+			_ => None,
+		};
+
+		if let Some(new_round) = advance {
+			self.advance_to_round(context, new_round);
+		}
+
+		Ok(Async::NotReady)
+	}
+
+	fn propose(
+		&mut self,
+		context: &C,
+		sending: &mut Sending<<C as TypeResolve>::Communication>
+	)
+		-> Result<(), <C::CreateProposal as Future>::Error>
+	{
+		if let LocalState::Start = self.local_state {
+			let mut propose = false;
+			if let &State::Begin = self.current_accumulator.state() {
+				let round_number = self.current_accumulator.round_number();
+				let primary = context.round_proposer(round_number);
+				propose = self.local_id == primary;
+			};
+
+			if !propose { return Ok(()) }
+
+			// obtain the proposal to broadcast.
+			let proposal = match self.locked {
+				Some(ref locked) => {
+					// TODO: it's possible but very unlikely that we don't have the
+					// corresponding proposal for what we are locked to.
+					//
+					// since this is an edge case on an edge case, it is fine
+					// to eat the round timeout for now, but it can be optimized by
+					// broadcasting an advance vote.
+					self.notable_candidates.get(locked.digest()).cloned()
+				}
+				None => {
+					let res = self.fetching_proposal
+						.get_or_insert_with(|| context.proposal())
+						.poll()?;
+
+					match res {
+						Async::Ready(p) => Some(p),
+						Async::NotReady => None,
+					}
+				}
+			};
+
+			if let Some(proposal) = proposal {
+				self.fetching_proposal = None;
+
+				let message = Message::Propose(
+					self.current_accumulator.round_number(),
+					proposal
+				);
+
+				self.import_and_send_message(message, context, sending);
+
+				// broadcast the justification along with the proposal if we are locked.
+				if let Some(ref locked) = self.locked {
+					sending.push(
+						Communication::Auxiliary(locked.justification.clone())
+					);
+				}
+
+				self.local_state = LocalState::Proposed;
+			}
+		}
+
+		Ok(())
+	}
+
+	fn prepare(
+		&mut self,
+		context: &C,
+		sending: &mut Sending<<C as TypeResolve>::Communication>
+	) {
+		// prepare only upon start or having proposed.
+		match self.local_state {
+			LocalState::Start | LocalState::Proposed => {},
+			_ => return
+		};
+
+		let mut prepare_for = None;
+
+		// we can't prepare until something was proposed.
+		if let &State::Proposed(ref candidate) = self.current_accumulator.state() {
+			let digest = context.candidate_digest(candidate);
+
+			// vote to prepare only if we believe the candidate to be valid and
+			// we are not locked on some other candidate.
+			match self.locked {
+				Some(ref locked) if locked.digest() != &digest => {}
+				Some(_) => {
+					// don't check validity if we are locked.
+					// this is necessary to preserve the liveness property.
+					prepare_for = Some(digest);
+				}
+				None => if context.candidate_valid(candidate) {
+					prepare_for = Some(digest);
+				}
+			}
+		}
+
+		if let Some(digest) = prepare_for {
+			let message = Message::Prepare(
+				self.current_accumulator.round_number(),
+				digest
+			);
+
+			self.import_and_send_message(message, context, sending);
+			self.local_state = LocalState::Prepared;
+		}
+	}
+
+	fn commit(
+		&mut self,
+		context: &C,
+		sending: &mut Sending<<C as TypeResolve>::Communication>
+	) {
+		// commit only if we haven't voted to advance or committed already
+		match self.local_state {
+			LocalState::Committed | LocalState::VoteAdvance => return,
+			_ => {}
+		}
+
+		let mut commit_for = None;
+
+		if let &State::Prepared(ref p_just) = self.current_accumulator.state() {
+			// we are now locked to this prepare justification.
+			let digest = p_just.digest.clone();
+			self.locked = Some(Locked { justification: p_just.clone() });
+			commit_for = Some(digest);
+		}
+
+		if let Some(digest) = commit_for {
+			let message = Message::Commit(
+				self.current_accumulator.round_number(),
+				digest
+			);
+
+			self.import_and_send_message(message, context, sending);
+			self.local_state = LocalState::Committed;
+		}
+	}
+
+	fn vote_advance(
+		&mut self,
+		context: &C,
+		sending: &mut Sending<<C as TypeResolve>::Communication>
+	)
+		-> Result<(), <C::RoundTimeout as Future>::Error>
+	{
+		// we can vote for advancement under all circumstances unless we have already.
+		if let LocalState::VoteAdvance = self.local_state { return Ok(()) }
+
+		// if we got f + 1 advance votes, or the timeout has fired, and we haven't
+		// sent an AdvanceRound message yet, do so.
+		let mut attempt_advance = self.current_accumulator.advance_votes() > self.max_faulty;
+
+		if let Async::Ready(_) = self.round_timeout.poll()? {
+			attempt_advance = true;
+		}
+
+		if attempt_advance {
+			let message = Message::AdvanceRound(
+				self.current_accumulator.round_number(),
+			);
+
+			self.import_and_send_message(message, context, sending);
+			self.local_state = LocalState::VoteAdvance;
+		}
+
+		Ok(())
+	}
+
+	fn advance_to_round(&mut self, context: &C, round: usize) {
+		assert!(round > self.current_accumulator.round_number());
+
+		let threshold = self.nodes - self.max_faulty;
+
+		self.fetching_proposal = None;
+		self.round_timeout = context.begin_round_timeout(round).fuse();
+		self.local_state = LocalState::Start;
+
+		let new_future = Accumulator::new(
+			round + 1,
+			threshold,
+			context.round_proposer(round + 1),
+		);
+
+		// when advancing from a round, store away the witnessed proposal.
+		//
+		// if we or other participants end up locked on that candidate,
+		// we will have it.
+		if let Some(proposal) = self.current_accumulator.proposal() {
+			let digest = context.candidate_digest(proposal);
+			self.notable_candidates.entry(digest).or_insert_with(|| proposal.clone());
+		}
+
+		// special case when advancing by a single round.
+		if self.future_accumulator.round_number() == round {
+			self.current_accumulator
+				= ::std::mem::replace(&mut self.future_accumulator, new_future);
+		} else {
+			self.future_accumulator = new_future;
+			self.current_accumulator = Accumulator::new(
+				round,
+				threshold,
+				context.round_proposer(round),
+			);
+		}
+	}
+
+	fn import_and_send_message(
+		&mut self,
+		message: Message<C::Candidate, C::Digest>,
+		context: &C,
+		sending: &mut Sending<<C as TypeResolve>::Communication>
+	) {
+		let signed_message = context.sign_local(message);
+		self.import_message(signed_message.clone());
+		sending.push(Communication::Consensus(signed_message));
+	}
+}
+
+/// Future that resolves upon BFT agreement for a candidate.
+#[must_use = "futures do nothing unless polled"]
+pub struct Agreement<C: Context, I, O> {
+	context: C,
+	input: I,
+	output: O,
+	concluded: Option<Committed<C::Candidate, C::Digest, C::Signature>>,
+	sending: Sending<<C as TypeResolve>::Communication>,
+	strategy: Strategy<C>,
+}
+
+impl<C, I, O, E> Future for Agreement<C, I, O>
+	where
+		C: Context,
+		C::RoundTimeout: Future<Error=E>,
+		C::CreateProposal: Future<Error=E>,
+		I: Stream<Item=<C as TypeResolve>::Communication,Error=E>,
+		O: Sink<SinkItem=<C as TypeResolve>::Communication,SinkError=E>,
+		E: From<InputStreamConcluded>,
+{
+	type Item = Committed<C::Candidate, C::Digest, C::Signature>;
+	type Error = E;
+
+	fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+		// even if we've observed the conclusion, wait until all
+		// pending outgoing messages are flushed.
+		if let Some(just) = self.concluded.take() {
+			return Ok(match self.sending.process_all(&mut self.output)? {
+				Async::Ready(()) => Async::Ready(just),
+				Async::NotReady => {
+					self.concluded = Some(just);
+					Async::NotReady
+				}
+			})
+		}
+
+		loop {
+			let message = match self.input.poll()? {
+				Async::Ready(msg) => msg.ok_or(InputStreamConcluded)?,
+				Async::NotReady => break,
+			};
+
+			match message {
+				Communication::Consensus(message) => self.strategy.import_message(message),
+				Communication::Auxiliary(lock_proof)
+					=> self.strategy.import_lock_proof(&self.context, lock_proof),
+			}
+		}
+
+		// try to process timeouts.
+		let state_machine_res = self.strategy.poll(&self.context, &mut self.sending)?;
+
+		// make progress on flushing all pending messages.
+		let _ = self.sending.process_all(&mut self.output)?;
+
+		match state_machine_res {
+			Async::Ready(just) => {
+				self.concluded = Some(just);
+				self.poll()
+			}
+			Async::NotReady => {
+
+				Ok(Async::NotReady)
+			}
+		}
+	}
+}
+
+/// Attempt to reach BFT agreement on a candidate.
+///
+/// `nodes` is the number of nodes in the system.
+/// `max_faulty` is the maximum number of faulty nodes. Should be less than
+/// 1/3 of `nodes`, otherwise agreement may never be reached.
+///
+/// The input stream should never logically conclude. The logic here assumes
+/// that messages flushed to the output stream will eventually reach other nodes.
+///
+/// Note that it is possible to witness agreement being reached without ever
+/// seeing the candidate. Any candidates seen will be checked for validity.
+///
+/// Although technically the agreement will always complete (given the eventual
+/// delivery of messages), in practice it is possible for this future to
+/// conclude without having witnessed the conclusion.
+/// In general, this future should be pre-empted by the import of a justification
+/// set for this block height.
+pub fn agree<C: Context, I, O, E>(context: C, nodes: usize, max_faulty: usize, input: I, output: O)
+	-> Agreement<C, I, O>
+	where
+		C: Context,
+		C::RoundTimeout: Future<Error=E>,
+		C::CreateProposal: Future<Error=E>,
+		I: Stream<Item=<C as TypeResolve>::Communication,Error=E>,
+		O: Sink<SinkItem=<C as TypeResolve>::Communication,SinkError=E>,
+		E: From<InputStreamConcluded>,
+{
+	let strategy = Strategy::create(&context, nodes, max_faulty);
+	Agreement {
+		context,
+		input,
+		output,
+		concluded: None,
+		sending: Sending::with_capacity(4),
+		strategy: strategy,
+	}
+}

substrate/substrate/bft/src/generic/tests.rs

@@ -0,0 +1,408 @@
+// Copyright 2017 Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// 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/>.
+
+//! Tests for the candidate agreement strategy.
+
+use super::*;
+
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+
+use futures::prelude::*;
+use futures::sync::{oneshot, mpsc};
+use futures::future::FutureResult;
+
+struct Network<T> {
+	endpoints: Vec<mpsc::UnboundedSender<T>>,
+	input: mpsc::UnboundedReceiver<(usize, T)>,
+}
+
+impl<T: Clone + Send + 'static> Network<T> {
+	fn new(nodes: usize)
+		-> (Self, Vec<mpsc::UnboundedSender<(usize, T)>>, Vec<mpsc::UnboundedReceiver<T>>)
+	{
+		let mut inputs = Vec::with_capacity(nodes);
+		let mut outputs = Vec::with_capacity(nodes);
+		let mut endpoints = Vec::with_capacity(nodes);
+
+		let (in_tx, in_rx) = mpsc::unbounded();
+		for _ in 0..nodes {
+			let (out_tx, out_rx) = mpsc::unbounded();
+			inputs.push(in_tx.clone());
+			outputs.push(out_rx);
+			endpoints.push(out_tx);
+		}
+
+		let network = Network {
+			endpoints,
+			input: in_rx,
+		};
+
+		(network, inputs, outputs)
+	}
+
+	fn route_on_thread(self) {
+		::std::thread::spawn(move || { let _ = self.wait(); });
+	}
+}
+
+impl<T: Clone> Future for Network<T> {
+	type Item = ();
+	type Error = ();
+
+	fn poll(&mut self) -> Poll<(), Self::Error> {
+		match try_ready!(self.input.poll()) {
+			None => Ok(Async::Ready(())),
+			Some((sender, item)) => {
+				{
+					let receiving_endpoints = self.endpoints
+						.iter()
+						.enumerate()
+						.filter(|&(i, _)| i != sender)
+						.map(|(_, x)| x);
+
+					for endpoint in receiving_endpoints {
+						let _ = endpoint.unbounded_send(item.clone());
+					}
+				}
+
+				self.poll()
+			}
+		}
+	}
+}
+
+#[derive(Debug, PartialEq, Eq, Clone, Hash)]
+struct Candidate(usize);
+
+#[derive(Debug, PartialEq, Eq, Clone, Hash)]
+struct Digest(usize);
+
+#[derive(Debug, PartialEq, Eq, Clone, Hash)]
+struct AuthorityId(usize);
+
+#[derive(Debug, PartialEq, Eq, Clone)]
+struct Signature(Message<Candidate, Digest>, AuthorityId);
+
+struct SharedContext {
+	node_count: usize,
+	current_round: usize,
+	awaiting_round_timeouts: HashMap<usize, Vec<oneshot::Sender<()>>>,
+}
+
+#[derive(Debug)]
+struct Error;
+
+impl From<InputStreamConcluded> for Error {
+	fn from(_: InputStreamConcluded) -> Error {
+		Error
+	}
+}
+
+impl SharedContext {
+	fn new(node_count: usize) -> Self {
+		SharedContext {
+			node_count,
+			current_round: 0,
+			awaiting_round_timeouts: HashMap::new()
+		}
+	}
+
+	fn round_timeout(&mut self, round: usize) -> Box<Future<Item=(),Error=Error>> {
+		let (tx, rx) = oneshot::channel();
+		if round < self.current_round {
+			tx.send(()).unwrap();
+		} else {
+			self.awaiting_round_timeouts
+				.entry(round)
+				.or_insert_with(Vec::new)
+				.push(tx);
+		}
+
+		Box::new(rx.map_err(|_| Error))
+	}
+
+	fn bump_round(&mut self) {
+		let awaiting_timeout = self.awaiting_round_timeouts
+			.remove(&self.current_round)
+			.unwrap_or_else(Vec::new);
+
+		for tx in awaiting_timeout {
+			let _ = tx.send(());
+		}
+
+		self.current_round += 1;
+	}
+
+	fn round_proposer(&self, round: usize) -> AuthorityId {
+		AuthorityId(round % self.node_count)
+	}
+}
+
+struct TestContext {
+	local_id: AuthorityId,
+	proposal: Mutex<usize>,
+	shared: Arc<Mutex<SharedContext>>,
+}
+
+impl Context for TestContext {
+	type Candidate = Candidate;
+	type Digest = Digest;
+	type AuthorityId = AuthorityId;
+	type Signature = Signature;
+	type RoundTimeout = Box<Future<Item=(), Error=Error>>;
+	type CreateProposal = FutureResult<Candidate, Error>;
+
+	fn local_id(&self) -> AuthorityId {
+		self.local_id.clone()
+	}
+
+	fn proposal(&self) -> Self::CreateProposal {
+		let proposal = {
+			let mut p = self.proposal.lock().unwrap();
+			let x = *p;
+			*p = (*p * 2) + 1;
+			x
+		};
+
+		Ok(Candidate(proposal)).into_future()
+	}
+
+	fn candidate_digest(&self, candidate: &Candidate) -> Digest {
+		Digest(candidate.0)
+	}
+
+	fn sign_local(&self, message: Message<Candidate, Digest>)
+		-> LocalizedMessage<Candidate, Digest, AuthorityId, Signature>
+	{
+		let signature = Signature(message.clone(), self.local_id.clone());
+		LocalizedMessage {
+			message,
+			signature,
+			sender: self.local_id.clone()
+		}
+	}
+
+	fn round_proposer(&self, round: usize) -> AuthorityId {
+		self.shared.lock().unwrap().round_proposer(round)
+	}
+
+	fn candidate_valid(&self, candidate: &Candidate) -> bool {
+		candidate.0 % 3 != 0
+	}
+
+	fn begin_round_timeout(&self, round: usize) -> Self::RoundTimeout {
+		self.shared.lock().unwrap().round_timeout(round)
+	}
+}
+
+fn timeout_in(t: Duration) -> oneshot::Receiver<()> {
+	let (tx, rx) = oneshot::channel();
+	::std::thread::spawn(move || {
+		::std::thread::sleep(t);
+		let _ = tx.send(());
+	});
+
+	rx
+}
+
+#[test]
+fn consensus_completes_with_minimum_good() {
+	let node_count = 10;
+	let max_faulty = 3;
+
+	let shared_context = Arc::new(Mutex::new(SharedContext::new(node_count)));
+
+	let (network, net_send, net_recv) = Network::new(node_count);
+	network.route_on_thread();
+
+	let nodes = net_send
+		.into_iter()
+		.zip(net_recv)
+		.take(node_count - max_faulty)
+		.enumerate()
+		.map(|(i, (tx, rx))| {
+			let ctx = TestContext {
+				local_id: AuthorityId(i),
+				proposal: Mutex::new(i),
+				shared: shared_context.clone(),
+			};
+
+			agree(
+				ctx,
+				node_count,
+				max_faulty,
+				rx.map_err(|_| Error),
+				tx.sink_map_err(|_| Error).with(move |t| Ok((i, t))),
+			)
+		})
+		.collect::<Vec<_>>();
+
+	::std::thread::spawn(move || {
+		let mut timeout = ::std::time::Duration::from_millis(50);
+		loop {
+			::std::thread::sleep(timeout.clone());
+			shared_context.lock().unwrap().bump_round();
+			timeout *= 2;
+		}
+	});
+
+	let timeout = timeout_in(Duration::from_millis(500)).map_err(|_| Error);
+	let results = ::futures::future::join_all(nodes)
+		.map(Some)
+		.select(timeout.map(|_| None))
+		.wait()
+		.map(|(i, _)| i)
+		.map_err(|(e, _)| e)
+		.expect("to complete")
+		.expect("to not time out");
+
+	for result in &results {
+		assert_eq!(&result.justification.digest, &results[0].justification.digest);
+	}
+}
+
+#[test]
+fn consensus_does_not_complete_without_enough_nodes() {
+	let node_count = 10;
+	let max_faulty = 3;
+
+	let shared_context = Arc::new(Mutex::new(SharedContext::new(node_count)));
+
+	let (network, net_send, net_recv) = Network::new(node_count);
+	network.route_on_thread();
+
+	let nodes = net_send
+		.into_iter()
+		.zip(net_recv)
+		.take(node_count - max_faulty - 1)
+		.enumerate()
+		.map(|(i, (tx, rx))| {
+			let ctx = TestContext {
+				local_id: AuthorityId(i),
+				proposal: Mutex::new(i),
+				shared: shared_context.clone(),
+			};
+
+			agree(
+				ctx,
+				node_count,
+				max_faulty,
+				rx.map_err(|_| Error),
+				tx.sink_map_err(|_| Error).with(move |t| Ok((i, t))),
+			)
+		})
+		.collect::<Vec<_>>();
+
+	let timeout = timeout_in(Duration::from_millis(500)).map_err(|_| Error);
+	let result = ::futures::future::join_all(nodes)
+		.map(Some)
+		.select(timeout.map(|_| None))
+		.wait()
+		.map(|(i, _)| i)
+		.map_err(|(e, _)| e)
+		.expect("to complete");
+
+	assert!(result.is_none(), "not enough online nodes");
+}
+
+#[test]
+fn threshold_plus_one_locked_on_proposal_only_one_with_candidate() {
+	let node_count = 10;
+	let max_faulty = 3;
+
+	let locked_proposal = Candidate(999_999_999);
+	let locked_digest = Digest(999_999_999);
+	let locked_round = 1;
+	let justification = UncheckedJustification {
+		round_number: locked_round,
+		digest: locked_digest.clone(),
+		signatures: (0..7)
+			.map(|i| Signature(Message::Prepare(locked_round, locked_digest.clone()), AuthorityId(i)))
+			.collect()
+	}.check(7, |_, _, s| Some(s.1.clone())).unwrap();
+
+	let mut shared_context = SharedContext::new(node_count);
+	shared_context.current_round = locked_round + 1;
+	let shared_context = Arc::new(Mutex::new(shared_context));
+
+	let (network, net_send, net_recv) = Network::new(node_count);
+	network.route_on_thread();
+
+	let nodes = net_send
+		.into_iter()
+		.zip(net_recv)
+		.enumerate()
+		.map(|(i, (tx, rx))| {
+			let ctx = TestContext {
+				local_id: AuthorityId(i),
+				proposal: Mutex::new(i),
+				shared: shared_context.clone(),
+			};
+
+			let mut agreement = agree(
+				ctx,
+				node_count,
+				max_faulty,
+				rx.map_err(|_| Error),
+				tx.sink_map_err(|_| Error).with(move |t| Ok((i, t))),
+			);
+
+			agreement.strategy.advance_to_round(
+				&agreement.context,
+				locked_round + 1
+			);
+
+			if i <= max_faulty {
+				agreement.strategy.locked = Some(Locked {
+					justification: justification.clone(),
+				})
+			}
+
+			if i == max_faulty {
+				agreement.strategy.notable_candidates.insert(
+					locked_digest.clone(),
+					locked_proposal.clone(),
+				);
+			}
+
+			agreement
+		})
+		.collect::<Vec<_>>();
+
+	::std::thread::spawn(move || {
+		let mut timeout = ::std::time::Duration::from_millis(50);
+		loop {
+			::std::thread::sleep(timeout.clone());
+			shared_context.lock().unwrap().bump_round();
+			timeout *= 2;
+		}
+	});
+
+	let timeout = timeout_in(Duration::from_millis(500)).map_err(|_| Error);
+	let results = ::futures::future::join_all(nodes)
+		.map(Some)
+		.select(timeout.map(|_| None))
+		.wait()
+		.map(|(i, _)| i)
+		.map_err(|(e, _)| e)
+		.expect("to complete")
+		.expect("to not time out");
+
+	for result in &results {
+		assert_eq!(&result.justification.digest, &locked_digest);
+	}
+}

substrate/substrate/bft/src/lib.rs

@@ -0,0 +1,473 @@
+// Copyright 2017 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/>.
+
+//! BFT Agreement based on a rotating proposer in different rounds.
+
+pub mod error;
+pub mod generic;
+
+extern crate substrate_codec as codec;
+extern crate substrate_client as client;
+extern crate substrate_primitives as primitives;
+extern crate substrate_state_machine as state_machine;
+extern crate ed25519;
+extern crate tokio_timer;
+extern crate parking_lot;
+
+#[macro_use]
+extern crate futures;
+
+#[macro_use]
+extern crate error_chain;
+
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use client::{BlockId, Client};
+use client::backend::Backend;
+use codec::Slicable;
+use ed25519::Signature;
+use primitives::block::{Block, Header, HeaderHash};
+use primitives::AuthorityId;
+use state_machine::CodeExecutor;
+
+use futures::{stream, task, Async, Sink, Future, IntoFuture};
+use futures::future::Executor;
+use futures::sync::oneshot;
+use tokio_timer::Timer;
+use parking_lot::Mutex;
+
+pub use generic::InputStreamConcluded;
+pub use error::{Error, ErrorKind};
+
+/// Messages over the proposal.
+/// Each message carries an associated round number.
+pub type Message = generic::Message<Block, HeaderHash>;
+
+/// Localized message type.
+pub type LocalizedMessage = generic::LocalizedMessage<
+	Block,
+	HeaderHash,
+	AuthorityId,
+	Signature
+>;
+
+/// Justification of some hash.
+pub type Justification = generic::Justification<HeaderHash, Signature>;
+
+/// Justification of a prepare message.
+pub type PrepareJustification = generic::PrepareJustification<HeaderHash, Signature>;
+
+/// Result of a committed round of BFT
+pub type Committed = generic::Committed<Block, HeaderHash, Signature>;
+
+/// Communication between BFT participants.
+pub type Communication = generic::Communication<Block, HeaderHash, AuthorityId, Signature>;
+
+/// Logic for a proposer.
+///
+/// This will encapsulate creation and evaluation of proposals at a specific
+/// block.
+pub trait Proposer: Sized {
+    type CreateProposal: IntoFuture<Item=Block,Error=Error>;
+
+    /// Initialize the proposal logic on top of a specific header.
+    // TODO: provide state context explicitly?
+    fn init(parent_header: &Header, sign_with: Arc<ed25519::Pair>) -> Self;
+
+    /// Create a proposal.
+    fn propose(&self) -> Self::CreateProposal;
+    /// Evaluate proposal. True means valid.
+	// TODO: change this to a future.
+    fn evaluate(&self, proposal: &Block) -> bool;
+}
+
+/// Block import trait.
+pub trait BlockImport {
+	/// Import a block alongside its corresponding justification.
+	fn import_block(&self, block: Block, justification: Justification);
+}
+
+/// Trait for getting the authorities at a given block.
+pub trait Authorities {
+	/// Get the authorities at the given block.
+	fn authorities(&self, at: &BlockId) -> Result<Vec<AuthorityId>, Error>;
+}
+
+impl<B, E> BlockImport for Client<B, E>
+	where
+		B: Backend,
+		E: CodeExecutor,
+		client::error::Error: From<<B::State as state_machine::backend::Backend>::Error>
+{
+	fn import_block(&self, block: Block, _justification: Justification) {
+		// TODO: use justification.
+		let _ = self.import_block(block.header, Some(block.transactions));
+	}
+}
+
+impl<B, E> Authorities for Client<B, E>
+	where
+		B: Backend,
+		E: CodeExecutor,
+		client::error::Error: From<<B::State as state_machine::backend::Backend>::Error>
+{
+	fn authorities(&self, at: &BlockId) -> Result<Vec<AuthorityId>, Error> {
+		self.authorities_at(at).map_err(|_| ErrorKind::StateUnavailable(*at).into())
+	}
+}
+
+
+/// Instance of BFT agreement.
+struct BftInstance<P> {
+	key: Arc<ed25519::Pair>,
+	authorities: Vec<AuthorityId>,
+	parent_hash: HeaderHash,
+	timer: Timer,
+	round_timeout_multiplier: u64,
+	proposer: P,
+}
+
+impl<P: Proposer> generic::Context for BftInstance<P> {
+	type AuthorityId = AuthorityId;
+	type Digest = HeaderHash;
+	type Signature = Signature;
+	type Candidate = Block;
+	type RoundTimeout = Box<Future<Item=(),Error=Error> + Send>;
+	type CreateProposal = <P::CreateProposal as IntoFuture>::Future;
+
+	fn local_id(&self) -> AuthorityId {
+		self.key.public().0
+	}
+
+	fn proposal(&self) -> Self::CreateProposal {
+		self.proposer.propose().into_future()
+	}
+
+	fn candidate_digest(&self, proposal: &Block) -> HeaderHash {
+		proposal.header.hash()
+	}
+
+	fn sign_local(&self, message: Message) -> LocalizedMessage {
+		use primitives::bft::{Message as PrimitiveMessage, Action as PrimitiveAction};
+
+		let action = match message.clone() {
+			::generic::Message::Propose(r, p) => PrimitiveAction::Propose(r as u32, p),
+			::generic::Message::Prepare(r, h) => PrimitiveAction::Prepare(r as u32, h),
+			::generic::Message::Commit(r, h) => PrimitiveAction::Commit(r as u32, h),
+			::generic::Message::AdvanceRound(r) => PrimitiveAction::AdvanceRound(r as u32),
+		};
+
+		let primitive = PrimitiveMessage {
+			parent: self.parent_hash,
+			action,
+		};
+
+		let to_sign = Slicable::encode(&primitive);
+		let signature = self.key.sign(&to_sign);
+
+		LocalizedMessage {
+			message,
+			signature,
+			sender: self.key.public().0
+		}
+	}
+
+	fn round_proposer(&self, round: usize) -> AuthorityId {
+		use primitives::hashing::blake2_256;
+
+		// repeat blake2_256 on parent hash round + 1 times.
+		// use as index into authorities vec.
+		// TODO: parent hash is really insecure as a randomness beacon as
+		// the prior can easily influence the block hash.
+		let hashed = (0..round + 1).fold(self.parent_hash.0, |a, _| {
+			blake2_256(&a[..])
+		});
+
+		let index = u32::decode(&mut &hashed[..])
+			.expect("there are more than 4 bytes in a 32 byte hash; qed");
+
+		self.authorities[(index as usize) % self.authorities.len()]
+	}
+
+	fn candidate_valid(&self, proposal: &Block) -> bool {
+		self.proposer.evaluate(proposal)
+	}
+
+	fn begin_round_timeout(&self, round: usize) -> Self::RoundTimeout {
+		use std::time::Duration;
+
+		let round = ::std::cmp::min(63, round) as u32;
+		let timeout = 1u64.checked_shl(round)
+			.unwrap_or_else(u64::max_value)
+			.saturating_mul(self.round_timeout_multiplier);
+
+		Box::new(self.timer.sleep(Duration::from_secs(timeout))
+			.map_err(|_| ErrorKind::FaultyTimer.into()))
+	}
+}
+
+type Input = stream::Empty<Communication, Error>;
+
+// "black hole" output sink.
+struct Output;
+
+impl Sink for Output {
+	type SinkItem = Communication;
+	type SinkError = Error;
+
+	fn start_send(&mut self, _item: Communication) -> ::futures::StartSend<Communication, Error> {
+		Ok(::futures::AsyncSink::Ready)
+	}
+
+	fn poll_complete(&mut self) -> ::futures::Poll<(), Error> {
+		Ok(Async::Ready(()))
+	}
+}
+
+/// A future that resolves either when canceled (witnessing a block from the network at same height)
+/// or when agreement completes.
+pub struct BftFuture<P: Proposer, I> {
+	inner: generic::Agreement<BftInstance<P>, Input, Output>,
+	cancel: Arc<AtomicBool>,
+	send_task: Option<oneshot::Sender<task::Task>>,
+	import: Arc<I>,
+}
+
+impl<P: Proposer, I: BlockImport> Future for BftFuture<P, I> {
+	type Item = ();
+	type Error = ();
+
+	fn poll(&mut self) -> ::futures::Poll<(), ()> {
+		// send the task to the bft service so this can be cancelled.
+		if let Some(sender) = self.send_task.take() {
+			let _ = sender.send(task::current());
+		}
+
+		// service has canceled the future. bail
+		if self.cancel.load(Ordering::Acquire) {
+			return Ok(Async::Ready(()))
+		}
+
+		// TODO: handle this error, at least by logging.
+		let committed = try_ready!(self.inner.poll().map_err(|_| ()));
+
+		// If we didn't see the proposal (very unlikely),
+		// we will get the block from the network later.
+		if let Some(justified_block) = committed.candidate {
+			self.import.import_block(justified_block, committed.justification)
+		}
+
+		Ok(Async::Ready(()))
+	}
+}
+
+struct AgreementHandle {
+	cancel: Arc<AtomicBool>,
+	task: Option<oneshot::Receiver<task::Task>>,
+}
+
+impl Drop for AgreementHandle {
+	fn drop(&mut self) {
+		let task = match self.task.take() {
+			Some(t) => t,
+			None => return,
+		};
+
+		// if this fails, the task is definitely not live anyway.
+		if let Ok(task) = task.wait() {
+			self.cancel.store(true, Ordering::Release);
+			task.notify();
+		}
+	}
+}
+
+/// The BftService kicks off the agreement process on top of any blocks it
+/// is notified of.
+pub struct BftService<P, E, I> {
+	client: Arc<I>,
+	executor: E,
+	live_agreements: Mutex<HashMap<HeaderHash, AgreementHandle>>,
+	timer: Timer,
+	round_timeout_multiplier: u64,
+	key: Arc<ed25519::Pair>, // TODO: key changing over time.
+	_marker: ::std::marker::PhantomData<P>,
+}
+
+impl<P, E, I> BftService<P, E, I>
+	where
+		P: Proposer,
+		E: Executor<BftFuture<P, I>>,
+		I: BlockImport + Authorities,
+{
+	/// Signal that a valid block with the given header has been imported.
+	///
+	/// This will begin the consensus process to build a block on top of it.
+	/// If the executor fails to run the future, an error will be returned.
+	pub fn build_upon(&self, header: &Header) -> Result<(), Error> {
+		let hash = header.hash();
+		let mut _preempted_consensus = None;
+
+		let proposer = P::init(header, self.key.clone());
+
+		// TODO: check key is one of the authorities.
+		let authorities = self.client.authorities(&BlockId::Hash(hash))?;
+		let n = authorities.len();
+		let max_faulty = n.saturating_sub(1) / 3;
+
+		let bft_instance = BftInstance {
+			proposer,
+			parent_hash: hash,
+			round_timeout_multiplier: self.round_timeout_multiplier,
+			timer: self.timer.clone(),
+			key: self.key.clone(),
+			authorities: authorities,
+		};
+
+		let agreement = generic::agree(
+			bft_instance,
+			n,
+			max_faulty,
+			stream::empty(),
+			Output,
+		);
+
+		let cancel = Arc::new(AtomicBool::new(false));
+		let (tx, rx) = oneshot::channel();
+
+		self.executor.execute(BftFuture {
+			inner: agreement,
+			cancel: cancel.clone(),
+			send_task: Some(tx),
+			import: self.client.clone(),
+		}).map_err(|e| e.kind()).map_err(ErrorKind::Executor)?;
+
+		{
+			let mut live = self.live_agreements.lock();
+			live.insert(hash, AgreementHandle {
+				task: Some(rx),
+				cancel,
+			});
+
+			// cancel any agreements attempted to build upon this block's parent
+			// as clearly agreement has already been reached.
+			_preempted_consensus = live.remove(&header.parent_hash);
+		}
+
+		Ok(())
+	}
+}
+
+#[cfg(test)]
+mod tests {
+	use super::*;
+	use std::collections::HashSet;
+	use primitives::block;
+	use self::tokio_core::reactor::{Core, Handle};
+	use self::keyring::Keyring;
+
+	extern crate substrate_keyring as keyring;
+	extern crate tokio_core;
+
+	struct FakeClient {
+		authorities: Vec<AuthorityId>,
+		imported_heights: Mutex<HashSet<block::Number>>
+	}
+
+	impl BlockImport for FakeClient {
+		fn import_block(&self, block: Block, _justification: Justification) {
+			assert!(self.imported_heights.lock().insert(block.header.number))
+		}
+	}
+
+	impl Authorities for FakeClient {
+		fn authorities(&self, _at: &BlockId) -> Result<Vec<AuthorityId>, Error> {
+			Ok(self.authorities.clone())
+		}
+	}
+
+	struct DummyProposer(block::Number);
+
+	impl Proposer for DummyProposer {
+	    type CreateProposal = Result<Block, Error>;
+
+	    fn init(parent_header: &Header, _sign_with: Arc<ed25519::Pair>) -> Self {
+			DummyProposer(parent_header.number + 1)
+		}
+
+    	fn propose(&self) -> Result<Block, Error> {
+			Ok(Block {
+				header: Header::from_block_number(self.0),
+				transactions: Default::default()
+			})
+		}
+
+		fn evaluate(&self, proposal: &Block) -> bool {
+			proposal.header.number == self.0
+		}
+	}
+
+	fn make_service(client: FakeClient, handle: Handle)
+		-> BftService<DummyProposer, Handle, FakeClient>
+	{
+		BftService {
+			client: Arc::new(client),
+			executor: handle,
+			live_agreements: Mutex::new(HashMap::new()),
+			timer: Timer::default(),
+			round_timeout_multiplier: 4,
+			key: Arc::new(Keyring::One.into()),
+			_marker: Default::default(),
+		}
+	}
+
+	#[test]
+	fn future_gets_preempted() {
+		let client = FakeClient {
+			authorities: vec![
+				Keyring::One.to_raw_public(),
+				Keyring::Two.to_raw_public(),
+				Keyring::Alice.to_raw_public(),
+				Keyring::Eve.to_raw_public(),
+			],
+			imported_heights: Mutex::new(HashSet::new()),
+		};
+
+		let mut core = Core::new().unwrap();
+
+		let service = make_service(client, core.handle());
+
+		let first = Header::from_block_number(2);
+		let first_hash = first.hash();
+
+		let mut second = Header::from_block_number(3);
+		second.parent_hash = first_hash;
+		let second_hash = second.hash();
+
+		service.build_upon(&first).unwrap();
+		assert!(service.live_agreements.lock().contains_key(&first_hash));
+
+		// turn the core so the future gets polled and sends its task to the
+		// service. otherwise it deadlocks.
+		core.turn(Some(::std::time::Duration::from_millis(100)));
+		service.build_upon(&second).unwrap();
+		assert!(!service.live_agreements.lock().contains_key(&first_hash));
+		assert!(service.live_agreements.lock().contains_key(&second_hash));
+
+		core.turn(Some(::std::time::Duration::from_millis(100)));
+	}
+}