implement set pausing

core/finality-grandpa/src/authorities.rs

@@ -38,13 +38,6 @@ impl<H, N> Clone for SharedAuthoritySet<H, N> {
 }
 
 impl<H, N> SharedAuthoritySet<H, N> {
-	/// The genesis authority set.
-	pub(crate) fn genesis(initial: Vec<(Ed25519AuthorityId, u64)>) -> Self {
-		SharedAuthoritySet {
-			inner: Arc::new(RwLock::new(AuthoritySet::genesis(initial)))
-		}
-	}
-
 	/// Acquire a reference to the inner read-write lock.
 	pub(crate) fn inner(&self) -> &RwLock<AuthoritySet<H, N>> {
 		&*self.inner
@@ -93,7 +86,6 @@ pub(crate) struct AuthoritySet<H, N> {
 	pub(crate) current_authorities: Vec<(Ed25519AuthorityId, u64)>,
 	pub(crate) set_id: u64,
 	pub(crate) pending_changes: Vec<PendingChange<H, N>>,
-	pub(crate) paused: bool,
 }
 
 impl<H, N> AuthoritySet<H, N> {
@@ -103,7 +95,6 @@ impl<H, N> AuthoritySet<H, N> {
 			current_authorities: initial,
 			set_id: 0,
 			pending_changes: Vec::new(),
-			paused: false,
 		}
 	}
 
@@ -194,7 +185,6 @@ where
 					new_set = Some(AuthoritySet {
 						current_authorities: change.next_authorities.clone(),
 						set_id: self.set_id + 1,
-						paused: false,
 						pending_changes: Vec::new(), // new set, new changes.
 					});
 

core/finality-grandpa/src/aux_schema.rs

@@ -43,6 +43,16 @@ pub enum VoterSetState<H, N> {
 	Live(u64, RoundState<H, N>),
 }
 
+impl<H: Clone, N: Clone> VoterSetState<H, N> {
+	/// Yields the current state.
+	pub(crate) fn round(&self) -> (u64, RoundState<H, N>) {
+		match *self {
+			VoterSetState::Paused(n, ref s) => (n, s.clone()),
+			VoterSetState::Live(n, ref s) => (n, s.clone()),
+		}
+	}
+}
+
 type V0VoterSetState<H, N> = (u64, RoundState<H, N>);
 
 #[derive(Debug, Clone, Encode, Decode, PartialEq)]
@@ -64,7 +74,6 @@ impl<H, N> Into<AuthoritySet<H, N>> for V0AuthoritySet<H, N> {
 	fn into(self) -> AuthoritySet<H, N> {
 		AuthoritySet {
 			current_authorities: self.current_authorities,
-			paused: false,
 			set_id: self.set_id,
 			pending_changes: self.pending_changes.into_iter().map(|old_change| PendingChange {
 				next_authorities: old_change.next_authorities,
@@ -109,7 +118,7 @@ pub(crate) fn load_persistent<B, H, N, G>(
 		N: Decode + Encode + Clone,
 		G: FnOnce() -> ClientResult<Vec<(Ed25519AuthorityId, u64)>>
 {
-	let version: Option<u64> = load_decode(backend, VERSION_KEY)?;
+	let version: Option<u32> = load_decode(backend, VERSION_KEY)?;
 	let consensus_changes = load_decode(backend, CONSENSUS_CHANGES_KEY)?
 		.unwrap_or_else(ConsensusChanges::<H, N>::empty);
 
@@ -233,3 +242,10 @@ pub(crate) fn update_consensus_changes<H, N, F, R>(
 {
 	write_aux(&[(CONSENSUS_CHANGES_KEY, set.encode().as_slice())])
 }
+
+#[cfg(test)]
+pub(crate) fn load_authorities<B: AuxStore, H: Decode, N: Decode>(backend: &B)
+	-> Option<AuthoritySet<H, N>> {
+	load_decode::<_, AuthoritySet<H, N>>(backend, AUTHORITY_SET_KEY)
+		.expect("backend error")
+}

core/finality-grandpa/src/communication.rs

@@ -268,6 +268,7 @@ pub(crate) fn checked_message_stream<Block: BlockT, S>(
 {
 	inner
 		.filter_map(|raw| {
+			println!("got incoming message");
 			let decoded = SignedMessage::<Block>::decode(&mut &raw[..]);
 			if decoded.is_none() {
 				debug!(target: "afg", "Skipping malformed message {:?}", raw);

core/finality-grandpa/src/environment.rs

@@ -20,6 +20,7 @@ use std::time::{Duration, Instant};
 use codec::Encode;
 use futures::prelude::*;
 use tokio::timer::Delay;
+use parking_lot::RwLock;
 
 use client::{
 	backend::Backend, BlockchainEvents, CallExecutor, Client, error::Error as ClientError
@@ -43,6 +44,33 @@ use consensus_changes::SharedConsensusChanges;
 use justification::GrandpaJustification;
 use until_imported::UntilVoteTargetImported;
 
+/// Data about a completed round.
+pub(crate) type CompletedRound<H, N> = (u64, RoundState<H, N>);
+
+/// A read-only view of the last completed round.
+pub(crate) struct LastCompletedRound<H, N> {
+	inner: RwLock<CompletedRound<H, N>>,
+}
+
+impl<H: Clone, N: Clone> LastCompletedRound<H, N> {
+	/// Create a new tracker based on some starting last-completed round.
+	pub(crate) fn new(round: CompletedRound<H, N>) -> Self {
+		LastCompletedRound { inner: RwLock::new(round) }
+	}
+
+	/// Read the last completed round.
+	pub(crate) fn read(&self) -> CompletedRound<H, N> {
+		self.inner.read().clone()
+	}
+
+	// NOTE: not exposed outside of this module intentionally.
+	fn with<F, R>(&self, f: F) -> R
+		where F: FnOnce(&mut CompletedRound<H, N>) -> R
+	{
+		f(&mut *self.inner.write())
+	}
+}
+
 /// The environment we run GRANDPA in.
 pub(crate) struct Environment<B, E, Block: BlockT, N: Network<Block>, RA> {
 	pub(crate) inner: Arc<Client<B, E, Block, RA>>,
@@ -52,6 +80,7 @@ pub(crate) struct Environment<B, E, Block: BlockT, N: Network<Block>, RA> {
 	pub(crate) consensus_changes: SharedConsensusChanges<Block::Hash, NumberFor<Block>>,
 	pub(crate) network: N,
 	pub(crate) set_id: u64,
+	pub(crate) last_completed: LastCompletedRound<Block::Hash, NumberFor<Block>>,
 }
 
 impl<Block: BlockT<Hash=H256>, B, E, N, RA> grandpa::Chain<Block::Hash, NumberFor<Block>> for Environment<B, E, Block, N, RA> where
@@ -248,7 +277,11 @@ impl<B, E, Block: BlockT<Hash=H256>, N, RA> voter::Environment<Block::Hash, Numb
 			state.finalized.as_ref().map(|e| e.1),
 		);
 
-		::aux_schema::complete_round(&**self.inner.backend(), round, state).map_err(Into::into)
+		self.last_completed.with(|last_completed| {
+			::aux_schema::complete_round(&**self.inner.backend(), round, state.clone())?;
+			*last_completed = (round, state); // after writing to DB successfully.
+			Ok(())
+		})
 	}
 
 	fn finalize_block(&self, hash: Block::Hash, number: NumberFor<Block>, round: u64, commit: Commit<Block>) -> Result<(), Self::Error> {
@@ -362,6 +395,7 @@ pub(crate) fn finalize_block<B, Block: BlockT<Hash=H256>, E, RA>(
 				return Err(e.into());
 			}
 		}
+
 		// NOTE: this code assumes that honest voters will never vote past a
 		// transition block, thus we don't have to worry about the case where
 		// we have a transition with `effective_block = N`, but we finalize
@@ -407,7 +441,7 @@ pub(crate) fn finalize_block<B, Block: BlockT<Hash=H256>, E, RA>(
 
 		// ideally some handle to a synchronization oracle would be used
 		// to avoid unconditionally notifying.
-		client.finalize_block(BlockId::Hash(hash), justification, true).map_err(|e| {
+		client.apply_finality(import_op, BlockId::Hash(hash), justification, true).map_err(|e| {
 			warn!(target: "finality", "Error applying finality to block {:?}: {:?}", (hash, number), e);
 			e
 		})?;

core/finality-grandpa/src/lib.rs

@@ -551,6 +551,7 @@ pub fn run_grandpa<B, E, Block: BlockT<Hash=H256>, N, RA>(
 		set_id: authority_set.set_id(),
 		authority_set: authority_set.clone(),
 		consensus_changes: consensus_changes.clone(),
+		last_completed: ::environment::LastCompletedRound::new(set_state.round()),
 	});
 
 	let initial_state = (initial_environment, set_state, voter_commands_rx.into_future());
@@ -589,7 +590,7 @@ pub fn run_grandpa<B, E, Block: BlockT<Hash=H256>, N, RA>(
 					last_finalized,
 				))
 			}
-			VoterSetState::Paused(r, s) => None,
+			VoterSetState::Paused(_, _) => None,
 		};
 
 		// needs to be combined with another future otherwise it can deadlock.
@@ -607,6 +608,9 @@ pub fn run_grandpa<B, E, Block: BlockT<Hash=H256>, N, RA>(
 		let handle_voter_command = move |command: VoterCommand<_, _>, voter_commands_rx| {
 			match command {
 				VoterCommand::ChangeAuthorities(new) => {
+					// start the new authority set using the block where the
+					// set changed (not where the signal happened!) as the base.
+					let genesis_state = RoundState::genesis((new.canon_hash, new.canon_number));
 					let env = Arc::new(Environment {
 						inner: client,
 						config,
@@ -615,18 +619,31 @@ pub fn run_grandpa<B, E, Block: BlockT<Hash=H256>, N, RA>(
 						network,
 						authority_set,
 						consensus_changes,
+						last_completed: ::environment::LastCompletedRound::new(
+							(0, genesis_state.clone())
+						),
 					});
 
-					// start the new authority set using the block where the
-					// set changed (not where the signal happened!) as the base.
+
 					let set_state = VoterSetState::Live(
 						0, // always start at round 0 when changing sets.
-						RoundState::genesis((new.canon_hash, new.canon_number)),
+						genesis_state,
 					);
 
 					Ok(FutureLoop::Continue((env, set_state, voter_commands_rx)))
 				}
-				VoterCommand::Pause(_) => unimplemented!(),
+				VoterCommand::Pause(reason) => {
+					info!("Pausing old validator set: {}", reason);
+
+					// not racing because old voter is shut down.
+					let (last_round_number, last_round_state) = env.last_completed.read();
+					let set_state = VoterSetState::Paused(
+						last_round_number,
+						last_round_state,
+					);
+
+					Ok(FutureLoop::Continue((env, set_state, voter_commands_rx)))
+				},
 			}
 		};
 

core/finality-grandpa/src/tests.rs

@@ -40,6 +40,7 @@ use runtime_primitives::ExecutionContext;
 use substrate_primitives::NativeOrEncoded;
 
 use authorities::AuthoritySet;
+use consensus_changes::ConsensusChanges;
 
 type PeerData =
 	Mutex<
@@ -462,6 +463,7 @@ fn run_to_completion(blocks: u64, net: Arc<Mutex<GrandpaTestNet>>, peers: &[Keyr
 
 #[test]
 fn finalize_3_voters_no_observers() {
+	::env_logger::init();
 	let peers = &[Keyring::Alice, Keyring::Bob, Keyring::Charlie];
 	let voters = make_ids(peers);
 
@@ -579,8 +581,9 @@ fn transition_3_voters_twice_1_observer() {
 		assert_eq!(peer.client().info().unwrap().chain.best_number, 1,
 					"Peer #{} failed to sync", i);
 
-		let set_raw = peer.client().backend().get_aux(::AUTHORITY_SET_KEY).unwrap().unwrap();
-		let set = AuthoritySet::<Hash, BlockNumber>::decode(&mut &set_raw[..]).unwrap();
+		let set: AuthoritySet<Hash, BlockNumber> = ::aux_schema::load_authorities(
+			&**peer.client().backend()
+		).unwrap();
 
 		assert_eq!(set.current(), (0, make_ids(peers_a).as_slice()));
 		assert_eq!(set.pending_changes().len(), 0);
@@ -665,8 +668,9 @@ fn transition_3_voters_twice_1_observer() {
 				.take_while(|n| Ok(n.header.number() < &30))
 				.for_each(move |_| Ok(()))
 				.map(move |()| {
-					let set_raw = client.backend().get_aux(::AUTHORITY_SET_KEY).unwrap().unwrap();
-					let set = AuthoritySet::<Hash, BlockNumber>::decode(&mut &set_raw[..]).unwrap();
+					let set: AuthoritySet<Hash, BlockNumber> = ::aux_schema::load_authorities(
+						&**client.backend()
+					).unwrap();
 
 					assert_eq!(set.current(), (2, make_ids(peers_c).as_slice()));
 					assert!(set.pending_changes().is_empty());
@@ -872,8 +876,9 @@ fn force_change_to_new_set() {
 			assert_eq!(peer.client().info().unwrap().chain.best_number, 26,
 					"Peer #{} failed to sync", i);
 
-			let set_raw = peer.client().backend().get_aux(::AUTHORITY_SET_KEY).unwrap().unwrap();
-			let set = AuthoritySet::<Hash, BlockNumber>::decode(&mut &set_raw[..]).unwrap();
+			let set: AuthoritySet<Hash, BlockNumber> = ::aux_schema::load_authorities(
+				&**peer.client().backend()
+			).unwrap();
 
 			assert_eq!(set.current(), (1, voters.as_slice()));
 			assert_eq!(set.pending_changes().len(), 0);