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

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//78// std9use std::sync::Arc;10use std::sync::Mutex;11use std::collections::BTreeMap;12use std::collections::HashMap;1314// Local Runtime Types15use nft_runtime::RuntimeApi;1617// Cumulus Imports18use cumulus_client_consensus_aura::{build_aura_consensus, BuildAuraConsensusParams, SlotProportion};19use cumulus_client_consensus_common::ParachainConsensus;20use cumulus_client_network::build_block_announce_validator;21use cumulus_client_service::{22	prepare_node_config, start_collator, start_full_node, StartCollatorParams, StartFullNodeParams,23};24use cumulus_primitives_core::ParaId;2526// Substrate Imports27use sc_client_api::ExecutorProvider;28pub use sc_executor::NativeExecutor;29use sc_executor::native_executor_instance;30use sc_network::NetworkService;31use sc_service::{BasePath, Configuration, PartialComponents, Role, TaskManager};32use sc_telemetry::{Telemetry, TelemetryHandle, TelemetryWorker, TelemetryWorkerHandle};33use sp_consensus::SlotData;34use sp_keystore::SyncCryptoStorePtr;35use sp_runtime::traits::BlakeTwo256;36use substrate_prometheus_endpoint::Registry;3738// Frontier Imports39use fc_rpc_core::types::FilterPool;40use fc_rpc_core::types::PendingTransactions;4142// Runtime type overrides43type BlockNumber = u32;44type Header = sp_runtime::generic::Header<BlockNumber, sp_runtime::traits::BlakeTwo256>;45pub type Block = sp_runtime::generic::Block<Header, sp_runtime::OpaqueExtrinsic>;46type Hash = sp_core::H256;4748// Native executor instance.49native_executor_instance!(50	pub ParachainRuntimeExecutor,51	nft_runtime::api::dispatch,52	nft_runtime::native_version,53	frame_benchmarking::benchmarking::HostFunctions,54);5556pub fn open_frontier_backend(config: &Configuration) -> Result<Arc<fc_db::Backend<Block>>, String> {57	let config_dir = config58		.base_path59		.as_ref()60		.map(|base_path| base_path.config_dir(config.chain_spec.id()))61		.unwrap_or_else(|| {62			BasePath::from_project("", "", "nft").config_dir(config.chain_spec.id())63		});64	let database_dir = config_dir.join("frontier").join("db");6566	Ok(Arc::new(fc_db::Backend::<Block>::new(67		&fc_db::DatabaseSettings {68			source: fc_db::DatabaseSettingsSrc::RocksDb {69				path: database_dir,70				cache_size: 0,71			},72		},73	)?))74}7576type Executor = ParachainRuntimeExecutor;7778type FullClient = sc_service::TFullClient<Block, RuntimeApi, Executor>;79type FullBackend = sc_service::TFullBackend<Block>;80type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;8182/// Starts a `ServiceBuilder` for a full service.83///84/// Use this macro if you don't actually need the full service, but just the builder in order to85/// be able to perform chain operations.86#[allow(clippy::type_complexity)]87pub fn new_partial<BIQ>(88	config: &Configuration,89	build_import_queue: BIQ,90) -> Result<91	PartialComponents<92		FullClient,93		FullBackend,94		FullSelectChain,95		sp_consensus::DefaultImportQueue<Block, FullClient>,96		sc_transaction_pool::FullPool<Block, FullClient>,97		(98			Option<Telemetry>,99			PendingTransactions,100			Option<FilterPool>,101			Arc<fc_db::Backend<Block>>,102			Option<TelemetryWorkerHandle>,103		),104	>,105	sc_service::Error,106>107where108	sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,109	Executor: sc_executor::NativeExecutionDispatch + 'static,110	BIQ: FnOnce(111		Arc<FullClient>,112		&Configuration,113		Option<TelemetryHandle>,114		&TaskManager,115	) -> Result<sp_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error>,116{117	let _telemetry = config118		.telemetry_endpoints119		.clone()120		.filter(|x| !x.is_empty())121		.map(|endpoints| -> Result<_, sc_telemetry::Error> {122			let worker = TelemetryWorker::new(16)?;123			let telemetry = worker.handle().new_telemetry(endpoints);124			Ok((worker, telemetry))125		})126		.transpose()?;127128	let telemetry = config129		.telemetry_endpoints130		.clone()131		.filter(|x| !x.is_empty())132		.map(|endpoints| -> Result<_, sc_telemetry::Error> {133			let worker = TelemetryWorker::new(16)?;134			let telemetry = worker.handle().new_telemetry(endpoints);135			Ok((worker, telemetry))136		})137		.transpose()?;138139	let (client, backend, keystore_container, task_manager) =140		sc_service::new_full_parts::<Block, RuntimeApi, Executor>(141			config,142			telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),143		)?;144	let client = Arc::new(client);145146	let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());147148	let telemetry = telemetry.map(|(worker, telemetry)| {149		task_manager.spawn_handle().spawn("telemetry", worker.run());150		telemetry151	});152153	let select_chain = sc_consensus::LongestChain::new(backend.clone());154155	let transaction_pool = sc_transaction_pool::BasicPool::new_full(156		config.transaction_pool.clone(),157		config.role.is_authority().into(),158		config.prometheus_registry(),159		task_manager.spawn_essential_handle(),160		client.clone(),161	);162163	let pending_transactions: PendingTransactions = Some(Arc::new(Mutex::new(HashMap::new())));164165	let filter_pool: Option<FilterPool> = Some(Arc::new(Mutex::new(BTreeMap::new())));166167	let frontier_backend = open_frontier_backend(config)?;168169	let import_queue = build_import_queue(170		client.clone(),171		config,172		telemetry.as_ref().map(|telemetry| telemetry.handle()),173		&task_manager,174	)?;175176	let params = PartialComponents {177		backend,178		client,179		import_queue,180		keystore_container,181		task_manager,182		transaction_pool,183		select_chain,184		other: (185			telemetry,186			pending_transactions,187			filter_pool,188			frontier_backend,189			telemetry_worker_handle,190		),191	};192193	Ok(params)194}195196/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.197///198/// This is the actual implementation that is abstract over the executor and the runtime api.199#[sc_tracing::logging::prefix_logs_with("Parachain")]200async fn start_node_impl<BIQ, BIC>(201	parachain_config: Configuration,202	polkadot_config: Configuration,203	id: ParaId,204	build_import_queue: BIQ,205	build_consensus: BIC,206) -> sc_service::error::Result<(TaskManager, Arc<FullClient>)>207where208	sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,209	Executor: sc_executor::NativeExecutionDispatch + 'static,210	BIQ: FnOnce(211		Arc<FullClient>,212		&Configuration,213		Option<TelemetryHandle>,214		&TaskManager,215	) -> Result<sp_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error>,216	BIC: FnOnce(217		Arc<FullClient>,218		Option<&Registry>,219		Option<TelemetryHandle>,220		&TaskManager,221		&polkadot_service::NewFull<polkadot_service::Client>,222		Arc<sc_transaction_pool::FullPool<Block, FullClient>>,223		Arc<NetworkService<Block, Hash>>,224		SyncCryptoStorePtr,225		bool,226	) -> Result<Box<dyn ParachainConsensus<Block>>, sc_service::Error>,227{228	if matches!(parachain_config.role, Role::Light) {229		return Err("Light client not supported!".into());230	}231232	let parachain_config = prepare_node_config(parachain_config);233234	let params = new_partial::<BIQ>(&parachain_config, build_import_queue)?;235	let (236		mut telemetry,237		pending_transactions,238		filter_pool,239		frontier_backend,240		telemetry_worker_handle,241	) = params.other;242243	let relay_chain_full_node =244		cumulus_client_service::build_polkadot_full_node(polkadot_config, telemetry_worker_handle)245			.map_err(|e| match e {246				polkadot_service::Error::Sub(x) => x,247				s => format!("{}", s).into(),248			})?;249250	let client = params.client.clone();251	let backend = params.backend.clone();252	let block_announce_validator = build_block_announce_validator(253		relay_chain_full_node.client.clone(),254		id,255		Box::new(relay_chain_full_node.network.clone()),256		relay_chain_full_node.backend.clone(),257	);258259	let force_authoring = parachain_config.force_authoring;260	let validator = parachain_config.role.is_authority();261	let prometheus_registry = parachain_config.prometheus_registry().cloned();262	let transaction_pool = params.transaction_pool.clone();263	let mut task_manager = params.task_manager;264	let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);265	let (network, system_rpc_tx, start_network) =266		sc_service::build_network(sc_service::BuildNetworkParams {267			config: &parachain_config,268			client: client.clone(),269			transaction_pool: transaction_pool.clone(),270			spawn_handle: task_manager.spawn_handle(),271			import_queue: import_queue.clone(),272			on_demand: None,273			block_announce_validator_builder: Some(Box::new(|_| block_announce_validator)),274		})?;275276	let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());277	let rpc_client = client.clone();278	let rpc_pool = transaction_pool.clone();279	let select_chain = params.select_chain.clone();280	let is_authority = parachain_config.role.clone().is_authority();281	let rpc_network = network.clone();282	let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {283		let full_deps = nft_rpc::FullDeps {284			backend: frontier_backend.clone(),285			deny_unsafe,286			client: rpc_client.clone(),287			pool: rpc_pool.clone(),288			// TODO: Unhardcode289			enable_dev_signer: false,290			filter_pool: filter_pool.clone(),291			network: rpc_network.clone(),292			pending_transactions: pending_transactions.clone(),293			select_chain: select_chain.clone(),294			is_authority,295			// TODO: Unhardcode296			max_past_logs: 10000,297		};298299		nft_rpc::create_full::<_, _, _, RuntimeApi, _>(full_deps, subscription_executor.clone())300	});301302	sc_service::spawn_tasks(sc_service::SpawnTasksParams {303		on_demand: None,304		remote_blockchain: None,305		rpc_extensions_builder,306		client: client.clone(),307		transaction_pool: transaction_pool.clone(),308		task_manager: &mut task_manager,309		config: parachain_config,310		keystore: params.keystore_container.sync_keystore(),311		backend: backend.clone(),312		network: network.clone(),313		system_rpc_tx,314		telemetry: telemetry.as_mut(),315	})?;316317	let announce_block = {318		let network = network.clone();319		Arc::new(move |hash, data| network.announce_block(hash, data))320	};321322	if validator {323		let parachain_consensus = build_consensus(324			client.clone(),325			prometheus_registry.as_ref(),326			telemetry.as_ref().map(|t| t.handle()),327			&task_manager,328			&relay_chain_full_node,329			transaction_pool,330			network,331			params.keystore_container.sync_keystore(),332			force_authoring,333		)?;334335		let spawner = task_manager.spawn_handle();336337		let params = StartCollatorParams {338			para_id: id,339			block_status: client.clone(),340			announce_block,341			client: client.clone(),342			task_manager: &mut task_manager,343			relay_chain_full_node,344			spawner,345			parachain_consensus,346			import_queue,347		};348349		start_collator(params).await?;350	} else {351		let params = StartFullNodeParams {352			client: client.clone(),353			announce_block,354			task_manager: &mut task_manager,355			para_id: id,356			relay_chain_full_node,357		};358359		start_full_node(params)?;360	}361362	start_network.start_network();363364	Ok((task_manager, client))365}366367/// Build the import queue for the the parachain runtime.368pub fn parachain_build_import_queue(369	client: Arc<FullClient>,370	config: &Configuration,371	telemetry: Option<TelemetryHandle>,372	task_manager: &TaskManager,373) -> Result<sp_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error> {374	let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;375376	cumulus_client_consensus_aura::import_queue::<377		sp_consensus_aura::sr25519::AuthorityPair,378		_,379		_,380		_,381		_,382		_,383		_,384	>(cumulus_client_consensus_aura::ImportQueueParams {385		block_import: client.clone(),386		client: client.clone(),387		create_inherent_data_providers: move |_, _| async move {388			let time = sp_timestamp::InherentDataProvider::from_system_time();389390			let slot =391				sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_duration(392					*time,393					slot_duration.slot_duration(),394				);395396			Ok((time, slot))397		},398		registry: config.prometheus_registry(),399		can_author_with: sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()),400		spawner: &task_manager.spawn_essential_handle(),401		telemetry,402	})403	.map_err(Into::into)404}405406/// Start a normal parachain node.407pub async fn start_node(408	parachain_config: Configuration,409	polkadot_config: Configuration,410	id: ParaId,411) -> sc_service::error::Result<(TaskManager, Arc<FullClient>)> {412	start_node_impl::<_, _>(413		parachain_config,414		polkadot_config,415		id,416		parachain_build_import_queue,417		|client,418		 prometheus_registry,419		 telemetry,420		 task_manager,421		 relay_chain_node,422		 transaction_pool,423		 sync_oracle,424		 keystore,425		 force_authoring| {426			let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;427428			let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(429				task_manager.spawn_handle(),430				client.clone(),431				transaction_pool,432				prometheus_registry,433				telemetry.clone(),434			);435436			let relay_chain_backend = relay_chain_node.backend.clone();437			let relay_chain_client = relay_chain_node.client.clone();438			Ok(build_aura_consensus::<439				sp_consensus_aura::sr25519::AuthorityPair,440				_,441				_,442				_,443				_,444				_,445				_,446				_,447				_,448				_,449			>(BuildAuraConsensusParams {450				proposer_factory,451				create_inherent_data_providers: move |_, (relay_parent, validation_data)| {452					let parachain_inherent =453					cumulus_primitives_parachain_inherent::ParachainInherentData::create_at_with_client(454						relay_parent,455						&relay_chain_client,456						&*relay_chain_backend,457						&validation_data,458						id,459					);460					async move {461						let time = sp_timestamp::InherentDataProvider::from_system_time();462463						let slot =464						sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_duration(465							*time,466							slot_duration.slot_duration(),467						);468469						let parachain_inherent = parachain_inherent.ok_or_else(|| {470							Box::<dyn std::error::Error + Send + Sync>::from(471								"Failed to create parachain inherent",472							)473						})?;474						Ok((time, slot, parachain_inherent))475					}476				},477				block_import: client.clone(),478				relay_chain_client: relay_chain_node.client.clone(),479				relay_chain_backend: relay_chain_node.backend.clone(),480				para_client: client,481				backoff_authoring_blocks: Option::<()>::None,482				sync_oracle,483				keystore,484				force_authoring,485				slot_duration,486				// We got around 500ms for proposing487				block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),488				telemetry,489			}))490		},491	)492	.await493}