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

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;13use std::time::Duration;14use futures::StreamExt;1516// Local Runtime Types17use nft_runtime::RuntimeApi;1819// Cumulus Imports20use cumulus_client_consensus_aura::{build_aura_consensus, BuildAuraConsensusParams, SlotProportion};21use cumulus_client_consensus_common::ParachainConsensus;22use cumulus_client_network::build_block_announce_validator;23use cumulus_client_service::{24	prepare_node_config, start_collator, start_full_node, StartCollatorParams, StartFullNodeParams,25};26use cumulus_primitives_core::ParaId;2728// Substrate Imports29use sc_client_api::ExecutorProvider;30pub use sc_executor::NativeExecutor;31use sc_executor::native_executor_instance;32use sc_network::NetworkService;33use sc_service::{BasePath, Configuration, PartialComponents, Role, TaskManager};34use sc_telemetry::{Telemetry, TelemetryHandle, TelemetryWorker, TelemetryWorkerHandle};35use sp_consensus::SlotData;36use sp_keystore::SyncCryptoStorePtr;37use sp_runtime::traits::BlakeTwo256;38use substrate_prometheus_endpoint::Registry;39use sc_client_api::BlockchainEvents;4041// Frontier Imports42use fc_rpc_core::types::FilterPool;43use fc_rpc_core::types::PendingTransactions;44use fc_mapping_sync::{MappingSyncWorker, SyncStrategy};4546// Runtime type overrides47type BlockNumber = u32;48type Header = sp_runtime::generic::Header<BlockNumber, sp_runtime::traits::BlakeTwo256>;49pub type Block = sp_runtime::generic::Block<Header, sp_runtime::OpaqueExtrinsic>;50type Hash = sp_core::H256;5152// Native executor instance.53native_executor_instance!(54	pub ParachainRuntimeExecutor,55	nft_runtime::api::dispatch,56	nft_runtime::native_version,57	frame_benchmarking::benchmarking::HostFunctions,58);5960pub fn open_frontier_backend(config: &Configuration) -> Result<Arc<fc_db::Backend<Block>>, String> {61	let config_dir = config62		.base_path63		.as_ref()64		.map(|base_path| base_path.config_dir(config.chain_spec.id()))65		.unwrap_or_else(|| {66			BasePath::from_project("", "", "nft").config_dir(config.chain_spec.id())67		});68	let database_dir = config_dir.join("frontier").join("db");6970	Ok(Arc::new(fc_db::Backend::<Block>::new(71		&fc_db::DatabaseSettings {72			source: fc_db::DatabaseSettingsSrc::RocksDb {73				path: database_dir,74				cache_size: 0,75			},76		},77	)?))78}7980type Executor = ParachainRuntimeExecutor;8182type FullClient = sc_service::TFullClient<Block, RuntimeApi, Executor>;83type FullBackend = sc_service::TFullBackend<Block>;84type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;8586/// Starts a `ServiceBuilder` for a full service.87///88/// Use this macro if you don't actually need the full service, but just the builder in order to89/// be able to perform chain operations.90#[allow(clippy::type_complexity)]91pub fn new_partial<BIQ>(92	config: &Configuration,93	build_import_queue: BIQ,94) -> Result<95	PartialComponents<96		FullClient,97		FullBackend,98		FullSelectChain,99		sc_consensus::DefaultImportQueue<Block, FullClient>,100		sc_transaction_pool::FullPool<Block, FullClient>,101		(102			Option<Telemetry>,103			PendingTransactions,104			Option<FilterPool>,105			Arc<fc_db::Backend<Block>>,106			Option<TelemetryWorkerHandle>,107		),108	>,109	sc_service::Error,110>111where112	sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,113	Executor: sc_executor::NativeExecutionDispatch + 'static,114	BIQ: FnOnce(115		Arc<FullClient>,116		&Configuration,117		Option<TelemetryHandle>,118		&TaskManager,119	) -> Result<sc_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error>,120{121	let _telemetry = config122		.telemetry_endpoints123		.clone()124		.filter(|x| !x.is_empty())125		.map(|endpoints| -> Result<_, sc_telemetry::Error> {126			let worker = TelemetryWorker::new(16)?;127			let telemetry = worker.handle().new_telemetry(endpoints);128			Ok((worker, telemetry))129		})130		.transpose()?;131132	let telemetry = config133		.telemetry_endpoints134		.clone()135		.filter(|x| !x.is_empty())136		.map(|endpoints| -> Result<_, sc_telemetry::Error> {137			let worker = TelemetryWorker::new(16)?;138			let telemetry = worker.handle().new_telemetry(endpoints);139			Ok((worker, telemetry))140		})141		.transpose()?;142143	let (client, backend, keystore_container, task_manager) =144		sc_service::new_full_parts::<Block, RuntimeApi, Executor>(145			config,146			telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),147		)?;148	let client = Arc::new(client);149150	let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());151152	let telemetry = telemetry.map(|(worker, telemetry)| {153		task_manager.spawn_handle().spawn("telemetry", worker.run());154		telemetry155	});156157	let select_chain = sc_consensus::LongestChain::new(backend.clone());158159	let transaction_pool = sc_transaction_pool::BasicPool::new_full(160		config.transaction_pool.clone(),161		config.role.is_authority().into(),162		config.prometheus_registry(),163		task_manager.spawn_essential_handle(),164		client.clone(),165	);166167	let pending_transactions: PendingTransactions = Some(Arc::new(Mutex::new(HashMap::new())));168169	let filter_pool: Option<FilterPool> = Some(Arc::new(Mutex::new(BTreeMap::new())));170171	let frontier_backend = open_frontier_backend(config)?;172173	let import_queue = build_import_queue(174		client.clone(),175		config,176		telemetry.as_ref().map(|telemetry| telemetry.handle()),177		&task_manager,178	)?;179180	let params = PartialComponents {181		backend,182		client,183		import_queue,184		keystore_container,185		task_manager,186		transaction_pool,187		select_chain,188		other: (189			telemetry,190			pending_transactions,191			filter_pool,192			frontier_backend,193			telemetry_worker_handle,194		),195	};196197	Ok(params)198}199200/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.201///202/// This is the actual implementation that is abstract over the executor and the runtime api.203#[sc_tracing::logging::prefix_logs_with("Parachain")]204async fn start_node_impl<BIQ, BIC>(205	parachain_config: Configuration,206	polkadot_config: Configuration,207	id: ParaId,208	build_import_queue: BIQ,209	build_consensus: BIC,210) -> sc_service::error::Result<(TaskManager, Arc<FullClient>)>211where212	sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,213	Executor: sc_executor::NativeExecutionDispatch + 'static,214	BIQ: FnOnce(215		Arc<FullClient>,216		&Configuration,217		Option<TelemetryHandle>,218		&TaskManager,219	) -> Result<sc_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error>,220	BIC: FnOnce(221		Arc<FullClient>,222		Option<&Registry>,223		Option<TelemetryHandle>,224		&TaskManager,225		&polkadot_service::NewFull<polkadot_service::Client>,226		Arc<sc_transaction_pool::FullPool<Block, FullClient>>,227		Arc<NetworkService<Block, Hash>>,228		SyncCryptoStorePtr,229		bool,230	) -> Result<Box<dyn ParachainConsensus<Block>>, sc_service::Error>,231{232	if matches!(parachain_config.role, Role::Light) {233		return Err("Light client not supported!".into());234	}235236	let parachain_config = prepare_node_config(parachain_config);237238	let params = new_partial::<BIQ>(&parachain_config, build_import_queue)?;239	let (240		mut telemetry,241		pending_transactions,242		filter_pool,243		frontier_backend,244		telemetry_worker_handle,245	) = params.other;246247	let relay_chain_full_node =248		cumulus_client_service::build_polkadot_full_node(polkadot_config, telemetry_worker_handle)249			.map_err(|e| match e {250				polkadot_service::Error::Sub(x) => x,251				s => format!("{}", s).into(),252			})?;253254	let client = params.client.clone();255	let backend = params.backend.clone();256	let block_announce_validator = build_block_announce_validator(257		relay_chain_full_node.client.clone(),258		id,259		Box::new(relay_chain_full_node.network.clone()),260		relay_chain_full_node.backend.clone(),261	);262263	let force_authoring = parachain_config.force_authoring;264	let validator = parachain_config.role.is_authority();265	let prometheus_registry = parachain_config.prometheus_registry().cloned();266	let transaction_pool = params.transaction_pool.clone();267	let mut task_manager = params.task_manager;268	let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);269270	let (network, system_rpc_tx, start_network) =271		sc_service::build_network(sc_service::BuildNetworkParams {272			config: &parachain_config,273			client: client.clone(),274			transaction_pool: transaction_pool.clone(),275			spawn_handle: task_manager.spawn_handle(),276			import_queue: import_queue.clone(),277			on_demand: None,278			block_announce_validator_builder: Some(Box::new(|_| block_announce_validator)),279			warp_sync: None,280		})?;281282	let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());283	let rpc_client = client.clone();284	let rpc_pool = transaction_pool.clone();285	let select_chain = params.select_chain.clone();286	let is_authority = parachain_config.role.clone().is_authority();287	let rpc_network = network.clone();288289	let rpc_frontier_backend = frontier_backend.clone();290	let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {291		let full_deps = nft_rpc::FullDeps {292			backend: rpc_frontier_backend.clone(),293			deny_unsafe,294			client: rpc_client.clone(),295			pool: rpc_pool.clone(),296			// TODO: Unhardcode297			enable_dev_signer: false,298			filter_pool: filter_pool.clone(),299			network: rpc_network.clone(),300			pending_transactions: pending_transactions.clone(),301			select_chain: select_chain.clone(),302			is_authority,303			// TODO: Unhardcode304			max_past_logs: 10000,305		};306307		Ok(nft_rpc::create_full::<_, _, _, RuntimeApi, _>(308			full_deps,309			subscription_executor.clone(),310		))311	});312313	task_manager.spawn_essential_handle().spawn(314		"frontier-mapping-sync-worker",315		MappingSyncWorker::new(316			client.import_notification_stream(),317			Duration::new(6, 0),318			client.clone(),319			backend.clone(),320			frontier_backend.clone(),321			SyncStrategy::Normal,322		)323		.for_each(|()| futures::future::ready(())),324	);325326	sc_service::spawn_tasks(sc_service::SpawnTasksParams {327		on_demand: None,328		remote_blockchain: None,329		rpc_extensions_builder,330		client: client.clone(),331		transaction_pool: transaction_pool.clone(),332		task_manager: &mut task_manager,333		config: parachain_config,334		keystore: params.keystore_container.sync_keystore(),335		backend: backend.clone(),336		network: network.clone(),337		system_rpc_tx,338		telemetry: telemetry.as_mut(),339	})?;340341	let announce_block = {342		let network = network.clone();343		Arc::new(move |hash, data| network.announce_block(hash, data))344	};345346	if validator {347		let parachain_consensus = build_consensus(348			client.clone(),349			prometheus_registry.as_ref(),350			telemetry.as_ref().map(|t| t.handle()),351			&task_manager,352			&relay_chain_full_node,353			transaction_pool,354			network,355			params.keystore_container.sync_keystore(),356			force_authoring,357		)?;358359		let spawner = task_manager.spawn_handle();360361		let params = StartCollatorParams {362			para_id: id,363			block_status: client.clone(),364			announce_block,365			client: client.clone(),366			task_manager: &mut task_manager,367			relay_chain_full_node,368			spawner,369			parachain_consensus,370			import_queue,371		};372373		start_collator(params).await?;374	} else {375		let params = StartFullNodeParams {376			client: client.clone(),377			announce_block,378			task_manager: &mut task_manager,379			para_id: id,380			relay_chain_full_node,381		};382383		start_full_node(params)?;384	}385386	start_network.start_network();387388	Ok((task_manager, client))389}390391/// Build the import queue for the the parachain runtime.392pub fn parachain_build_import_queue(393	client: Arc<FullClient>,394	config: &Configuration,395	telemetry: Option<TelemetryHandle>,396	task_manager: &TaskManager,397) -> Result<sc_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error> {398	let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;399400	cumulus_client_consensus_aura::import_queue::<401		sp_consensus_aura::sr25519::AuthorityPair,402		_,403		_,404		_,405		_,406		_,407		_,408	>(cumulus_client_consensus_aura::ImportQueueParams {409		block_import: client.clone(),410		client: client.clone(),411		create_inherent_data_providers: move |_, _| async move {412			let time = sp_timestamp::InherentDataProvider::from_system_time();413414			let slot =415				sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_duration(416					*time,417					slot_duration.slot_duration(),418				);419420			Ok((time, slot))421		},422		registry: config.prometheus_registry(),423		can_author_with: sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()),424		spawner: &task_manager.spawn_essential_handle(),425		telemetry,426	})427	.map_err(Into::into)428}429430/// Start a normal parachain node.431pub async fn start_node(432	parachain_config: Configuration,433	polkadot_config: Configuration,434	id: ParaId,435) -> sc_service::error::Result<(TaskManager, Arc<FullClient>)> {436	start_node_impl::<_, _>(437		parachain_config,438		polkadot_config,439		id,440		parachain_build_import_queue,441		|client,442		 prometheus_registry,443		 telemetry,444		 task_manager,445		 relay_chain_node,446		 transaction_pool,447		 sync_oracle,448		 keystore,449		 force_authoring| {450			let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;451452			let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(453				task_manager.spawn_handle(),454				client.clone(),455				transaction_pool,456				prometheus_registry,457				telemetry.clone(),458			);459460			let relay_chain_backend = relay_chain_node.backend.clone();461			let relay_chain_client = relay_chain_node.client.clone();462			Ok(build_aura_consensus::<463				sp_consensus_aura::sr25519::AuthorityPair,464				_,465				_,466				_,467				_,468				_,469				_,470				_,471				_,472				_,473			>(BuildAuraConsensusParams {474				proposer_factory,475				create_inherent_data_providers: move |_, (relay_parent, validation_data)| {476					let parachain_inherent =477					cumulus_primitives_parachain_inherent::ParachainInherentData::create_at_with_client(478						relay_parent,479						&relay_chain_client,480						&*relay_chain_backend,481						&validation_data,482						id,483					);484					async move {485						let time = sp_timestamp::InherentDataProvider::from_system_time();486487						let slot =488						sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_duration(489							*time,490							slot_duration.slot_duration(),491						);492493						let parachain_inherent = parachain_inherent.ok_or_else(|| {494							Box::<dyn std::error::Error + Send + Sync>::from(495								"Failed to create parachain inherent",496							)497						})?;498						Ok((time, slot, parachain_inherent))499					}500				},501				block_import: client.clone(),502				relay_chain_client: relay_chain_node.client.clone(),503				relay_chain_backend: relay_chain_node.backend.clone(),504				para_client: client,505				backoff_authoring_blocks: Option::<()>::None,506				sync_oracle,507				keystore,508				force_authoring,509				slot_duration,510				// We got around 500ms for proposing511				block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),512				telemetry,513				max_block_proposal_slot_portion: None,514			}))515		},516	)517	.await518}