BFT delay adjustments (#593)

* force delay only on votes

* set proposal timestamp forward

* Adjusted timeout formula

consensus/src/lib.rs

@@ -64,7 +64,7 @@ extern crate substrate_keyring;
 
 use std::collections::{HashMap, HashSet};
 use std::sync::Arc;
-use std::time::{Duration, Instant};
+use std::time::{self, Duration, Instant};
 
 use codec::{Decode, Encode};
 use extrinsic_store::Store as ExtrinsicStore;
@@ -274,6 +274,9 @@ impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
 	) -> Result<(Self::Proposer, Self::Input, Self::Output), Error> {
 		use runtime_primitives::traits::{Hash as HashT, BlakeTwo256};
 
+		// force delay in evaluation this long.
+		const FORCE_DELAY: Timestamp = 5;
+
 		let parent_hash = parent_header.hash().into();
 
 		let id = BlockId::hash(parent_hash);
@@ -343,6 +346,7 @@ impl<C, N, P> bft::Environment<Block> for ProposerFactory<C, N, P>
 			transaction_pool: self.transaction_pool.clone(),
 			offline: self.offline.clone(),
 			validators,
+			minimum_timestamp: current_timestamp() + FORCE_DELAY,
 			_drop_signal: drop_signal,
 		};
 
@@ -422,6 +426,7 @@ pub struct Proposer<C: PolkadotApi + Send + Sync> {
 	transaction_pool: Arc<TransactionPool<C>>,
 	offline: SharedOfflineTracker,
 	validators: Vec<AccountId>,
+	minimum_timestamp: u64,
 	_drop_signal: exit_future::Signal,
 }
 
@@ -473,6 +478,7 @@ impl<C> bft::Proposer<Block> for Proposer<C>
 			table: self.table.clone(),
 			offline: self.offline.clone(),
 			validators: self.validators.clone(),
+			minimum_timestamp: self.minimum_timestamp,
 			timing,
 		})
 	}
@@ -525,9 +531,11 @@ impl<C> bft::Proposer<Block> for Proposer<C>
 			);
 
 			// the duration until the given timestamp is current
-			let proposed_timestamp = proposal.timestamp();
+			let proposed_timestamp = ::std::cmp::max(self.minimum_timestamp, proposal.timestamp());
 			let timestamp_delay = if proposed_timestamp > current_timestamp {
-				Some(now + Duration::from_secs(proposed_timestamp - current_timestamp))
+				let delay_s = proposed_timestamp - current_timestamp;
+				debug!(target: "bft", "Delaying evaluation of proposal for {} seconds", delay_s);
+				Some(now + Duration::from_secs(delay_s))
 			} else {
 				None
 			};
@@ -677,8 +685,6 @@ impl<C> bft::Proposer<Block> for Proposer<C>
 }
 
 fn current_timestamp() -> Timestamp {
-	use std::time;
-
 	time::SystemTime::now().duration_since(time::UNIX_EPOCH)
 		.expect("now always later than unix epoch; qed")
 		.as_secs()
@@ -732,6 +738,7 @@ pub struct CreateProposal<C: PolkadotApi + Send + Sync>  {
 	timing: ProposalTiming,
 	validators: Vec<AccountId>,
 	offline: SharedOfflineTracker,
+	minimum_timestamp: Timestamp,
 }
 
 impl<C> CreateProposal<C> where C: PolkadotApi + Send + Sync {
@@ -743,7 +750,7 @@ impl<C> CreateProposal<C> where C: PolkadotApi + Send + Sync {
 		const MAX_VOTE_OFFLINE_SECONDS: Duration = Duration::from_secs(60);
 
 		// TODO: handle case when current timestamp behind that in state.
-		let timestamp = current_timestamp();
+		let timestamp = ::std::cmp::max(self.minimum_timestamp, current_timestamp());
 
 		let elapsed_since_start = self.timing.dynamic_inclusion.started_at().elapsed();
 		let offline_indices = if elapsed_since_start > MAX_VOTE_OFFLINE_SECONDS {

consensus/src/service.rs

@@ -39,7 +39,7 @@ use extrinsic_store::Store as ExtrinsicStore;
 use tokio::executor::current_thread::TaskExecutor as LocalThreadHandle;
 use tokio::runtime::TaskExecutor as ThreadPoolHandle;
 use tokio::runtime::current_thread::Runtime as LocalRuntime;
-use tokio::timer::{Delay, Interval};
+use tokio::timer::Interval;
 
 use super::{Network, Collators, ProposerFactory};
 use error;
@@ -59,26 +59,11 @@ fn start_bft<F, C>(
 	<F::Proposer as bft::Proposer<Block>>::Error: ::std::fmt::Display + Into<error::Error>,
 	<F as bft::Environment<Block>>::Error: ::std::fmt::Display
 {
-	const DELAY_UNTIL: Duration = Duration::from_millis(5000);
-
 	let mut handle = LocalThreadHandle::current();
 	match bft_service.build_upon(&header) {
-		Ok(Some(bft_work)) => {
-			// do not poll work for some amount of time.
-			let work = Delay::new(Instant::now() + DELAY_UNTIL).then(move |res| {
-				if let Err(e) = res {
-					warn!(target: "bft", "Failed to force delay of consensus: {:?}", e);
-				}
-
-				debug!(target: "bft", "Starting agreement. After forced delay for {:?}",
-					DELAY_UNTIL);
-
-				bft_work
-			});
-			if let Err(e) = handle.spawn_local(Box::new(work)) {
+		Ok(Some(bft_work)) => if let Err(e) = handle.spawn_local(Box::new(bft_work)) {
 		    warn!(target: "bft", "Couldn't initialize BFT agreement: {:?}", e);
 		}
-		}
 		Ok(None) => trace!(target: "bft", "Could not start agreement on top of {}", header.hash()),
 		Err(e) => warn!(target: "bft", "BFT agreement error: {}", e),
 	}