sc-consensus-beefy: pump gossip engine while waiting for initialization conditions (#3435)

As part of BEEFY worker/voter initialization the task waits for certain
chain and backend conditions to be fulfilled:
- BEEFY consensus enabled on-chain & GRANDPA best finalized higher than
on-chain BEEFY genesis block,
- backend has synced headers for BEEFY mandatory blocks between best
BEEFY and best GRANDPA.

During this waiting time, any messages gossiped on the BEEFY topic for
current chain get enqueued in the gossip engine, leading to RAM bloating
and output warning/error messages when the wait time is non-negligible
(like during a clean sync).

This PR adds logic to pump the gossip engine while waiting for other
things to make sure gossiped messages get consumed (practically
discarded until worker is fully initialized).

Also raises the warning threshold for enqueued messages from 10k to
100k. This is in line with the other gossip protocols on the node.

Fixes https://github.com/paritytech/polkadot-sdk/issues/3390

---------

Signed-off-by: Adrian Catangiu <adrian@parity.io>

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

@@ -31,7 +31,7 @@ use crate::{
 	import::BeefyBlockImport,
 	metrics::register_metrics,
 };
-use futures::{stream::Fuse, StreamExt};
+use futures::{stream::Fuse, FutureExt, StreamExt};
 use log::{debug, error, info, warn};
 use parking_lot::Mutex;
 use prometheus::Registry;
@@ -40,17 +40,21 @@ use sc_consensus::BlockImport;
 use sc_network::{NetworkRequest, NotificationService, ProtocolName};
 use sc_network_gossip::{GossipEngine, Network as GossipNetwork, Syncing as GossipSyncing};
 use sp_api::ProvideRuntimeApi;
-use sp_blockchain::{
-	Backend as BlockchainBackend, Error as ClientError, HeaderBackend, Result as ClientResult,
-};
+use sp_blockchain::{Backend as BlockchainBackend, HeaderBackend};
 use sp_consensus::{Error as ConsensusError, SyncOracle};
 use sp_consensus_beefy::{
-	ecdsa_crypto::AuthorityId, BeefyApi, MmrRootHash, PayloadProvider, ValidatorSet,
+	ecdsa_crypto::AuthorityId, BeefyApi, ConsensusLog, MmrRootHash, PayloadProvider, ValidatorSet,
+	BEEFY_ENGINE_ID,
 };
 use sp_keystore::KeystorePtr;
 use sp_mmr_primitives::MmrApi;
 use sp_runtime::traits::{Block, Header as HeaderT, NumberFor, Zero};
-use std::{collections::BTreeMap, marker::PhantomData, sync::Arc, time::Duration};
+use std::{
+	collections::{BTreeMap, VecDeque},
+	marker::PhantomData,
+	sync::Arc,
+	time::Duration,
+};
 
 mod aux_schema;
 mod error;
@@ -63,9 +67,19 @@ pub mod communication;
 pub mod import;
 pub mod justification;
 
+use crate::{
+	communication::{gossip::GossipValidator, peers::PeerReport},
+	justification::BeefyVersionedFinalityProof,
+	keystore::BeefyKeystore,
+	metrics::VoterMetrics,
+	round::Rounds,
+	worker::{BeefyWorker, PersistedState},
+};
 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)]
 mod tests;
@@ -209,6 +223,247 @@ pub struct BeefyParams<B: Block, BE, C, N, P, R, S> {
 	/// Handler for incoming BEEFY justifications requests from a remote peer.
 	pub on_demand_justifications_handler: BeefyJustifsRequestHandler<B, C>,
 }
+/// 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 gossip_engine: GossipEngine<B>,
+	pub gossip_validator: Arc<GossipValidator<B>>,
+	pub gossip_report_stream: TracingUnboundedReceiver<PeerReport>,
+	pub on_demand_justifications: OnDemandJustificationsEngine<B>,
+}
+
+/// Helper builder object for building [worker::BeefyWorker].
+///
+/// It has to do it in two steps: initialization and build, because the first step can sleep waiting
+/// for certain chain and backend conditions, and while sleeping we still need to pump the
+/// GossipEngine. Once initialization is done, the GossipEngine (and other pieces) are added to get
+/// the complete [worker::BeefyWorker] object.
+pub(crate) struct BeefyWorkerBuilder<B: Block, BE, RuntimeApi> {
+	// utilities
+	backend: Arc<BE>,
+	runtime: Arc<RuntimeApi>,
+	key_store: BeefyKeystore<AuthorityId>,
+	// voter metrics
+	metrics: Option<VoterMetrics>,
+	persisted_state: PersistedState<B>,
+}
+
+impl<B, BE, R> BeefyWorkerBuilder<B, BE, R>
+where
+	B: Block + codec::Codec,
+	BE: Backend<B>,
+	R: ProvideRuntimeApi<B>,
+	R::Api: BeefyApi<B, AuthorityId>,
+{
+	/// This will wait for the chain to enable BEEFY (if not yet enabled) and also wait for the
+	/// backend to sync all headers required by the voter to build a contiguous chain of mandatory
+	/// justifications. Then it builds the initial voter state using a combination of previously
+	/// persisted state in AUX DB and latest chain information/progress.
+	///
+	/// Returns a sane `BeefyWorkerBuilder` that can build the `BeefyWorker`.
+	pub async fn async_initialize(
+		backend: Arc<BE>,
+		runtime: Arc<R>,
+		key_store: BeefyKeystore<AuthorityId>,
+		metrics: Option<VoterMetrics>,
+		min_block_delta: u32,
+		gossip_validator: Arc<GossipValidator<B>>,
+		finality_notifications: &mut Fuse<FinalityNotifications<B>>,
+	) -> Result<Self, Error> {
+		// Wait for BEEFY pallet to be active before starting voter.
+		let (beefy_genesis, best_grandpa) =
+			wait_for_runtime_pallet(&*runtime, finality_notifications).await?;
+
+		let persisted_state = Self::load_or_init_state(
+			beefy_genesis,
+			best_grandpa,
+			min_block_delta,
+			backend.clone(),
+			runtime.clone(),
+			&key_store,
+			&metrics,
+		)
+		.await?;
+		// Update the gossip validator with the right starting round and set id.
+		persisted_state
+			.gossip_filter_config()
+			.map(|f| gossip_validator.update_filter(f))?;
+
+		Ok(BeefyWorkerBuilder { backend, runtime, key_store, metrics, persisted_state })
+	}
+
+	/// Takes rest of missing pieces as params and builds the `BeefyWorker`.
+	pub fn build<P, S>(
+		self,
+		payload_provider: P,
+		sync: Arc<S>,
+		comms: BeefyComms<B>,
+		links: BeefyVoterLinks<B>,
+		pending_justifications: BTreeMap<NumberFor<B>, BeefyVersionedFinalityProof<B>>,
+	) -> BeefyWorker<B, BE, P, R, S> {
+		BeefyWorker {
+			backend: self.backend,
+			runtime: self.runtime,
+			key_store: self.key_store,
+			metrics: self.metrics,
+			persisted_state: self.persisted_state,
+			payload_provider,
+			sync,
+			comms,
+			links,
+			pending_justifications,
+		}
+	}
+
+	// If no persisted state present, walk back the chain from first GRANDPA notification to either:
+	//  - latest BEEFY finalized block, or if none found on the way,
+	//  - BEEFY pallet genesis;
+	// Enqueue any BEEFY mandatory blocks (session boundaries) found on the way, for voter to
+	// finalize.
+	async fn init_state(
+		beefy_genesis: NumberFor<B>,
+		best_grandpa: <B as Block>::Header,
+		min_block_delta: u32,
+		backend: Arc<BE>,
+		runtime: Arc<R>,
+	) -> Result<PersistedState<B>, Error> {
+		let blockchain = backend.blockchain();
+
+		let beefy_genesis = runtime
+			.runtime_api()
+			.beefy_genesis(best_grandpa.hash())
+			.ok()
+			.flatten()
+			.filter(|genesis| *genesis == beefy_genesis)
+			.ok_or_else(|| Error::Backend("BEEFY pallet expected to be active.".into()))?;
+		// Walk back the imported blocks and initialize voter either, at the last block with
+		// a BEEFY justification, or at pallet genesis block; voter will resume from there.
+		let mut sessions = VecDeque::new();
+		let mut header = best_grandpa.clone();
+		let state = loop {
+			if let Some(true) = blockchain
+				.justifications(header.hash())
+				.ok()
+				.flatten()
+				.map(|justifs| justifs.get(BEEFY_ENGINE_ID).is_some())
+			{
+				debug!(
+					target: LOG_TARGET,
+					"🥩 Initialize BEEFY voter at last BEEFY finalized block: {:?}.",
+					*header.number()
+				);
+				let best_beefy = *header.number();
+				// If no session boundaries detected so far, just initialize new rounds here.
+				if sessions.is_empty() {
+					let active_set =
+						expect_validator_set(runtime.as_ref(), backend.as_ref(), &header).await?;
+					let mut rounds = Rounds::new(best_beefy, active_set);
+					// Mark the round as already finalized.
+					rounds.conclude(best_beefy);
+					sessions.push_front(rounds);
+				}
+				let state = PersistedState::checked_new(
+					best_grandpa,
+					best_beefy,
+					sessions,
+					min_block_delta,
+					beefy_genesis,
+				)
+				.ok_or_else(|| Error::Backend("Invalid BEEFY chain".into()))?;
+				break state
+			}
+
+			if *header.number() == beefy_genesis {
+				// We've reached BEEFY genesis, initialize voter here.
+				let genesis_set =
+					expect_validator_set(runtime.as_ref(), backend.as_ref(), &header).await?;
+				info!(
+					target: LOG_TARGET,
+					"🥩 Loading BEEFY voter state from genesis on what appears to be first startup. \
+					Starting voting rounds at block {:?}, genesis validator set {:?}.",
+					beefy_genesis,
+					genesis_set,
+				);
+
+				sessions.push_front(Rounds::new(beefy_genesis, genesis_set));
+				break PersistedState::checked_new(
+					best_grandpa,
+					Zero::zero(),
+					sessions,
+					min_block_delta,
+					beefy_genesis,
+				)
+				.ok_or_else(|| Error::Backend("Invalid BEEFY chain".into()))?
+			}
+
+			if let Some(active) = find_authorities_change::<B>(&header) {
+				debug!(
+					target: LOG_TARGET,
+					"🥩 Marking block {:?} as BEEFY Mandatory.",
+					*header.number()
+				);
+				sessions.push_front(Rounds::new(*header.number(), active));
+			}
+
+			// Move up the chain.
+			header = wait_for_parent_header(blockchain, header, HEADER_SYNC_DELAY).await?;
+		};
+
+		aux_schema::write_current_version(backend.as_ref())?;
+		aux_schema::write_voter_state(backend.as_ref(), &state)?;
+		Ok(state)
+	}
+
+	async fn load_or_init_state(
+		beefy_genesis: NumberFor<B>,
+		best_grandpa: <B as Block>::Header,
+		min_block_delta: u32,
+		backend: Arc<BE>,
+		runtime: Arc<R>,
+		key_store: &BeefyKeystore<AuthorityId>,
+		metrics: &Option<VoterMetrics>,
+	) -> Result<PersistedState<B>, Error> {
+		// Initialize voter state from AUX DB if compatible.
+		if let Some(mut state) = crate::aux_schema::load_persistent(backend.as_ref())?
+			// Verify state pallet genesis matches runtime.
+			.filter(|state| state.pallet_genesis() == beefy_genesis)
+		{
+			// Overwrite persisted state with current best GRANDPA block.
+			state.set_best_grandpa(best_grandpa.clone());
+			// Overwrite persisted data with newly provided `min_block_delta`.
+			state.set_min_block_delta(min_block_delta);
+			debug!(target: LOG_TARGET, "🥩 Loading BEEFY voter state from db: {:?}.", state);
+
+			// Make sure that all the headers that we need have been synced.
+			let mut new_sessions = vec![];
+			let mut header = best_grandpa.clone();
+			while *header.number() > state.best_beefy() {
+				if state.voting_oracle().can_add_session(*header.number()) {
+					if let Some(active) = find_authorities_change::<B>(&header) {
+						new_sessions.push((active, *header.number()));
+					}
+				}
+				header =
+					wait_for_parent_header(backend.blockchain(), header, HEADER_SYNC_DELAY).await?;
+			}
+
+			// Make sure we didn't miss any sessions during node restart.
+			for (validator_set, new_session_start) in new_sessions.drain(..).rev() {
+				debug!(
+					target: LOG_TARGET,
+					"🥩 Handling missed BEEFY session after node restart: {:?}.",
+					new_session_start
+				);
+				state.init_session_at(new_session_start, validator_set, key_store, metrics);
+			}
+			return Ok(state)
+		}
+
+		// No valid voter-state persisted, re-initialize from pallet genesis.
+		Self::init_state(beefy_genesis, best_grandpa, min_block_delta, backend, runtime).await
+	}
+}
 
 /// Start the BEEFY gadget.
 ///
@@ -277,7 +532,7 @@ pub async fn start_beefy_gadget<B, BE, C, N, P, R, S>(
 		known_peers,
 		prometheus_registry.clone(),
 	);
-	let mut beefy_comms = worker::BeefyComms {
+	let mut beefy_comms = BeefyComms {
 		gossip_engine,
 		gossip_validator,
 		gossip_report_stream,
@@ -287,57 +542,45 @@ pub async fn start_beefy_gadget<B, BE, C, N, P, R, S>(
 	// We re-create and re-run the worker in this loop in order to quickly reinit and resume after
 	// select recoverable errors.
 	loop {
-		// Wait for BEEFY pallet to be active before starting voter.
-		let (beefy_genesis, best_grandpa) = match wait_for_runtime_pallet(
-			&*runtime,
-			&mut beefy_comms.gossip_engine,
-			&mut finality_notifications,
-		)
-		.await
-		{
-			Ok(res) => res,
-			Err(e) => {
-				error!(target: LOG_TARGET, "Error: {:?}. Terminating.", e);
-				return
-			},
+		// 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,
+				).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 mut worker_base = worker::BeefyWorkerBase {
-			backend: backend.clone(),
-			runtime: runtime.clone(),
-			key_store: key_store.clone().into(),
-			metrics: metrics.clone(),
-			_phantom: Default::default(),
-		};
-
-		let persisted_state = match worker_base
-			.load_or_init_state(beefy_genesis, best_grandpa, min_block_delta)
-			.await
-		{
-			Ok(state) => state,
-			Err(e) => {
-				error!(target: LOG_TARGET, "Error: {:?}. Terminating.", e);
-				return
-			},
-		};
-		// Update the gossip validator with the right starting round and set id.
-		if let Err(e) = persisted_state
-			.gossip_filter_config()
-			.map(|f| beefy_comms.gossip_validator.update_filter(f))
-		{
-			error!(target: LOG_TARGET, "Error: {:?}. Terminating.", e);
-			return
-		}
-
-		let worker = worker::BeefyWorker {
-			base: worker_base,
-			payload_provider: payload_provider.clone(),
-			sync: sync.clone(),
-			comms: beefy_comms,
-			links: links.clone(),
-			pending_justifications: BTreeMap::new(),
-			persisted_state,
-		};
+		let worker = worker_builder.build(
+			payload_provider.clone(),
+			sync.clone(),
+			beefy_comms,
+			links.clone(),
+			BTreeMap::new(),
+		);
 
 		match futures::future::select(
 			Box::pin(worker.run(&mut block_import_justif, &mut finality_notifications)),
@@ -404,9 +647,8 @@ where
 /// Should be called only once during worker initialization.
 async fn wait_for_runtime_pallet<B, R>(
 	runtime: &R,
-	mut gossip_engine: &mut GossipEngine<B>,
 	finality: &mut Fuse<FinalityNotifications<B>>,
-) -> ClientResult<(NumberFor<B>, <B as Block>::Header)>
+) -> Result<(NumberFor<B>, <B as Block>::Header), Error>
 where
 	B: Block,
 	R: ProvideRuntimeApi<B>,
@@ -414,33 +656,24 @@ where
 {
 	info!(target: LOG_TARGET, "🥩 BEEFY gadget waiting for BEEFY pallet to become available...");
 	loop {
-		futures::select! {
-			notif = finality.next() => {
-				let notif = match notif {
-					Some(notif) => notif,
-					None => break
-				};
-				let at = notif.header.hash();
-				if let Some(start) = runtime.runtime_api().beefy_genesis(at).ok().flatten() {
-					if *notif.header.number() >= start {
-						// Beefy pallet available, return header for best grandpa at the time.
-						info!(
-							target: LOG_TARGET,
-							"🥩 BEEFY pallet available: block {:?} beefy genesis {:?}",
-							notif.header.number(), start
-						);
-						return Ok((start, notif.header))
-					}
-				}
-			},
-			_ = gossip_engine => {
-				break
+		let notif = finality.next().await.ok_or_else(|| {
+			let err_msg = "🥩 Finality stream has unexpectedly terminated.".into();
+			error!(target: LOG_TARGET, "{}", err_msg);
+			Error::Backend(err_msg)
+		})?;
+		let at = notif.header.hash();
+		if let Some(start) = runtime.runtime_api().beefy_genesis(at).ok().flatten() {
+			if *notif.header.number() >= start {
+				// Beefy pallet available, return header for best grandpa at the time.
+				info!(
+					target: LOG_TARGET,
+					"🥩 BEEFY pallet available: block {:?} beefy genesis {:?}",
+					notif.header.number(), start
+				);
+				return Ok((start, notif.header))
 			}
 		}
 	}
-	let err_msg = "🥩 Gossip engine has unexpectedly terminated.".into();
-	error!(target: LOG_TARGET, "{}", err_msg);
-	Err(ClientError::Backend(err_msg))
 }
 
 /// Provides validator set active `at_header`. It tries to get it from state, otherwise falls
@@ -474,7 +707,7 @@ where
 		if let Ok(Some(active)) = runtime.runtime_api().validator_set(header.hash()) {
 			return Ok(active)
 		} else {
-			match worker::find_authorities_change::<B>(&header) {
+			match find_authorities_change::<B>(&header) {
 				Some(active) => return Ok(active),
 				// Move up the chain. Ultimately we'll get it from chain genesis state, or error out
 				// there.
@@ -486,3 +719,18 @@ where
 		}
 	}
 }
+
+/// Scan the `header` digest log for a BEEFY validator set change. Return either the new
+/// validator set or `None` in case no validator set change has been signaled.
+pub(crate) fn find_authorities_change<B>(header: &B::Header) -> Option<ValidatorSet<AuthorityId>>
+where
+	B: Block,
+{
+	let id = OpaqueDigestItemId::Consensus(&BEEFY_ENGINE_ID);
+
+	let filter = |log: ConsensusLog<AuthorityId>| match log {
+		ConsensusLog::AuthoritiesChange(validator_set) => Some(validator_set),
+		_ => None,
+	};
+	header.digest().convert_first(|l| l.try_to(id).and_then(filter))
+}