git.delta.rocks / unique-network / refs/commits / d9a94fdfefd5

1//! Service and ServiceFactory implementation. Specialized wrapper over substrate service.23//4// This file is subject to the terms and conditions defined in5// file 'LICENSE', which is part of this source code package.6//78use std::{sync::{Arc, Mutex}, cell::RefCell, time::Duration, collections::{HashMap, BTreeMap}};9use fc_rpc::EthTask;10use fc_rpc_core::types::{FilterPool, PendingTransactions};11use sc_client_api::{ExecutorProvider, RemoteBackend, BlockchainEvents};12use fc_consensus::FrontierBlockImport;13use fc_mapping_sync::MappingSyncWorker;14use nft_runtime::{self, opaque::Block, RuntimeApi, SLOT_DURATION};15use sc_service::{error::Error as ServiceError, Configuration, TaskManager, BasePath};16use sp_inherents::{InherentDataProviders, ProvideInherentData, InherentIdentifier, InherentData};17use sc_executor::native_executor_instance;18pub use sc_executor::NativeExecutor;19use sc_consensus_aura::{ImportQueueParams, StartAuraParams, SlotProportion};20use sp_consensus_aura::sr25519::{AuthorityPair as AuraPair};21use sc_finality_grandpa::SharedVoterState;22use sp_timestamp::InherentError;23use sc_telemetry::{Telemetry, TelemetryWorker};24use sc_cli::SubstrateCli;25use futures::StreamExt;2627use crate::cli::Cli;2829// Our native executor instance.30native_executor_instance!(31	pub Executor,32	nft_runtime::api::dispatch,33	nft_runtime::native_version,34	frame_benchmarking::benchmarking::HostFunctions,35);3637type FullClient = sc_service::TFullClient<Block, RuntimeApi, Executor>;38type FullBackend = sc_service::TFullBackend<Block>;39type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;4041pub type ConsensusResult = (42	sc_consensus_aura::AuraBlockImport<43		Block,44		FullClient,45		FrontierBlockImport<46			Block,47			sc_finality_grandpa::GrandpaBlockImport<FullBackend, Block, FullClient, FullSelectChain>,48			FullClient49		>,50		AuraPair51	>,52	sc_finality_grandpa::LinkHalf<Block, FullClient, FullSelectChain>53);5455/// Provide a mock duration starting at 0 in millisecond for timestamp inherent.56/// Each call will increment timestamp by slot_duration making Aura think time has passed.57pub struct MockTimestampInherentDataProvider;5859pub const INHERENT_IDENTIFIER: InherentIdentifier = *b"timstap0";6061thread_local!(static TIMESTAMP: RefCell<u64> = RefCell::new(0));6263impl ProvideInherentData for MockTimestampInherentDataProvider {64	fn inherent_identifier(&self) -> &'static InherentIdentifier {65		&INHERENT_IDENTIFIER66	}6768	fn provide_inherent_data(69		&self,70		inherent_data: &mut InherentData,71	) -> Result<(), sp_inherents::Error> {72		TIMESTAMP.with(|x| {73			*x.borrow_mut() += SLOT_DURATION;74			inherent_data.put_data(INHERENT_IDENTIFIER, &*x.borrow())75		})76	}7778	fn error_to_string(&self, error: &[u8]) -> Option<String> {79		InherentError::try_from(&INHERENT_IDENTIFIER, error).map(|e| format!("{:?}", e))80	}81}8283pub fn open_frontier_backend(config: &Configuration) -> Result<Arc<fc_db::Backend<Block>>, String> {84	let config_dir = config.base_path.as_ref()85		.map(|base_path| base_path.config_dir(config.chain_spec.id()))86		.unwrap_or_else(|| {87			BasePath::from_project("", "", &crate::cli::Cli::executable_name())88				.config_dir(config.chain_spec.id())89		});90	let database_dir = config_dir.join("frontier").join("db");9192	Ok(Arc::new(fc_db::Backend::<Block>::new(&fc_db::DatabaseSettings {93		source: fc_db::DatabaseSettingsSrc::RocksDb {94			path: database_dir,95			cache_size: 0,96		}97	})?))98}99100pub fn new_partial(config: &Configuration, #[allow(unused_variables)] cli: &Cli) -> Result<101	sc_service::PartialComponents<102		FullClient, FullBackend, FullSelectChain,103		sp_consensus::import_queue::BasicQueue<Block, sp_api::TransactionFor<FullClient, Block>>,104		sc_transaction_pool::FullPool<Block, FullClient>,105		(ConsensusResult, PendingTransactions, Option<FilterPool>, Arc<fc_db::Backend<Block>>, Option<Telemetry>),106>, ServiceError> {107	let inherent_data_providers = sp_inherents::InherentDataProviders::new();108109	let telemetry = config.telemetry_endpoints.clone()110		.filter(|x| !x.is_empty())111		.map(|endpoints| -> Result<_, sc_telemetry::Error> {112			let worker = TelemetryWorker::new(16)?;113			let telemetry = worker.handle().new_telemetry(endpoints);114			Ok((worker, telemetry))115		})116		.transpose()?;117118	let (client, backend, keystore_container, task_manager) =119		sc_service::new_full_parts::<Block, RuntimeApi, Executor>(120			&config,121			telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),122		)?;123	let client = Arc::new(client);124125	let telemetry = telemetry126		.map(|(worker, telemetry)| {127			task_manager.spawn_handle().spawn("telemetry", worker.run());128			telemetry129		});130131	let select_chain = sc_consensus::LongestChain::new(backend.clone());132133	let transaction_pool = sc_transaction_pool::BasicPool::new_full(134		config.transaction_pool.clone(),135		config.role.is_authority().into(),136		config.prometheus_registry(),137		task_manager.spawn_handle(),138		client.clone(),139	);140141	let pending_transactions: PendingTransactions142		= Some(Arc::new(Mutex::new(HashMap::new())));143144	let filter_pool: Option<FilterPool>145		= Some(Arc::new(Mutex::new(BTreeMap::new())));146147	let frontier_backend = open_frontier_backend(config)?;148149	let (grandpa_block_import, grandpa_link) = sc_finality_grandpa::block_import(150		client.clone(),151		&(client.clone() as Arc<_>),152		select_chain.clone(),153		telemetry.as_ref().map(|x| x.handle()),154	)?;155156	let frontier_block_import = FrontierBlockImport::new(157		grandpa_block_import.clone(),158		client.clone(),159		frontier_backend.clone(),160	);161162	let aura_block_import = sc_consensus_aura::AuraBlockImport::<_, _, _, AuraPair>::new(163		frontier_block_import, client.clone(),164	);165166	let import_queue = sc_consensus_aura::import_queue::<AuraPair, _, _, _, _, _>(167		ImportQueueParams {168			slot_duration: sc_consensus_aura::slot_duration(&*client)?,169			block_import: aura_block_import.clone(),170			justification_import: Some(Box::new(grandpa_block_import.clone())),171			client: client.clone(),172			inherent_data_providers: inherent_data_providers.clone(),173			spawner: &task_manager.spawn_essential_handle(),174			registry: config.prometheus_registry(),175			can_author_with: sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()),176			check_for_equivocation: Default::default(),177			telemetry: telemetry.as_ref().map(|x| x.handle()),178		}179	)?;180181	Ok(sc_service::PartialComponents {182		client, backend, task_manager, import_queue, keystore_container,183		select_chain, transaction_pool, inherent_data_providers,184		other: (185			(aura_block_import, grandpa_link),186			pending_transactions,187			filter_pool,188			frontier_backend,189			telemetry,190		)191	})192}193194/// Builds a new service for a full client.195pub fn new_full(196	config: Configuration,197	cli: &Cli,198) -> Result<TaskManager, ServiceError> {199	let enable_dev_signer = cli.run.shared_params.dev;200201	let sc_service::PartialComponents {202		client, backend, mut task_manager, import_queue, keystore_container,203		select_chain, transaction_pool, inherent_data_providers,204		other: (consensus_result, pending_transactions, filter_pool, frontier_backend, mut telemetry),205	} = new_partial(&config, cli)?;206207	let (network, network_status_sinks, system_rpc_tx, network_starter) =208		sc_service::build_network(sc_service::BuildNetworkParams {209			config: &config,210			client: Arc::clone(&client),211			transaction_pool: Arc::clone(&transaction_pool),212			spawn_handle: task_manager.spawn_handle(),213			import_queue,214			on_demand: None,215			block_announce_validator_builder: None,216		})?;217218	if config.offchain_worker.enabled {219		sc_service::build_offchain_workers(220			&config, task_manager.spawn_handle(), Arc::clone(&client), network.clone(),221		);222	}223224	let role = config.role.clone();225	let force_authoring = config.force_authoring;226	let backoff_authoring_blocks: Option<()> = None;227	let name = config.network.node_name.clone();228	let enable_grandpa = !config.disable_grandpa;229	let prometheus_registry = config.prometheus_registry().cloned();230	let is_authority = role.is_authority();231	let subscription_task_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());232233	let rpc_extensions_builder = {234		let client = Arc::clone(&client);235		let pool = Arc::clone(&transaction_pool);236		let network = network.clone();237		let pending = pending_transactions.clone();238		let filter_pool = filter_pool.clone();239		let frontier_backend = frontier_backend.clone();240241		Box::new(move |deny_unsafe, _| {242			let deps = crate::rpc::FullDeps {243				client: Arc::clone(&client),244				pool: Arc::clone(&pool),245				deny_unsafe,246				is_authority,247				enable_dev_signer,248				network: network.clone(),249				pending_transactions: pending.clone(),250				filter_pool: filter_pool.clone(),251				backend: frontier_backend.clone(),252			};253			crate::rpc::create_full(254				deps,255				subscription_task_executor.clone()256			)257		})258	};259260	task_manager.spawn_essential_handle().spawn(261		"frontier-mapping-sync-worker",262		MappingSyncWorker::new(263			client.import_notification_stream(),264			Duration::new(6, 0),265			Arc::clone(&client),266			Arc::clone(&backend),267			frontier_backend.clone(),268		).for_each(|()| futures::future::ready(()))269	);270271	let _rpc_handlers = sc_service::spawn_tasks(sc_service::SpawnTasksParams {272		network: network.clone(),273		client: Arc::clone(&client),274		keystore: keystore_container.sync_keystore(),275		task_manager: &mut task_manager,276		transaction_pool: Arc::clone(&transaction_pool),277		rpc_extensions_builder: rpc_extensions_builder,278		on_demand: None,279		remote_blockchain: None,280		backend, network_status_sinks, system_rpc_tx, config, telemetry: telemetry.as_mut(),281	})?;282283	// Spawn Frontier EthFilterApi maintenance task.284	if let Some(filter_pool) = filter_pool {285		// Each filter is allowed to stay in the pool for 100 blocks.286		const FILTER_RETAIN_THRESHOLD: u64 = 100;287		task_manager.spawn_essential_handle().spawn(288			"frontier-filter-pool",289			EthTask::filter_pool_task(290					Arc::clone(&client),291					filter_pool,292					FILTER_RETAIN_THRESHOLD,293			)294		);295	}296297	// Spawn Frontier pending transactions maintenance task (as essential, otherwise we leak).298	if let Some(pending_transactions) = pending_transactions {299		const TRANSACTION_RETAIN_THRESHOLD: u64 = 5;300		task_manager.spawn_essential_handle().spawn(301			"frontier-pending-transactions",302			EthTask::pending_transaction_task(303				Arc::clone(&client),304					pending_transactions,305					TRANSACTION_RETAIN_THRESHOLD,306				)307		);308	}309310	let (aura_block_import, grandpa_link) = consensus_result;311312	if role.is_authority() {313		let proposer = sc_basic_authorship::ProposerFactory::new(314			task_manager.spawn_handle(),315			Arc::clone(&client),316			Arc::clone(&transaction_pool),317			prometheus_registry.as_ref(),318			telemetry.as_ref().map(|x| x.handle()),319		);320321		let can_author_with =322			sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone());323		let aura = sc_consensus_aura::start_aura::<AuraPair, _, _, _, _, _, _, _, _, _>(324			StartAuraParams {325				slot_duration: sc_consensus_aura::slot_duration(&*client)?,326				client: Arc::clone(&client),327				select_chain,328				block_import: aura_block_import,329				proposer_factory: proposer,330				sync_oracle: network.clone(),331				inherent_data_providers: inherent_data_providers.clone(),332				force_authoring,333				backoff_authoring_blocks,334				keystore: keystore_container.sync_keystore(),335				can_author_with,336				block_proposal_slot_portion: SlotProportion::new(2f32 / 3f32),337				telemetry: telemetry.as_ref().map(|x| x.handle()),338			}339		)?;340341		// the AURA authoring task is considered essential, i.e. if it342		// fails we take down the service with it.343		task_manager.spawn_essential_handle().spawn_blocking("aura", aura);344345		// if the node isn't actively participating in consensus then it doesn't346		// need a keystore, regardless of which protocol we use below.347		let keystore = if role.is_authority() {348			Some(keystore_container.sync_keystore())349		} else {350			None351		};352353		let grandpa_config = sc_finality_grandpa::Config {354			// FIXME #1578 make this available through chainspec355			gossip_duration: Duration::from_millis(333),356			justification_period: 512,357			name: Some(name),358			observer_enabled: false,359			keystore,360			is_authority: role.is_authority(),361			telemetry: telemetry.as_ref().map(|x| x.handle()),362		};363364		if enable_grandpa {365			// start the full GRANDPA voter366			// NOTE: non-authorities could run the GRANDPA observer protocol, but at367			// this point the full voter should provide better guarantees of block368			// and vote data availability than the observer. The observer has not369			// been tested extensively yet and having most nodes in a network run it370			// could lead to finality stalls.371			let grandpa_config = sc_finality_grandpa::GrandpaParams {372				config: grandpa_config,373				link: grandpa_link,374				network,375				telemetry: telemetry.as_ref().map(|x| x.handle()),376				voting_rule: sc_finality_grandpa::VotingRulesBuilder::default().build(),377				prometheus_registry,378				shared_voter_state: SharedVoterState::empty(),379			};380381			// the GRANDPA voter task is considered infallible, i.e.382			// if it fails we take down the service with it.383			task_manager.spawn_essential_handle().spawn_blocking(384				"grandpa-voter",385				sc_finality_grandpa::run_grandpa_voter(grandpa_config)?386			);387		}388	}389390	network_starter.start_network();391	Ok(task_manager)392}393/// Builds a new service for a light client.394pub fn new_light(config: Configuration) -> Result<TaskManager, ServiceError> {395	let telemetry = config.telemetry_endpoints.clone()396		.filter(|x| !x.is_empty())397		.map(|endpoints| -> Result<_, sc_telemetry::Error> {398			let worker = TelemetryWorker::new(16)?;399			let telemetry = worker.handle().new_telemetry(endpoints);400			Ok((worker, telemetry))401		})402		.transpose()?;403404	let (client, backend, keystore_container, mut task_manager, on_demand) =405		sc_service::new_light_parts::<Block, RuntimeApi, Executor>(406			&config,407			telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),408		)?;409410	let mut telemetry = telemetry411		.map(|(worker, telemetry)| {412			task_manager.spawn_handle().spawn("telemetry", worker.run());413			telemetry414		});415416	let select_chain = sc_consensus::LongestChain::new(backend.clone());417418	let transaction_pool = Arc::new(sc_transaction_pool::BasicPool::new_light(419		config.transaction_pool.clone(),420		config.prometheus_registry(),421		task_manager.spawn_handle(),422		client.clone(),423		on_demand.clone(),424	));425426	let (grandpa_block_import, _) = sc_finality_grandpa::block_import(427		client.clone(),428		&(client.clone() as Arc<_>),429		select_chain.clone(),430		telemetry.as_ref().map(|x| x.handle()),431	)?;432433	let import_queue = sc_consensus_aura::import_queue::<AuraPair, _, _, _, _, _>(434		ImportQueueParams {435			slot_duration: sc_consensus_aura::slot_duration(&*client)?,436			block_import: grandpa_block_import.clone(),437			justification_import: Some(Box::new(grandpa_block_import)),438			client: client.clone(),439			inherent_data_providers: InherentDataProviders::new(),440			spawner: &task_manager.spawn_essential_handle(),441			registry: config.prometheus_registry(),442			can_author_with: sp_consensus::NeverCanAuthor,443			check_for_equivocation: Default::default(),444			telemetry: telemetry.as_ref().map(|x| x.handle()),445		}446	)?;447448	let light_deps = crate::rpc::LightDeps {449		remote_blockchain: backend.remote_blockchain(),450		fetcher: on_demand.clone(),451		client: client.clone(),452		pool: transaction_pool.clone(),453	};454455	let rpc_extensions = crate::rpc::create_light(light_deps);456457	let (network, network_status_sinks, system_rpc_tx, network_starter) =458		sc_service::build_network(sc_service::BuildNetworkParams {459			config: &config,460			client: client.clone(),461			transaction_pool: transaction_pool.clone(),462			spawn_handle: task_manager.spawn_handle(),463			import_queue,464			on_demand: Some(on_demand.clone()),465			block_announce_validator_builder: None,466		})?;467468	if config.offchain_worker.enabled {469		sc_service::build_offchain_workers(470			&config, task_manager.spawn_handle(), client.clone(), network.clone(),471		);472	}473474	sc_service::spawn_tasks(sc_service::SpawnTasksParams {475		remote_blockchain: Some(backend.remote_blockchain()),476		transaction_pool,477		task_manager: &mut task_manager,478		on_demand: Some(on_demand),479		rpc_extensions_builder: Box::new(sc_service::NoopRpcExtensionBuilder(rpc_extensions)),480		config,481		client,482		keystore: keystore_container.sync_keystore(),483		backend,484		network,485		network_status_sinks,486		system_rpc_tx,487		telemetry: telemetry.as_mut(),488	})?;489490	network_starter.start_network();491492	Ok(task_manager)493}