// This file is part of Substrate. // Copyright (C) 2017-2020 Parity Technologies (UK) Ltd. // SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 // This program 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. // This program 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 this program. If not, see . use crate::{ Service, NetworkStatus, NetworkState, error::Error, DEFAULT_PROTOCOL_ID, MallocSizeOfWasm, start_rpc_servers, build_network_future, TransactionPoolAdapter, TaskManager, SpawnTaskHandle, status_sinks, metrics::MetricsService, client::{light, Client, ClientConfig}, config::{Configuration, KeystoreConfig, PrometheusConfig, OffchainWorkerConfig}, }; use sc_client_api::{ self, light::RemoteBlockchain, execution_extensions::ExtensionsFactory, ExecutorProvider, CallExecutor, ForkBlocks, BadBlocks, CloneableSpawn, UsageProvider, backend::RemoteBackend, }; use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedSender, TracingUnboundedReceiver}; use sc_chain_spec::get_extension; use sp_consensus::{ block_validation::{BlockAnnounceValidator, DefaultBlockAnnounceValidator}, import_queue::ImportQueue, }; use futures::{ Future, FutureExt, StreamExt, future::ready, }; use jsonrpc_pubsub::manager::SubscriptionManager; use sc_keystore::Store as Keystore; use log::{info, warn, error}; use sc_network::config::{Role, FinalityProofProvider, OnDemand, BoxFinalityProofRequestBuilder}; use sc_network::NetworkService; use parking_lot::{Mutex, RwLock}; use sp_runtime::generic::BlockId; use sp_runtime::traits::{ Block as BlockT, NumberFor, SaturatedConversion, HashFor, }; use sp_api::ProvideRuntimeApi; use sc_executor::{NativeExecutor, NativeExecutionDispatch, RuntimeInfo}; use std::{ collections::HashMap, io::{Read, Write, Seek}, marker::PhantomData, sync::Arc, pin::Pin }; use wasm_timer::SystemTime; use sc_telemetry::{telemetry, SUBSTRATE_INFO}; use sp_transaction_pool::{LocalTransactionPool, MaintainedTransactionPool}; use prometheus_endpoint::Registry; use sc_client_db::{Backend, DatabaseSettings}; use sp_core::traits::CodeExecutor; use sp_runtime::BuildStorage; use sc_client_api::execution_extensions::ExecutionExtensions; use sp_core::storage::Storage; pub type BackgroundTask = Pin + Send>>; /// Aggregator for the components required to build a service. /// /// # Usage /// /// Call [`ServiceBuilder::new_full`] or [`ServiceBuilder::new_light`], then call the various /// `with_` methods to add the required components that you built yourself: /// /// - [`with_select_chain`](ServiceBuilder::with_select_chain) /// - [`with_import_queue`](ServiceBuilder::with_import_queue) /// - [`with_finality_proof_provider`](ServiceBuilder::with_finality_proof_provider) /// - [`with_transaction_pool`](ServiceBuilder::with_transaction_pool) /// /// After this is done, call [`build`](ServiceBuilder::build) to construct the service. /// /// The order in which the `with_*` methods are called doesn't matter, as the correct binding of /// generics is done when you call `build`. /// pub struct ServiceBuilder { config: Configuration, pub (crate) client: Arc, backend: Arc, task_manager: TaskManager, keystore: Arc>, fetcher: Option, select_chain: Option, pub (crate) import_queue: TImpQu, finality_proof_request_builder: Option, finality_proof_provider: Option, transaction_pool: Arc, rpc_extensions_builder: Box + Send>, remote_backend: Option>>, marker: PhantomData<(TBl, TRtApi)>, block_announce_validator_builder: Option) -> Box + Send> + Send>>, informant_prefix: String, } /// A utility trait for building an RPC extension given a `DenyUnsafe` instance. /// This is useful since at service definition time we don't know whether the /// specific interface where the RPC extension will be exposed is safe or not. /// This trait allows us to lazily build the RPC extension whenever we bind the /// service to an interface. pub trait RpcExtensionBuilder { /// The type of the RPC extension that will be built. type Output: sc_rpc::RpcExtension; /// Returns an instance of the RPC extension for a particular `DenyUnsafe` /// value, e.g. the RPC extension might not expose some unsafe methods. fn build(&self, deny: sc_rpc::DenyUnsafe) -> Self::Output; } impl RpcExtensionBuilder for F where F: Fn(sc_rpc::DenyUnsafe) -> R, R: sc_rpc::RpcExtension, { type Output = R; fn build(&self, deny: sc_rpc::DenyUnsafe) -> Self::Output { (*self)(deny) } } /// A utility struct for implementing an `RpcExtensionBuilder` given a cloneable /// `RpcExtension`, the resulting builder will simply ignore the provided /// `DenyUnsafe` instance and return a static `RpcExtension` instance. struct NoopRpcExtensionBuilder(R); impl RpcExtensionBuilder for NoopRpcExtensionBuilder where R: Clone + sc_rpc::RpcExtension, { type Output = R; fn build(&self, _deny: sc_rpc::DenyUnsafe) -> Self::Output { self.0.clone() } } impl From for NoopRpcExtensionBuilder where R: sc_rpc::RpcExtension, { fn from(e: R) -> NoopRpcExtensionBuilder { NoopRpcExtensionBuilder(e) } } /// Full client type. pub type TFullClient = Client< TFullBackend, TFullCallExecutor, TBl, TRtApi, >; /// Full client backend type. pub type TFullBackend = sc_client_db::Backend; /// Full client call executor type. pub type TFullCallExecutor = crate::client::LocalCallExecutor< sc_client_db::Backend, NativeExecutor, >; /// Light client type. pub type TLightClient = Client< TLightBackend, TLightCallExecutor, TBl, TRtApi, >; /// Light client backend type. pub type TLightBackend = sc_light::Backend< sc_client_db::light::LightStorage, HashFor, >; /// Light call executor type. pub type TLightCallExecutor = sc_light::GenesisCallExecutor< sc_light::Backend< sc_client_db::light::LightStorage, HashFor >, crate::client::LocalCallExecutor< sc_light::Backend< sc_client_db::light::LightStorage, HashFor >, NativeExecutor >, >; type TFullParts = ( TFullClient, Arc>, Arc>, TaskManager, ); /// Creates a new full client for the given config. pub fn new_full_client( config: &Configuration, ) -> Result, Error> where TBl: BlockT, TExecDisp: NativeExecutionDispatch + 'static, { new_full_parts(config).map(|parts| parts.0) } fn new_full_parts( config: &Configuration, ) -> Result, Error> where TBl: BlockT, TExecDisp: NativeExecutionDispatch + 'static, { let keystore = match &config.keystore { KeystoreConfig::Path { path, password } => Keystore::open( path.clone(), password.clone() )?, KeystoreConfig::InMemory => Keystore::new_in_memory(), }; let task_manager = { let registry = config.prometheus_config.as_ref().map(|cfg| &cfg.registry); TaskManager::new(config.task_executor.clone(), registry)? }; let executor = NativeExecutor::::new( config.wasm_method, config.default_heap_pages, config.max_runtime_instances, ); let chain_spec = &config.chain_spec; let fork_blocks = get_extension::>(chain_spec.extensions()) .cloned() .unwrap_or_default(); let bad_blocks = get_extension::>(chain_spec.extensions()) .cloned() .unwrap_or_default(); let (client, backend) = { let db_config = sc_client_db::DatabaseSettings { state_cache_size: config.state_cache_size, state_cache_child_ratio: config.state_cache_child_ratio.map(|v| (v, 100)), pruning: config.pruning.clone(), source: config.database.clone(), }; let extensions = sc_client_api::execution_extensions::ExecutionExtensions::new( config.execution_strategies.clone(), Some(keystore.clone()), ); new_client( db_config, executor, chain_spec.as_storage_builder(), fork_blocks, bad_blocks, extensions, Box::new(task_manager.spawn_handle()), config.prometheus_config.as_ref().map(|config| config.registry.clone()), ClientConfig { offchain_worker_enabled : config.offchain_worker.enabled , offchain_indexing_api: config.offchain_worker.indexing_enabled, }, )? }; Ok((client, backend, keystore, task_manager)) } /// Create an instance of db-backed client. pub fn new_client( settings: DatabaseSettings, executor: E, genesis_storage: &dyn BuildStorage, fork_blocks: ForkBlocks, bad_blocks: BadBlocks, execution_extensions: ExecutionExtensions, spawn_handle: Box, prometheus_registry: Option, config: ClientConfig, ) -> Result<( crate::client::Client< Backend, crate::client::LocalCallExecutor, E>, Block, RA, >, Arc>, ), sp_blockchain::Error, > where Block: BlockT, E: CodeExecutor + RuntimeInfo, { const CANONICALIZATION_DELAY: u64 = 4096; let backend = Arc::new(Backend::new(settings, CANONICALIZATION_DELAY)?); let executor = crate::client::LocalCallExecutor::new(backend.clone(), executor, spawn_handle, config.clone()); Ok(( crate::client::Client::new( backend.clone(), executor, genesis_storage, fork_blocks, bad_blocks, execution_extensions, prometheus_registry, config, )?, backend, )) } impl ServiceBuilder<(), (), (), (), (), (), (), (), (), (), ()> { /// Start the service builder with a configuration. pub fn new_full( config: Configuration, ) -> Result, Arc>, (), (), BoxFinalityProofRequestBuilder, Arc>, (), (), TFullBackend, >, Error> { let (client, backend, keystore, task_manager) = new_full_parts(&config)?; let client = Arc::new(client); Ok(ServiceBuilder { config, client, backend, keystore, task_manager, fetcher: None, select_chain: None, import_queue: (), finality_proof_request_builder: None, finality_proof_provider: None, transaction_pool: Arc::new(()), rpc_extensions_builder: Box::new(|_| ()), remote_backend: None, block_announce_validator_builder: None, informant_prefix: Default::default(), marker: PhantomData, }) } /// Start the service builder with a configuration. pub fn new_light( config: Configuration, ) -> Result, Arc>, (), (), BoxFinalityProofRequestBuilder, Arc>, (), (), TLightBackend, >, Error> { let task_manager = { let registry = config.prometheus_config.as_ref().map(|cfg| &cfg.registry); TaskManager::new(config.task_executor.clone(), registry)? }; let keystore = match &config.keystore { KeystoreConfig::Path { path, password } => Keystore::open( path.clone(), password.clone() )?, KeystoreConfig::InMemory => Keystore::new_in_memory(), }; let executor = NativeExecutor::::new( config.wasm_method, config.default_heap_pages, config.max_runtime_instances, ); let db_storage = { let db_settings = sc_client_db::DatabaseSettings { state_cache_size: config.state_cache_size, state_cache_child_ratio: config.state_cache_child_ratio.map(|v| (v, 100)), pruning: config.pruning.clone(), source: config.database.clone(), }; sc_client_db::light::LightStorage::new(db_settings)? }; let light_blockchain = sc_light::new_light_blockchain(db_storage); let fetch_checker = Arc::new( sc_light::new_fetch_checker::<_, TBl, _>( light_blockchain.clone(), executor.clone(), Box::new(task_manager.spawn_handle()), ), ); let fetcher = Arc::new(sc_network::config::OnDemand::new(fetch_checker)); let backend = sc_light::new_light_backend(light_blockchain); let remote_blockchain = backend.remote_blockchain(); let client = Arc::new(light::new_light( backend.clone(), config.chain_spec.as_storage_builder(), executor, Box::new(task_manager.spawn_handle()), config.prometheus_config.as_ref().map(|config| config.registry.clone()), )?); Ok(ServiceBuilder { config, client, backend, task_manager, keystore, fetcher: Some(fetcher.clone()), select_chain: None, import_queue: (), finality_proof_request_builder: None, finality_proof_provider: None, transaction_pool: Arc::new(()), rpc_extensions_builder: Box::new(|_| ()), remote_backend: Some(remote_blockchain), block_announce_validator_builder: None, informant_prefix: Default::default(), marker: PhantomData, }) } } impl ServiceBuilder< TBl, TRtApi, TCl, TFchr, TSc, TImpQu, TFprb, TFpp, TExPool, TRpc, Backend > { /// Returns a reference to the configuration that was stored in this builder. pub fn config(&self) -> &Configuration { &self.config } /// Returns a reference to the optional prometheus registry that was stored in this builder. pub fn prometheus_registry(&self) -> Option<&Registry> { self.config.prometheus_config.as_ref().map(|config| &config.registry) } /// Returns a reference to the client that was stored in this builder. pub fn client(&self) -> &Arc { &self.client } /// Returns a reference to the backend that was used in this builder. pub fn backend(&self) -> &Arc { &self.backend } /// Returns a reference to the select-chain that was stored in this builder. pub fn select_chain(&self) -> Option<&TSc> { self.select_chain.as_ref() } /// Returns a reference to the keystore pub fn keystore(&self) -> Arc> { self.keystore.clone() } /// Returns a reference to the transaction pool stored in this builder pub fn pool(&self) -> Arc { self.transaction_pool.clone() } /// Returns a reference to the fetcher, only available if builder /// was created with `new_light`. pub fn fetcher(&self) -> Option where TFchr: Clone { self.fetcher.clone() } /// Returns a reference to the remote_backend, only available if builder /// was created with `new_light`. pub fn remote_backend(&self) -> Option>> { self.remote_backend.clone() } /// Defines which head-of-chain strategy to use. pub fn with_opt_select_chain( self, select_chain_builder: impl FnOnce( &Configuration, &Arc, ) -> Result, Error> ) -> Result, Error> { let select_chain = select_chain_builder(&self.config, &self.backend)?; Ok(ServiceBuilder { config: self.config, client: self.client, backend: self.backend, task_manager: self.task_manager, keystore: self.keystore, fetcher: self.fetcher, select_chain, import_queue: self.import_queue, finality_proof_request_builder: self.finality_proof_request_builder, finality_proof_provider: self.finality_proof_provider, transaction_pool: self.transaction_pool, rpc_extensions_builder: self.rpc_extensions_builder, remote_backend: self.remote_backend, block_announce_validator_builder: self.block_announce_validator_builder, informant_prefix: self.informant_prefix, marker: self.marker, }) } /// Defines which head-of-chain strategy to use. pub fn with_select_chain( self, builder: impl FnOnce(&Configuration, &Arc) -> Result, ) -> Result, Error> { self.with_opt_select_chain(|cfg, b| builder(cfg, b).map(Option::Some)) } /// Defines which import queue to use. pub fn with_import_queue( self, builder: impl FnOnce(&Configuration, Arc, Option, Arc, &SpawnTaskHandle, Option<&Registry>) -> Result ) -> Result, Error> where TSc: Clone { let import_queue = builder( &self.config, self.client.clone(), self.select_chain.clone(), self.transaction_pool.clone(), &self.task_manager.spawn_handle(), self.config.prometheus_config.as_ref().map(|config| &config.registry), )?; Ok(ServiceBuilder { config: self.config, client: self.client, backend: self.backend, task_manager: self.task_manager, keystore: self.keystore, fetcher: self.fetcher, select_chain: self.select_chain, import_queue, finality_proof_request_builder: self.finality_proof_request_builder, finality_proof_provider: self.finality_proof_provider, transaction_pool: self.transaction_pool, rpc_extensions_builder: self.rpc_extensions_builder, remote_backend: self.remote_backend, block_announce_validator_builder: self.block_announce_validator_builder, informant_prefix: self.informant_prefix, marker: self.marker, }) } /// Defines which strategy to use for providing finality proofs. pub fn with_opt_finality_proof_provider( self, builder: impl FnOnce(Arc, Arc) -> Result>>, Error> ) -> Result>, TExPool, TRpc, Backend, >, Error> { let finality_proof_provider = builder(self.client.clone(), self.backend.clone())?; Ok(ServiceBuilder { config: self.config, client: self.client, backend: self.backend, task_manager: self.task_manager, keystore: self.keystore, fetcher: self.fetcher, select_chain: self.select_chain, import_queue: self.import_queue, finality_proof_request_builder: self.finality_proof_request_builder, finality_proof_provider, transaction_pool: self.transaction_pool, rpc_extensions_builder: self.rpc_extensions_builder, remote_backend: self.remote_backend, block_announce_validator_builder: self.block_announce_validator_builder, informant_prefix: self.informant_prefix, marker: self.marker, }) } /// Defines which strategy to use for providing finality proofs. pub fn with_finality_proof_provider( self, build: impl FnOnce(Arc, Arc) -> Result>, Error> ) -> Result>, TExPool, TRpc, Backend, >, Error> { self.with_opt_finality_proof_provider(|client, backend| build(client, backend).map(Option::Some)) } /// Defines which import queue to use. pub fn with_import_queue_and_opt_fprb( self, builder: impl FnOnce( &Configuration, Arc, Arc, Option, Option, Arc, &SpawnTaskHandle, Option<&Registry>, ) -> Result<(UImpQu, Option), Error> ) -> Result, Error> where TSc: Clone, TFchr: Clone { let (import_queue, fprb) = builder( &self.config, self.client.clone(), self.backend.clone(), self.fetcher.clone(), self.select_chain.clone(), self.transaction_pool.clone(), &self.task_manager.spawn_handle(), self.config.prometheus_config.as_ref().map(|config| &config.registry), )?; Ok(ServiceBuilder { config: self.config, client: self.client, backend: self.backend, task_manager: self.task_manager, keystore: self.keystore, fetcher: self.fetcher, select_chain: self.select_chain, import_queue, finality_proof_request_builder: fprb, finality_proof_provider: self.finality_proof_provider, transaction_pool: self.transaction_pool, rpc_extensions_builder: self.rpc_extensions_builder, remote_backend: self.remote_backend, block_announce_validator_builder: self.block_announce_validator_builder, informant_prefix: self.informant_prefix, marker: self.marker, }) } /// Defines which import queue to use. pub fn with_import_queue_and_fprb( self, builder: impl FnOnce( &Configuration, Arc, Arc, Option, Option, Arc, &SpawnTaskHandle, Option<&Registry>, ) -> Result<(UImpQu, UFprb), Error> ) -> Result, Error> where TSc: Clone, TFchr: Clone { self.with_import_queue_and_opt_fprb(|cfg, cl, b, f, sc, tx, tb, pr| builder(cfg, cl, b, f, sc, tx, tb, pr) .map(|(q, f)| (q, Some(f))) ) } /// Defines which transaction pool to use. pub fn with_transaction_pool( self, transaction_pool_builder: impl FnOnce( &Self, ) -> Result<(UExPool, Option), Error>, ) -> Result, Error> where TSc: Clone, TFchr: Clone { let (transaction_pool, background_task) = transaction_pool_builder(&self)?; if let Some(background_task) = background_task{ self.task_manager.spawn_handle().spawn("txpool-background", background_task); } Ok(ServiceBuilder { config: self.config, client: self.client, task_manager: self.task_manager, backend: self.backend, keystore: self.keystore, fetcher: self.fetcher, select_chain: self.select_chain, import_queue: self.import_queue, finality_proof_request_builder: self.finality_proof_request_builder, finality_proof_provider: self.finality_proof_provider, transaction_pool: Arc::new(transaction_pool), rpc_extensions_builder: self.rpc_extensions_builder, remote_backend: self.remote_backend, block_announce_validator_builder: self.block_announce_validator_builder, informant_prefix: self.informant_prefix, marker: self.marker, }) } /// Defines the RPC extension builder to use. Unlike `with_rpc_extensions`, /// this method is useful in situations where the RPC extensions need to /// access to a `DenyUnsafe` instance to avoid exposing sensitive methods. pub fn with_rpc_extensions_builder( self, rpc_extensions_builder: impl FnOnce(&Self) -> Result, ) -> Result< ServiceBuilder, Error, > where TSc: Clone, TFchr: Clone, URpcBuilder: RpcExtensionBuilder + Send + 'static, URpc: sc_rpc::RpcExtension, { let rpc_extensions_builder = rpc_extensions_builder(&self)?; Ok(ServiceBuilder { config: self.config, client: self.client, backend: self.backend, task_manager: self.task_manager, keystore: self.keystore, fetcher: self.fetcher, select_chain: self.select_chain, import_queue: self.import_queue, finality_proof_request_builder: self.finality_proof_request_builder, finality_proof_provider: self.finality_proof_provider, transaction_pool: self.transaction_pool, rpc_extensions_builder: Box::new(rpc_extensions_builder), remote_backend: self.remote_backend, block_announce_validator_builder: self.block_announce_validator_builder, informant_prefix: self.informant_prefix, marker: self.marker, }) } /// Defines the RPC extensions to use. pub fn with_rpc_extensions( self, rpc_extensions: impl FnOnce(&Self) -> Result, ) -> Result< ServiceBuilder, Error, > where TSc: Clone, TFchr: Clone, URpc: Clone + sc_rpc::RpcExtension + Send + 'static, { let rpc_extensions = rpc_extensions(&self)?; self.with_rpc_extensions_builder(|_| Ok(NoopRpcExtensionBuilder::from(rpc_extensions))) } /// Defines the `BlockAnnounceValidator` to use. `DefaultBlockAnnounceValidator` will be used by /// default. pub fn with_block_announce_validator( self, block_announce_validator_builder: impl FnOnce(Arc) -> Box + Send> + Send + 'static, ) -> Result, Error> where TSc: Clone, TFchr: Clone { Ok(ServiceBuilder { config: self.config, client: self.client, backend: self.backend, task_manager: self.task_manager, keystore: self.keystore, fetcher: self.fetcher, select_chain: self.select_chain, import_queue: self.import_queue, finality_proof_request_builder: self.finality_proof_request_builder, finality_proof_provider: self.finality_proof_provider, transaction_pool: self.transaction_pool, rpc_extensions_builder: self.rpc_extensions_builder, remote_backend: self.remote_backend, block_announce_validator_builder: Some(Box::new(block_announce_validator_builder)), informant_prefix: self.informant_prefix, marker: self.marker, }) } /// Defines the informant's prefix for the logs. An empty string by default. /// /// By default substrate will show logs without a prefix. Example: /// /// ```text /// 2020-05-28 15:11:06 ✨ Imported #2 (0xc21c…2ca8) /// 2020-05-28 15:11:07 💤 Idle (0 peers), best: #2 (0xc21c…2ca8), finalized #0 (0x7299…e6df), ⬇ 0 ⬆ 0 /// ``` /// /// But you can define a prefix by using this function. Example: /// /// ```rust,ignore /// service.with_informant_prefix("[Prefix] ".to_string()); /// ``` /// /// This will output: /// /// ```text /// 2020-05-28 15:11:06 ✨ [Prefix] Imported #2 (0xc21c…2ca8) /// 2020-05-28 15:11:07 💤 [Prefix] Idle (0 peers), best: #2 (0xc21c…2ca8), finalized #0 (0x7299…e6df), ⬇ 0 ⬆ 0 /// ``` pub fn with_informant_prefix( self, informant_prefix: String, ) -> Result, Error> where TSc: Clone, TFchr: Clone { Ok(ServiceBuilder { informant_prefix: informant_prefix, ..self }) } } /// Implemented on `ServiceBuilder`. Allows running block commands, such as import/export/validate /// components to the builder. pub trait ServiceBuilderCommand { /// Block type this API operates on. type Block: BlockT; /// Native execution dispatch required by some commands. type NativeDispatch: NativeExecutionDispatch + 'static; /// Starts the process of importing blocks. fn import_blocks( self, input: impl Read + Seek + Send + 'static, force: bool, binary: bool, ) -> Pin> + Send>>; /// Performs the blocks export. fn export_blocks( self, output: impl Write + 'static, from: NumberFor, to: Option>, binary: bool ) -> Pin>>>; /// Performs a revert of `blocks` blocks. fn revert_chain( &self, blocks: NumberFor ) -> Result<(), Error>; /// Re-validate known block. fn check_block( self, block: BlockId ) -> Pin> + Send>>; /// Export the raw state at the given `block`. If `block` is `None`, the /// best block will be used. fn export_raw_state( &self, block: Option>, ) -> Result; } impl ServiceBuilder< TBl, TRtApi, Client, Arc>, TSc, TImpQu, BoxFinalityProofRequestBuilder, Arc>, TExPool, TRpc, TBackend, > where Client: ProvideRuntimeApi, as ProvideRuntimeApi>::Api: sp_api::Metadata + sc_offchain::OffchainWorkerApi + sp_transaction_pool::runtime_api::TaggedTransactionQueue + sp_session::SessionKeys + sp_api::ApiErrorExt + sp_api::ApiExt, TBl: BlockT, TRtApi: 'static + Send + Sync, TBackend: 'static + sc_client_api::backend::Backend + Send, TExec: 'static + CallExecutor + Send + Sync + Clone, TSc: Clone, TImpQu: 'static + ImportQueue, TExPool: MaintainedTransactionPool::Hash> + MallocSizeOfWasm + 'static, TRpc: sc_rpc::RpcExtension, { /// Set an ExecutionExtensionsFactory pub fn with_execution_extensions_factory(self, execution_extensions_factory: Box) -> Result { self.client.execution_extensions().set_extensions_factory(execution_extensions_factory); Ok(self) } fn build_common(self) -> Result, TSc, NetworkStatus, NetworkService::Hash>, TExPool, sc_offchain::OffchainWorkers< Client, TBackend::OffchainStorage, TBl >, >, Error> where TExec: CallExecutor, { let ServiceBuilder { marker: _, mut config, client, task_manager, fetcher: on_demand, backend, keystore, select_chain, import_queue, finality_proof_request_builder, finality_proof_provider, transaction_pool, rpc_extensions_builder, remote_backend, block_announce_validator_builder, informant_prefix, } = self; sp_session::generate_initial_session_keys( client.clone(), &BlockId::Hash(client.chain_info().best_hash), config.dev_key_seed.clone().map(|s| vec![s]).unwrap_or_default(), )?; // A side-channel for essential tasks to communicate shutdown. let (essential_failed_tx, essential_failed_rx) = tracing_unbounded("mpsc_essential_tasks"); let chain_info = client.chain_info(); info!("📦 Highest known block at #{}", chain_info.best_number); telemetry!( SUBSTRATE_INFO; "node.start"; "height" => chain_info.best_number.saturated_into::(), "best" => ?chain_info.best_hash ); let spawn_handle = task_manager.spawn_handle(); let (system_rpc_tx, system_rpc_rx) = tracing_unbounded("mpsc_system_rpc"); let (network, network_status_sinks, network_future) = build_network( &config, client.clone(), transaction_pool.clone(), Clone::clone(&spawn_handle), on_demand.clone(), block_announce_validator_builder, finality_proof_request_builder, finality_proof_provider, system_rpc_rx, import_queue )?; // 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", network_future ); let offchain_storage = backend.offchain_storage(); let offchain_workers = match (config.offchain_worker.clone(), offchain_storage.clone()) { (OffchainWorkerConfig {enabled: true, .. }, Some(db)) => { Some(Arc::new(sc_offchain::OffchainWorkers::new(client.clone(), db))) }, (OffchainWorkerConfig {enabled: true, .. }, None) => { warn!("Offchain workers disabled, due to lack of offchain storage support in backend."); None }, _ => None, }; // Inform the tx pool about imported and finalized blocks. spawn_handle.spawn( "txpool-notifications", sc_transaction_pool::notification_future(client.clone(), transaction_pool.clone()), ); // Inform the offchain worker about new imported blocks if let Some(offchain) = offchain_workers.clone() { spawn_handle.spawn( "offchain-notifications", sc_offchain::notification_future( config.role.is_authority(), client.clone(), offchain, task_manager.spawn_handle(), network.clone() ) ); } spawn_handle.spawn( "on-transaction-imported", extrinsic_notifications(transaction_pool.clone(), network.clone()), ); // Prometheus metrics. let metrics_service = if let Some(PrometheusConfig { port, registry }) = config.prometheus_config.clone() { // Set static metrics. let metrics = MetricsService::with_prometheus( ®istry, &config.network.node_name, &config.impl_version, &config.role, )?; spawn_handle.spawn( "prometheus-endpoint", prometheus_endpoint::init_prometheus(port, registry).map(drop) ); metrics } else { MetricsService::new() }; // Periodically notify the telemetry. spawn_handle.spawn("telemetry-periodic-send", telemetry_periodic_send( client.clone(), transaction_pool.clone(), metrics_service, network_status_sinks.clone() )); // Periodically send the network state to the telemetry. spawn_handle.spawn( "telemetry-periodic-network-state", telemetry_periodic_network_state(network_status_sinks.clone()), ); // RPC let gen_handler = |deny_unsafe: sc_rpc::DenyUnsafe| gen_handler( deny_unsafe, &config, &task_manager, client.clone(), transaction_pool.clone(), keystore.clone(), on_demand.clone(), remote_backend.clone(), &*rpc_extensions_builder, offchain_storage.clone(), system_rpc_tx.clone() ); let rpc = start_rpc_servers(&config, gen_handler)?; // This is used internally, so don't restrict access to unsafe RPC let rpc_handlers = gen_handler(sc_rpc::DenyUnsafe::No); let telemetry_connection_sinks: Arc>>> = Default::default(); // Telemetry let telemetry = config.telemetry_endpoints.clone().map(|endpoints| { let (telemetry, future) = build_telemetry( &mut config, endpoints, telemetry_connection_sinks.clone(), network.clone() ); spawn_handle.spawn( "telemetry-worker", future, ); telemetry }); // Instrumentation if let Some(tracing_targets) = config.tracing_targets.as_ref() { let subscriber = sc_tracing::ProfilingSubscriber::new( config.tracing_receiver, tracing_targets ); match tracing::subscriber::set_global_default(subscriber) { Ok(_) => (), Err(e) => error!(target: "tracing", "Unable to set global default subscriber {}", e), } } // Spawn informant task spawn_handle.spawn("informant", sc_informant::build( client.clone(), network_status_sinks.clone(), transaction_pool.clone(), sc_informant::OutputFormat { enable_color: true, prefix: informant_prefix }, )); Ok(Service { client, task_manager, network, network_status_sinks, select_chain, transaction_pool, essential_failed_tx, essential_failed_rx, rpc_handlers, _rpc: rpc, _telemetry: telemetry, _offchain_workers: offchain_workers, _telemetry_on_connect_sinks: telemetry_connection_sinks.clone(), keystore, marker: PhantomData::, prometheus_registry: config.prometheus_config.map(|config| config.registry), _base_path: config.base_path.map(Arc::new), }) } /// Builds the light service. pub fn build_light(self) -> Result, TSc, NetworkStatus, NetworkService::Hash>, TExPool, sc_offchain::OffchainWorkers< Client, TBackend::OffchainStorage, TBl >, >, Error> where TExec: CallExecutor, { self.build_common() } } impl ServiceBuilder< TBl, TRtApi, Client, Arc>, TSc, TImpQu, BoxFinalityProofRequestBuilder, Arc>, TExPool, TRpc, TBackend, > where Client: ProvideRuntimeApi, as ProvideRuntimeApi>::Api: sp_api::Metadata + sc_offchain::OffchainWorkerApi + sp_transaction_pool::runtime_api::TaggedTransactionQueue + sp_session::SessionKeys + sp_api::ApiErrorExt + sp_api::ApiExt, TBl: BlockT, TRtApi: 'static + Send + Sync, TBackend: 'static + sc_client_api::backend::Backend + Send, TExec: 'static + CallExecutor + Send + Sync + Clone, TSc: Clone, TImpQu: 'static + ImportQueue, TExPool: MaintainedTransactionPool::Hash> + LocalTransactionPool::Hash> + MallocSizeOfWasm + 'static, TRpc: sc_rpc::RpcExtension, { /// Builds the full service. pub fn build_full(self) -> Result, TSc, NetworkStatus, NetworkService::Hash>, TExPool, sc_offchain::OffchainWorkers< Client, TBackend::OffchainStorage, TBl >, >, Error> where TExec: CallExecutor, { // make transaction pool available for off-chain runtime calls. self.client.execution_extensions() .register_transaction_pool(Arc::downgrade(&self.transaction_pool) as _); self.build_common() } } async fn extrinsic_notifications( transaction_pool: Arc, network: Arc::Hash>> ) where TBl: BlockT, TExPool: MaintainedTransactionPool::Hash>, { // extrinsic notifications transaction_pool.import_notification_stream() .for_each(move |hash| { network.propagate_extrinsic(hash); let status = transaction_pool.status(); telemetry!(SUBSTRATE_INFO; "txpool.import"; "ready" => status.ready, "future" => status.future ); ready(()) }) .await; } // Periodically notify the telemetry. async fn telemetry_periodic_send( client: Arc>, transaction_pool: Arc, mut metrics_service: MetricsService, network_status_sinks: Arc, NetworkState)>>> ) where TBl: BlockT, TExec: CallExecutor, Client: ProvideRuntimeApi, TExPool: MaintainedTransactionPool::Hash>, TBackend: sc_client_api::backend::Backend, { let (state_tx, state_rx) = tracing_unbounded::<(NetworkStatus<_>, NetworkState)>("mpsc_netstat1"); network_status_sinks.lock().push(std::time::Duration::from_millis(5000), state_tx); state_rx.for_each(move |(net_status, _)| { let info = client.usage_info(); metrics_service.tick( &info, &transaction_pool.status(), &net_status, ); ready(()) }).await; } async fn telemetry_periodic_network_state( network_status_sinks: Arc, NetworkState)>>> ) { // Periodically send the network state to the telemetry. let (netstat_tx, netstat_rx) = tracing_unbounded::<(NetworkStatus<_>, NetworkState)>("mpsc_netstat2"); network_status_sinks.lock().push(std::time::Duration::from_secs(30), netstat_tx); netstat_rx.for_each(move |(_, network_state)| { telemetry!( SUBSTRATE_INFO; "system.network_state"; "state" => network_state, ); ready(()) }).await; } fn build_telemetry( config: &mut Configuration, endpoints: sc_telemetry::TelemetryEndpoints, telemetry_connection_sinks: Arc>>>, network: Arc::Hash>> ) -> (sc_telemetry::Telemetry, Pin + Send>>) { let is_authority = config.role.is_authority(); let network_id = network.local_peer_id().to_base58(); let name = config.network.node_name.clone(); let impl_name = config.impl_name.to_owned(); let version = config.impl_version; let chain_name = config.chain_spec.name().to_owned(); let telemetry = sc_telemetry::init_telemetry(sc_telemetry::TelemetryConfig { endpoints, wasm_external_transport: config.telemetry_external_transport.take(), }); let startup_time = SystemTime::UNIX_EPOCH.elapsed() .map(|dur| dur.as_millis()) .unwrap_or(0); let future = telemetry.clone() .for_each(move |event| { // Safe-guard in case we add more events in the future. let sc_telemetry::TelemetryEvent::Connected = event; telemetry!(SUBSTRATE_INFO; "system.connected"; "name" => name.clone(), "implementation" => impl_name.clone(), "version" => version, "config" => "", "chain" => chain_name.clone(), "authority" => is_authority, "startup_time" => startup_time, "network_id" => network_id.clone() ); telemetry_connection_sinks.lock().retain(|sink| { sink.unbounded_send(()).is_ok() }); ready(()) }) .boxed(); (telemetry, future) } fn gen_handler( deny_unsafe: sc_rpc::DenyUnsafe, config: &Configuration, task_manager: &TaskManager, client: Arc>, transaction_pool: Arc, keystore: Arc>, on_demand: Option>>, remote_backend: Option>>, rpc_extensions_builder: &(dyn RpcExtensionBuilder + Send), offchain_storage: Option<>::OffchainStorage>, system_rpc_tx: TracingUnboundedSender> ) -> jsonrpc_pubsub::PubSubHandler where TBl: BlockT, TExec: CallExecutor + Send + Sync + 'static, TRtApi: Send + Sync + 'static, Client: ProvideRuntimeApi, TExPool: MaintainedTransactionPool::Hash> + 'static, TBackend: sc_client_api::backend::Backend + 'static, TRpc: sc_rpc::RpcExtension, as ProvideRuntimeApi>::Api: sp_session::SessionKeys + sp_api::Metadata, { use sc_rpc::{chain, state, author, system, offchain}; let system_info = sc_rpc::system::SystemInfo { chain_name: config.chain_spec.name().into(), impl_name: config.impl_name.into(), impl_version: config.impl_version.into(), properties: config.chain_spec.properties(), chain_type: config.chain_spec.chain_type(), }; let subscriptions = SubscriptionManager::new(Arc::new(task_manager.spawn_handle())); let (chain, state, child_state) = if let (Some(remote_backend), Some(on_demand)) = (remote_backend, on_demand) { // Light clients let chain = sc_rpc::chain::new_light( client.clone(), subscriptions.clone(), remote_backend.clone(), on_demand.clone() ); let (state, child_state) = sc_rpc::state::new_light( client.clone(), subscriptions.clone(), remote_backend.clone(), on_demand.clone() ); (chain, state, child_state) } else { // Full nodes let chain = sc_rpc::chain::new_full(client.clone(), subscriptions.clone()); let (state, child_state) = sc_rpc::state::new_full(client.clone(), subscriptions.clone()); (chain, state, child_state) }; let author = sc_rpc::author::Author::new( client.clone(), transaction_pool.clone(), subscriptions, keystore.clone(), deny_unsafe, ); let system = system::System::new(system_info, system_rpc_tx.clone(), deny_unsafe); let maybe_offchain_rpc = offchain_storage.clone() .map(|storage| { let offchain = sc_rpc::offchain::Offchain::new(storage, deny_unsafe); // FIXME: Use plain Option (don't collect into HashMap) when we upgrade to jsonrpc 14.1 // https://github.com/paritytech/jsonrpc/commit/20485387ed06a48f1a70bf4d609a7cde6cf0accf let delegate = offchain::OffchainApi::to_delegate(offchain); delegate.into_iter().collect::>() }).unwrap_or_default(); sc_rpc_server::rpc_handler(( state::StateApi::to_delegate(state), state::ChildStateApi::to_delegate(child_state), chain::ChainApi::to_delegate(chain), maybe_offchain_rpc, author::AuthorApi::to_delegate(author), system::SystemApi::to_delegate(system), rpc_extensions_builder.build(deny_unsafe), )) } fn build_network( config: &Configuration, client: Arc>, transaction_pool: Arc, spawn_handle: SpawnTaskHandle, on_demand: Option>>, block_announce_validator_builder: Option>) -> Box + Send> + Send >>, finality_proof_request_builder: Option>, finality_proof_provider: Option>>, system_rpc_rx: TracingUnboundedReceiver>, import_queue: TImpQu ) -> Result< ( Arc::Hash>>, Arc, NetworkState)>>>, Pin + Send>> ), Error > where TBl: BlockT, TExec: CallExecutor + Send + Sync + 'static, TRtApi: Send + Sync + 'static, Client: ProvideRuntimeApi, TExPool: MaintainedTransactionPool::Hash> + 'static, TBackend: sc_client_api::backend::Backend + 'static, TImpQu: ImportQueue + 'static, { let transaction_pool_adapter = Arc::new(TransactionPoolAdapter { imports_external_transactions: !matches!(config.role, Role::Light), pool: transaction_pool.clone(), client: client.clone(), }); let protocol_id = { let protocol_id_full = match config.chain_spec.protocol_id() { Some(pid) => pid, None => { warn!("Using default protocol ID {:?} because none is configured in the \ chain specs", DEFAULT_PROTOCOL_ID ); DEFAULT_PROTOCOL_ID } }.as_bytes(); sc_network::config::ProtocolId::from(protocol_id_full) }; let block_announce_validator = if let Some(f) = block_announce_validator_builder { f(client.clone()) } else { Box::new(DefaultBlockAnnounceValidator::new(client.clone())) }; let network_params = sc_network::config::Params { role: config.role.clone(), executor: { Some(Box::new(move |fut| { spawn_handle.spawn("libp2p-node", fut); })) }, network_config: config.network.clone(), chain: client.clone(), finality_proof_provider, finality_proof_request_builder, on_demand: on_demand.clone(), transaction_pool: transaction_pool_adapter.clone() as _, import_queue: Box::new(import_queue), protocol_id, block_announce_validator, metrics_registry: config.prometheus_config.as_ref().map(|config| config.registry.clone()) }; let has_bootnodes = !network_params.network_config.boot_nodes.is_empty(); let network_mut = sc_network::NetworkWorker::new(network_params)?; let network = network_mut.service().clone(); let network_status_sinks = Arc::new(Mutex::new(status_sinks::StatusSinks::new())); let future = build_network_future( config.role.clone(), network_mut, client.clone(), network_status_sinks.clone(), system_rpc_rx, has_bootnodes, config.announce_block, ).boxed(); Ok((network, network_status_sinks, future)) }