client/beefy: initialize OnDemandJustificationsEngine

client/beefy/src/communication/request_response/incoming_handler.rs

@@ -17,9 +17,10 @@
 //! Helper for handling (i.e. answering) BEEFY justifications requests from a remote peer.
 
 use beefy_primitives::BEEFY_ENGINE_ID;
-use log::debug;
+use log::{debug, trace};
 use sc_client_api::BlockBackend;
 use sc_network::{config as netconfig, config::RequestResponseConfig};
+use sc_network_common::protocol::ProtocolName;
 use sp_runtime::{generic::BlockId, traits::Block};
 use std::{marker::PhantomData, sync::Arc};
 
@@ -29,16 +30,16 @@ use crate::communication::request_response::{
 
 /// Handler for incoming BEEFY justifications requests from a remote peer.
 pub struct BeefyJustifsRequestHandler<B, Client> {
-	request_receiver: IncomingRequestReceiver,
-	/// Blockchain client.
-	client: Arc<Client>,
-	_block: PhantomData<B>,
+	pub(crate) request_receiver: IncomingRequestReceiver,
+	pub(crate) justif_protocol_name: ProtocolName,
+	pub(crate) client: Arc<Client>,
+	pub(crate) _block: PhantomData<B>,
 }
 
 impl<B, Client> BeefyJustifsRequestHandler<B, Client>
 where
 	B: Block,
-	Client: BlockBackend<B> + Send + Sync + 'static,
+	Client: BlockBackend<B> + Send + Sync,
 {
 	/// Create a new [`BeefyJustifsRequestHandler`].
 	pub fn new(fork_id: Option<&str>, client: Arc<Client>) -> (Self, RequestResponseConfig) {
@@ -48,8 +49,13 @@ where
 			.flatten()
 			.expect("Genesis block exists; qed");
 		let (request_receiver, config) = justif_protocol_config(genesis_hash, fork_id);
+		let justif_protocol_name = config.name.clone();
 
-		(Self { client, request_receiver, _block: PhantomData }, config)
+		(Self { request_receiver, justif_protocol_name, client, _block: PhantomData }, config)
+	}
+
+	pub fn protocol_name(&self) -> ProtocolName {
+		self.justif_protocol_name.clone()
 	}
 
 	async fn handle_request(&self, request: IncomingRequest<B>) -> Result<(), Error> {
@@ -76,20 +82,22 @@ where
 
 	/// Run [`BeefyJustifsRequestHandler`].
 	pub async fn run(mut self) {
+		trace!(target: "beefy::sync", "🥩 Running BeefyJustifsRequestHandler");
+
 		while let Ok(request) = self.request_receiver.recv(|| vec![]).await {
 			let peer = request.peer;
 			match self.handle_request(request).await {
 				Ok(()) => {
 					debug!(
 						target: "beefy::sync",
-						"Handled BEEFY justification request from {}.", peer
+						"🥩 Handled BEEFY justification request from {}.", peer
 					)
 				},
 				Err(e) => {
 					// TODO: handle reputation changes here
 					debug!(
 						target: "beefy::sync",
-						"Failed to handle BEEFY justification request from {}: {}", peer, e,
+						"🥩 Failed to handle BEEFY justification request from {}: {}", peer, e,
 					)
 				},
 			}

client/beefy/src/communication/request_response/mod.rs

@@ -44,7 +44,7 @@ const JUSTIF_REQUEST_TIMEOUT: Duration = Duration::from_secs(3);
 ///
 /// Returns a receiver for messages received on this protocol and the requested
 /// `ProtocolConfig`.
-pub fn justif_protocol_config<Hash: AsRef<[u8]>>(
+pub(crate) fn justif_protocol_config<Hash: AsRef<[u8]>>(
 	genesis_hash: Hash,
 	fork_id: Option<&str>,
 ) -> (IncomingRequestReceiver, RequestResponseConfig) {
@@ -73,7 +73,7 @@ pub struct JustificationRequest<B: Block> {
 
 /// A request coming in, including a sender for sending responses.
 #[derive(Debug)]
-pub struct IncomingRequest<B: Block> {
+pub(crate) struct IncomingRequest<B: Block> {
 	/// `PeerId` of sending peer.
 	pub peer: PeerId,
 	/// The sent request.
@@ -123,10 +123,10 @@ impl<B: Block> IncomingRequest<B> {
 	}
 }
 
-/// Receiver for incoming requests.
+/// Receiver for incoming BEEFY justifications requests.
 ///
 /// Takes care of decoding and handling of invalid encoded requests.
-pub struct IncomingRequestReceiver {
+pub(crate) struct IncomingRequestReceiver {
 	raw: mpsc::Receiver<netconfig::IncomingRequest>,
 }
 

client/beefy/src/lib.rs

@@ -19,17 +19,18 @@
 use beefy_primitives::{BeefyApi, MmrRootHash};
 use parking_lot::Mutex;
 use prometheus::Registry;
-use sc_client_api::{Backend, BlockchainEvents, Finalizer};
+use sc_client_api::{Backend, BlockBackend, BlockchainEvents, Finalizer};
 use sc_consensus::BlockImport;
 use sc_network::ProtocolName;
+use sc_network_common::service::NetworkRequest;
 use sc_network_gossip::Network as GossipNetwork;
 use sp_api::ProvideRuntimeApi;
 use sp_blockchain::HeaderBackend;
 use sp_consensus::{Error as ConsensusError, SyncOracle};
 use sp_keystore::SyncCryptoStorePtr;
 use sp_mmr_primitives::MmrApi;
 use sp_runtime::traits::Block;
-use std::sync::Arc;
+use std::{marker::PhantomData, sync::Arc};
 
 mod error;
 mod keystore;
@@ -51,14 +52,14 @@ use crate::{
 			BeefyVersionedFinalityProofStream,
 		},
 		peers::KnownPeers,
+		request_response::outgoing_request::OnDemandJustificationsEngine,
 	},
 	import::BeefyBlockImport,
 };
 
 pub use communication::beefy_protocol_name::{
 	gossip_protocol_name, justifications_protocol_name as justifs_protocol_name,
 };
-use sc_network_common::service::NetworkRequest;
 
 /// A convenience BEEFY client trait that defines all the type bounds a BEEFY client
 /// has to satisfy. Ideally that should actually be a trait alias. Unfortunately as
@@ -149,6 +150,24 @@ where
 	(import, voter_links, rpc_links)
 }
 
+/// BEEFY gadget network parameters.
+pub struct BeefyNetworkParams<B, N>
+where
+	B: Block,
+	N: GossipNetwork<B> + NetworkRequest + Clone + SyncOracle + Send + Sync + 'static,
+{
+	/// Network implementing gossip, requests and sync-oracle.
+	pub network: N,
+	/// Chain specific BEEFY gossip protocol name. See
+	/// [`beefy_protocol_name::gossip_protocol_name`].
+	pub gossip_protocol_name: ProtocolName,
+	/// Chain specific BEEFY on-demand justifications protocol name. See
+	/// [`beefy_protocol_name::justifications_protocol_name`].
+	pub justifications_protocol_name: ProtocolName,
+
+	_phantom: PhantomData<B>,
+}
+
 /// BEEFY gadget initialization parameters.
 pub struct BeefyParams<B, BE, C, N, R>
 where
@@ -167,15 +186,12 @@ where
 	pub runtime: Arc<R>,
 	/// Local key store
 	pub key_store: Option<SyncCryptoStorePtr>,
-	/// Gossip network
-	pub network: N,
+	/// BEEFY voter network params
+	pub network_params: BeefyNetworkParams<B, N>,
 	/// Minimal delta between blocks, BEEFY should vote for
 	pub min_block_delta: u32,
 	/// Prometheus metric registry
 	pub prometheus_registry: Option<Registry>,
-	/// Chain specific BEEFY gossip protocol name. See
-	/// [`beefy_protocol_name::gossip_protocol_name`].
-	pub protocol_name: ProtocolName,
 	/// Links between the block importer, the background voter and the RPC layer.
 	pub links: BeefyVoterLinks<B>,
 }
@@ -187,7 +203,7 @@ pub async fn start_beefy_gadget<B, BE, C, N, R>(beefy_params: BeefyParams<B, BE,
 where
 	B: Block,
 	BE: Backend<B>,
-	C: Client<B, BE>,
+	C: Client<B, BE> + BlockBackend<B>,
 	R: ProvideRuntimeApi<B>,
 	R::Api: BeefyApi<B> + MmrApi<B, MmrRootHash>,
 	N: GossipNetwork<B> + NetworkRequest + Clone + SyncOracle + Send + Sync + 'static,
@@ -197,23 +213,31 @@ where
 		backend,
 		runtime,
 		key_store,
-		network,
+		network_params,
 		min_block_delta,
 		prometheus_registry,
-		protocol_name,
 		links,
 	} = beefy_params;
 
+	let BeefyNetworkParams { network, gossip_protocol_name, justifications_protocol_name, .. } =
+		network_params;
+
 	let known_peers = Arc::new(Mutex::new(KnownPeers::new()));
 	let gossip_validator =
 		Arc::new(communication::gossip::GossipValidator::new(known_peers.clone()));
 	let gossip_engine = sc_network_gossip::GossipEngine::new(
 		network.clone(),
-		protocol_name,
+		gossip_protocol_name,
 		gossip_validator.clone(),
 		None,
 	);
 
+	let on_demand_justifications = OnDemandJustificationsEngine::new(
+		network.clone(),
+		runtime.clone(),
+		justifications_protocol_name,
+	);
+
 	let metrics =
 		prometheus_registry.as_ref().map(metrics::Metrics::register).and_then(
 			|result| match result {
@@ -237,6 +261,7 @@ where
 		known_peers,
 		gossip_engine,
 		gossip_validator,
+		on_demand_justifications,
 		links,
 		metrics,
 		min_block_delta,

client/beefy/src/worker.rs

@@ -192,6 +192,7 @@ pub(crate) struct WorkerParams<B: Block, BE, C, R, N> {
 	pub known_peers: Arc<Mutex<KnownPeers<B>>>,
 	pub gossip_engine: GossipEngine<B>,
 	pub gossip_validator: Arc<GossipValidator<B>>,
+	pub on_demand_justifications: OnDemandJustificationsEngine<B, N, R>,
 	pub links: BeefyVoterLinks<B>,
 	pub metrics: Option<Metrics>,
 	pub min_block_delta: u32,
@@ -255,6 +256,7 @@ where
 			network,
 			gossip_engine,
 			gossip_validator,
+			on_demand_justifications,
 			known_peers,
 			links,
 			metrics,
@@ -266,13 +268,6 @@ where
 			.expect_header(BlockId::number(backend.blockchain().info().finalized_number))
 			.expect("latest block always has header available; qed.");
 
-		let on_demand_justifications = OnDemandJustificationsEngine::new(
-			network.clone(),
-			runtime.clone(),
-			// FIXME: use right protocol name.
-			"TODO: FIXME: proto-name-here".into(),
-		);
-
 		BeefyWorker {
 			client: client.clone(),
 			backend,