Initial version of the polite-grandpa networking protocol (#2110)

* Consensus status packet

* Allow for repropagation after status

* More generic gossip

* add a basic view struct and gossip module

* move gossip stuff to the gossip module

* integrate view into gossip

* some reshuffling

* alter rules for keeping one commit at a time in view

* Allow sending addressed messages

* don't cast outgoing votes if we know that we voted before

* Handle one hop messages

* initial run at polite grandpa

* build WASM

* handle neighbor messages

* refactor validator's internals into an Inner struct

* gossip only knows to keep or discard messages. optimize should_send_to

* Periodic rebroadcast

* implement `should_send_to` and message_expired

* track peers' best received commit height

* Pass peer id to topic steam

* kill rebroadcasting network

* Notify about existing peers

* clean up network APIs a bunch

* implement gossip::send_message for direct messages

* refactor network trait

* implement gossip::send_message for direct messages

* get all non set-change tests passing

* treat unknown rebroadcasts as broadcasts

* get all other main tests passing

* remove unimplemented test

* everything compiles

* treat unknown rebroadcasts as broadcasts

* Rebradcast interval

* Apply suggestions from code review

Style

Co-Authored-By: arkpar <arkady.paronyan@gmail.com>

* Style

* some module docs

* address some grumbles + docs

* allow rebroadcast every few minutes

* send_topic && generic context

* some tests for view change

* more grumbles & tests

* use send_peer

core/finality-grandpa/src/environment.rs

@@ -82,7 +82,7 @@ pub(crate) struct Environment<B, E, Block: BlockT, N: Network<Block>, RA> {
 	pub(crate) config: Config,
 	pub(crate) authority_set: SharedAuthoritySet<Block::Hash, NumberFor<Block>>,
 	pub(crate) consensus_changes: SharedConsensusChanges<Block::Hash, NumberFor<Block>>,
-	pub(crate) network: N,
+	pub(crate) network: crate::communication::NetworkBridge<Block, N>,
 	pub(crate) set_id: u64,
 	pub(crate) last_completed: LastCompletedRound<Block::Hash, NumberFor<Block>>,
 }
@@ -231,32 +231,24 @@ impl<B, E, Block: BlockT<Hash=H256>, N, RA> voter::Environment<Block::Hash, Numb
 		let prevote_timer = Delay::new(now + self.config.gossip_duration * 2);
 		let precommit_timer = Delay::new(now + self.config.gossip_duration * 4);
 
-		let incoming = crate::communication::checked_message_stream::<Block, _>(
-			self.network.messages_for(round, self.set_id),
-			self.voters.clone(),
-		);
-
 		let local_key = self.config.local_key.as_ref()
 			.filter(|pair| self.voters.contains_key(&pair.public().into()));
 
-		let (out_rx, outgoing) = crate::communication::outgoing_messages::<Block, _>(
-			round,
-			self.set_id,
-			local_key.cloned(),
+		let (incoming, outgoing) = self.network.round_communication(
+			crate::communication::Round(round),
+			crate::communication::SetId(self.set_id),
 			self.voters.clone(),
-			self.network.clone(),
+			local_key.cloned(),
+			crate::communication::HasVoted::No,
 		);
 
 		// schedule incoming messages from the network to be held until
 		// corresponding blocks are imported.
-		let incoming = UntilVoteTargetImported::new(
+		let incoming = Box::new(UntilVoteTargetImported::new(
 			self.inner.import_notification_stream(),
 			self.inner.clone(),
 			incoming,
-		);
-
-		// join incoming network messages with locally originating ones.
-		let incoming = Box::new(out_rx.select(incoming).map_err(Into::into));
+		).map_err(Into::into));
 
 		// schedule network message cleanup when sink drops.
 		let outgoing = Box::new(outgoing.sink_map_err(Into::into));
@@ -305,15 +297,21 @@ impl<B, E, Block: BlockT<Hash=H256>, N, RA> voter::Environment<Block::Hash, Numb
 			return Ok(());
 		}
 
-		finalize_block(
+		let res = finalize_block(
 			&*self.inner,
 			&self.authority_set,
 			&self.consensus_changes,
 			Some(As::sa(self.config.justification_period)),
 			hash,
 			number,
 			(round, commit).into(),
-		)
+		);
+
+		if let Ok(_) = res {
+			self.network.note_commit_finalized(number);
+		}
+
+		res
 	}
 
 	fn round_commit_timer(&self) -> Self::Timer {