begin BftService

polkadot/consensus/src/lib.rs

@@ -371,81 +371,6 @@ impl<C: Context> SharedTable<C> {
 	}
 }
 
-/// Errors that can occur during agreement.
-#[derive(Debug, Clone, Copy, PartialEq)]
-pub enum Error {
-	IoTerminated,
-	FaultyTimer,
-	CannotPropose,
-}
-
-impl From<bft::InputStreamConcluded> for Error {
-	fn from(_: bft::InputStreamConcluded) -> Error {
-		Error::IoTerminated
-	}
-}
-
-/// Context owned by the BFT future necessary to execute the logic.
-pub struct BftContext<C: Context> {
-	context: C,
-	table: SharedTable<C>,
-	timer: Timer,
-	round_timeout_multiplier: u64,
-}
-
-impl<C: Context> bft::Context for BftContext<C>
-	where C::Proposal: 'static,
-{
-	type AuthorityId = C::AuthorityId;
-	type Digest = C::Digest;
-	type Signature = C::Signature;
-	type Candidate = C::Proposal;
-	type RoundTimeout = Box<Future<Item=(),Error=Error>>;
-	type CreateProposal = Box<Future<Item=Self::Candidate,Error=Error>>;
-
-	fn local_id(&self) -> Self::AuthorityId {
-		self.context.local_id()
-	}
-
-	fn proposal(&self) -> Self::CreateProposal {
-		Box::new(self.table.get_proposal().map_err(|_| Error::CannotPropose))
-	}
-
-	fn candidate_digest(&self, candidate: &Self::Candidate) -> Self::Digest {
-		C::proposal_digest(candidate)
-	}
-
-	fn sign_local(&self, message: bft::Message<Self::Candidate, Self::Digest>)
-		-> bft::LocalizedMessage<Self::Candidate, Self::Digest, Self::AuthorityId, Self::Signature>
-	{
-		let sender = self.local_id();
-		let signature = self.context.sign_bft_message(&message);
-		bft::LocalizedMessage {
-			message,
-			sender,
-			signature,
-		}
-	}
-
-	fn round_proposer(&self, round: usize) -> Self::AuthorityId {
-		self.context.round_proposer(round)
-	}
-
-	fn candidate_valid(&self, proposal: &Self::Candidate) -> bool {
-		self.table.proposal_valid(proposal)
-	}
-
-	fn begin_round_timeout(&self, round: usize) -> Self::RoundTimeout {
-		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(|_| Error::FaultyTimer))
-	}
-}
-
 
 /// Parameters necessary for agreement.
 pub struct AgreementParams<C: Context> {
@@ -477,22 +402,6 @@ pub trait MessageRecovery<C: Context> {
 	fn check_message(&self, Self::UncheckedMessage) -> Option<CheckedMessage<C>>;
 }
 
-/// A batch of statements to send out.
-pub trait StatementBatch<V, T> {
-	/// Get the target authorities of these statements.
-	fn targets(&self) -> &[V];
-
-	/// If the batch is empty.
-	fn is_empty(&self) -> bool;
-
-	/// Push a statement onto the batch. Returns false when the batch is full.
-	///
-	/// This is meant to do work like incrementally serializing the statements
-	/// into a vector of bytes while making sure the length is below a certain
-	/// amount.
-	fn push(&mut self, statement: T) -> bool;
-}
-
 /// Recovered and fully checked messages.
 pub enum CheckedMessage<C: Context> {
 	/// Messages meant for the BFT agreement logic.

substrate/bft/Cargo.toml

@@ -5,5 +5,7 @@ authors = ["Parity Technologies <[email protected]>"]
 
 [dependencies]
 futures = "0.1.17"
+substrate-client = { path = "../client" }
 substrate-primitives = { path = "../primitives" }
 ed25519 = { path = "../ed25519" }
+tokio-timer = "0.1.2"

substrate/bft/src/lib.rs

@@ -21,14 +21,18 @@ pub mod generic;
 
 #[cfg_attr(test, macro_use)]
 extern crate futures;
+extern crate substrate_client as client;
 extern crate substrate_primitives as primitives;
 extern crate ed25519;
+extern crate tokio_timer;
 
+use client::Client;
 use ed25519::Signature;
-use primitives::block::HeaderHash;
-use primitives::{Block, AuthorityId};
+use primitives::block::{Block, Header, HeaderHash};
+use primitives::AuthorityId;
 
 use futures::{Stream, Sink, Future};
+use tokio_timer::Timer;
 
 pub use generic::InputStreamConcluded;
 
@@ -56,96 +60,94 @@ pub type Committed = generic::Committed<Block, HeaderHash, Signature>;
 /// Communication between BFT participants.
 pub type Communication = generic::Communication<Block, HeaderHash, AuthorityId, Signature>;
 
-/// Context necessary for agreement.
-pub trait Context {
-	/// 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=Block>;
-
-	/// Get the local authority ID.
-	fn local_id(&self) -> AuthorityId;
-
-	/// Get the best proposal.
-	fn proposal(&self) -> Self::CreateProposal;
-
-	/// Get the digest of a candidate.
-	fn candidate_digest(&self, candidate: &Block) -> HeaderHash;
-
-	/// Sign a message using the local authority ID.
-	fn sign_local(&self, message: Message) -> LocalizedMessage;
+/// Errors that can occur during agreement.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub enum Error {
+	/// Io streams terminated.
+	IoTerminated,
+	/// Timer failed to fire.
+	FaultyTimer,
+	/// Unable to propose for some reason.
+	CannotPropose,
+}
 
-	/// Get the proposer for a given round of consensus.
-	fn round_proposer(&self, round: usize) -> AuthorityId;
+impl From<InputStreamConcluded> for Error {
+	fn from(_: InputStreamConcluded) -> Error {
+		Error::IoTerminated;
+	}
+}
 
-	/// Whether the candidate is valid.
-	fn candidate_valid(&self, candidate: &Block) -> bool;
+/// 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: 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;
+}
 
-	/// 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;
+/// Instance of BFT agreement.
+pub struct BftInstance<P> {
+	key: ed25519::Pair, // TODO (now): key changing over time.
+	authorities: Vec<AuthorityId>,
+	parent_hash: HeaderHash,
+	timer: Timer,
+	round_timeout_multiplier: u64,
+	proposer: P,
 }
 
-impl<T: Context> generic::Context for T {
-	type Candidate = Block;
-	type Digest = HeaderHash;
+impl<P: Proposer> generic::Context for BftInstance<P> {
 	type AuthorityId = AuthorityId;
+	type Digest = HeaderHash;
 	type Signature = Signature;
-	type RoundTimeout = <Self as Context>::RoundTimeout;
-	type CreateProposal = <Self as Context>::CreateProposal;
+	type Candidate = Block;
+	type RoundTimeout = Box<Future<Item=(),Error=Error>>;
+	type CreateProposal = P::CreateProposal;
 
-	fn local_id(&self) -> AuthorityId { Context::local_id(self) }
+	fn local_id(&self) -> AuthorityId {
+		self.key.public()
+	}
 
-	fn proposal(&self) -> Self::CreateProposal { Context::proposal(self) }
+	fn proposal(&self) -> P::CreateProposal {
+		self.proposer.propose()
+	}
 
-	fn candidate_digest(&self, candidate: &Block) -> HeaderHash {
-		Context::candidate_digest(self, candidate)
+	fn candidate_digest(&self, proposal: &Block) -> HeaderHash {
+		unimplemented!() // TODO: calculate header hash.
 	}
 
 	fn sign_local(&self, message: Message) -> LocalizedMessage {
-		Context::sign_local(self, message)
+		unimplemented!() // TODO: figure out message encoding.
 	}
 
 	fn round_proposer(&self, round: usize) -> AuthorityId {
-		Context::round_proposer(self, round)
+		// repeat blake2_256 on parent hash round + 1 times.
+		// use as index into authorities vec.
+		unimplemented!()
 	}
 
-	fn candidate_valid(&self, candidate: &Block) -> bool {
-		Context::candidate_valid(self, candidate)
+	fn candidate_valid(&self, proposal: &Block) -> bool {
+		self.proposer.evaluate(proposal)
 	}
 
 	fn begin_round_timeout(&self, round: usize) -> Self::RoundTimeout {
-		Context::begin_round_timeout(self, round)
+		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(|_| Error::FaultyTimer))
 	}
 }
 
-/// 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)
-	-> generic::Agreement<C, I, O>
-	where
-		C: Context,
-		C::RoundTimeout: Future<Error=E>,
-		C::CreateProposal: Future<Error=E>,
-		I: Stream<Item=Communication,Error=E>,
-		O: Sink<SinkItem=Communication,SinkError=E>,
-		E: From<InputStreamConcluded>,
-{
-	generic::agree(context, nodes, max_faulty, input, output)
-}
+/// Bft service built around

substrate/primitives/src/bft.rs

@@ -0,0 +1,54 @@
+// 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 formats for the BFT consensus layer.
+
+use block::{Block, HeaderHash};
+use codec::{Slicable, Input};
+
+#[derive(Clone, Copy, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(Debug))]
+#[repr(u8)]
+enum ActionKind {
+	Propose = 1,
+	Prepare = 2,
+	Commit = 3,
+	AdvanceRound = 4,
+}
+
+/// Actions which can be taken during the BFT process.
+#[derive(Clone, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(Debug))]
+pub enum Action {
+	/// Proposal of a block candidate.
+	Propose(usize, Block),
+	/// Preparation to commit for a candidate.
+	Prepare(usize, HeaderHash),
+	/// Vote to commit to a candidate.
+	Commit(usize, HeaderHash),
+	/// Vote to advance round after inactive primary.
+	AdvanceRound(usize),
+}
+
+/// Messages exchanged between participants in the BFT consensus.
+#[derive(Clone, PartialEq, Eq)]
+#[cfg_attr(feature = "std", derive(Debug))]
+pub struct Message {
+	/// The parent header hash this action is relative to.
+	pub parent: HeaderHash,
+	/// The action being broadcasted.
+	pub action: Action,
+}

substrate/primitives/src/block.rs

@@ -149,6 +149,11 @@ impl Header {
 			digest: Default::default(),
 		}
 	}
+
+	/// Get the blake2-256 hash of this header.
+	pub fn hash(&self) -> HeaderHash {
+		hashing::blake2_256(Slicable::to_vec(self).as_slice())
+	}
 }
 
 impl Slicable for Header {