core/finality-grandpa: Request block sync from network after import timeout

core/finality-grandpa/src/communication/mod.rs

@@ -38,7 +38,7 @@ use network::{consensus_gossip as network_gossip, NetworkService};
 use network_gossip::ConsensusMessage;
 use codec::{Encode, Decode};
 use primitives::Pair;
-use sr_primitives::traits::{Block as BlockT, Hash as HashT, Header as HeaderT};
+use sr_primitives::traits::{Block as BlockT, Hash as HashT, Header as HeaderT, NumberFor};
 use substrate_telemetry::{telemetry, CONSENSUS_DEBUG, CONSENSUS_INFO};
 use tokio_executor::Executor;
 
@@ -128,6 +128,9 @@ pub trait Network<Block: BlockT>: Clone + Send + 'static {
 
 	/// Inform peers that a block with given hash should be downloaded.
 	fn announce(&self, block: Block::Hash, associated_data: Vec<u8>);
+
+	/// Configure an explicit fork sync request.
+	fn set_sync_fork_request(&self, peers: Vec<network::PeerId>, hash: Block::Hash, number: NumberFor<Block>);
 }
 
 /// Create a unique topic for a round and set-id combo.
@@ -209,6 +212,10 @@ impl<B, S, H> Network<B> for Arc<NetworkService<B, S, H>> where
 	fn announce(&self, block: B::Hash, associated_data: Vec<u8>) {
 		self.announce_block(block, associated_data)
 	}
+
+	fn set_sync_fork_request(&self, peers: Vec<network::PeerId>, hash: B::Hash, number: NumberFor<B>) {
+		NetworkService::set_sync_fork_request(self, peers, hash, number)
+	}
 }
 
 /// A stream used by NetworkBridge in its implementation of Network. Given a oneshot that eventually returns a channel
@@ -458,6 +465,8 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 			format!("Failed to receive on unbounded receiver for round {}", round.0)
 		));
 
+		// Combine incoming votes from external Grandpa nodes with outgoing votes from our own Grandpa voter to have a
+		// single vote-import-pipeline.
 		let incoming = incoming.select(out_rx);
 
 		(incoming, outgoing)
@@ -504,6 +513,10 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 
 		(incoming, outgoing)
 	}
+
+	pub(crate) fn set_sync_fork_request(&self, peers: Vec<network::PeerId>, hash: B::Hash, number: NumberFor<B>){
+		self.service.set_sync_fork_request(peers, hash, number)
+	}
 }
 
 fn incoming_global<B: BlockT, N: Network<B>>(

core/finality-grandpa/src/communication/tests.rs

@@ -25,6 +25,7 @@ use tokio::runtime::current_thread;
 use std::sync::Arc;
 use keyring::Ed25519Keyring;
 use codec::Encode;
+use sr_primitives::traits::NumberFor;
 
 use crate::environment::SharedVoterSetState;
 use super::gossip::{self, GossipValidator};
@@ -91,6 +92,8 @@ impl super::Network<Block> for TestNetwork {
 	fn announce(&self, block: Hash, _associated_data: Vec<u8>) {
 		let _ = self.sender.unbounded_send(Event::Announce(block));
 	}
+
+	fn set_sync_fork_request(&self, peers: Vec<network::PeerId>, hash: Hash, number: NumberFor<Block>) {}
 }
 
 impl network_gossip::ValidatorContext<Block> for TestNetwork {
@@ -296,7 +299,7 @@ fn good_commit_leads_to_relay() {
 			// when the commit comes in, we'll tell the callback it was good.
 			let handle_commit = commits_in.into_future()
 				.map(|(item, _)| {
-					match item.unwrap() {
+					match item.unwrap().1 {
 						grandpa::voter::CommunicationIn::Commit(_, _, mut callback) => {
 							callback.run(grandpa::voter::CommitProcessingOutcome::good());
 						},
@@ -411,7 +414,7 @@ fn bad_commit_leads_to_report() {
 			// when the commit comes in, we'll tell the callback it was good.
 			let handle_commit = commits_in.into_future()
 				.map(|(item, _)| {
-					match item.unwrap() {
+					match item.unwrap().1 {
 						grandpa::voter::CommunicationIn::Commit(_, _, mut callback) => {
 							callback.run(grandpa::voter::CommitProcessingOutcome::bad());
 						},

core/finality-grandpa/src/environment.rs

@@ -370,6 +370,8 @@ impl<Block: BlockT> SharedVoterSetState<Block> {
 
 /// The environment we run GRANDPA in.
 pub(crate) struct Environment<B, E, Block: BlockT, N: Network<Block>, RA, SC, VR> {
+	// TODO: As far as I can tell this does not follow the pattern of having an outer struct protecting an inner struct
+	// via a lock in order to be Sync. How about renaming this to `client` like we do in many other places?
 	pub(crate) inner: Arc<Client<B, E, Block, RA>>,
 	pub(crate) select_chain: SC,
 	pub(crate) voters: Arc<VoterSet<AuthorityId>>,
@@ -602,6 +604,7 @@ where
 		// corresponding blocks are imported.
 		let incoming = Box::new(UntilVoteTargetImported::new(
 			self.inner.import_notification_stream(),
+			self.network.clone(),
 			self.inner.clone(),
 			incoming,
 			"round",

core/finality-grandpa/src/lib.rs

@@ -22,11 +22,11 @@
 //!
 //! # Usage
 //!
-//! First, create a block-import wrapper with the `block_import` function.
-//! The GRANDPA worker needs to be linked together with this block import object,
-//! so a `LinkHalf` is returned as well. All blocks imported (from network or consensus or otherwise)
-//! must pass through this wrapper, otherwise consensus is likely to break in
-//! unexpected ways.
+//! First, create a block-import wrapper with the `block_import` function. The
+//! GRANDPA worker needs to be linked together with this block import object, so
+//! a `LinkHalf` is returned as well. All blocks imported (from network or
+//! consensus or otherwise) must pass through this wrapper, otherwise consensus
+//! is likely to break in unexpected ways.
 //!
 //! Next, use the `LinkHalf` and a local configuration to `run_grandpa_voter`.
 //! This requires a `Network` implementation. The returned future should be
@@ -242,7 +242,7 @@ impl From<ClientError> for Error {
 }
 
 /// Something which can determine if a block is known.
-pub trait BlockStatus<Block: BlockT> {
+pub(crate) trait BlockStatus<Block: BlockT> {
 	/// Return `Ok(Some(number))` or `Ok(None)` depending on whether the block
 	/// is definitely known and has been imported.
 	/// If an unexpected error occurs, return that.
@@ -261,6 +261,20 @@ impl<B, E, Block: BlockT<Hash=H256>, RA> BlockStatus<Block> for Arc<Client<B, E,
 	}
 }
 
+/// Something that one can ask to do a block sync request.
+pub(crate) trait BlockSyncRequester<Block: BlockT> {
+	  fn set_sync_fork_request(&self, peers: Vec<network::PeerId>, hash: Block::Hash, number: NumberFor<Block>);
+}
+
+impl<Block, N> BlockSyncRequester<Block> for NetworkBridge<Block, N> where
+	Block: BlockT,
+	N: communication::Network<Block>,
+{
+	fn set_sync_fork_request(&self, peers: Vec<network::PeerId>, hash: Block::Hash, number: NumberFor<Block>){
+		NetworkBridge::set_sync_fork_request(self, peers, hash, number)
+	}
+}
+
 /// A new authority set along with the canonical block it changed at.
 #[derive(Debug)]
 pub(crate) struct NewAuthoritySet<H, N> {
@@ -429,6 +443,7 @@ fn global_communication<Block: BlockT<Hash=H256>, B, E, N, RA>(
 	// block commit and catch up messages until relevant blocks are imported.
 	let global_in = UntilGlobalMessageBlocksImported::new(
 		client.import_notification_stream(),
+		network.clone(),
 		client.clone(),
 		global_in,
 		"global",

core/finality-grandpa/src/tests.rs

@@ -1185,7 +1185,7 @@ fn voter_persists_its_votes() {
 		runtime.spawn(round_rx.for_each(move |signed| {
 			if state.compare_and_swap(0, 1, Ordering::SeqCst) == 0 {
 				// the first message we receive should be a prevote from alice.
-				let prevote = match signed.message {
+				let prevote = match signed.1.message {
 					grandpa::Message::Prevote(prevote) => prevote,
 					_ => panic!("voter should prevote."),
 				};
@@ -1225,7 +1225,7 @@ fn voter_persists_its_votes() {
 
 			} else if state.compare_and_swap(1, 2, Ordering::SeqCst) == 1 {
 				// the next message we receive should be our own prevote
-				let prevote = match signed.message {
+				let prevote = match signed.1.message {
 					grandpa::Message::Prevote(prevote) => prevote,
 					_ => panic!("We should receive our own prevote."),
 				};
@@ -1242,7 +1242,7 @@ fn voter_persists_its_votes() {
 			} else if state.compare_and_swap(2, 3, Ordering::SeqCst) == 2 {
 				// we then receive a precommit from alice for block 15
 				// even though we casted a prevote for block 30
-				let precommit = match signed.message {
+				let precommit = match signed.1.message {
 					grandpa::Message::Precommit(precommit) => precommit,
 					_ => panic!("voter should precommit."),
 				};