note future changes when importing block

substrate/core/finality-grandpa/Cargo.toml

@@ -8,14 +8,17 @@ futures = "0.1.17"
 parity-codec = "2.1"
 parity-codec-derive = "2.0"
 sr-primitives = { path = "../sr-primitives" }
+substrate-consensus-common = { path = "../consensus/common" }
 substrate-primitives = { path = "../primitives" }
 substrate-client = { path = "../client" }
 substrate-network = { path = "../network" }
 log = "0.4"
+parking_lot = "0.4"
 tokio = "0.1.7"
 
 [dependencies.finality-grandpa]
-version = "0.2.0"
+#version = "0.3.0"
+git = "https://github.com/paritytech/finality-grandpa"
 features = ["derive-codec"]
 
 [dev-dependencies]

substrate/core/finality-grandpa/src/authorities.rs

@@ -16,23 +16,33 @@
 
 //! Utilities for dealing with authorities, authority sets, and handoffs.
 
+use parking_lot::RwLock;
 use substrate_primitives::AuthorityId;
+
+use std::cmp::Ord;
 use std::ops::Add;
+use std::sync::Arc;
 
 /// A shared authority set.
 pub(crate) struct SharedAuthoritySet<H, N> {
-	inner: RwLock<AuthoritySet<H, N>>,
+	inner: Arc<RwLock<AuthoritySet<H, N>>>,
+}
+
+impl<H, N> Clone for SharedAuthoritySet<H, N> {
+	fn clone(&self) -> Self {
+		SharedAuthoritySet { inner: self.inner.clone() }
+	}
 }
 
 impl<H, N> SharedAuthoritySet<H, N> {
 	/// The genesis authority set.
-	pub(crate) fn genesis(initial: Vec<(AuthorityId, usize)>) -> Self {
+	pub(crate) fn genesis(initial: Vec<(AuthorityId, u64)>) -> Self {
 		SharedAuthoritySet {
-			inner: RwLock::new(AuthoritySet {
+			inner: Arc::new(RwLock::new(AuthoritySet {
 				current_authorities: initial,
 				set_id: 0,
 				pending_changes: Vec::new(),
-			})
+			}))
 		}
 	}
 
@@ -42,33 +52,35 @@ impl<H, N> SharedAuthoritySet<H, N> {
 	{
 		f(&*self.inner.read())
 	}
+}
 
-impl<H, N: Add + Clone> SharedAuthoritySet<H, N> {
+impl<H, N: Add<Output=N> + Ord + Clone> SharedAuthoritySet<H, N> {
 	/// Note an upcoming pending transition.
 	pub(crate) fn add_pending_change(&self, pending: PendingChange<H, N>) {
-		let idx = self.pending_changes
-			.binary_search_by_key(|change| change.effective_number())
+		let mut inner = self.inner.write();
+		let idx = inner.pending_changes
+			.binary_search_by_key(&pending.effective_number(), |change| change.effective_number())
 			.unwrap_or_else(|i| i);
 
-		self.pending_changes.insert(idx);
+		inner.pending_changes.insert(idx, pending);
 	}
 
 	/// Get the earliest limit-block number, if any.
-	pub(crate) fn current_limit(&self) -> Option<&N> {
-		self.pending_changes.get(0).map(|change| &change.effective_number());
+	pub(crate) fn current_limit(&self) -> Option<N> {
+		self.inner.read().pending_changes.get(0).map(|change| change.effective_number().clone())
 	}
 }
 
 impl<H, N> From<AuthoritySet<H, N>> for SharedAuthoritySet<H, N> {
 	fn from(set: AuthoritySet<H, N>) -> Self {
-		SharedAuthoritySet { inner: RwLock::new(set) }
+		SharedAuthoritySet { inner: Arc::new(RwLock::new(set)) }
 	}
 }
 
 /// A set of authorities.
 #[derive(Encode, Decode)]
 pub(crate) struct AuthoritySet<H, N> {
-	current_authorities: Vec<(AuthorityId, usize)>,
+	current_authorities: Vec<(AuthorityId, u64)>,
 	set_id: u64,
 	pending_changes: Vec<PendingChange<H, N>>,
 }
@@ -80,7 +92,7 @@ pub(crate) struct AuthoritySet<H, N> {
 #[derive(Encode, Decode)]
 pub(crate) struct PendingChange<H, N> {
 	/// The new authorities and weights to apply.
-	pub(crate) next_authorities: Vec<(AuthorityId, usize)>,
+	pub(crate) next_authorities: Vec<(AuthorityId, u64)>,
 	/// How deep in the finalized chain the announcing block must be
 	/// before the change is applied.
 	pub(crate) finalization_depth: N,
@@ -90,9 +102,9 @@ pub(crate) struct PendingChange<H, N> {
 	pub(crate) canon_hash: H,
 }
 
-impl<H, N: Add + Clone> PendingChange<H, N> {
+impl<H, N: Add<Output=N> + Clone> PendingChange<H, N> {
 	/// Returns the effective number.
 	fn effective_number(&self) -> N {
-		self.canon_height + self.finalization_depth
+		self.canon_height.clone() + self.finalization_depth.clone()
 	}
 }

substrate/core/finality-grandpa/src/lib.rs

@@ -22,9 +22,11 @@ extern crate finality_grandpa as grandpa;
 extern crate futures;
 extern crate substrate_client as client;
 extern crate sr_primitives as runtime_primitives;
+extern crate substrate_consensus_common as consensus_common;
 extern crate substrate_primitives;
 extern crate substrate_network as network;
 extern crate tokio;
+extern crate parking_lot;
 extern crate parity_codec as codec;
 
 #[macro_use]
@@ -47,10 +49,11 @@ use futures::stream::Fuse;
 use futures::sync::mpsc;
 use client::{Client, ImportNotifications, backend::Backend, CallExecutor};
 use codec::{Encode, Decode};
+use consensus_common::BlockImport;
 use runtime_primitives::traits::{
 	As, NumberFor, Block as BlockT, Header as HeaderT, DigestItemFor,
 };
-use runtime_primitives::generic::BlockId;
+use runtime_primitives::{generic::BlockId, Justification};
 use substrate_primitives::{ed25519, AuthorityId, Blake2Hasher};
 use tokio::timer::Interval;
 
@@ -61,6 +64,8 @@ use std::collections::{VecDeque, HashMap};
 use std::sync::Arc;
 use std::time::{Instant, Duration};
 
+use authorities::SharedAuthoritySet;
+
 mod authorities;
 
 const LAST_COMPLETED_KEY: &[u8] = b"grandpa_completed_round";
@@ -404,8 +409,9 @@ fn outgoing_messages<Block: BlockT, N: Network>(
 /// The environment we run GRANDPA in.
 pub struct Environment<B, E, Block: BlockT, N: Network> {
 	inner: Arc<Client<B, E, Block>>,
-	voters: HashMap<AuthorityId, usize>,
+	voters: HashMap<AuthorityId, u64>,
 	config: Config,
+	authority_set: SharedAuthoritySet<Block::Hash, NumberFor<Block>>,
 	network: N,
 }
 
@@ -461,14 +467,23 @@ impl<Block: BlockT, B, E, N> grandpa::Chain<Block::Hash> for Environment<B, E, B
 	}
 }
 
+/// A scheduled change of authority set.
+#[derive(Debug, PartialEq)]
+pub struct ScheduledChange<N> {
+	/// The new authorities after the change, along with their respective weights.
+	pub next_authorities: Vec<(AuthorityId, u64)>,
+	/// The number of blocks to delay.
+	pub delay: N,
+}
+
 /// A GRANDPA-compatible DigestItem. This can describe when GRANDPA set changes
 /// are scheduled.
 // TODO: with specialization, do a blanket implementation so this trait
 // doesn't have to be implemented by users.
 pub trait CompatibleDigestItem<N> {
-	/// If this digest item notes a GRANDPA set change, return the number of
-	/// blocks the change should occur after.
-	fn scheduled_change_in(&self) -> Option<N> { None }
+	/// If this digest item notes a GRANDPA set change, return information about
+	/// the scheduled change.
+	fn scheduled_change(&self) -> Option<ScheduledChange<N>> { None }
 }
 
 impl<B, E, Block: BlockT, N> voter::Environment<Block::Hash> for Environment<B, E, Block, N> where
@@ -487,6 +502,7 @@ impl<B, E, Block: BlockT, N> voter::Environment<Block::Hash> for Environment<B,
 	type Out = Box<Sink<SinkItem = ::grandpa::Message<Block::Hash>, SinkError = Self::Error>>;
 	type Error = Error;
 
+	#[allow(unreachable_code)]
 	fn round_data(
 		&self,
 		round: u64
@@ -498,7 +514,9 @@ impl<B, E, Block: BlockT, N> voter::Environment<Block::Hash> for Environment<B,
 		let prevote_timer = Delay::new(now + self.config.gossip_duration * 2);
 		let precommit_timer = Delay::new(now + self.config.gossip_duration * 4);
 
+		// TODO [now]: Get from shared authority set.
 		let set_id = unimplemented!();
+
 		// TODO: dispatch this with `mpsc::spawn`.
 		let incoming = checked_message_stream::<Block, _>(
 			round,
@@ -511,7 +529,7 @@ impl<B, E, Block: BlockT, N> voter::Environment<Block::Hash> for Environment<B,
 			round,
 			set_id,
 			self.config.local_key.clone(),
-			self.config.voters.clone(),
+			self.config.genesis_voters.clone(),
 			self.network.clone(),
 		);
 
@@ -584,13 +602,57 @@ impl<B, E, Block: BlockT, N> voter::Environment<Block::Hash> for Environment<B,
 	}
 }
 
-/// Run a GRANDPA voter as a task. The returned future should be executed in a tokio runtime.
+/// A block-import handler for GRANDPA.
+///
+/// This scans each imported block for signals of changing authority set.
+/// When using GRANDPA, the block import worker should be using this block import
+/// object.
+pub struct GrandpaBlockImport<B, E, Block: BlockT> {
+	inner: Arc<Client<B, E, Block>>,
+	authority_set: SharedAuthoritySet<Block::Hash, NumberFor<Block>>,
+}
+
+impl<B, E, Block: BlockT> BlockImport<Block> for GrandpaBlockImport<B, E, Block> where
+	B: Backend<Block, Blake2Hasher> + 'static,
+	E: CallExecutor<Block, Blake2Hasher> + 'static,
+	DigestItemFor<Block>: CompatibleDigestItem<NumberFor<Block>>,
+{
+	fn import_block(&self, block: Block, _justification: Justification, _authorities: &[AuthorityId]) -> bool {
+		use runtime_primitives::traits::Digest;
+		use authorities::PendingChange;
+
+		let maybe_event = block.header().digest().logs().iter()
+			.filter_map(|log| log.scheduled_change())
+			.next()
+			.map(|change| (block.header().hash(), *block.header().number(), change));
+
+		// TODO [now]: use import-block trait for client when implemented
+		let result = self.inner.import_block(unimplemented!(), unimplemented!()).is_ok();
+		if let (true, Some((hash, number, change))) = (result, maybe_event) {
+			self.authority_set.add_pending_change(PendingChange {
+				next_authorities: change.next_authorities,
+				finalization_depth: number + change.delay,
+				canon_height: number,
+				canon_hash: hash,
+			});
+
+			// TODO [now]: write to DB, and what to do on failure?
+		}
+		result
+	}
+}
+
+/// Run a GRANDPA voter as a task. This returns two pieces of data: a task to run,
+/// and a `BlockImport` implementation.
 pub fn run_grandpa<B, E, Block: BlockT, N>(
 	config: Config,
 	client: Arc<Client<B, E, Block>>,
-	voters: HashMap<AuthorityId, usize>,
+	voters: HashMap<AuthorityId, u64>,
 	network: N,
-) -> Result<impl Future<Item=(),Error=()>,client::error::Error> where
+) -> ::client::error::Result<(
+	impl Future<Item=(),Error=()>,
+	GrandpaBlockImport<B, E, Block>,
+)> where
 	Block::Hash: Ord,
 	B: Backend<Block, Blake2Hasher> + 'static,
 	E: CallExecutor<Block, Blake2Hasher> + 'static,
@@ -614,11 +676,22 @@ pub fn run_grandpa<B, E, Block: BlockT, N>(
 			))?
 	};
 
+	// TODO: attempt to load from disk.
+	let authority_set = SharedAuthoritySet::genesis(
+		voters.iter().map(|(&id, &weight)| (id, weight)).collect(),
+	);
+
+	let block_import = GrandpaBlockImport {
+		inner: client.clone(),
+		authority_set: authority_set.clone(),
+	};
+
 	let environment = Arc::new(Environment {
 		inner: client,
 		config,
 		voters,
 		network,
+		authority_set,
 	});
 
 	let voter = voter::Voter::new(
@@ -628,7 +701,9 @@ pub fn run_grandpa<B, E, Block: BlockT, N>(
 		last_finalized,
 	);
 
-	Ok(voter.map_err(|e| warn!("GRANDPA Voter failed: {:?}", e)))
+	let work = voter.map_err(|e| warn!("GRANDPA Voter failed: {:?}", e));
+
+	Ok((work, block_import))
 }
 
 #[cfg(test)]