sc-beefy-consensus: Remove unneeded stream. (#4015)

The stream was just used to communicate from the validator the peer
reports back to the gossip engine. Internally the gossip engine just
forwards these reports to the networking engine. So, we can just do this
directly.

The reporting stream was also pumped [in the worker behind the
engine](https://github.com/paritytech/polkadot-sdk/blob/9d6261892814fa27c97881c0321c008d7340b54b/substrate/client/consensus/beefy/src/worker.rs#L939).
This means if there was a lot of data incoming over the engine, the
reporting stream was almost never processed and thus, it could have
started to grow and we have seen issues around this.

Partly Closes: https://github.com/paritytech/polkadot-sdk/issues/3945

substrate/client/consensus/beefy/src/lib.rs

@@ -68,7 +68,7 @@ pub mod import;
 pub mod justification;
 
 use crate::{
-	communication::{gossip::GossipValidator, peers::PeerReport},
+	communication::gossip::GossipValidator,
 	justification::BeefyVersionedFinalityProof,
 	keystore::BeefyKeystore,
 	metrics::VoterMetrics,
@@ -78,7 +78,6 @@ use crate::{
 pub use communication::beefy_protocol_name::{
 	gossip_protocol_name, justifications_protocol_name as justifs_protocol_name,
 };
-use sc_utils::mpsc::TracingUnboundedReceiver;
 use sp_runtime::generic::OpaqueDigestItemId;
 
 #[cfg(test)]
@@ -228,10 +227,9 @@ pub struct BeefyParams<B: Block, BE, C, N, P, R, S> {
 /// Helper object holding BEEFY worker communication/gossip components.
 ///
 /// These are created once, but will be reused if worker is restarted/reinitialized.
-pub(crate) struct BeefyComms<B: Block> {
+pub(crate) struct BeefyComms<B: Block, N> {
 	pub gossip_engine: GossipEngine<B>,
-	pub gossip_validator: Arc<GossipValidator<B>>,
-	pub gossip_report_stream: TracingUnboundedReceiver<PeerReport>,
+	pub gossip_validator: Arc<GossipValidator<B, N>>,
 	pub on_demand_justifications: OnDemandJustificationsEngine<B>,
 }
 
@@ -264,13 +262,13 @@ where
 	/// persisted state in AUX DB and latest chain information/progress.
 	///
 	/// Returns a sane `BeefyWorkerBuilder` that can build the `BeefyWorker`.
-	pub async fn async_initialize(
+	pub async fn async_initialize<N>(
 		backend: Arc<BE>,
 		runtime: Arc<R>,
 		key_store: BeefyKeystore<AuthorityId>,
 		metrics: Option<VoterMetrics>,
 		min_block_delta: u32,
-		gossip_validator: Arc<GossipValidator<B>>,
+		gossip_validator: Arc<GossipValidator<B, N>>,
 		finality_notifications: &mut Fuse<FinalityNotifications<B>>,
 		is_authority: bool,
 	) -> Result<Self, Error> {
@@ -298,15 +296,15 @@ where
 	}
 
 	/// Takes rest of missing pieces as params and builds the `BeefyWorker`.
-	pub fn build<P, S>(
+	pub fn build<P, S, N>(
 		self,
 		payload_provider: P,
 		sync: Arc<S>,
-		comms: BeefyComms<B>,
+		comms: BeefyComms<B, N>,
 		links: BeefyVoterLinks<B>,
 		pending_justifications: BTreeMap<NumberFor<B>, BeefyVersionedFinalityProof<B>>,
 		is_authority: bool,
-	) -> BeefyWorker<B, BE, P, R, S> {
+	) -> BeefyWorker<B, BE, P, R, S, N> {
 		BeefyWorker {
 			backend: self.backend,
 			runtime: self.runtime,
@@ -526,8 +524,8 @@ pub async fn start_beefy_gadget<B, BE, C, N, P, R, S>(
 	let known_peers = Arc::new(Mutex::new(KnownPeers::new()));
 	// Default votes filter is to discard everything.
 	// Validator is updated later with correct starting round and set id.
-	let (gossip_validator, gossip_report_stream) =
-		communication::gossip::GossipValidator::new(known_peers.clone());
+	let gossip_validator =
+		communication::gossip::GossipValidator::new(known_peers.clone(), network.clone());
 	let gossip_validator = Arc::new(gossip_validator);
 	let gossip_engine = GossipEngine::new(
 		network.clone(),
@@ -546,46 +544,35 @@ pub async fn start_beefy_gadget<B, BE, C, N, P, R, S>(
 		known_peers,
 		prometheus_registry.clone(),
 	);
-	let mut beefy_comms = BeefyComms {
-		gossip_engine,
-		gossip_validator,
-		gossip_report_stream,
-		on_demand_justifications,
-	};
+	let mut beefy_comms = BeefyComms { gossip_engine, gossip_validator, on_demand_justifications };
 
 	// We re-create and re-run the worker in this loop in order to quickly reinit and resume after
 	// select recoverable errors.
 	loop {
 		// Make sure to pump gossip engine while waiting for initialization conditions.
-		let worker_builder = loop {
-			futures::select! {
-				builder_init_result = BeefyWorkerBuilder::async_initialize(
-					backend.clone(),
-					runtime.clone(),
-					key_store.clone().into(),
-					metrics.clone(),
-					min_block_delta,
-					beefy_comms.gossip_validator.clone(),
-					&mut finality_notifications,
-					is_authority,
-				).fuse() => {
-					match builder_init_result {
-						Ok(builder) => break builder,
-						Err(e) => {
-							error!(target: LOG_TARGET, "🥩 Error: {:?}. Terminating.", e);
-							return
-						},
-					}
-				},
-				// Pump peer reports
-				_ = &mut beefy_comms.gossip_report_stream.next() => {
-					continue
-				},
-				// Pump gossip engine.
-				_ = &mut beefy_comms.gossip_engine => {
-					error!(target: LOG_TARGET, "🥩 Gossip engine has unexpectedly terminated.");
-					return
+		let worker_builder = futures::select! {
+			builder_init_result = BeefyWorkerBuilder::async_initialize(
+				backend.clone(),
+				runtime.clone(),
+				key_store.clone().into(),
+				metrics.clone(),
+				min_block_delta,
+				beefy_comms.gossip_validator.clone(),
+				&mut finality_notifications,
+				is_authority,
+			).fuse() => {
+				match builder_init_result {
+					Ok(builder) => builder,
+					Err(e) => {
+						error!(target: LOG_TARGET, "🥩 Error: {:?}. Terminating.", e);
+						return
+					},
 				}
+			},
+			// Pump gossip engine.
+			_ = &mut beefy_comms.gossip_engine => {
+				error!(target: LOG_TARGET, "🥩 Gossip engine has unexpectedly terminated.");
+				return
 			}
 		};