// Copyright 2019 Joystream Contributors // This file is part of Joystream node. // Joystream node 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. // Joystream node 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 Joystream node. If not, see . #![warn(unused_extern_crates)] // Substrate implementation issue. #![allow(clippy::redundant_closure_call)] // Substrate implementation issue. #![allow(clippy::type_complexity)] //! Service implementation. Specialized wrapper over substrate service. use crate::node_executor; use crate::node_rpc; use node_runtime::opaque::Block; use node_runtime::RuntimeApi; use futures::prelude::*; use node_executor::Executor; use sc_client_api::{ExecutorProvider, RemoteBackend}; use sc_finality_grandpa as grandpa; use sc_finality_grandpa::FinalityProofProvider as GrandpaFinalityProofProvider; use sc_network::{Event, NetworkService}; use sc_service::{ config::{Configuration, Role}, error::Error as ServiceError, RpcHandlers, TaskManager, }; use sp_core::traits::BareCryptoStorePtr; use sp_inherents::InherentDataProviders; use sp_runtime::traits::Block as BlockT; use std::sync::Arc; 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_service::PartialComponents< FullClient, FullBackend, FullSelectChain, sp_consensus::DefaultImportQueue, 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, Arc>, ), ), >, ServiceError, > { let (client, backend, keystore, task_manager) = sc_service::new_full_parts::(&config)?; let client = Arc::new(client); let select_chain = sc_consensus::LongestChain::new(backend.clone()); let transaction_pool = sc_transaction_pool::BasicPool::new_full( config.transaction_pool.clone(), 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(), )?; 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)), None, client.clone(), select_chain.clone(), inherent_data_providers.clone(), &task_manager.spawn_handle(), config.prometheus_registry(), sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()), )?; 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 finality_proof_provider = GrandpaFinalityProofProvider::new_for_service(backend.clone(), client.clone()); let rpc_setup = (shared_voter_state.clone(), finality_proof_provider.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.clone(); 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(), 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, select_chain, import_queue, transaction_pool, inherent_data_providers, other: (rpc_extensions_builder, import_setup, rpc_setup), }) } 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( 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, select_chain, transaction_pool, inherent_data_providers, other: (rpc_extensions_builder, import_setup, rpc_setup), } = new_partial(&config)?; let (shared_voter_state, finality_proof_provider) = rpc_setup; 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, finality_proof_request_builder: None, finality_proof_provider: Some(finality_proof_provider), })?; if config.offchain_worker.enabled { sc_service::build_offchain_workers( &config, backend.clone(), task_manager.spawn_handle(), client.clone(), network.clone(), ); } let role = config.role.clone(); let force_authoring = config.force_authoring; let name = config.network.node_name.clone(); let enable_grandpa = !config.disable_grandpa; let prometheus_registry = config.prometheus_registry().cloned(); let telemetry_connection_sinks = sc_service::TelemetryConnectionSinks::default(); sc_service::spawn_tasks(sc_service::SpawnTasksParams { config, backend, client: client.clone(), keystore: keystore.clone(), 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, telemetry_connection_sinks: telemetry_connection_sinks.clone(), network_status_sinks: network_status_sinks.clone(), system_rpc_tx, })?; 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( client.clone(), transaction_pool.clone(), prometheus_registry.as_ref(), ); let can_author_with = sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()); let babe_config = sc_consensus_babe::BabeParams { keystore: keystore.clone(), client: client.clone(), select_chain, env: proposer, block_import, sync_oracle: network.clone(), inherent_data_providers: inherent_data_providers.clone(), force_authoring, babe_link, can_author_with, }; let babe = sc_consensus_babe::start_babe(babe_config)?; task_manager .spawn_essential_handle() .spawn_blocking("babe-proposer", babe); } // Spawn authority discovery module. if matches!(role, Role::Authority { .. } | Role::Sentry { .. }) { let (sentries, authority_discovery_role) = match role { sc_service::config::Role::Authority { ref sentry_nodes } => ( sentry_nodes.clone(), sc_authority_discovery::Role::Authority(keystore.clone()), ), sc_service::config::Role::Sentry { .. } => { (vec![], sc_authority_discovery::Role::Sentry) } _ => unreachable!("Due to outer matches! constraint; qed."), }; let dht_event_stream = network .event_stream("authority-discovery") .filter_map(|e| async move { match e { Event::Dht(e) => Some(e), _ => None, } }) .boxed(); let (authority_discovery_worker, _service) = sc_authority_discovery::new_worker_and_service( client.clone(), network.clone(), sentries, dht_event_stream, authority_discovery_role, prometheus_registry.clone(), ); task_manager .spawn_handle() .spawn("authority-discovery-worker", authority_discovery_worker); } // 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 as BareCryptoStorePtr) } 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_network_authority(), }; 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(), inherent_data_providers: inherent_data_providers.clone(), telemetry_on_connect: Some(telemetry_connection_sinks.on_connect_stream()), 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)?); } else { grandpa::setup_disabled_grandpa(client.clone(), &inherent_data_providers, network.clone())?; } 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) } // Substrate implementation issue. #[allow(clippy::type_complexity)] pub fn new_light_base( config: Configuration, ) -> Result< ( TaskManager, RpcHandlers, Arc, Arc::Hash>>, Arc< sc_transaction_pool::LightPool>, >, ), ServiceError, > { let (client, backend, keystore, mut task_manager, on_demand) = sc_service::new_light_parts::(&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::light_block_import( client.clone(), backend.clone(), &(client.clone() as Arc<_>), Arc::new(on_demand.checker().clone()), )?; let finality_proof_import = grandpa_block_import.clone(); let finality_proof_request_builder = finality_proof_import.create_finality_proof_request_builder(); 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, None, Some(Box::new(finality_proof_import)), client.clone(), select_chain, inherent_data_providers, &task_manager.spawn_handle(), config.prometheus_registry(), sp_consensus::NeverCanAuthor, )?; let finality_proof_provider = GrandpaFinalityProofProvider::new_for_service(backend.clone(), client.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: Some(on_demand.clone()), block_announce_validator_builder: None, finality_proof_request_builder: Some(finality_proof_request_builder), finality_proof_provider: Some(finality_proof_provider), })?; network_starter.start_network(); if config.offchain_worker.enabled { sc_service::build_offchain_workers( &config, backend.clone(), 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(), config, keystore, backend, network_status_sinks, system_rpc_tx, network: network.clone(), telemetry_connection_sinks: sc_service::TelemetryConnectionSinks::default(), task_manager: &mut task_manager, })?; 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 crate::service::{new_full_base, new_light_base, NewFullBase}; use codec::Encode; use node_runtime::opaque::Block; use node_runtime::{currency::CENTS, SLOT_DURATION}; use node_runtime::{Address, BalancesCall, Call, UncheckedExtrinsic}; use node_runtime::{DigestItem, Signature}; use sc_client_api::BlockBackend; use sc_consensus_babe::{BabeIntermediate, CompatibleDigestItem, INTERMEDIATE_KEY}; use sc_consensus_epochs::descendent_query; use sc_service_test::TestNetNode; use sp_consensus::{ BlockImport, BlockImportParams, BlockOrigin, Environment, ForkChoiceStrategy, Proposer, RecordProof, }; use sp_core::{crypto::Pair as CryptoPair, H256}; use sp_finality_tracker; use sp_keyring::AccountKeyring; use sp_runtime::traits::IdentifyAccount; use sp_runtime::{ generic::{BlockId, Digest, Era, SignedPayload}, traits::Verify, traits::{Block as BlockT, Header as HeaderT}, }; use sp_timestamp; use sp_transaction_pool::{ChainEvent, MaintainedTransactionPool}; use std::{any::Any, borrow::Cow, sync::Arc}; 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 = sc_keystore::Store::open(keystore_path.path(), None).expect("Creates keystore"); let alice = keystore .write() .insert_ephemeral_from_seed::("//Alice") .expect("Creates authority pair"); let chain_spec = crate::chain_spec::tests::integration_test_config_with_single_authority(); // For the block factory let mut slot_num = 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."); inherent_data.replace_data(sp_finality_tracker::INHERENT_IDENTIFIER, &1u64); 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.client(), service.transaction_pool(), None, ); let epoch_descriptor = babe_link .epoch_changes() .lock() .epoch_descriptor_for_child_of( descendent_query(&*service.client()), &parent_hash, parent_number, slot_num, ) .unwrap() .unwrap(); 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 = loop { inherent_data.replace_data( sp_timestamp::INHERENT_IDENTIFIER, &(slot_num * SLOT_DURATION), ); if let Some(babe_pre_digest) = sc_consensus_babe::test_helpers::claim_slot( slot_num, &parent_header, &*service.client(), &keystore, &babe_link, ) { break babe_pre_digest; } slot_num += 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), RecordProof::Yes, ) .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 = alice.sign(&to_sign[..]); let item = ::babe_seal(signature.into()); slot_num += 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); 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()], ) } }