Vote out offline authorities (#524)

* notify when an authority appears to have missed their block

* Runtime API

* offline tracker

* Move to consensus

* generating reports of offline indices

* stubbed-out evaluation logic

* Slashing data pathwat

* usize -> u32

* Slash bad validators.

* update to rhododendron 0.3

* fix compilation of polkadot-consensus

* Support offline noting in checked_block

* include offline reports in block authorship voting

* do not vote validators offline after some time

* add test for offline-tracker

* fix test build

* bump spec version

* update wasm

* Only allow validators that are possible to slash

* Fix grumble

* More idiomatic

* New Wasm.

* update rhododendron

* improve logging and reduce round time exponent

* format offline validators in ss58

substrate/substrate/bft/Cargo.toml

@@ -15,7 +15,7 @@ tokio = "0.1.7"
 parking_lot = "0.4"
 error-chain = "0.12"
 log = "0.3"
-rhododendron = "0.2"
+rhododendron = "0.3"
 
 [dev-dependencies]
 substrate-keyring = { path = "../keyring" }

substrate/substrate/bft/src/lib.rs

@@ -56,6 +56,7 @@ extern crate error_chain;
 use std::mem;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
+use std::time::{Instant, Duration};
 
 use codec::Encode;
 use ed25519::LocalizedSignature;
@@ -69,7 +70,7 @@ use futures::sync::oneshot;
 use tokio::timer::Delay;
 use parking_lot::Mutex;
 
-pub use rhododendron::InputStreamConcluded;
+pub use rhododendron::{InputStreamConcluded, AdvanceRoundReason};
 pub use error::{Error, ErrorKind};
 
 /// Messages over the proposal.
@@ -184,6 +185,9 @@ pub trait Proposer<B: Block> {
 	/// Determine the proposer for a given round. This should be a deterministic function
 	/// with consistent results across all authorities.
 	fn round_proposer(&self, round_number: usize, authorities: &[AuthorityId]) -> AuthorityId;
+
+	/// Hook called when a BFT round advances without a proposal.
+	fn on_round_end(&self, _round_number: usize, _proposed: bool) { }
 }
 
 /// Block import trait.
@@ -207,6 +211,26 @@ struct BftInstance<B: Block, P> {
 	proposer: P,
 }
 
+impl<B: Block, P: Proposer<B>> BftInstance<B, P>
+	where
+		B: Clone + Eq,
+		B::Hash: ::std::hash::Hash
+
+{
+	fn round_timeout_duration(&self, round: usize) -> Duration {
+		const ROUND_INCREMENT_STEP: usize = 10000;
+
+		let round = round / ROUND_INCREMENT_STEP;
+		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);
+
+		Duration::from_secs(timeout)
+	}
+}
+
 impl<B: Block, P: Proposer<B>> rhododendron::Context for BftInstance<B, P>
 	where
 		B: Clone + Eq,
@@ -246,20 +270,39 @@ impl<B: Block, P: Proposer<B>> rhododendron::Context for BftInstance<B, P>
 	}
 
 	fn begin_round_timeout(&self, round: usize) -> Self::RoundTimeout {
-		use std::time::{Instant, Duration};
-
-		let round = round / 3;
-		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);
-
-		let fut = Delay::new(Instant::now() + Duration::from_secs(timeout))
+		let timeout = self.round_timeout_duration(round);
+		let fut = Delay::new(Instant::now() + timeout)
 			.map_err(|e| Error::from(ErrorKind::FaultyTimer(e)))
 			.map_err(Into::into);
 
 		Box::new(fut)
 	}
+
+	fn on_advance_round(
+		&self,
+		accumulator: &::rhododendron::Accumulator<B, B::Hash, Self::AuthorityId, Self::Signature>,
+		round: usize,
+		next_round: usize,
+		reason: AdvanceRoundReason,
+	) {
+		use std::collections::HashSet;
+
+		let collect_pubkeys = |participants: HashSet<&Self::AuthorityId>| participants.into_iter()
+			.map(|p| ::ed25519::Public::from_raw(p.0))
+			.collect::<Vec<_>>();
+
+		let round_timeout = self.round_timeout_duration(next_round);
+		debug!(target: "bft", "Advancing to round {} from {}", next_round, round);
+		debug!(target: "bft", "Participating authorities: {:?}",
+			collect_pubkeys(accumulator.participants()));
+		debug!(target: "bft", "Voting authorities: {:?}",
+			collect_pubkeys(accumulator.voters()));
+		debug!(target: "bft", "Round {} should end in at most {} seconds from now", next_round, round_timeout.as_secs());
+
+		if let AdvanceRoundReason::Timeout = reason {
+			self.proposer.on_round_end(round, accumulator.proposal().is_some());
+		}
+	}
 }
 
 /// A future that resolves either when canceled (witnessing a block from the network at same height)
@@ -303,6 +346,9 @@ impl<B, P, I, InStream, OutSink> Future for BftFuture<B, P, I, InStream, OutSink
 		// TODO: handle and log this error in a way which isn't noisy on exit.
 		let committed = try_ready!(self.inner.poll().map_err(|_| ()));
 
+		// if something was committed, the round leader must have proposed.
+		self.inner.context().proposer.on_round_end(committed.round_number, true);
+
 		// 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 {

substrate/substrate/misbehavior-check/Cargo.toml

@@ -11,7 +11,7 @@ substrate-runtime-io = { path = "../runtime-io", default-features = false }
 
 [dev-dependencies]
 substrate-bft = { path = "../bft" }
-rhododendron = "0.2"
+rhododendron = "0.3"
 substrate-keyring = { path = "../keyring" }
 
 [features]

substrate/substrate/runtime/contract/src/tests.rs

@@ -54,6 +54,7 @@ impl staking::Trait for Test {
 	type OnAccountKill = Contract;
 }
 impl session::Trait for Test {
+	const NOTE_OFFLINE_POSITION: u32 = 1;
 	type ConvertAccountIdToSessionKey = Identity;
 	type OnSessionChange = Staking;
 }

substrate/substrate/runtime/council/src/lib.rs

@@ -649,6 +649,7 @@ mod tests {
 		type Header = Header;
 	}
 	impl session::Trait for Test {
+		const NOTE_OFFLINE_POSITION: u32 = 1;
 		type ConvertAccountIdToSessionKey = Identity;
 		type OnSessionChange = staking::Module<Test>;
 	}

substrate/substrate/runtime/democracy/src/lib.rs

@@ -392,6 +392,7 @@ mod tests {
 		type Header = Header;
 	}
 	impl session::Trait for Test {
+		const NOTE_OFFLINE_POSITION: u32 = 1;
 		type ConvertAccountIdToSessionKey = Identity;
 		type OnSessionChange = staking::Module<Test>;
 	}
@@ -495,7 +496,7 @@ mod tests {
 			assert_eq!(Democracy::tally(r), (10, 0));
 
 			assert_eq!(Democracy::end_block(System::block_number()), Ok(()));
-			Staking::on_session_change(true, 0);
+			Staking::on_session_change(0, Vec::new());
 
 			assert_eq!(Staking::era_length(), 2);
 		});
@@ -573,19 +574,19 @@ mod tests {
 			System::set_block_number(1);
 			assert_ok!(Democracy::vote(&1, 0, true));
 			assert_eq!(Democracy::end_block(System::block_number()), Ok(()));
-			Staking::on_session_change(true, 0);
+			Staking::on_session_change(0, Vec::new());
 			assert_eq!(Staking::bonding_duration(), 4);
 
 			System::set_block_number(2);
 			assert_ok!(Democracy::vote(&1, 1, true));
 			assert_eq!(Democracy::end_block(System::block_number()), Ok(()));
-			Staking::on_session_change(true, 0);
+			Staking::on_session_change(0, Vec::new());
 			assert_eq!(Staking::bonding_duration(), 3);
 
 			System::set_block_number(3);
 			assert_ok!(Democracy::vote(&1, 2, true));
 			assert_eq!(Democracy::end_block(System::block_number()), Ok(()));
-			Staking::on_session_change(true, 0);
+			Staking::on_session_change(0, Vec::new());
 			assert_eq!(Staking::bonding_duration(), 2);
 		});
 	}
@@ -606,7 +607,7 @@ mod tests {
 			assert_eq!(Democracy::tally(r), (10, 0));
 
 			assert_eq!(Democracy::end_block(System::block_number()), Ok(()));
-			Staking::on_session_change(true, 0);
+			Staking::on_session_change(0, Vec::new());
 
 			assert_eq!(Staking::era_length(), 2);
 		});
@@ -621,7 +622,7 @@ mod tests {
 			assert_ok!(Democracy::cancel_referendum(r));
 
 			assert_eq!(Democracy::end_block(System::block_number()), Ok(()));
-			Staking::on_session_change(true, 0);
+			Staking::on_session_change(0, Vec::new());
 
 			assert_eq!(Staking::era_length(), 1);
 		});
@@ -639,7 +640,7 @@ mod tests {
 			assert_eq!(Democracy::tally(r), (0, 10));
 
 			assert_eq!(Democracy::end_block(System::block_number()), Ok(()));
-			Staking::on_session_change(true, 0);
+			Staking::on_session_change(0, Vec::new());
 
 			assert_eq!(Staking::era_length(), 1);
 		});
@@ -660,7 +661,7 @@ mod tests {
 			assert_eq!(Democracy::tally(r), (110, 100));
 
 			assert_eq!(Democracy::end_block(System::block_number()), Ok(()));
-			Staking::on_session_change(true, 0);
+			Staking::on_session_change(0, Vec::new());
 
 			assert_eq!(Staking::era_length(), 2);
 		});
@@ -677,7 +678,7 @@ mod tests {
 			assert_eq!(Democracy::tally(r), (60, 50));
 
 			assert_eq!(Democracy::end_block(System::block_number()), Ok(()));
-			Staking::on_session_change(true, 0);
+			Staking::on_session_change(0, Vec::new());
 
 			assert_eq!(Staking::era_length(), 1);
 		});
@@ -698,7 +699,7 @@ mod tests {
 			assert_eq!(Democracy::tally(r), (100, 50));
 
 			assert_eq!(Democracy::end_block(System::block_number()), Ok(()));
-			Staking::on_session_change(true, 0);
+			Staking::on_session_change(0, Vec::new());
 
 			assert_eq!(Staking::era_length(), 2);
 		});

substrate/substrate/runtime/executive/src/lib.rs

@@ -252,6 +252,7 @@ mod tests {
 		type Header = Header;
 	}
 	impl session::Trait for Test {
+		const NOTE_OFFLINE_POSITION: u32 = 1;
 		type ConvertAccountIdToSessionKey = Identity;
 		type OnSessionChange = staking::Module<Test>;
 	}

substrate/substrate/runtime/session/src/lib.rs

@@ -46,23 +46,26 @@ extern crate substrate_runtime_system as system;
 extern crate substrate_runtime_timestamp as timestamp;
 
 use rstd::prelude::*;
-use primitives::traits::{Zero, One, RefInto, Executable, Convert, As};
+use primitives::traits::{Zero, One, RefInto, MaybeEmpty, Executable, Convert, As};
 use runtime_support::{StorageValue, StorageMap};
 use runtime_support::dispatch::Result;
 
 /// A session has changed.
-pub trait OnSessionChange<T> {
+pub trait OnSessionChange<T, A> {
 	/// Session has changed.
-	fn on_session_change(normal_rotation: bool, time_elapsed: T);
+	fn on_session_change(time_elapsed: T, bad_validators: Vec<A>);
 }
 
-impl<T> OnSessionChange<T> for () {
-	fn on_session_change(_: bool, _: T) {}
+impl<T, A> OnSessionChange<T, A> for () {
+	fn on_session_change(_: T, _: Vec<A>) {}
 }
 
 pub trait Trait: timestamp::Trait {
+	// the position of the required timestamp-set extrinsic.
+	const NOTE_OFFLINE_POSITION: u32;
+
 	type ConvertAccountIdToSessionKey: Convert<Self::AccountId, Self::SessionKey>;
-	type OnSessionChange: OnSessionChange<Self::Moment>;
+	type OnSessionChange: OnSessionChange<Self::Moment, Self::AccountId>;
 }
 
 decl_module! {
@@ -71,6 +74,7 @@ decl_module! {
 	#[cfg_attr(feature = "std", derive(Serialize, Deserialize))]
 	pub enum Call where aux: T::PublicAux {
 		fn set_key(aux, key: T::SessionKey) -> Result = 0;
+		fn note_offline(aux, offline_val_indices: Vec<u32>) -> Result = 1;
 	}
 
 	#[cfg_attr(feature = "std", derive(Serialize, Deserialize))]
@@ -94,6 +98,10 @@ decl_storage! {
 	// Percent by which the session must necessarily finish late before we early-exit the session.
 	pub BrokenPercentLate get(broken_percent_late): b"ses:broken_percent_late" => required T::Moment;
 
+	// Opinions of the current validator set about the activeness of their peers.
+	// Gets cleared when the validator set changes.
+	pub BadValidators get(bad_validators): b"ses:bad_validators" => Vec<T::AccountId>;
+
 	// New session is being forced is this entry exists; in which case, the boolean value is whether
 	// the new session should be considered a normal rotation (rewardable) or exceptional (slashable).
 	pub ForcingNewSession get(forcing_new_session): b"ses:forcing_new_session" => bool;
@@ -136,6 +144,20 @@ impl<T: Trait> Module<T> {
 		Ok(())
 	}
 
+	/// Notes which of the validators appear to be online from the point of the view of the block author.
+	pub fn note_offline(aux: &T::PublicAux, offline_val_indices: Vec<u32>) -> Result {
+		assert!(aux.is_empty());
+		assert!(
+			<system::Module<T>>::extrinsic_index() == T::NOTE_OFFLINE_POSITION,
+			"note_offline extrinsic must be at position {} in the block",
+			T::NOTE_OFFLINE_POSITION
+		);
+
+		let vs = Self::validators();
+		<BadValidators<T>>::put(offline_val_indices.into_iter().map(|i| vs[i as usize].clone()).collect::<Vec<T::AccountId>>());
+		Ok(())
+	}
+
 	// INTERNAL API (available to other runtime modules)
 
 	/// Set the current set of validators.
@@ -156,17 +178,15 @@ impl<T: Trait> Module<T> {
 		// check block number and call next_session if necessary.
 		let block_number = <system::Module<T>>::block_number();
 		let is_final_block = ((block_number - Self::last_length_change()) % Self::length()).is_zero();
-		let broken_validation = Self::broken_validation();
-		if let Some(normal_rotation) = Self::forcing_new_session() {
-			Self::rotate_session(normal_rotation, is_final_block);
-			<ForcingNewSession<T>>::kill();
-		} else if is_final_block || broken_validation {
-			Self::rotate_session(!broken_validation, is_final_block);
+		let bad_validators = <BadValidators<T>>::take().unwrap_or_default();
+		let should_end_session = <ForcingNewSession<T>>::take().is_some() || !bad_validators.is_empty() || is_final_block;
+		if should_end_session {
+			Self::rotate_session(is_final_block, bad_validators);
 		}
 	}
 
 	/// Move onto next session: register the new authority set.
-	pub fn rotate_session(normal_rotation: bool, is_final_block: bool) {
+	pub fn rotate_session(is_final_block: bool, bad_validators: Vec<T::AccountId>) {
 		let now = <timestamp::Module<T>>::get();
 		let time_elapsed = now.clone() - Self::current_start();
 
@@ -186,7 +206,7 @@ impl<T: Trait> Module<T> {
 			<LastLengthChange<T>>::put(block_number);
 		}
 
-		T::OnSessionChange::on_session_change(normal_rotation, time_elapsed);
+		T::OnSessionChange::on_session_change(time_elapsed, bad_validators);
 
 		// Update any changes in session keys.
 		Self::validators().iter().enumerate().for_each(|(i, v)| {
@@ -301,6 +321,7 @@ mod tests {
 		type Moment = u64;
 	}
 	impl Trait for Test {
+		const NOTE_OFFLINE_POSITION: u32 = 1;
 		type ConvertAccountIdToSessionKey = Identity;
 		type OnSessionChange = ();
 	}
@@ -350,7 +371,7 @@ mod tests {
 			assert_eq!(Session::ideal_session_duration(), 15);
 			// ideal end = 0 + 15 * 3 = 15
 			// broken_limit = 15 * 130 / 100 = 19
-		
+
 			System::set_block_number(3);
 			assert_eq!(Session::blocks_remaining(), 2);
 			Timestamp::set_timestamp(9);				// earliest end = 9 + 2 * 5 = 19; OK.
@@ -378,7 +399,7 @@ mod tests {
 			assert_eq!(Session::blocks_remaining(), 0);
 			Session::check_rotate_session();
 			assert_eq!(Session::length(), 10);
-		
+
 			System::set_block_number(7);
 			assert_eq!(Session::current_index(), 1);
 			assert_eq!(Session::blocks_remaining(), 5);

substrate/substrate/runtime/staking/src/lib.rs

@@ -177,6 +177,9 @@ decl_storage! {
 	// The current era stake threshold
 	pub StakeThreshold get(stake_threshold): b"sta:stake_threshold" => required T::Balance;
 
+	// The current bad validator slash.
+	pub CurrentSlash get(current_slash): b"sta:current_slash" => default T::Balance;
+
 	// The next free enumeration set.
 	pub NextEnumSet get(next_enum_set): b"sta:next_enum" => required T::AccountIndex;
 	// The enumeration sets.
@@ -589,10 +592,30 @@ impl<T: Trait> Module<T> {
 
 	/// Session has just changed. We need to determine whether we pay a reward, slash and/or
 	/// move to a new era.
-	fn new_session(normal_rotation: bool, actual_elapsed: T::Moment) {
+	fn new_session(actual_elapsed: T::Moment, bad_validators: Vec<T::AccountId>) {
 		let session_index = <session::Module<T>>::current_index();
+		let early_exit_era = !bad_validators.is_empty();
+
+		if early_exit_era {
+			// slash
+			let slash = Self::current_slash() + Self::early_era_slash();
+			<CurrentSlash<T>>::put(&slash);
+			for v in bad_validators.into_iter() {
+				if let Some(rem) = Self::slash(&v, slash) {
+					let noms = Self::current_nominators_for(&v);
+					let total = noms.iter().map(Self::voting_balance).fold(T::Balance::zero(), |acc, x| acc + x);
+					if !total.is_zero() {
+						let safe_mul_rational = |b| b * rem / total;// TODO: avoid overflow
+						for n in noms.iter() {
+							let _ = Self::slash(n, safe_mul_rational(Self::voting_balance(n)));	// best effort - not much that can be done on fail.
+						}
+					}
+				}
+			}
+		} else {
+			// Zero any cumulative slash since we're healthy now.
+			<CurrentSlash<T>>::kill();
 
-		if normal_rotation {
 			// reward
 			let ideal_elapsed = <session::Module<T>>::ideal_session_duration();
 			let per65536: u64 = (T::Moment::sa(65536u64) * ideal_elapsed.clone() / actual_elapsed.max(ideal_elapsed)).as_();
@@ -609,25 +632,10 @@ impl<T: Trait> Module<T> {
 					let _ = Self::reward(v, safe_mul_rational(Self::voting_balance(v)));
 				}
 			}
-		} else {
-			// slash
-			let early_era_slash = Self::early_era_slash();
-			for v in <session::Module<T>>::validators().iter() {
-				if let Some(rem) = Self::slash(v, early_era_slash) {
-					let noms = Self::current_nominators_for(v);
-					let total = noms.iter().map(Self::voting_balance).fold(T::Balance::zero(), |acc, x| acc + x);
-					if !total.is_zero() {
-						let safe_mul_rational = |b| b * rem / total;// TODO: avoid overflow
-						for n in noms.iter() {
-							let _ = Self::slash(n, safe_mul_rational(Self::voting_balance(n)));	// best effort - not much that can be done on fail.
-						}
-					}
-				}
-			}
 		}
 		if <ForcingNewEra<T>>::take().is_some()
 			|| ((session_index - Self::last_era_length_change()) % Self::sessions_per_era()).is_zero()
-			|| !normal_rotation
+			|| early_exit_era
 		{
 			Self::new_era();
 		}
@@ -654,6 +662,8 @@ impl<T: Trait> Module<T> {
 			}
 		}
 
+		let minimum_allowed = Self::early_era_slash();
+
 		// evaluate desired staking amounts and nominations and optimise to find the best
 		// combination of validators, then use session::internal::set_validators().
 		// for now, this just orders would-be stakers by their balances and chooses the top-most
@@ -662,11 +672,12 @@ impl<T: Trait> Module<T> {
 		let mut intentions = <Intentions<T>>::get()
 			.into_iter()
 			.map(|v| (Self::voting_balance(&v) + Self::nomination_balance(&v), v))
+			.filter(|&(b, _)| b >= minimum_allowed)
 			.collect::<Vec<_>>();
 		intentions.sort_unstable_by(|&(ref b1, _), &(ref b2, _)| b2.cmp(&b1));
 
 		<StakeThreshold<T>>::put(
-			if intentions.len() > 0 {
+			if !intentions.is_empty() {
 				let i = (<ValidatorCount<T>>::get() as usize).min(intentions.len() - 1);
 				intentions[i].0.clone()
 			} else { Zero::zero() }
@@ -797,9 +808,9 @@ impl<T: Trait> Executable for Module<T> {
 	}
 }
 
-impl<T: Trait> OnSessionChange<T::Moment> for Module<T> {
-	fn on_session_change(normal_rotation: bool, elapsed: T::Moment) {
-		Self::new_session(normal_rotation, elapsed);
+impl<T: Trait> OnSessionChange<T::Moment, T::AccountId> for Module<T> {
+	fn on_session_change(elapsed: T::Moment, bad_validators: Vec<T::AccountId>) {
+		Self::new_session(elapsed, bad_validators);
 	}
 }
 

substrate/substrate/runtime/staking/src/mock.rs

@@ -45,6 +45,7 @@ impl system::Trait for Test {
 	type Header = Header;
 }
 impl session::Trait for Test {
+	const NOTE_OFFLINE_POSITION: u32 = 1;
 	type ConvertAccountIdToSessionKey = Identity;
 	type OnSessionChange = Staking;
 }

substrate/substrate/test-client/Cargo.toml

@@ -4,7 +4,7 @@ version = "0.1.0"
 authors = ["Parity Technologies <admin@parity.io>"]
 
 [dependencies]
-rhododendron = "0.2"
+rhododendron = "0.3"
 substrate-bft = { path = "../bft" }
 substrate-client = { path = "../client" }
 substrate-codec = { path = "../codec" }