core/finality-grandpa: Minor refactorings (#3825)

* core/finality-grandpa: Improve code comments

* core/finality-grandpa: Rename VoteOrPrecommit to PrevoteOrPrecommit

According to the Grandpa paper [1]:

> A vote is a block hash, together with some metadata such as round
number and the type of vote, such as prevote or precommit, all signed
with a voter’s private key.

To reduce confusion this patch makes the code consistent with the
research paper.

[1] https://github.com/w3f/consensus/blob/master/pdf/grandpa.pdf

* core/finality-grandpa: Add comment for NetworkStream concept

* core/finality-grandpa: Improve round_communication doc comment

* core/finality-grandpa: Rename PrevoteOrPrecommit to Vote

* core/finality-grandpa: Represent NetworkStream state machine as enum

* core/finality-grandpa: Improve KeepTopics comment

Author: mxinden

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

@@ -52,7 +52,7 @@ pub type AuthorityWeight = u64;
 /// The index of an authority.
 pub type AuthorityIndex = u64;
 
-/// The identifier of a GRANDPA set.
+/// The monotonic identifier of a GRANDPA set of authorities.
 pub type SetId = u64;
 
 /// The round indicator.

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

@@ -183,7 +183,13 @@ impl<N: Ord> View<N> {
 
 const KEEP_RECENT_ROUNDS: usize = 3;
 
-/// Tracks topics we keep messages for.
+/// Tracks gossip topics that we are keeping messages for. We keep topics of:
+///
+/// - the last `KEEP_RECENT_ROUNDS` complete GRANDPA rounds,
+///
+/// - the topic for the current and next round,
+///
+/// - and a global topic for commit and catch-up messages.
 struct KeepTopics<B: BlockT> {
 	current_set: SetId,
 	rounds: VecDeque<(Round, SetId)>,
@@ -256,7 +262,7 @@ fn neighbor_topics<B: BlockT>(view: &View<NumberFor<B>>) -> Vec<B::Hash> {
 #[derive(Debug, Encode, Decode)]
 pub(super) enum GossipMessage<Block: BlockT> {
 	/// Grandpa message with round and set info.
-	VoteOrPrecommit(VoteOrPrecommitMessage<Block>),
+	Vote(VoteMessage<Block>),
 	/// Grandpa commit message with round and set info.
 	Commit(FullCommitMessage<Block>),
 	/// A neighbor packet. Not repropagated.
@@ -273,9 +279,9 @@ impl<Block: BlockT> From<NeighborPacket<NumberFor<Block>>> for GossipMessage<Blo
 	}
 }
 
-/// Network level message with topic information.
+/// Network level vote message with topic information.
 #[derive(Debug, Encode, Decode)]
-pub(super) struct VoteOrPrecommitMessage<Block: BlockT> {
+pub(super) struct VoteMessage<Block: BlockT> {
 	/// The round this message is from.
 	pub(super) round: Round,
 	/// The voter set ID this message is from.
@@ -612,7 +618,7 @@ impl<Block: BlockT> Inner<Block> {
 		cost::PAST_REJECTION
 	}
 
-	fn validate_round_message(&self, who: &PeerId, full: &VoteOrPrecommitMessage<Block>)
+	fn validate_round_message(&self, who: &PeerId, full: &VoteMessage<Block>)
 		-> Action<Block::Hash>
 	{
 		match self.consider_vote(full.round, full.set_id) {
@@ -1003,7 +1009,7 @@ impl<Block: BlockT> GossipValidator<Block> {
 
 		let action = {
 			match GossipMessage::<Block>::decode(&mut data) {
-				Ok(GossipMessage::VoteOrPrecommit(ref message))
+				Ok(GossipMessage::Vote(ref message))
 					=> self.inner.write().validate_round_message(who, message),
 				Ok(GossipMessage::Commit(ref message)) => self.inner.write().validate_commit_message(who, message),
 				Ok(GossipMessage::Neighbor(update)) => {
@@ -1163,7 +1169,7 @@ impl<Block: BlockT> network_gossip::Validator<Block> for GossipValidator<Block>
 				Ok(GossipMessage::Neighbor(_)) => false,
 				Ok(GossipMessage::CatchUpRequest(_)) => false,
 				Ok(GossipMessage::CatchUp(_)) => false,
-				Ok(GossipMessage::VoteOrPrecommit(_)) => false, // should not be the case.
+				Ok(GossipMessage::Vote(_)) => false, // should not be the case.
 			}
 		})
 	}
@@ -1478,7 +1484,7 @@ mod tests {
 		val.note_round(Round(1), |_, _| {});
 
 		let inner = val.inner.read();
-		let unknown_voter = inner.validate_round_message(&peer, &VoteOrPrecommitMessage {
+		let unknown_voter = inner.validate_round_message(&peer, &VoteMessage {
 			round: Round(1),
 			set_id: SetId(set_id),
 			message: SignedMessage::<Block> {
@@ -1491,7 +1497,7 @@ mod tests {
 			}
 		});
 
-		let bad_sig = inner.validate_round_message(&peer, &VoteOrPrecommitMessage {
+		let bad_sig = inner.validate_round_message(&peer, &VoteMessage {
 			round: Round(1),
 			set_id: SetId(set_id),
 			message: SignedMessage::<Block> {

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

@@ -48,7 +48,7 @@ use crate::{
 };
 use crate::environment::HasVoted;
 use gossip::{
-	GossipMessage, FullCatchUpMessage, FullCommitMessage, VoteOrPrecommitMessage, GossipValidator
+	GossipMessage, FullCatchUpMessage, FullCommitMessage, VoteMessage, GossipValidator
 };
 use fg_primitives::{
 	AuthorityPair, AuthorityId, AuthoritySignature, SetId as SetIdNumber, RoundNumber,
@@ -148,12 +148,21 @@ impl<B, S, H> Network<B> for Arc<NetworkService<B, S, H>> where
 	type In = NetworkStream;
 
 	fn messages_for(&self, topic: B::Hash) -> Self::In {
+		// Given that one can only communicate with the Substrate network via the `NetworkService` via message-passing,
+		// and given that methods on the network consensus gossip are not exposed but only reachable by passing a
+		// closure into `with_gossip` on the `NetworkService` this function needs to make use of the `NetworkStream`
+		// construction.
+		//
+		// We create a oneshot channel and pass the sender within a closure to the network. At some point in the future
+		// the network passes the message channel back through the oneshot channel. But the consumer of this function
+		// expects a stream, not a stream within a oneshot. This complexity is abstracted within `NetworkStream`,
+		// waiting for the oneshot to resolve and from there on acting like a normal message channel.
 		let (tx, rx) = oneshot::channel();
 		self.with_gossip(move |gossip, _| {
 			let inner_rx = gossip.messages_for(GRANDPA_ENGINE_ID, topic);
 			let _ = tx.send(inner_rx);
 		});
-		NetworkStream { outer: rx, inner: None }
+		NetworkStream::PollingOneshot(rx)
 	}
 
 	fn register_validator(&self, validator: Arc<dyn network_gossip::Validator<B>>) {
@@ -202,28 +211,40 @@ impl<B, S, H> Network<B> for Arc<NetworkService<B, S, H>> where
 	}
 }
 
-/// A stream used by NetworkBridge in its implementation of Network.
-pub struct NetworkStream {
-	inner: Option<mpsc::UnboundedReceiver<network_gossip::TopicNotification>>,
-	outer: oneshot::Receiver<mpsc::UnboundedReceiver<network_gossip::TopicNotification>>
+/// A stream used by NetworkBridge in its implementation of Network. Given a oneshot that eventually returns a channel
+/// which eventually returns messages, instead of:
+///
+/// 1. polling the oneshot until it returns a message channel
+///
+/// 2. polling the message channel for messages
+///
+/// `NetworkStream` combines the two steps into one, requiring a consumer to only poll `NetworkStream` to retrieve
+/// messages directly.
+pub enum NetworkStream {
+	PollingOneshot(oneshot::Receiver<mpsc::UnboundedReceiver<network_gossip::TopicNotification>>),
+	PollingTopicNotifications(mpsc::UnboundedReceiver<network_gossip::TopicNotification>),
 }
 
 impl Stream for NetworkStream {
 	type Item = network_gossip::TopicNotification;
 	type Error = ();
 
 	fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
-		if let Some(ref mut inner) = self.inner {
-			return inner.poll();
-		}
-		match self.outer.poll() {
-			Ok(futures::Async::Ready(mut inner)) => {
-				let poll_result = inner.poll();
-				self.inner = Some(inner);
-				poll_result
+		match self {
+			NetworkStream::PollingOneshot(oneshot) => {
+				match oneshot.poll() {
+					Ok(futures::Async::Ready(mut stream)) => {
+						let poll_result = stream.poll();
+						*self = NetworkStream::PollingTopicNotifications(stream);
+						poll_result
+					},
+					Ok(futures::Async::NotReady) => Ok(futures::Async::NotReady),
+					Err(_) => Err(())
+				}
+			},
+			NetworkStream::PollingTopicNotifications(stream) => {
+				stream.poll()
 			},
-			Ok(futures::Async::NotReady) => Ok(futures::Async::NotReady),
-			Err(_) => Err(())
 		}
 	}
 }
@@ -275,8 +296,8 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 				validator.note_round(Round(round.number), |_, _| {});
 
 				for signed in round.votes.iter() {
-					let message = gossip::GossipMessage::VoteOrPrecommit(
-						gossip::VoteOrPrecommitMessage::<B> {
+					let message = gossip::GossipMessage::Vote(
+						gossip::VoteMessage::<B> {
 							message: signed.clone(),
 							round: Round(round.number),
 							set_id: SetId(set_id),
@@ -341,7 +362,8 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 		);
 	}
 
-	/// Get the round messages for a round in the current set ID. These are signature-checked.
+	/// Get a stream of signature-checked round messages from the network as well as a sink for round messages to the
+	/// network all within the current set.
 	pub(crate) fn round_communication(
 		&self,
 		round: Round,
@@ -379,7 +401,7 @@ impl<B: BlockT, N: Network<B>> NetworkBridge<B, N> {
 			})
 			.and_then(move |msg| {
 				match msg {
-					GossipMessage::VoteOrPrecommit(msg) => {
+					GossipMessage::Vote(msg) => {
 						// check signature.
 						if !voters.contains_key(&msg.message.id) {
 							debug!(target: "afg", "Skipping message from unknown voter {}", msg.message.id);
@@ -707,7 +729,7 @@ impl<Block: BlockT, N: Network<Block>> Sink for OutgoingMessages<Block, N>
 				id: local_id.clone(),
 			};
 
-			let message = GossipMessage::VoteOrPrecommit(VoteOrPrecommitMessage::<Block> {
+			let message = GossipMessage::Vote(VoteMessage::<Block> {
 				message: signed.clone(),
 				round: Round(self.round),
 				set_id: SetId(self.set_id),