// This file is part of Substrate. // Copyright (C) 2018-2021 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 . #![warn(unused_extern_crates)] //! Service implementation. Specialized wrapper over substrate service. use std::sync::Arc; use sc_consensus_babe; use node_primitives::Block; use node_runtime::RuntimeApi; use sc_service::{ config::Configuration, error::Error as ServiceError, RpcHandlers, TaskManager, }; use sp_inherents::InherentDataProviders; use sc_network::{Event, NetworkService}; use sp_runtime::traits::Block as BlockT; use futures::prelude::*; use sc_client_api::{ExecutorProvider, RemoteBackend}; use node_executor::Executor; use sc_telemetry::{Telemetry, TelemetryWorker}; use sc_consensus_babe::SlotProportion; type FullClient = sc_service::TFullClient; type FullBackend = sc_service::TFullBackend; type FullSelectChain = sc_consensus::LongestChain; type FullGrandpaBlockImport = grandpa::GrandpaBlockImport; type LightClient = sc_service::TLightClient; pub fn new_partial( config: &Configuration, ) -> Result, sc_transaction_pool::FullPool, ( impl Fn( node_rpc::DenyUnsafe, sc_rpc::SubscriptionTaskExecutor, ) -> node_rpc::IoHandler, ( sc_consensus_babe::BabeBlockImport, grandpa::LinkHalf, sc_consensus_babe::BabeLink, ), grandpa::SharedVoterState, Option, ) >, ServiceError> { let telemetry = config.telemetry_endpoints.clone() .filter(|x| !x.is_empty()) .map(|endpoints| -> Result<_, sc_telemetry::Error> { let worker = TelemetryWorker::new(16)?; let telemetry = worker.handle().new_telemetry(endpoints); Ok((worker, telemetry)) }) .transpose()?; let (client, backend, keystore_container, task_manager) = sc_service::new_full_parts::( &config, telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()), )?; let client = Arc::new(client); let telemetry = telemetry .map(|(worker, telemetry)| { task_manager.spawn_handle().spawn("telemetry", worker.run()); telemetry }); let select_chain = sc_consensus::LongestChain::new(backend.clone()); let transaction_pool = sc_transaction_pool::BasicPool::new_full( config.transaction_pool.clone(), config.role.is_authority().into(), config.prometheus_registry(), task_manager.spawn_handle(), client.clone(), ); let (grandpa_block_import, grandpa_link) = grandpa::block_import( client.clone(), &(client.clone() as Arc<_>), select_chain.clone(), telemetry.as_ref().map(|x| x.handle()), )?; let justification_import = grandpa_block_import.clone(); let (block_import, babe_link) = sc_consensus_babe::block_import( sc_consensus_babe::Config::get_or_compute(&*client)?, grandpa_block_import, client.clone(), )?; let inherent_data_providers = sp_inherents::InherentDataProviders::new(); let import_queue = sc_consensus_babe::import_queue( babe_link.clone(), block_import.clone(), Some(Box::new(justification_import)), client.clone(), select_chain.clone(), inherent_data_providers.clone(), &task_manager.spawn_essential_handle(), config.prometheus_registry(), sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()), telemetry.as_ref().map(|x| x.handle()), )?; let import_setup = (block_import, grandpa_link, babe_link); let (rpc_extensions_builder, rpc_setup) = { let (_, grandpa_link, babe_link) = &import_setup; let justification_stream = grandpa_link.justification_stream(); let shared_authority_set = grandpa_link.shared_authority_set().clone(); let shared_voter_state = grandpa::SharedVoterState::empty(); let rpc_setup = shared_voter_state.clone(); let finality_proof_provider = grandpa::FinalityProofProvider::new_for_service( backend.clone(), Some(shared_authority_set.clone()), ); let babe_config = babe_link.config().clone(); let shared_epoch_changes = babe_link.epoch_changes().clone(); let client = client.clone(); let pool = transaction_pool.clone(); let select_chain = select_chain.clone(); let keystore = keystore_container.sync_keystore(); let chain_spec = config.chain_spec.cloned_box(); let rpc_extensions_builder = move |deny_unsafe, subscription_executor| { let deps = node_rpc::FullDeps { client: client.clone(), pool: pool.clone(), select_chain: select_chain.clone(), chain_spec: chain_spec.cloned_box(), deny_unsafe, babe: node_rpc::BabeDeps { babe_config: babe_config.clone(), shared_epoch_changes: shared_epoch_changes.clone(), keystore: keystore.clone(), }, grandpa: node_rpc::GrandpaDeps { shared_voter_state: shared_voter_state.clone(), shared_authority_set: shared_authority_set.clone(), justification_stream: justification_stream.clone(), subscription_executor, finality_provider: finality_proof_provider.clone(), }, }; node_rpc::create_full(deps) }; (rpc_extensions_builder, rpc_setup) }; Ok(sc_service::PartialComponents { client, backend, task_manager, keystore_container, select_chain, import_queue, transaction_pool, inherent_data_providers, other: (rpc_extensions_builder, import_setup, rpc_setup, telemetry), }) } pub struct NewFullBase { pub task_manager: TaskManager, pub inherent_data_providers: InherentDataProviders, pub client: Arc, pub network: Arc::Hash>>, pub network_status_sinks: sc_service::NetworkStatusSinks, pub transaction_pool: Arc>, } /// Creates a full service from the configuration. pub fn new_full_base( mut config: Configuration, with_startup_data: impl FnOnce( &sc_consensus_babe::BabeBlockImport, &sc_consensus_babe::BabeLink, ) ) -> Result { let sc_service::PartialComponents { client, backend, mut task_manager, import_queue, keystore_container, select_chain, transaction_pool, inherent_data_providers, other: (rpc_extensions_builder, import_setup, rpc_setup, mut telemetry), } = new_partial(&config)?; let shared_voter_state = rpc_setup; config.network.extra_sets.push(grandpa::grandpa_peers_set_config()); #[cfg(feature = "cli")] config.network.request_response_protocols.push( sc_finality_grandpa_warp_sync::request_response_config_for_chain( &config, task_manager.spawn_handle(), backend.clone(), import_setup.1.shared_authority_set().clone(), ) ); let (network, network_status_sinks, system_rpc_tx, network_starter) = sc_service::build_network(sc_service::BuildNetworkParams { config: &config, client: client.clone(), transaction_pool: transaction_pool.clone(), spawn_handle: task_manager.spawn_handle(), import_queue, on_demand: None, block_announce_validator_builder: None, })?; if config.offchain_worker.enabled { sc_service::build_offchain_workers( &config, task_manager.spawn_handle(), client.clone(), network.clone(), ); } let role = config.role.clone(); let force_authoring = config.force_authoring; let backoff_authoring_blocks = Some(sc_consensus_slots::BackoffAuthoringOnFinalizedHeadLagging::default()); let name = config.network.node_name.clone(); let enable_grandpa = !config.disable_grandpa; let prometheus_registry = config.prometheus_registry().cloned(); let _rpc_handlers = sc_service::spawn_tasks( sc_service::SpawnTasksParams { config, backend: backend.clone(), client: client.clone(), keystore: keystore_container.sync_keystore(), network: network.clone(), rpc_extensions_builder: Box::new(rpc_extensions_builder), transaction_pool: transaction_pool.clone(), task_manager: &mut task_manager, on_demand: None, remote_blockchain: None, network_status_sinks: network_status_sinks.clone(), system_rpc_tx, telemetry: telemetry.as_mut(), }, )?; let (block_import, grandpa_link, babe_link) = import_setup; (with_startup_data)(&block_import, &babe_link); if let sc_service::config::Role::Authority { .. } = &role { let proposer = sc_basic_authorship::ProposerFactory::new( task_manager.spawn_handle(), client.clone(), transaction_pool.clone(), prometheus_registry.as_ref(), telemetry.as_ref().map(|x| x.handle()), ); let can_author_with = sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()); let babe_config = sc_consensus_babe::BabeParams { keystore: keystore_container.sync_keystore(), client: client.clone(), select_chain, env: proposer, block_import, sync_oracle: network.clone(), inherent_data_providers: inherent_data_providers.clone(), force_authoring, backoff_authoring_blocks, babe_link, can_author_with, block_proposal_slot_portion: SlotProportion::new(0.5), telemetry: telemetry.as_ref().map(|x| x.handle()), }; let babe = sc_consensus_babe::start_babe(babe_config)?; task_manager.spawn_essential_handle().spawn_blocking("babe-proposer", babe); } // Spawn authority discovery module. if role.is_authority() { let authority_discovery_role = sc_authority_discovery::Role::PublishAndDiscover( keystore_container.keystore(), ); let dht_event_stream = network.event_stream("authority-discovery") .filter_map(|e| async move { match e { Event::Dht(e) => Some(e), _ => None, }}); let (authority_discovery_worker, _service) = sc_authority_discovery::new_worker_and_service( client.clone(), network.clone(), Box::pin(dht_event_stream), authority_discovery_role, prometheus_registry.clone(), ); task_manager.spawn_handle().spawn("authority-discovery-worker", authority_discovery_worker.run()); } // if the node isn't actively participating in consensus then it doesn't // need a keystore, regardless of which protocol we use below. let keystore = if role.is_authority() { Some(keystore_container.sync_keystore()) } else { None }; let config = grandpa::Config { // FIXME #1578 make this available through chainspec gossip_duration: std::time::Duration::from_millis(333), justification_period: 512, name: Some(name), observer_enabled: false, keystore, is_authority: role.is_authority(), telemetry: telemetry.as_ref().map(|x| x.handle()), }; if enable_grandpa { // start the full GRANDPA voter // NOTE: non-authorities could run the GRANDPA observer protocol, but at // this point the full voter should provide better guarantees of block // and vote data availability than the observer. The observer has not // been tested extensively yet and having most nodes in a network run it // could lead to finality stalls. let grandpa_config = grandpa::GrandpaParams { config, link: grandpa_link, network: network.clone(), telemetry: telemetry.as_ref().map(|x| x.handle()), voting_rule: grandpa::VotingRulesBuilder::default().build(), prometheus_registry, shared_voter_state, }; // the GRANDPA voter task is considered infallible, i.e. // if it fails we take down the service with it. task_manager.spawn_essential_handle().spawn_blocking( "grandpa-voter", grandpa::run_grandpa_voter(grandpa_config)? ); } network_starter.start_network(); Ok(NewFullBase { task_manager, inherent_data_providers, client, network, network_status_sinks, transaction_pool, }) } /// Builds a new service for a full client. pub fn new_full( config: Configuration, ) -> Result { new_full_base(config, |_, _| ()).map(|NewFullBase { task_manager, .. }| { task_manager }) } pub fn new_light_base( mut config: Configuration, ) -> Result<( TaskManager, RpcHandlers, Arc, Arc::Hash>>, Arc>> ), ServiceError> { let telemetry = config.telemetry_endpoints.clone() .filter(|x| !x.is_empty()) .map(|endpoints| -> Result<_, sc_telemetry::Error> { #[cfg(feature = "browser")] let transport = Some( sc_telemetry::ExtTransport::new(libp2p_wasm_ext::ffi::websocket_transport()) ); #[cfg(not(feature = "browser"))] let transport = None; let worker = TelemetryWorker::with_transport(16, transport)?; let telemetry = worker.handle().new_telemetry(endpoints); Ok((worker, telemetry)) }) .transpose()?; let (client, backend, keystore_container, mut task_manager, on_demand) = sc_service::new_light_parts::( &config, telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()), )?; let mut telemetry = telemetry .map(|(worker, telemetry)| { task_manager.spawn_handle().spawn("telemetry", worker.run()); telemetry }); config.network.extra_sets.push(grandpa::grandpa_peers_set_config()); let select_chain = sc_consensus::LongestChain::new(backend.clone()); let transaction_pool = Arc::new(sc_transaction_pool::BasicPool::new_light( config.transaction_pool.clone(), config.prometheus_registry(), task_manager.spawn_handle(), client.clone(), on_demand.clone(), )); let (grandpa_block_import, _) = grandpa::block_import( client.clone(), &(client.clone() as Arc<_>), select_chain.clone(), telemetry.as_ref().map(|x| x.handle()), )?; let justification_import = grandpa_block_import.clone(); let (babe_block_import, babe_link) = sc_consensus_babe::block_import( sc_consensus_babe::Config::get_or_compute(&*client)?, grandpa_block_import, client.clone(), )?; let inherent_data_providers = sp_inherents::InherentDataProviders::new(); let import_queue = sc_consensus_babe::import_queue( babe_link, babe_block_import, Some(Box::new(justification_import)), client.clone(), select_chain.clone(), inherent_data_providers.clone(), &task_manager.spawn_essential_handle(), config.prometheus_registry(), sp_consensus::NeverCanAuthor, telemetry.as_ref().map(|x| x.handle()), )?; let (network, network_status_sinks, system_rpc_tx, network_starter) = sc_service::build_network(sc_service::BuildNetworkParams { config: &config, client: client.clone(), transaction_pool: transaction_pool.clone(), spawn_handle: task_manager.spawn_handle(), import_queue, on_demand: Some(on_demand.clone()), block_announce_validator_builder: None, })?; network_starter.start_network(); if config.offchain_worker.enabled { sc_service::build_offchain_workers( &config, task_manager.spawn_handle(), client.clone(), network.clone(), ); } let light_deps = node_rpc::LightDeps { remote_blockchain: backend.remote_blockchain(), fetcher: on_demand.clone(), client: client.clone(), pool: transaction_pool.clone(), }; let rpc_extensions = node_rpc::create_light(light_deps); let rpc_handlers = sc_service::spawn_tasks(sc_service::SpawnTasksParams { on_demand: Some(on_demand), remote_blockchain: Some(backend.remote_blockchain()), rpc_extensions_builder: Box::new(sc_service::NoopRpcExtensionBuilder(rpc_extensions)), client: client.clone(), transaction_pool: transaction_pool.clone(), keystore: keystore_container.sync_keystore(), config, backend, network_status_sinks, system_rpc_tx, network: network.clone(), task_manager: &mut task_manager, telemetry: telemetry.as_mut(), })?; Ok(( task_manager, rpc_handlers, client, network, transaction_pool, )) } /// Builds a new service for a light client. pub fn new_light( config: Configuration, ) -> Result { new_light_base(config).map(|(task_manager, _, _, _, _)| { task_manager }) } #[cfg(test)] mod tests { use std::{sync::Arc, borrow::Cow, convert::TryInto}; use sc_consensus_babe::{CompatibleDigestItem, BabeIntermediate, INTERMEDIATE_KEY}; use sc_consensus_epochs::descendent_query; use sp_consensus::{ Environment, Proposer, BlockImportParams, BlockOrigin, ForkChoiceStrategy, BlockImport, }; use node_primitives::{Block, DigestItem, Signature}; use node_runtime::{BalancesCall, Call, UncheckedExtrinsic, Address}; use node_runtime::constants::{currency::CENTS, time::SLOT_DURATION}; use codec::Encode; use sp_core::{ crypto::Pair as CryptoPair, H256, Public }; use sp_keystore::{SyncCryptoStorePtr, SyncCryptoStore}; use sp_runtime::{ generic::{BlockId, Era, Digest, SignedPayload}, traits::{Block as BlockT, Header as HeaderT}, traits::Verify, }; use sp_timestamp; use sp_keyring::AccountKeyring; use sc_service_test::TestNetNode; use crate::service::{new_full_base, new_light_base, NewFullBase}; use sp_runtime::{key_types::BABE, traits::IdentifyAccount, RuntimeAppPublic}; use sp_transaction_pool::{MaintainedTransactionPool, ChainEvent}; use sc_client_api::BlockBackend; use sc_keystore::LocalKeystore; type AccountPublic = ::Signer; #[test] // It is "ignored", but the node-cli ignored tests are running on the CI. // This can be run locally with `cargo test --release -p node-cli test_sync -- --ignored`. #[ignore] fn test_sync() { let keystore_path = tempfile::tempdir().expect("Creates keystore path"); let keystore: SyncCryptoStorePtr = Arc::new(LocalKeystore::open(keystore_path.path(), None) .expect("Creates keystore")); let alice: sp_consensus_babe::AuthorityId = SyncCryptoStore::sr25519_generate_new(&*keystore, BABE, Some("//Alice")) .expect("Creates authority pair").into(); let chain_spec = crate::chain_spec::tests::integration_test_config_with_single_authority(); // For the block factory let mut slot = 1u64; // For the extrinsics factory let bob = Arc::new(AccountKeyring::Bob.pair()); let charlie = Arc::new(AccountKeyring::Charlie.pair()); let mut index = 0; sc_service_test::sync( chain_spec, |config| { let mut setup_handles = None; let NewFullBase { task_manager, inherent_data_providers, client, network, transaction_pool, .. } = new_full_base(config, | block_import: &sc_consensus_babe::BabeBlockImport, babe_link: &sc_consensus_babe::BabeLink, | { setup_handles = Some((block_import.clone(), babe_link.clone())); } )?; let node = sc_service_test::TestNetComponents::new( task_manager, client, network, transaction_pool ); Ok((node, (inherent_data_providers, setup_handles.unwrap()))) }, |config| { let (keep_alive, _, client, network, transaction_pool) = new_light_base(config)?; Ok(sc_service_test::TestNetComponents::new(keep_alive, client, network, transaction_pool)) }, |service, &mut (ref inherent_data_providers, (ref mut block_import, ref babe_link))| { let mut inherent_data = inherent_data_providers .create_inherent_data() .expect("Creates inherent data."); let parent_id = BlockId::number(service.client().chain_info().best_number); let parent_header = service.client().header(&parent_id).unwrap().unwrap(); let parent_hash = parent_header.hash(); let parent_number = *parent_header.number(); futures::executor::block_on( service.transaction_pool().maintain( ChainEvent::NewBestBlock { hash: parent_header.hash(), tree_route: None, }, ) ); let mut proposer_factory = sc_basic_authorship::ProposerFactory::new( service.spawn_handle(), service.client(), service.transaction_pool(), None, None, ); let mut digest = Digest::::default(); // even though there's only one authority some slots might be empty, // so we must keep trying the next slots until we can claim one. let (babe_pre_digest, epoch_descriptor) = loop { inherent_data.replace_data( sp_timestamp::INHERENT_IDENTIFIER, &(slot * SLOT_DURATION), ); let epoch_descriptor = babe_link.epoch_changes().shared_data().epoch_descriptor_for_child_of( descendent_query(&*service.client()), &parent_hash, parent_number, slot.into(), ).unwrap().unwrap(); let epoch = babe_link.epoch_changes().shared_data().epoch_data( &epoch_descriptor, |slot| sc_consensus_babe::Epoch::genesis(&babe_link.config(), slot), ).unwrap(); if let Some(babe_pre_digest) = sc_consensus_babe::authorship::claim_slot( slot.into(), &epoch, &keystore, ).map(|(digest, _)| digest) { break (babe_pre_digest, epoch_descriptor) } slot += 1; }; digest.push(::babe_pre_digest(babe_pre_digest)); let new_block = futures::executor::block_on(async move { let proposer = proposer_factory.init(&parent_header).await; proposer.unwrap().propose( inherent_data, digest, std::time::Duration::from_secs(1), ).await }).expect("Error making test block").block; let (new_header, new_body) = new_block.deconstruct(); let pre_hash = new_header.hash(); // sign the pre-sealed hash of the block and then // add it to a digest item. let to_sign = pre_hash.encode(); let signature = SyncCryptoStore::sign_with( &*keystore, sp_consensus_babe::AuthorityId::ID, &alice.to_public_crypto_pair(), &to_sign, ).unwrap().unwrap().try_into().unwrap(); let item = ::babe_seal( signature, ); slot += 1; let mut params = BlockImportParams::new(BlockOrigin::File, new_header); params.post_digests.push(item); params.body = Some(new_body); params.intermediates.insert( Cow::from(INTERMEDIATE_KEY), Box::new(BabeIntermediate:: { epoch_descriptor }) as Box<_>, ); params.fork_choice = Some(ForkChoiceStrategy::LongestChain); futures::executor::block_on(block_import.import_block(params, Default::default())) .expect("error importing test block"); }, |service, _| { let amount = 5 * CENTS; let to: Address = AccountPublic::from(bob.public()).into_account().into(); let from: Address = AccountPublic::from(charlie.public()).into_account().into(); let genesis_hash = service.client().block_hash(0).unwrap().unwrap(); let best_block_id = BlockId::number(service.client().chain_info().best_number); let (spec_version, transaction_version) = { let version = service.client().runtime_version_at(&best_block_id).unwrap(); (version.spec_version, version.transaction_version) }; let signer = charlie.clone(); let function = Call::Balances(BalancesCall::transfer(to.into(), amount)); let check_spec_version = frame_system::CheckSpecVersion::new(); let check_tx_version = frame_system::CheckTxVersion::new(); let check_genesis = frame_system::CheckGenesis::new(); let check_era = frame_system::CheckEra::from(Era::Immortal); let check_nonce = frame_system::CheckNonce::from(index); let check_weight = frame_system::CheckWeight::new(); let payment = pallet_transaction_payment::ChargeTransactionPayment::from(0); let extra = ( check_spec_version, check_tx_version, check_genesis, check_era, check_nonce, check_weight, payment, ); let raw_payload = SignedPayload::from_raw( function, extra, (spec_version, transaction_version, genesis_hash, genesis_hash, (), (), ()) ); let signature = raw_payload.using_encoded(|payload| { signer.sign(payload) }); let (function, extra, _) = raw_payload.deconstruct(); index += 1; UncheckedExtrinsic::new_signed( function, from.into(), signature.into(), extra, ).into() }, ); } #[test] #[ignore] fn test_consensus() { sc_service_test::consensus( crate::chain_spec::tests::integration_test_config_with_two_authorities(), |config| { let NewFullBase { task_manager, client, network, transaction_pool, .. } = new_full_base(config,|_, _| ())?; Ok(sc_service_test::TestNetComponents::new(task_manager, client, network, transaction_pool)) }, |config| { let (keep_alive, _, client, network, transaction_pool) = new_light_base(config)?; Ok(sc_service_test::TestNetComponents::new(keep_alive, client, network, transaction_pool)) }, vec![ "//Alice".into(), "//Bob".into(), ], ) } }