// Copyright 2022 Parity Technologies (UK) Ltd. // This file is part of Cumulus. // Cumulus is free software: you can redistribute it and/or modify // it under the terms of the GNU General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // Cumulus is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU General Public License for more details. // You should have received a copy of the GNU General Public License // along with Cumulus. If not, see . use polkadot_core_primitives::{Block, Hash}; use polkadot_service::{BlockT, NumberFor}; use polkadot_node_network_protocol::PeerId; use sc_network::{NetworkService, SyncState}; use sc_client_api::HeaderBackend; use sc_consensus::{BlockImportError, BlockImportStatus, JustificationSyncLink, Link}; use sc_network_common::{ config::{ NonDefaultSetConfig, NonReservedPeerMode, NotificationHandshake, ProtocolId, SetConfig, }, protocol::role::Roles, service::NetworkSyncForkRequest, sync::{ message::{BlockAnnouncesHandshake, BlockRequest}, BadPeer, Metrics, OnBlockData, PollBlockAnnounceValidation, SyncStatus, }, }; use sc_service::{error::Error, Configuration, NetworkStarter, SpawnTaskHandle}; use std::{iter, sync::Arc}; use crate::BlockChainRpcClient; pub(crate) struct BuildCollatorNetworkParams<'a> { /// The service configuration. pub config: &'a Configuration, /// A shared client returned by `new_full_parts`. pub client: Arc, /// A handle for spawning tasks. pub spawn_handle: SpawnTaskHandle, /// Genesis hash pub genesis_hash: Hash, } /// Build the network service, the network status sinks and an RPC sender. pub(crate) fn build_collator_network( params: BuildCollatorNetworkParams, ) -> Result<(Arc>, NetworkStarter), Error> { let BuildCollatorNetworkParams { config, client, spawn_handle, genesis_hash } = params; let protocol_id = config.protocol_id(); let chain_sync = DummyChainSync; let block_announce_config = chain_sync.get_block_announce_proto_config::( protocol_id.clone(), &None, Roles::from(&config.role), client.info().best_number, client.info().best_hash, genesis_hash, ); let network_params = sc_network::config::Params { role: config.role.clone(), executor: { let spawn_handle = Clone::clone(&spawn_handle); Box::new(move |fut| { spawn_handle.spawn("libp2p-node", Some("networking"), fut); }) }, fork_id: None, chain_sync: Box::new(chain_sync), network_config: config.network.clone(), chain: client.clone(), protocol_id, metrics_registry: config.prometheus_config.as_ref().map(|config| config.registry.clone()), block_announce_config, chain_sync_service: Box::new(DummyChainSyncService::(Default::default())), request_response_protocol_configs: Vec::new(), }; let network_worker = sc_network::NetworkWorker::new(network_params)?; let network_service = network_worker.service().clone(); let (network_start_tx, network_start_rx) = futures::channel::oneshot::channel(); // The network worker is responsible for gathering all network messages and processing // them. This is quite a heavy task, and at the time of the writing of this comment it // frequently happens that this future takes several seconds or in some situations // even more than a minute until it has processed its entire queue. This is clearly an // issue, and ideally we would like to fix the network future to take as little time as // possible, but we also take the extra harm-prevention measure to execute the networking // future using `spawn_blocking`. spawn_handle.spawn_blocking("network-worker", Some("networking"), async move { if network_start_rx.await.is_err() { tracing::warn!( "The NetworkStart returned as part of `build_network` has been silently dropped" ); // This `return` might seem unnecessary, but we don't want to make it look like // everything is working as normal even though the user is clearly misusing the API. return } network_worker.await }); let network_starter = NetworkStarter::new(network_start_tx); Ok((network_service, network_starter)) } /// Empty ChainSync shell. Syncing code is not necessary for /// the minimal node, but network currently requires it. So /// we provide a noop implementation. struct DummyChainSync; impl DummyChainSync { pub fn get_block_announce_proto_config( &self, protocol_id: ProtocolId, fork_id: &Option, roles: Roles, best_number: NumberFor, best_hash: B::Hash, genesis_hash: B::Hash, ) -> NonDefaultSetConfig { let block_announces_protocol = { let genesis_hash = genesis_hash.as_ref(); if let Some(ref fork_id) = fork_id { format!( "/{}/{}/block-announces/1", array_bytes::bytes2hex("", genesis_hash), fork_id ) } else { format!("/{}/block-announces/1", array_bytes::bytes2hex("", genesis_hash)) } }; NonDefaultSetConfig { notifications_protocol: block_announces_protocol.into(), fallback_names: iter::once( format!("/{}/block-announces/1", protocol_id.as_ref()).into(), ) .collect(), max_notification_size: 1024 * 1024, handshake: Some(NotificationHandshake::new(BlockAnnouncesHandshake::::build( roles, best_number, best_hash, genesis_hash, ))), // NOTE: `set_config` will be ignored by `protocol.rs` as the block announcement // protocol is still hardcoded into the peerset. set_config: SetConfig { in_peers: 0, out_peers: 0, reserved_nodes: Vec::new(), non_reserved_mode: NonReservedPeerMode::Deny, }, } } } impl sc_network_common::sync::ChainSync for DummyChainSync { fn peer_info(&self, _who: &PeerId) -> Option> { None } fn status(&self) -> sc_network_common::sync::SyncStatus { SyncStatus { state: SyncState::Idle, best_seen_block: None, num_peers: 0, queued_blocks: 0, state_sync: None, warp_sync: None, } } fn num_sync_requests(&self) -> usize { 0 } fn num_downloaded_blocks(&self) -> usize { 0 } fn num_peers(&self) -> usize { 0 } fn new_peer( &mut self, _who: PeerId, _best_hash: ::Hash, _best_number: polkadot_service::NumberFor, ) -> Result< Option>, sc_network_common::sync::BadPeer, > { Ok(None) } fn update_chain_info( &mut self, _best_hash: &::Hash, _best_number: polkadot_service::NumberFor, ) { } fn request_justification( &mut self, _hash: &::Hash, _number: polkadot_service::NumberFor, ) { } fn clear_justification_requests(&mut self) {} fn set_sync_fork_request( &mut self, _peers: Vec, _hash: &::Hash, _number: polkadot_service::NumberFor, ) { } fn on_block_data( &mut self, _who: &PeerId, _request: Option>, _response: sc_network_common::sync::message::BlockResponse, ) -> Result, sc_network_common::sync::BadPeer> { unimplemented!("Not supported on the RPC collator") } fn on_block_justification( &mut self, _who: PeerId, _response: sc_network_common::sync::message::BlockResponse, ) -> Result, sc_network_common::sync::BadPeer> { unimplemented!("Not supported on the RPC collator") } fn on_justification_import( &mut self, _hash: ::Hash, _number: polkadot_service::NumberFor, _success: bool, ) { } fn on_block_finalized( &mut self, _hash: &::Hash, _number: polkadot_service::NumberFor, ) { } fn push_block_announce_validation( &mut self, _who: PeerId, _hash: ::Hash, _announce: sc_network_common::sync::message::BlockAnnounce<::Header>, _is_best: bool, ) { } fn poll_block_announce_validation( &mut self, _cx: &mut std::task::Context, ) -> std::task::Poll::Header>> { std::task::Poll::Pending } fn peer_disconnected(&mut self, _who: &PeerId) {} fn metrics(&self) -> sc_network_common::sync::Metrics { Metrics { queued_blocks: 0, fork_targets: 0, justifications: sc_network_common::sync::metrics::Metrics { pending_requests: 0, active_requests: 0, importing_requests: 0, failed_requests: 0, }, } } fn block_response_into_blocks( &self, _request: &sc_network_common::sync::message::BlockRequest, _response: sc_network_common::sync::OpaqueBlockResponse, ) -> Result>, String> { unimplemented!("Not supported on the RPC collator") } fn poll( &mut self, _cx: &mut std::task::Context, ) -> std::task::Poll> { std::task::Poll::Pending } fn send_block_request(&mut self, _who: PeerId, _request: BlockRequest) { unimplemented!("Not supported on the RPC collator") } fn num_active_peers(&self) -> usize { 0 } fn process_block_response_data(&mut self, _blocks_to_import: Result, BadPeer>) {} } struct DummyChainSyncService(std::marker::PhantomData); impl NetworkSyncForkRequest> for DummyChainSyncService { fn set_sync_fork_request(&self, _peers: Vec, _hash: B::Hash, _number: NumberFor) {} } impl JustificationSyncLink for DummyChainSyncService { fn request_justification(&self, _hash: &B::Hash, _number: NumberFor) {} fn clear_justification_requests(&self) {} } impl Link for DummyChainSyncService { fn blocks_processed( &mut self, _imported: usize, _count: usize, _results: Vec<(Result>, BlockImportError>, B::Hash)>, ) { } fn justification_imported( &mut self, _who: PeerId, _hash: &B::Hash, _number: NumberFor, _success: bool, ) { } fn request_justification(&mut self, _hash: &B::Hash, _number: NumberFor) {} }