Proposal creation and evaluation to plug into BFT (#77)

* 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

* Block builder (substrate)

* take polkadot-consensus down to the core.

* test for preemption

* fix test build

* Fix wasm build

* Bulid on any block

* Test for block builder.

* Block import tests for client.

* Tidy ups

* clean up block builder instantiation

* justification verification logic

* JustifiedHeader and import

* Propert block generation for tests

* network and tablerouter trait

* use statement import to drive creation of further statements

* Fixed rpc tests

* custom error type for consensus

* create proposer

* asynchronous proposal evaluation

* inherent transactions in polkadot runtime

* fix tests to match real polkadot block constraints

* implicitly generate inherent functions

* add inherent transaction functionality to block body

* block builder logic for polkadot

* some tests for the polkadot API

substrate/substrate/bft/src/error.rs

@@ -47,6 +47,12 @@ error_chain! {
 			description("Unable to dispatch agreement future"),
 			display("Unable to dispatch agreement future: {:?}", e),
 		}
+
+		/// Some other error.
+		Other(e: Box<::std::error::Error + Send>) {
+			description("Other error")
+			display("Other error: {}", e.description())
+		}
 	}
 }
 

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

@@ -74,6 +74,8 @@ pub struct LocalizedMessage<C, D, V, S> {
 /// Provides necessary types for protocol messages, and functions necessary for a
 /// participant to evaluate and create those messages.
 pub trait Context {
+	/// Errors which can occur from the futures in this context.
+	type Error: From<InputStreamConcluded>;
 	/// Candidate proposed.
 	type Candidate: Debug + Eq + Clone;
 	/// Candidate digest.
@@ -83,9 +85,11 @@ pub trait Context {
 	/// Signature.
 	type Signature: Debug + Eq + Clone;
 	/// A future that resolves when a round timeout is concluded.
-	type RoundTimeout: Future<Item=()>;
+	type RoundTimeout: Future<Item=(), Error=Self::Error>;
 	/// A future that resolves when a proposal is ready.
-	type CreateProposal: Future<Item=Self::Candidate>;
+	type CreateProposal: Future<Item=Self::Candidate, Error=Self::Error>;
+	/// A future that resolves when a proposal has been evaluated.
+	type EvaluateProposal: Future<Item=bool, Error=Self::Error>;
 
 	/// Get the local authority ID.
 	fn local_id(&self) -> Self::AuthorityId;
@@ -103,8 +107,8 @@ pub trait Context {
 	/// 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;
+	/// Whether the proposal is valid.
+	fn proposal_valid(&self, proposal: &Self::Candidate) -> Self::EvaluateProposal;
 
 	/// Create a round timeout. The context will determine the correct timeout
 	/// length, and create a future that will resolve when the timeout is
@@ -246,6 +250,7 @@ struct Strategy<C: Context> {
 	nodes: usize,
 	max_faulty: usize,
 	fetching_proposal: Option<C::CreateProposal>,
+	evaluating_proposal: Option<C::EvaluateProposal>,
 	round_timeout: future::Fuse<C::RoundTimeout>,
 	local_state: LocalState,
 	locked: Option<Locked<C::Digest, C::Signature>>,
@@ -278,6 +283,7 @@ impl<C: Context> Strategy<C> {
 			current_accumulator,
 			future_accumulator,
 			fetching_proposal: None,
+			evaluating_proposal: None,
 			local_state: LocalState::Start,
 			locked: None,
 			notable_candidates: HashMap::new(),
@@ -324,15 +330,12 @@ impl<C: Context> Strategy<C> {
 	// rounds if necessary.
 	//
 	// only call within the context of a `Task`.
-	fn poll<E>(
+	fn poll(
 		&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>,
+		-> Poll<Committed<C::Candidate, C::Digest, C::Signature>, C::Error>
 	{
 		let mut last_watermark = (
 			self.current_accumulator.round_number(),
@@ -361,18 +364,15 @@ impl<C: Context> Strategy<C> {
 
 	// 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>(
+	fn poll_once(
 		&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>,
+		-> Poll<Committed<C::Candidate, C::Digest, C::Signature>, C::Error>
 	{
 		self.propose(context, sending)?;
-		self.prepare(context, sending);
+		self.prepare(context, sending)?;
 		self.commit(context, sending);
 		self.vote_advance(context, sending)?;
 
@@ -423,7 +423,7 @@ impl<C: Context> Strategy<C> {
 		context: &C,
 		sending: &mut Sending<<C as TypeResolve>::Communication>
 	)
-		-> Result<(), <C::CreateProposal as Future>::Error>
+		-> Result<(), C::Error>
 	{
 		if let LocalState::Start = self.local_state {
 			let mut propose = false;
@@ -486,11 +486,13 @@ impl<C: Context> Strategy<C> {
 		&mut self,
 		context: &C,
 		sending: &mut Sending<<C as TypeResolve>::Communication>
-	) {
+	)
+		-> Result<(), C::Error>
+	{
 		// prepare only upon start or having proposed.
 		match self.local_state {
 			LocalState::Start | LocalState::Proposed => {},
-			_ => return
+			_ => return Ok(())
 		};
 
 		let mut prepare_for = None;
@@ -508,8 +510,17 @@ impl<C: Context> Strategy<C> {
 					// this is necessary to preserve the liveness property.
 					prepare_for = Some(digest);
 				}
-				None => if context.candidate_valid(candidate) {
-					prepare_for = Some(digest);
+				None => {
+					let res = self.evaluating_proposal
+						.get_or_insert_with(|| context.proposal_valid(candidate))
+						.poll()?;
+
+					if let Async::Ready(valid) = res {
+						self.evaluating_proposal = None;
+						if valid {
+							prepare_for = Some(digest);
+						}
+					}
 				}
 			}
 		}
@@ -523,6 +534,8 @@ impl<C: Context> Strategy<C> {
 			self.import_and_send_message(message, context, sending);
 			self.local_state = LocalState::Prepared;
 		}
+
+		Ok(())
 	}
 
 	fn commit(
@@ -561,7 +574,7 @@ impl<C: Context> Strategy<C> {
 		context: &C,
 		sending: &mut Sending<<C as TypeResolve>::Communication>
 	)
-		-> Result<(), <C::RoundTimeout as Future>::Error>
+		-> Result<(), C::Error>
 	{
 		// we can vote for advancement under all circumstances unless we have already.
 		if let LocalState::VoteAdvance = self.local_state { return Ok(()) }
@@ -592,6 +605,7 @@ impl<C: Context> Strategy<C> {
 		let threshold = self.nodes - self.max_faulty;
 
 		self.fetching_proposal = None;
+		self.evaluating_proposal = None;
 		self.round_timeout = context.begin_round_timeout(round).fuse();
 		self.local_state = LocalState::Start;
 
@@ -647,17 +661,14 @@ pub struct Agreement<C: Context, I, O> {
 	strategy: Strategy<C>,
 }
 
-impl<C, I, O, E> Future for Agreement<C, I, O>
+impl<C, I, O> 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>,
+		I: Stream<Item=<C as TypeResolve>::Communication,Error=C::Error>,
+		O: Sink<SinkItem=<C as TypeResolve>::Communication,SinkError=C::Error>,
 {
 	type Item = Committed<C::Candidate, C::Digest, C::Signature>;
-	type Error = E;
+	type Error = C::Error;
 
 	fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
 		// even if we've observed the conclusion, wait until all
@@ -721,15 +732,12 @@ impl<C, I, O, E> Future for Agreement<C, I, O>
 /// 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)
+pub fn agree<C: Context, I, O>(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>,
+		I: Stream<Item=<C as TypeResolve>::Communication,Error=C::Error>,
+		O: Sink<SinkItem=<C as TypeResolve>::Communication,SinkError=C::Error>,
 {
 	let strategy = Strategy::create(&context, nodes, max_faulty);
 	Agreement {

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

@@ -159,12 +159,14 @@ struct TestContext {
 }
 
 impl Context for TestContext {
+	type Error = Error;
 	type Candidate = Candidate;
 	type Digest = Digest;
 	type AuthorityId = AuthorityId;
 	type Signature = Signature;
 	type RoundTimeout = Box<Future<Item=(), Error=Error>>;
 	type CreateProposal = FutureResult<Candidate, Error>;
+	type EvaluateProposal = FutureResult<bool, Error>;
 
 	fn local_id(&self) -> AuthorityId {
 		self.local_id.clone()
@@ -200,8 +202,8 @@ impl Context for TestContext {
 		self.shared.lock().unwrap().round_proposer(round)
 	}
 
-	fn candidate_valid(&self, candidate: &Candidate) -> bool {
-		candidate.0 % 3 != 0
+	fn proposal_valid(&self, proposal: &Candidate) -> FutureResult<bool, Error> {
+		Ok(proposal.0 % 3 != 0).into_future()
 	}
 
 	fn begin_round_timeout(&self, round: usize) -> Self::RoundTimeout {

substrate/substrate/bft/src/lib.rs

@@ -100,22 +100,35 @@ pub type Committed = generic::Committed<Block, HeaderHash, LocalizedSignature>;
 /// Communication between BFT participants.
 pub type Communication = generic::Communication<Block, HeaderHash, AuthorityId, LocalizedSignature>;
 
+/// Proposer factory. Can be used to create a proposer instance.
+pub trait ProposerFactory {
+	/// The proposer type this creates.
+	type Proposer: Proposer;
+	/// Error which can occur upon creation.
+	type Error: From<Error>;
+
+	/// Initialize the proposal logic on top of a specific header.
+	// TODO: provide state context explicitly?
+	fn init(&self, parent_header: &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: Sized {
-    type CreateProposal: IntoFuture<Item=Block,Error=Error>;
+pub trait Proposer {
+	/// Error type which can occur when proposing or evaluating.
+	type Error: From<Error> + From<InputStreamConcluded> + 'static;
+	/// Future that resolves to a created proposal.
+	type Create: IntoFuture<Item=Block,Error=Self::Error>;
+	/// Future that resolves when a proposal is evaluated.
+	type Evaluate: IntoFuture<Item=bool,Error=Self::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.
+	/// Create a proposal.
+	fn propose(&self) -> Self::Create;
+	/// Evaluate proposal. True means valid.
 	// TODO: change this to a future.
-    fn evaluate(&self, proposal: &Block) -> bool;
+	fn evaluate(&self, proposal: &Block) -> Self::Evaluate;
 }
 
 /// Block import trait.
@@ -141,12 +154,14 @@ struct BftInstance<P> {
 }
 
 impl<P: Proposer> generic::Context for BftInstance<P> {
+	type Error = P::Error;
 	type AuthorityId = AuthorityId;
 	type Digest = HeaderHash;
 	type Signature = LocalizedSignature;
 	type Candidate = Block;
-	type RoundTimeout = Box<Future<Item=(),Error=Error> + Send>;
-	type CreateProposal = <P::CreateProposal as IntoFuture>::Future;
+	type RoundTimeout = Box<Future<Item=(),Error=Self::Error> + Send>;
+	type CreateProposal = <P::Create as IntoFuture>::Future;
+	type EvaluateProposal = <P::Evaluate as IntoFuture>::Future;
 
 	fn local_id(&self) -> AuthorityId {
 		self.key.public().0
@@ -181,8 +196,8 @@ impl<P: Proposer> generic::Context for BftInstance<P> {
 		self.authorities[(index as usize) % self.authorities.len()]
 	}
 
-	fn candidate_valid(&self, proposal: &Block) -> bool {
-		self.proposer.evaluate(proposal)
+	fn proposal_valid(&self, proposal: &Block) -> Self::EvaluateProposal {
+		self.proposer.evaluate(proposal).into_future()
 	}
 
 	fn begin_round_timeout(&self, round: usize) -> Self::RoundTimeout {
@@ -194,24 +209,25 @@ impl<P: Proposer> generic::Context for BftInstance<P> {
 			.saturating_mul(self.round_timeout_multiplier);
 
 		Box::new(self.timer.sleep(Duration::from_secs(timeout))
-			.map_err(|_| ErrorKind::FaultyTimer.into()))
+			.map_err(|_| Error::from(ErrorKind::FaultyTimer))
+			.map_err(Into::into))
 	}
 }
 
-type Input = stream::Empty<Communication, Error>;
+type Input<E> = stream::Empty<Communication, E>;
 
 // "black hole" output sink.
-struct Output;
+struct Output<E>(::std::marker::PhantomData<E>);
 
-impl Sink for Output {
+impl<E> Sink for Output<E> {
 	type SinkItem = Communication;
-	type SinkError = Error;
+	type SinkError = E;
 
-	fn start_send(&mut self, _item: Communication) -> ::futures::StartSend<Communication, Error> {
+	fn start_send(&mut self, _item: Communication) -> ::futures::StartSend<Communication, E> {
 		Ok(::futures::AsyncSink::Ready)
 	}
 
-	fn poll_complete(&mut self) -> ::futures::Poll<(), Error> {
+	fn poll_complete(&mut self) -> ::futures::Poll<(), E> {
 		Ok(Async::Ready(()))
 	}
 }
@@ -219,7 +235,7 @@ impl Sink for Output {
 /// 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>,
+	inner: generic::Agreement<BftInstance<P>, Input<P::Error>, Output<P::Error>>,
 	cancel: Arc<AtomicBool>,
 	send_task: Option<oneshot::Sender<task::Task>>,
 	import: Arc<I>,
@@ -282,30 +298,38 @@ pub struct BftService<P, E, I> {
 	timer: Timer,
 	round_timeout_multiplier: u64,
 	key: Arc<ed25519::Pair>, // TODO: key changing over time.
-	_marker: ::std::marker::PhantomData<P>,
+	factory: P,
 }
 
 impl<P, E, I> BftService<P, E, I>
 	where
-		P: Proposer,
-		E: Executor<BftFuture<P, I>>,
+		P: ProposerFactory,
+		E: Executor<BftFuture<P::Proposer, 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> {
+	/// If the local signing key is an authority, 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<(), P::Error> {
 		let hash = header.hash();
-		let mut _preempted_consensus = None;
+		let mut _preempted_consensus = None; // defers drop of live to the end.
 
-		let proposer = P::init(header, self.key.clone());
+		let authorities = self.client.authorities(&BlockId::Hash(hash))?;
 
 		// TODO: check key is one of the authorities.
-		let authorities = self.client.authorities(&BlockId::Hash(hash))?;
 		let n = authorities.len();
 		let max_faulty = max_faulty_of(n);
 
+		let local_id = self.key.public().0;
+
+		if !authorities.contains(&local_id) {
+			self.live_agreements.lock().remove(&header.parent_hash);
+			return Ok(())
+		}
+
+		let proposer = self.factory.init(header, &authorities, self.key.clone())?;
+
 		let bft_instance = BftInstance {
 			proposer,
 			parent_hash: hash,
@@ -320,7 +344,7 @@ impl<P, E, I> BftService<P, E, I>
 			n,
 			max_faulty,
 			stream::empty(),
-			Output,
+			Output(Default::default()),
 		);
 
 		let cancel = Arc::new(AtomicBool::new(false));
@@ -331,7 +355,7 @@ impl<P, E, I> BftService<P, E, I>
 			cancel: cancel.clone(),
 			send_task: Some(tx),
 			import: self.client.clone(),
-		}).map_err(|e| e.kind()).map_err(ErrorKind::Executor)?;
+		}).map_err(|e| e.kind()).map_err(ErrorKind::Executor).map_err(Error::from)?;
 
 		{
 			let mut live = self.live_agreements.lock();
@@ -455,29 +479,37 @@ mod tests {
 		}
 	}
 
+	struct DummyFactory;
 	struct DummyProposer(block::Number);
 
-	impl Proposer for DummyProposer {
-	    type CreateProposal = Result<Block, Error>;
+	impl ProposerFactory for DummyFactory {
+		type Proposer = DummyProposer;
+		type Error = Error;
 
-	    fn init(parent_header: &Header, _sign_with: Arc<ed25519::Pair>) -> Self {
-			DummyProposer(parent_header.number + 1)
+		fn init(&self, parent_header: &Header, _authorities: &[AuthorityId], _sign_with: Arc<ed25519::Pair>) -> Result<DummyProposer, Error> {
+			Ok(DummyProposer(parent_header.number + 1))
 		}
+	}
 
-    	fn propose(&self) -> Result<Block, Error> {
+	impl Proposer for DummyProposer {
+		type Error = Error;
+		type Create = Result<Block, Error>;
+		type Evaluate = Result<bool, Error>;
+
+		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 evaluate(&self, proposal: &Block) -> Result<bool, Error> {
+			Ok(proposal.header.number == self.0)
 		}
 	}
 
 	fn make_service(client: FakeClient, handle: Handle)
-		-> BftService<DummyProposer, Handle, FakeClient>
+		-> BftService<DummyFactory, Handle, FakeClient>
 	{
 		BftService {
 			client: Arc::new(client),
@@ -486,7 +518,7 @@ mod tests {
 			timer: Timer::default(),
 			round_timeout_multiplier: 4,
 			key: Arc::new(Keyring::One.into()),
-			_marker: Default::default(),
+			factory: DummyFactory
 		}
 	}