grandpa: fix handling of catch-up requests (#3956)

* grandpa: fix handling of catch-up requests

* grandpa: fix tests

* grandpa: add test for catch-up handling when observer disabled

* grandpa: extend doc comment

* grandpa: rename existing catch up test

substrate/core/finality-grandpa/src/communication/gossip.rs

@@ -512,6 +512,40 @@ enum PendingCatchUp {
 	},
 }
 
+/// Configuration for the round catch-up mechanism.
+enum CatchUpConfig {
+	/// Catch requests are enabled, our node will issue them whenever it sees a
+	/// neighbor packet for a round further than `CATCH_UP_THRESHOLD`. If
+	/// `only_from_authorities` is set, the node will only send catch-up
+	/// requests to other authorities it is connected to. This is useful if the
+	/// GRANDPA observer protocol is live on the network, in which case full
+	/// nodes (non-authorities) don't have the necessary round data to answer
+	/// catch-up requests.
+	Enabled { only_from_authorities: bool },
+	/// Catch-up requests are disabled, our node will never issue them. This is
+	/// useful for the GRANDPA observer mode, where we are only interested in
+	/// commit messages and don't need to follow the full round protocol.
+	Disabled,
+}
+
+impl CatchUpConfig {
+	fn enabled(only_from_authorities: bool) -> CatchUpConfig {
+		CatchUpConfig::Enabled { only_from_authorities }
+	}
+
+	fn disabled() -> CatchUpConfig {
+		CatchUpConfig::Disabled
+	}
+
+	fn request_allowed<N>(&self, peer: &PeerInfo<N>) -> bool {
+		match self {
+			CatchUpConfig::Disabled => false,
+			CatchUpConfig::Enabled { only_from_authorities, .. } =>
+				!only_from_authorities || peer.roles.is_authority(),
+		}
+	}
+}
+
 struct Inner<Block: BlockT> {
 	local_view: Option<View<NumberFor<Block>>>,
 	peers: Peers<NumberFor<Block>>,
@@ -520,13 +554,30 @@ struct Inner<Block: BlockT> {
 	config: crate::Config,
 	next_rebroadcast: Instant,
 	pending_catch_up: PendingCatchUp,
-	catch_up_enabled: bool,
+	catch_up_config: CatchUpConfig,
 }
 
 type MaybeMessage<Block> = Option<(Vec<PeerId>, NeighborPacket<NumberFor<Block>>)>;
 
 impl<Block: BlockT> Inner<Block> {
-	fn new(config: crate::Config, catch_up_enabled: bool) -> Self {
+	fn new(config: crate::Config) -> Self {
+		let catch_up_config = if config.observer_enabled {
+			if config.is_authority {
+				// since the observer protocol is enabled, we will only issue
+				// catch-up requests if we are an authority (and only to other
+				// authorities).
+				CatchUpConfig::enabled(true)
+			} else {
+				// otherwise, we are running the observer protocol and don't
+				// care about catch-up requests.
+				CatchUpConfig::disabled()
+			}
+		} else {
+			// if the observer protocol isn't enabled, then any full node should
+			// be able to answer catch-up requests.
+			CatchUpConfig::enabled(false)
+		};
+
 		Inner {
 			local_view: None,
 			peers: Peers::default(),
@@ -534,7 +585,7 @@ impl<Block: BlockT> Inner<Block> {
 			next_rebroadcast: Instant::now() + REBROADCAST_AFTER,
 			authorities: Vec::new(),
 			pending_catch_up: PendingCatchUp::None,
-			catch_up_enabled,
+			catch_up_config,
 			config,
 		}
 	}
@@ -823,10 +874,6 @@ impl<Block: BlockT> Inner<Block> {
 	}
 
 	fn try_catch_up(&mut self, who: &PeerId) -> (Option<GossipMessage<Block>>, Option<Report>) {
-		if !self.catch_up_enabled {
-			return (None, None);
-		}
-
 		let mut catch_up = None;
 		let mut report = None;
 
@@ -836,7 +883,7 @@ impl<Block: BlockT> Inner<Block> {
 		// won't be able to reply since they don't follow the full GRANDPA
 		// protocol and therefore might not have the vote data available.
 		if let (Some(peer), Some(local_view)) = (self.peers.peer(who), &self.local_view) {
-			if peer.roles.is_authority() &&
+			if self.catch_up_config.request_allowed(&peer) &&
 				peer.view.set_id == local_view.set_id &&
 				peer.view.round.0.saturating_sub(CATCH_UP_THRESHOLD) > local_view.round.0
 			{
@@ -949,11 +996,10 @@ impl<Block: BlockT> GossipValidator<Block> {
 	pub(super) fn new(
 		config: crate::Config,
 		set_state: environment::SharedVoterSetState<Block>,
-		catch_up_enabled: bool,
 	) -> (GossipValidator<Block>, ReportStream)	{
 		let (tx, rx) = mpsc::unbounded();
 		let val = GossipValidator {
-			inner: parking_lot::RwLock::new(Inner::new(config, catch_up_enabled)),
+			inner: parking_lot::RwLock::new(Inner::new(config)),
 			set_state,
 			report_sender: tx,
 		};
@@ -1274,6 +1320,8 @@ mod tests {
 			justification_period: 256,
 			keystore: None,
 			name: None,
+			is_authority: true,
+			observer_enabled: true,
 		}
 	}
 
@@ -1438,7 +1486,6 @@ mod tests {
 		let (val, _) = GossipValidator::<Block>::new(
 			config(),
 			voter_set_state(),
-			true,
 		);
 
 		let set_id = 1;
@@ -1474,7 +1521,6 @@ mod tests {
 		let (val, _) = GossipValidator::<Block>::new(
 			config(),
 			voter_set_state(),
-			true,
 		);
 		let set_id = 1;
 		let auth = AuthorityId::from_slice(&[1u8; 32]);
@@ -1519,7 +1565,6 @@ mod tests {
 		let (val, _) = GossipValidator::<Block>::new(
 			config(),
 			voter_set_state(),
-			true,
 		);
 
 		let set_id = 1;
@@ -1588,7 +1633,6 @@ mod tests {
 		let (val, _) = GossipValidator::<Block>::new(
 			config(),
 			set_state.clone(),
-			true,
 		);
 
 		let set_id = 1;
@@ -1643,7 +1687,6 @@ mod tests {
 		let (val, _) = GossipValidator::<Block>::new(
 			config(),
 			set_state.clone(),
-			true,
 		);
 
 		// the validator starts at set id 2
@@ -1723,7 +1766,6 @@ mod tests {
 		let (val, _) = GossipValidator::<Block>::new(
 			config(),
 			voter_set_state(),
-			true,
 		);
 
 		// the validator starts at set id 1.
@@ -1783,10 +1825,20 @@ mod tests {
 	#[test]
 	fn doesnt_send_catch_up_requests_when_disabled() {
 		// we create a gossip validator with catch up requests disabled.
+		let config = {
+			let mut c = config();
+
+			// if the observer protocol is enabled and we are not an authority,
+			// then we don't issue any catch-up requests.
+			c.is_authority = false;
+			c.observer_enabled = true;
+
+			c
+		};
+
 		let (val, _) = GossipValidator::<Block>::new(
-			config(),
+			config,
 			voter_set_state(),
-			false,
 		);
 
 		// the validator starts at set id 1.
@@ -1816,11 +1868,10 @@ mod tests {
 	}
 
 	#[test]
-	fn doesnt_send_catch_up_requests_to_non_authorities() {
+	fn doesnt_send_catch_up_requests_to_non_authorities_when_observer_enabled() {
 		let (val, _) = GossipValidator::<Block>::new(
 			config(),
 			voter_set_state(),
-			true,
 		);
 
 		// the validator starts at set id 1.
@@ -1865,13 +1916,59 @@ mod tests {
 		}
 	}
 
+	#[test]
+	fn sends_catch_up_requests_to_non_authorities_when_observer_disabled() {
+		let config = {
+			let mut c = config();
+
+			// if the observer protocol is disable any full-node should be able
+			// to answer catch-up requests.
+			c.observer_enabled = false;
+
+			c
+		};
+
+		let (val, _) = GossipValidator::<Block>::new(
+			config,
+			voter_set_state(),
+		);
+
+		// the validator starts at set id 1.
+		val.note_set(SetId(1), Vec::new(), |_, _| {});
+
+		// add the peer making the requests to the validator, otherwise it is
+		// discarded.
+		let peer_full = PeerId::random();
+		val.inner.write().peers.new_peer(peer_full.clone(), Roles::FULL);
+
+		let (_, _, catch_up_request, _) = val.inner.write().import_neighbor_message(
+			&peer_full,
+			NeighborPacket {
+				round: Round(42),
+				set_id: SetId(1),
+				commit_finalized_height: 50,
+			},
+		);
+
+		// importing a neighbor message from a peer in the same set in a later
+		// round should lead to a catch up request, the node is not an
+		// authority, but since the observer protocol is disabled we should
+		// issue a catch-up request to it anyway.
+		match catch_up_request {
+			Some(GossipMessage::CatchUpRequest(request)) => {
+				assert_eq!(request.set_id, SetId(1));
+				assert_eq!(request.round, Round(41));
+			},
+			_ => panic!("expected catch up message"),
+		}
+	}
+
 	#[test]
 	fn doesnt_expire_next_round_messages() {
 		// NOTE: this is a regression test
 		let (val, _) = GossipValidator::<Block>::new(
 			config(),
 			voter_set_state(),
-			true,
 		);
 
 		// the validator starts at set id 1.

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

@@ -289,7 +289,6 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 		config: crate::Config,
 		set_state: crate::environment::SharedVoterSetState<B>,
 		on_exit: impl Future<Item = (), Error = ()> + Clone + Send + 'static,
-		catch_up_enabled: bool,
 	) -> (
 		Self,
 		impl Future<Item = (), Error = ()> + Send + 'static,
@@ -298,7 +297,6 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 		let (validator, report_stream) = GossipValidator::new(
 			config,
 			set_state.clone(),
-			catch_up_enabled,
 		);
 
 		let validator = Arc::new(validator);

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

@@ -139,6 +139,8 @@ fn config() -> crate::Config {
 		justification_period: 256,
 		keystore: None,
 		name: None,
+		is_authority: true,
+		observer_enabled: true,
 	}
 }
 
@@ -186,7 +188,6 @@ fn make_test_network() -> (
 		config(),
 		voter_set_state(),
 		Exit,
-		true,
 	);
 
 	(

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

@@ -201,6 +201,13 @@ pub struct Config {
 	/// at least every justification_period blocks. There are some other events which might cause
 	/// justification generation.
 	pub justification_period: u32,
+	/// Whether the GRANDPA observer protocol is live on the network and thereby
+	/// a full-node not running as a validator is running the GRANDPA observer
+	/// protocol (we will only issue catch-up requests to authorities when the
+	/// observer protocol is enabled).
+	pub observer_enabled: bool,
+	/// Whether the node is running as an authority (i.e. running the full GRANDPA protocol).
+	pub is_authority: bool,
 	/// Some local identifier of the voter.
 	pub name: Option<String>,
 	/// The keystore that manages the keys of this node.
@@ -548,7 +555,6 @@ pub fn run_grandpa_voter<B, E, Block: BlockT<Hash=H256>, N, RA, SC, VR, X>(
 		config.clone(),
 		persistent_data.set_state.clone(),
 		on_exit.clone(),
-		true,
 	);
 
 	register_finality_tracker_inherent_data_provider(client.clone(), &inherent_data_providers)?;

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

@@ -176,7 +176,6 @@ pub fn run_grandpa_observer<B, E, Block: BlockT<Hash=H256>, N, RA, SC>(
 		config.clone(),
 		persistent_data.set_state.clone(),
 		on_exit.clone(),
-		false,
 	);
 
 	let observer_work = ObserverWork::new(

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

@@ -379,6 +379,8 @@ fn run_to_completion_with<F>(
 				justification_period: 32,
 				keystore: Some(keystore),
 				name: Some(format!("peer#{}", peer_id)),
+				is_authority: true,
+				observer_enabled: true,
 			},
 			link: link,
 			network: net_service,
@@ -511,6 +513,8 @@ fn finalize_3_voters_1_full_observer() {
 				justification_period: 32,
 				keystore,
 				name: Some(format!("peer#{}", peer_id)),
+				is_authority: true,
+				observer_enabled: true,
 			},
 			link: link,
 			network: net_service,
@@ -672,6 +676,8 @@ fn transition_3_voters_twice_1_full_observer() {
 				justification_period: 32,
 				keystore: Some(keystore),
 				name: Some(format!("peer#{}", peer_id)),
+				is_authority: true,
+				observer_enabled: true,
 			},
 			link: link,
 			network: net_service,
@@ -1095,6 +1101,8 @@ fn voter_persists_its_votes() {
 								justification_period: 32,
 								keystore: Some(self.keystore.clone()),
 								name: Some(format!("peer#{}", 0)),
+								is_authority: true,
+								observer_enabled: true,
 							},
 							link,
 							network: self.net.lock().peers[0].network_service().clone(),
@@ -1150,6 +1158,8 @@ fn voter_persists_its_votes() {
 			justification_period: 32,
 			keystore: Some(keystore),
 			name: Some(format!("peer#{}", 1)),
+			is_authority: true,
+			observer_enabled: true,
 		};
 
 		let set_state = {
@@ -1164,7 +1174,6 @@ fn voter_persists_its_votes() {
 			config.clone(),
 			set_state,
 			Exit,
-			true,
 		);
 		runtime.block_on(routing_work).unwrap();
 
@@ -1299,6 +1308,8 @@ fn finalize_3_voters_1_light_observer() {
 					justification_period: 32,
 					keystore: None,
 					name: Some("observer".to_string()),
+					is_authority: false,
+					observer_enabled: true,
 				},
 				link,
 				net.lock().peers[3].network_service().clone(),
@@ -1426,6 +1437,8 @@ fn voter_catches_up_to_latest_round_when_behind() {
 				justification_period: 32,
 				keystore,
 				name: Some(format!("peer#{}", peer_id)),
+				is_authority: true,
+				observer_enabled: true,
 			},
 			link,
 			network: net.lock().peer(peer_id).network_service().clone(),
@@ -1542,6 +1555,8 @@ fn grandpa_environment_respects_voting_rules() {
 			justification_period: 32,
 			keystore: None,
 			name: None,
+			is_authority: true,
+			observer_enabled: true,
 		};
 
 		let (network, _) = NetworkBridge::new(
@@ -1549,7 +1564,6 @@ fn grandpa_environment_respects_voting_rules() {
 			config.clone(),
 			set_state.clone(),
 			Exit,
-			true,
 		);
 
 		Environment {

substrate/node-template/src/service.rs

@@ -125,6 +125,8 @@ pub fn new_full<C: Send + Default + 'static>(config: Configuration<C, GenesisCon
 		justification_period: 512,
 		name: Some(name),
 		keystore: Some(service.keystore()),
+		observer_enabled: true,
+		is_authority,
 	};
 
 	match (is_authority, disable_grandpa) {

substrate/node/cli/src/service.rs

@@ -177,6 +177,8 @@ macro_rules! new_full {
 			justification_period: 512,
 			name: Some(name),
 			keystore: Some(service.keystore()),
+			observer_enabled: true,
+			is_authority,
 		};
 
 		match (is_authority, disable_grandpa) {