Some gossip improvements (#1892)

* queue messages in future

* use new gossip API in GRANDPA

* implement message_expired for grandpa

* fix indent

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

@@ -253,6 +253,42 @@ struct TopicTracker {
 	set_id: u64,
 }
 
+impl TopicTracker {
+	fn is_expired(&self, round: u64, set_id: u64) -> bool {
+		if set_id < self.set_id {
+			trace!(target: "afg", "Expired: Message with expired set_id {} (ours {})", set_id, self.set_id);
+			telemetry!(CONSENSUS_TRACE; "afg.expired_set_id";
+				"set_id" => ?set_id, "ours" => ?self.set_id
+			);
+			return true;
+		} else if set_id == self.set_id + 1 {
+			// allow a few first rounds of future set.
+			if round > MESSAGE_ROUND_TOLERANCE {
+				trace!(target: "afg", "Expired: Message too far in the future set, round {} (ours set_id {})", round, self.set_id);
+				telemetry!(CONSENSUS_TRACE; "afg.expired_msg_too_far_in_future_set";
+					"round" => ?round, "ours" => ?self.set_id
+				);
+				return true;
+			}
+		} else if set_id == self.set_id {
+			if round < self.min_live_round.saturating_sub(MESSAGE_ROUND_TOLERANCE) {
+				trace!(target: "afg", "Expired: Message round is out of bounds {} (ours {}-{})", round, self.min_live_round, self.max_round);
+				telemetry!(CONSENSUS_TRACE; "afg.msg_round_oob";
+					"round" => ?round, "our_min_live_round" => ?self.min_live_round, "our_max_round" => ?self.max_round
+				);
+				return true;
+			}
+		} else {
+			trace!(target: "afg", "Expired: Message in invalid future set {} (ours {})", set_id, self.set_id);
+			telemetry!(CONSENSUS_TRACE; "afg.expired_msg_in_invalid_future_set";
+				"set_id" => ?set_id, "ours" => ?self.set_id
+			);
+			return true;
+		}
+		false
+	}
+}
+
 struct GossipValidator<Block: BlockT> {
 	rounds: parking_lot::RwLock<TopicTracker>,
 	_marker: ::std::marker::PhantomData<Block>,
@@ -294,38 +330,7 @@ impl<Block: BlockT> GossipValidator<Block> {
 	}
 
 	fn is_expired(&self, round: u64, set_id: u64) -> bool {
-		let rounds = self.rounds.read();
+		self.rounds.read().is_expired(round, set_id)
-		if set_id < rounds.set_id {
-			trace!(target: "afg", "Expired: Message with expired set_id {} (ours {})", set_id, rounds.set_id);
-			telemetry!(CONSENSUS_TRACE; "afg.expired_set_id";
-				"set_id" => ?set_id, "ours" => ?rounds.set_id
-			);
-			return true;
-		} else if set_id == rounds.set_id + 1 {
-			// allow a few first rounds of future set.
-			if round > MESSAGE_ROUND_TOLERANCE {
-				trace!(target: "afg", "Expired: Message too far in the future set, round {} (ours set_id {})", round, rounds.set_id);
-				telemetry!(CONSENSUS_TRACE; "afg.expired_msg_too_far_in_future_set";
-					"round" => ?round, "ours" => ?rounds.set_id
-				);
-				return true;
-			}
-		} else if set_id == rounds.set_id {
-			if round < rounds.min_live_round.saturating_sub(MESSAGE_ROUND_TOLERANCE) {
-				trace!(target: "afg", "Expired: Message round is out of bounds {} (ours {}-{})", round, rounds.min_live_round, rounds.max_round);
-				telemetry!(CONSENSUS_TRACE; "afg.msg_round_oob";
-					"round" => ?round, "our_min_live_round" => ?rounds.min_live_round, "our_max_round" => ?rounds.max_round
-				);
-				return true;
-			}
-		} else {
-			trace!(target: "afg", "Expired: Message in invalid future set {} (ours {})", set_id, rounds.set_id);
-			telemetry!(CONSENSUS_TRACE; "afg.expired_msg_in_invalid_future_set";
-				"set_id" => ?set_id, "ours" => ?rounds.set_id
-			);
-			return true;
-		}
-		false
 	}
 
 	fn validate_round_message(&self, full: VoteOrPrecommitMessage<Block>)
@@ -401,6 +406,18 @@ impl<Block: BlockT> network_gossip::Validator<Block::Hash> for GossipValidator<B
 			}
 		}
 	}
+
+	fn message_expired<'a>(&'a self) -> Box<FnMut(Block::Hash, &[u8]) -> bool + 'a> {
+		let rounds = self.rounds.read();
+		Box::new(move |_topic, mut data| {
+			match GossipMessage::<Block>::decode(&mut data) {
+				None => true,
+				Some(GossipMessage::Commit(full)) => rounds.is_expired(full.round, full.set_id),
+				Some(GossipMessage::VoteOrPrecommit(full)) =>
+					rounds.is_expired(full.round, full.set_id),
+			}
+		})
+	}
 }
 
 /// A handle to the network. This is generally implemented by providing some
@@ -476,7 +493,7 @@ impl<B: BlockT, S: network::specialization::NetworkSpecialization<B>,> Network<B
 		self.validator.note_round(round, set_id);
 		let (tx, rx) = sync::oneshot::channel();
 		self.service.with_gossip(move |gossip, _| {
-			let inner_rx = gossip.messages_for(message_topic::<B>(round, set_id));
+			let inner_rx = gossip.messages_for(GRANDPA_ENGINE_ID, message_topic::<B>(round, set_id));
 			let _ = tx.send(inner_rx);
 		});
 		NetworkStream { outer: rx, inner: None }
@@ -501,7 +518,7 @@ impl<B: BlockT, S: network::specialization::NetworkSpecialization<B>,> Network<B
 		self.validator.note_set(set_id);
 		let (tx, rx) = sync::oneshot::channel();
 		self.service.with_gossip(move |gossip, _| {
-			let inner_rx = gossip.messages_for(commit_topic::<B>(set_id));
+			let inner_rx = gossip.messages_for(GRANDPA_ENGINE_ID, commit_topic::<B>(set_id));
 			let _ = tx.send(inner_rx);
 		});
 		NetworkStream { outer: rx, inner: None }

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

@@ -181,7 +181,10 @@ impl Network<Block> for MessageRouting {
 		self.validator.note_round(round, set_id);
 		let inner = self.inner.lock();
 		let peer = inner.peer(self.peer_id);
-		let messages = peer.consensus_gossip_messages_for(make_topic(round, set_id));
+		let messages = peer.consensus_gossip_messages_for(
+			GRANDPA_ENGINE_ID,
+			make_topic(round, set_id),
+		);
 
 		let messages = messages.map_err(
 			move |_| panic!("Messages for round {} dropped too early", round)
@@ -211,7 +214,10 @@ impl Network<Block> for MessageRouting {
 		self.validator.note_set(set_id);
 		let inner = self.inner.lock();
 		let peer = inner.peer(self.peer_id);
-        let messages = peer.consensus_gossip_messages_for(make_commit_topic(set_id));
+        let messages = peer.consensus_gossip_messages_for(
+			GRANDPA_ENGINE_ID,
+			make_commit_topic(set_id),
+		);
 
 		let messages = messages.map_err(
 			move |_| panic!("Commit messages for set {} dropped too early", set_id)

substrate/core/network/src/consensus_gossip.rs

@@ -40,17 +40,26 @@ struct PeerConsensus<H> {
 	is_authority: bool,
 }
 
+#[derive(Clone, Copy)]
+enum Status {
+	Live,
+	Future,
+}
+
 struct MessageEntry<B: BlockT> {
 	message_hash: B::Hash,
 	topic: B::Hash,
 	message: ConsensusMessage,
 	timestamp: Instant,
+	status: Status,
 }
 
 /// Message validation result.
 pub enum ValidationResult<H> {
 	/// Message is valid with this topic.
 	Valid(H),
+	/// Message is future with this topic.
+	Future(H),
 	/// Invalid message.
 	Invalid,
 	/// Obsolete message.
@@ -61,12 +70,20 @@ pub enum ValidationResult<H> {
 pub trait Validator<H> {
 	/// Validate consensus message.
 	fn validate(&self, data: &[u8]) -> ValidationResult<H>;
+
+	/// Produce a closure for validating messages on a given topic.
+	fn message_expired<'a>(&'a self) -> Box<FnMut(H, &[u8]) -> bool + 'a> {
+		Box::new(move |_topic, data| match self.validate(data) {
+			ValidationResult::Valid(_) | ValidationResult::Future(_) => false,
+			ValidationResult::Invalid | ValidationResult::Expired => true,
+		})
+	}
 }
 
 /// Consensus network protocol handler. Manages statements and candidate requests.
 pub struct ConsensusGossip<B: BlockT> {
 	peers: HashMap<NodeIndex, PeerConsensus<B::Hash>>,
-	live_message_sinks: HashMap<B::Hash, Vec<mpsc::UnboundedSender<Vec<u8>>>>,
+	live_message_sinks: HashMap<(ConsensusEngineId, B::Hash), Vec<mpsc::UnboundedSender<Vec<u8>>>>,
 	messages: Vec<MessageEntry<B>>,
 	known_messages: LruCache<B::Hash, ()>,
 	validators: HashMap<ConsensusEngineId, Arc<Validator<B::Hash>>>,
@@ -102,7 +119,9 @@ impl<B: BlockT> ConsensusGossip<B> {
 			// Send out all known messages to authorities.
 			let mut known_messages = HashSet::new();
 			for entry in self.messages.iter() {
-				if entry.timestamp + MESSAGE_LIFETIME < now { continue };
+				if entry.timestamp + MESSAGE_LIFETIME < now { continue }
+				if let Status::Future = entry.status { continue }
+
 				known_messages.insert(entry.message_hash);
 				protocol.send_message(who, Message::Consensus(entry.message.clone()));
 			}
@@ -161,18 +180,23 @@ impl<B: BlockT> ConsensusGossip<B> {
 		}
 	}
 
-	fn register_message<F>(&mut self, message_hash: B::Hash, topic: B::Hash, get_message: F)
+	fn register_message<F>(
+		&mut self,
+		message_hash: B::Hash,
+		topic: B::Hash,
+		status: Status,
+		get_message: F,
+	)
 		where F: Fn() -> ConsensusMessage
 	{
-		if self.known_messages.insert(message_hash, ()).is_none()
+		if self.known_messages.insert(message_hash, ()).is_none() {
-		{
 			self.messages.push(MessageEntry {
 				topic,
 				message_hash,
 				message: get_message(),
 				timestamp: Instant::now(),
+				status,
 			});
-
 		}
 	}
 
@@ -184,6 +208,8 @@ impl<B: BlockT> ConsensusGossip<B> {
 	/// Prune old or no longer relevant consensus messages. Provide a predicate
 	/// for pruning, which returns `false` when the items with a given topic should be pruned.
 	pub fn collect_garbage(&mut self) {
+		use std::collections::hash_map::Entry;
+
 		self.live_message_sinks.retain(|_, sinks| {
 			sinks.retain(|sink| !sink.is_closed());
 			!sinks.is_empty()
@@ -194,15 +220,23 @@ impl<B: BlockT> ConsensusGossip<B> {
 		let validators = &self.validators;
 		let now = Instant::now();
 
-		self.messages.retain(|entry| {
+		let mut check_fns = HashMap::new();
-			entry.timestamp + MESSAGE_LIFETIME >= now
+		let mut message_expired = move |entry: &MessageEntry<B>| {
-			&& match validators.get(&entry.message.engine_id)
+			let engine_id = entry.message.engine_id;
-				.map(|v| v.validate(&entry.message.data))
+			let check_fn = match check_fns.entry(engine_id) {
-			{
+				Entry::Occupied(entry) => entry.into_mut(),
-				Some(ValidationResult::Valid(_)) => true,
+				Entry::Vacant(vacant) => match validators.get(&engine_id) {
-				_ => false,
+					None => return true, // treat all messages with no validator as expired
-			}
+					Some(validator) => vacant.insert(validator.message_expired()),
-		});
+				}
+			};
+
+			(check_fn)(entry.topic, &entry.message.data)
+		};
+
+		self.messages.retain(|entry|
+			entry.timestamp + MESSAGE_LIFETIME >= now && !message_expired(entry)
+		);
 
 		trace!(target: "gossip", "Cleaned up {} stale messages, {} left ({} known)",
 			before - self.messages.len(),
@@ -216,12 +250,46 @@ impl<B: BlockT> ConsensusGossip<B> {
 	}
 
 	/// Get data of valid, incoming messages for a topic (but might have expired meanwhile)
-	pub fn messages_for(&mut self, topic: B::Hash) -> mpsc::UnboundedReceiver<Vec<u8>> {
+	pub fn messages_for(&mut self, engine_id: ConsensusEngineId, topic: B::Hash)
+		-> mpsc::UnboundedReceiver<Vec<u8>>
+	{
 		let (tx, rx) = mpsc::unbounded();
-		for entry in self.messages.iter().filter(|e| e.topic == topic) {
+
-			tx.unbounded_send(entry.message.data.clone()).expect("receiver known to be live; qed");
+		let validator = match self.validators.get(&engine_id) {
+			None => {
+				self.live_message_sinks.entry((engine_id, topic)).or_default().push(tx);
+				return rx;
+			}
+			Some(v) => v,
+		};
+
+		for entry in self.messages.iter_mut()
+			.filter(|e| e.topic == topic && e.message.engine_id == engine_id)
+		{
+			let live = match entry.status {
+				Status::Live => true,
+				Status::Future => match validator.validate(&entry.message.data) {
+					ValidationResult::Valid(_) => {
+						entry.status = Status::Live;
+						true
+					}
+					_ => {
+						// don't send messages considered to be future still.
+						// if messages are considered expired they'll be cleaned up when we
+						// collect garbage.
+						false
+					}
+				}
+			};
+
+			if live {
+				entry.status = Status::Live;
+				tx.unbounded_send(entry.message.data.clone())
+					.expect("receiver known to be live; qed");
+			}
 		}
-		self.live_message_sinks.entry(topic).or_default().push(tx);
+
+		self.live_message_sinks.entry((engine_id, topic)).or_default().push(tx);
 
 		rx
 	}
@@ -247,11 +315,13 @@ impl<B: BlockT> ConsensusGossip<B> {
 		if let Some(ref mut peer) = self.peers.get_mut(&who) {
 			use std::collections::hash_map::Entry;
 
+			let engine_id = message.engine_id;
 			//validate the message
-			let topic = match self.validators.get(&message.engine_id)
+			let (topic, status) = match self.validators.get(&engine_id)
 				.map(|v| v.validate(&message.data))
 			{
-				Some(ValidationResult::Valid(topic)) => topic,
+				Some(ValidationResult::Valid(topic)) => (topic, Status::Live),
+				Some(ValidationResult::Future(topic)) => (topic, Status::Future),
 				Some(ValidationResult::Invalid) => {
 					trace!(target:"gossip", "Invalid message from {}", who);
 					protocol.report_peer(
@@ -275,13 +345,14 @@ impl<B: BlockT> ConsensusGossip<B> {
 						who,
 						Severity::Useless(format!("Sent unknown consensus engine id")),
 					);
-					trace!(target:"gossip", "Unknown message engine id {:?} from {}", message.engine_id, who);
+					trace!(target:"gossip", "Unknown message engine id {:?} from {}",
+						engine_id, who);
 					return None;
 				}
 			};
 
 			peer.known_messages.insert(message_hash);
-			if let Entry::Occupied(mut entry) = self.live_message_sinks.entry(topic) {
+			if let Entry::Occupied(mut entry) = self.live_message_sinks.entry((engine_id, topic)) {
 				debug!(target: "gossip", "Pushing consensus message to sinks for {}.", topic);
 				entry.get_mut().retain(|sink| {
 					if let Err(e) = sink.unbounded_send(message.data.clone()) {
@@ -293,7 +364,7 @@ impl<B: BlockT> ConsensusGossip<B> {
 					entry.remove_entry();
 				}
 			}
-			self.multicast_inner(protocol, message_hash, topic, || message.clone());
+			self.multicast_inner(protocol, message_hash, topic, status, || message.clone());
 			Some((topic, message))
 		} else {
 			trace!(target:"gossip", "Ignored statement from unregistered peer {}", who);
@@ -309,7 +380,7 @@ impl<B: BlockT> ConsensusGossip<B> {
 		message: ConsensusMessage,
 	) {
 		let message_hash = HashFor::<B>::hash(&message.data);
-		self.multicast_inner(protocol, message_hash, topic, || message.clone());
+		self.multicast_inner(protocol, message_hash, topic, Status::Live, || message.clone());
 	}
 
 	fn multicast_inner<F>(
@@ -317,12 +388,15 @@ impl<B: BlockT> ConsensusGossip<B> {
 		protocol: &mut Context<B>,
 		message_hash: B::Hash,
 		topic: B::Hash,
+		status: Status,
 		get_message: F,
 	)
 		where F: Fn() -> ConsensusMessage
 	{
-		self.register_message(message_hash, topic, &get_message);
+		self.register_message(message_hash, topic, status, &get_message);
-		self.propagate(protocol, message_hash, get_message);
+		if let Status::Live = status {
+			self.propagate(protocol, message_hash, get_message);
+		}
 	}
 
 	/// Note new consensus session.
@@ -335,6 +409,8 @@ impl<B: BlockT> ConsensusGossip<B> {
 mod tests {
 	use runtime_primitives::testing::{H256, Block as RawBlock, ExtrinsicWrapper};
 	use std::time::Instant;
+	use futures::Stream;
+
 	use super::*;
 
 	type Block = RawBlock<ExtrinsicWrapper<u64>>;
@@ -347,21 +423,21 @@ mod tests {
 					message_hash: $hash,
 					message: ConsensusMessage { data: $m, engine_id: [0, 0, 0, 0]},
 					timestamp: $now,
+					status: Status::Live,
 				});
 			}
 		}
 	}
 
+	struct AllowAll;
+	impl Validator<H256> for AllowAll {
+		fn validate(&self, _data: &[u8]) -> ValidationResult<H256> {
+			ValidationResult::Valid(H256::default())
+		}
+	}
+
 	#[test]
 	fn collects_garbage() {
-
-		struct AllowAll;
-		impl Validator<H256> for AllowAll {
-			fn validate(&self, _data: &[u8]) -> ValidationResult<H256> {
-				ValidationResult::Valid(H256::default())
-			}
-		}
-
 		struct AllowOne;
 		impl Validator<H256> for AllowOne {
 			fn validate(&self, data: &[u8]) -> ValidationResult<H256> {
@@ -417,14 +493,15 @@ mod tests {
 		use futures::Stream;
 
 		let mut consensus = ConsensusGossip::<Block>::new();
+		consensus.register_validator([0, 0, 0, 0], Arc::new(AllowAll));
 
 		let message = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 0] };
 
 		let message_hash = HashFor::<Block>::hash(&message.data);
 		let topic = HashFor::<Block>::hash(&[1,2,3]);
 
-		consensus.register_message(message_hash, topic, || message.clone());
+		consensus.register_message(message_hash, topic, Status::Live, || message.clone());
-		let stream = consensus.messages_for(topic);
+		let stream = consensus.messages_for([0, 0, 0, 0], topic);
 
 		assert_eq!(stream.wait().next(), Some(Ok(message.data)));
 	}
@@ -437,29 +514,47 @@ mod tests {
 		let msg_a = ConsensusMessage { data: vec![1, 2, 3], engine_id: [0, 0, 0, 0] };
 		let msg_b = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 0] };
 
-		consensus.register_message(HashFor::<Block>::hash(&msg_a.data), topic, || msg_a.clone());
+		consensus.register_message(HashFor::<Block>::hash(&msg_a.data), topic, Status::Live, || msg_a.clone());
-		consensus.register_message(HashFor::<Block>::hash(&msg_b.data), topic, || msg_b.clone());
+		consensus.register_message(HashFor::<Block>::hash(&msg_b.data), topic, Status::Live, || msg_b.clone());
 
 		assert_eq!(consensus.messages.len(), 2);
 	}
 
 	#[test]
 	fn can_keep_multiple_subscribers_per_topic() {
-		use futures::Stream;
-
 		let mut consensus = ConsensusGossip::<Block>::new();
+		consensus.register_validator([0, 0, 0, 0], Arc::new(AllowAll));
 
 		let message = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 0] };
 
 		let message_hash = HashFor::<Block>::hash(&message.data);
 		let topic = HashFor::<Block>::hash(&[1,2,3]);
 
-		consensus.register_message(message_hash, topic, || message.clone());
+		consensus.register_message(message_hash, topic, Status::Live, || message.clone());
 
-		let stream1 = consensus.messages_for(topic);
+		let stream1 = consensus.messages_for([0, 0, 0, 0], topic);
-		let stream2 = consensus.messages_for(topic);
+		let stream2 = consensus.messages_for([0, 0, 0, 0], topic);
 
 		assert_eq!(stream1.wait().next(), Some(Ok(message.data.clone())));
 		assert_eq!(stream2.wait().next(), Some(Ok(message.data)));
 	}
+
+	#[test]
+	fn topics_are_localized_to_engine_id() {
+		let mut consensus = ConsensusGossip::<Block>::new();
+		consensus.register_validator([0, 0, 0, 0], Arc::new(AllowAll));
+
+		let topic = [1; 32].into();
+		let msg_a = ConsensusMessage { data: vec![1, 2, 3], engine_id: [0, 0, 0, 0] };
+		let msg_b = ConsensusMessage { data: vec![4, 5, 6], engine_id: [0, 0, 0, 1] };
+
+		consensus.register_message(HashFor::<Block>::hash(&msg_a.data), topic, Status::Live, || msg_a.clone());
+		consensus.register_message(HashFor::<Block>::hash(&msg_b.data), topic, Status::Live, || msg_b.clone());
+
+		let mut stream = consensus.messages_for([0, 0, 0, 0], topic).wait();
+
+		assert_eq!(stream.next(), Some(Ok(vec![1, 2, 3])));
+		let _ = consensus.live_message_sinks.remove(&([0, 0, 0, 0], topic));
+		assert_eq!(stream.next(), None);
+	}
 }

substrate/core/network/src/test/mod.rs

@@ -369,11 +369,12 @@ impl<D, S: NetworkSpecialization<Block> + Clone> Peer<D, S> {
 	/// access the underlying consensus gossip handler
 	pub fn consensus_gossip_messages_for(
 		&self,
+		engine_id: ConsensusEngineId,
 		topic: <Block as BlockT>::Hash,
 	) -> mpsc::UnboundedReceiver<Vec<u8>> {
 		let (tx, rx) = oneshot::channel();
 		self.with_gossip(move |gossip, _| {
-			let inner_rx = gossip.messages_for(topic);
+			let inner_rx = gossip.messages_for(engine_id, topic);
 			let _ = tx.send(inner_rx);
 		});
 		rx.wait().ok().expect("1. Network is running, 2. it should handle the above closure successfully")