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

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();240		// TODO: Tune241		let max_past_logs = 10000;242243		Box::new(move |deny_unsafe, _| {244			let deps = crate::rpc::FullDeps {245				client: Arc::clone(&client),246				pool: Arc::clone(&pool),247				deny_unsafe,248				is_authority,249				enable_dev_signer,250				network: network.clone(),251				pending_transactions: pending.clone(),252				filter_pool: filter_pool.clone(),253				backend: frontier_backend.clone(),254				max_past_logs,255			};256			crate::rpc::create_full(257				deps,258				subscription_task_executor.clone()259			)260		})261	};262263	task_manager.spawn_essential_handle().spawn(264		"frontier-mapping-sync-worker",265		MappingSyncWorker::new(266			client.import_notification_stream(),267			Duration::new(6, 0),268			Arc::clone(&client),269			Arc::clone(&backend),270			frontier_backend.clone(),271		).for_each(|()| futures::future::ready(()))272	);273274	let _rpc_handlers = sc_service::spawn_tasks(sc_service::SpawnTasksParams {275		network: network.clone(),276		client: Arc::clone(&client),277		keystore: keystore_container.sync_keystore(),278		task_manager: &mut task_manager,279		transaction_pool: Arc::clone(&transaction_pool),280		rpc_extensions_builder: rpc_extensions_builder,281		on_demand: None,282		remote_blockchain: None,283		backend, network_status_sinks, system_rpc_tx, config, telemetry: telemetry.as_mut(),284	})?;285286	// Spawn Frontier EthFilterApi maintenance task.287	if let Some(filter_pool) = filter_pool {288		// Each filter is allowed to stay in the pool for 100 blocks.289		const FILTER_RETAIN_THRESHOLD: u64 = 100;290		task_manager.spawn_essential_handle().spawn(291			"frontier-filter-pool",292			EthTask::filter_pool_task(293					Arc::clone(&client),294					filter_pool,295					FILTER_RETAIN_THRESHOLD,296			)297		);298	}299300	// Spawn Frontier pending transactions maintenance task (as essential, otherwise we leak).301	if let Some(pending_transactions) = pending_transactions {302		const TRANSACTION_RETAIN_THRESHOLD: u64 = 5;303		task_manager.spawn_essential_handle().spawn(304			"frontier-pending-transactions",305			EthTask::pending_transaction_task(306				Arc::clone(&client),307					pending_transactions,308					TRANSACTION_RETAIN_THRESHOLD,309				)310		);311	}312313	let (aura_block_import, grandpa_link) = consensus_result;314315	if role.is_authority() {316		let proposer = sc_basic_authorship::ProposerFactory::new(317			task_manager.spawn_handle(),318			Arc::clone(&client),319			Arc::clone(&transaction_pool),320			prometheus_registry.as_ref(),321			telemetry.as_ref().map(|x| x.handle()),322		);323324		let can_author_with =325			sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone());326		let aura = sc_consensus_aura::start_aura::<AuraPair, _, _, _, _, _, _, _, _, _>(327			StartAuraParams {328				slot_duration: sc_consensus_aura::slot_duration(&*client)?,329				client: Arc::clone(&client),330				select_chain,331				block_import: aura_block_import,332				proposer_factory: proposer,333				sync_oracle: network.clone(),334				inherent_data_providers: inherent_data_providers.clone(),335				force_authoring,336				backoff_authoring_blocks,337				keystore: keystore_container.sync_keystore(),338				can_author_with,339				block_proposal_slot_portion: SlotProportion::new(2f32 / 3f32),340				telemetry: telemetry.as_ref().map(|x| x.handle()),341			}342		)?;343344		// the AURA authoring task is considered essential, i.e. if it345		// fails we take down the service with it.346		task_manager.spawn_essential_handle().spawn_blocking("aura", aura);347348		// if the node isn't actively participating in consensus then it doesn't349		// need a keystore, regardless of which protocol we use below.350		let keystore = if role.is_authority() {351			Some(keystore_container.sync_keystore())352		} else {353			None354		};355356		let grandpa_config = sc_finality_grandpa::Config {357			// FIXME #1578 make this available through chainspec358			gossip_duration: Duration::from_millis(333),359			justification_period: 512,360			name: Some(name),361			observer_enabled: false,362			keystore,363			is_authority: role.is_authority(),364			telemetry: telemetry.as_ref().map(|x| x.handle()),365		};366367		if enable_grandpa {368			// start the full GRANDPA voter369			// NOTE: non-authorities could run the GRANDPA observer protocol, but at370			// this point the full voter should provide better guarantees of block371			// and vote data availability than the observer. The observer has not372			// been tested extensively yet and having most nodes in a network run it373			// could lead to finality stalls.374			let grandpa_config = sc_finality_grandpa::GrandpaParams {375				config: grandpa_config,376				link: grandpa_link,377				network,378				telemetry: telemetry.as_ref().map(|x| x.handle()),379				voting_rule: sc_finality_grandpa::VotingRulesBuilder::default().build(),380				prometheus_registry,381				shared_voter_state: SharedVoterState::empty(),382			};383384			// the GRANDPA voter task is considered infallible, i.e.385			// if it fails we take down the service with it.386			task_manager.spawn_essential_handle().spawn_blocking(387				"grandpa-voter",388				sc_finality_grandpa::run_grandpa_voter(grandpa_config)?389			);390		}391	}392393	network_starter.start_network();394	Ok(task_manager)395}396/// Builds a new service for a light client.397pub fn new_light(config: Configuration) -> Result<TaskManager, ServiceError> {398	let telemetry = config.telemetry_endpoints.clone()399		.filter(|x| !x.is_empty())400		.map(|endpoints| -> Result<_, sc_telemetry::Error> {401			let worker = TelemetryWorker::new(16)?;402			let telemetry = worker.handle().new_telemetry(endpoints);403			Ok((worker, telemetry))404		})405		.transpose()?;406407	let (client, backend, keystore_container, mut task_manager, on_demand) =408		sc_service::new_light_parts::<Block, RuntimeApi, Executor>(409			&config,410			telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),411		)?;412413	let mut telemetry = telemetry414		.map(|(worker, telemetry)| {415			task_manager.spawn_handle().spawn("telemetry", worker.run());416			telemetry417		});418419	let select_chain = sc_consensus::LongestChain::new(backend.clone());420421	let transaction_pool = Arc::new(sc_transaction_pool::BasicPool::new_light(422		config.transaction_pool.clone(),423		config.prometheus_registry(),424		task_manager.spawn_handle(),425		client.clone(),426		on_demand.clone(),427	));428429	let (grandpa_block_import, _) = sc_finality_grandpa::block_import(430		client.clone(),431		&(client.clone() as Arc<_>),432		select_chain.clone(),433		telemetry.as_ref().map(|x| x.handle()),434	)?;435436	let import_queue = sc_consensus_aura::import_queue::<AuraPair, _, _, _, _, _>(437		ImportQueueParams {438			slot_duration: sc_consensus_aura::slot_duration(&*client)?,439			block_import: grandpa_block_import.clone(),440			justification_import: Some(Box::new(grandpa_block_import)),441			client: client.clone(),442			inherent_data_providers: InherentDataProviders::new(),443			spawner: &task_manager.spawn_essential_handle(),444			registry: config.prometheus_registry(),445			can_author_with: sp_consensus::NeverCanAuthor,446			check_for_equivocation: Default::default(),447			telemetry: telemetry.as_ref().map(|x| x.handle()),448		}449	)?;450451	let light_deps = crate::rpc::LightDeps {452		remote_blockchain: backend.remote_blockchain(),453		fetcher: on_demand.clone(),454		client: client.clone(),455		pool: transaction_pool.clone(),456	};457458	let rpc_extensions = crate::rpc::create_light(light_deps);459460	let (network, network_status_sinks, system_rpc_tx, network_starter) =461		sc_service::build_network(sc_service::BuildNetworkParams {462			config: &config,463			client: client.clone(),464			transaction_pool: transaction_pool.clone(),465			spawn_handle: task_manager.spawn_handle(),466			import_queue,467			on_demand: Some(on_demand.clone()),468			block_announce_validator_builder: None,469		})?;470471	if config.offchain_worker.enabled {472		sc_service::build_offchain_workers(473			&config, task_manager.spawn_handle(), client.clone(), network.clone(),474		);475	}476477	sc_service::spawn_tasks(sc_service::SpawnTasksParams {478		remote_blockchain: Some(backend.remote_blockchain()),479		transaction_pool,480		task_manager: &mut task_manager,481		on_demand: Some(on_demand),482		rpc_extensions_builder: Box::new(sc_service::NoopRpcExtensionBuilder(rpc_extensions)),483		config,484		client,485		keystore: keystore_container.sync_keystore(),486		backend,487		network,488		network_status_sinks,489		system_rpc_tx,490		telemetry: telemetry.as_mut(),491	})?;492493	network_starter.start_network();494495	Ok(task_manager)496}