git.delta.rocks / unique-network / refs/commits / 0660dd124ff7

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::time::Duration;13use fc_rpc_core::types::FeeHistoryCache;14use futures::StreamExt;1516use unique_rpc::overrides_handle;17// Local Runtime Types18use unique_runtime::RuntimeApi;1920// Cumulus Imports21use cumulus_client_consensus_aura::{AuraConsensus, BuildAuraConsensusParams, SlotProportion};22use cumulus_client_consensus_common::ParachainConsensus;23use cumulus_client_service::{24	prepare_node_config, start_collator, start_full_node, StartCollatorParams, StartFullNodeParams,25};26use cumulus_client_network::BlockAnnounceValidator;27use cumulus_primitives_core::ParaId;28use cumulus_relay_chain_interface::RelayChainInterface;29use cumulus_relay_chain_local::build_relay_chain_interface;3031// Substrate Imports32use sc_client_api::ExecutorProvider;33use sc_executor::NativeElseWasmExecutor;34use sc_executor::NativeExecutionDispatch;35use sc_network::NetworkService;36use sc_service::{BasePath, Configuration, PartialComponents, Role, TaskManager};37use sc_telemetry::{Telemetry, TelemetryHandle, TelemetryWorker, TelemetryWorkerHandle};38use sp_consensus::SlotData;39use sp_keystore::SyncCryptoStorePtr;40use sp_runtime::traits::BlakeTwo256;41use substrate_prometheus_endpoint::Registry;42use sc_client_api::BlockchainEvents;4344// Frontier Imports45use fc_rpc_core::types::FilterPool;46use fc_mapping_sync::{MappingSyncWorker, SyncStrategy};4748// Runtime type overrides49type BlockNumber = u32;50type Header = sp_runtime::generic::Header<BlockNumber, sp_runtime::traits::BlakeTwo256>;51pub type Block = sp_runtime::generic::Block<Header, sp_runtime::OpaqueExtrinsic>;52type Hash = sp_core::H256;5354/// Native executor instance.55pub struct ParachainRuntimeExecutor;5657impl NativeExecutionDispatch for ParachainRuntimeExecutor {58	type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;5960	fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {61		unique_runtime::api::dispatch(method, data)62	}6364	fn native_version() -> sc_executor::NativeVersion {65		unique_runtime::native_version()66	}67}6869pub fn open_frontier_backend(config: &Configuration) -> Result<Arc<fc_db::Backend<Block>>, String> {70	let config_dir = config71		.base_path72		.as_ref()73		.map(|base_path| base_path.config_dir(config.chain_spec.id()))74		.unwrap_or_else(|| {75			BasePath::from_project("", "", "unique").config_dir(config.chain_spec.id())76		});77	let database_dir = config_dir.join("frontier").join("db");7879	Ok(Arc::new(fc_db::Backend::<Block>::new(80		&fc_db::DatabaseSettings {81			source: fc_db::DatabaseSettingsSrc::RocksDb {82				path: database_dir,83				cache_size: 0,84			},85		},86	)?))87}8889type ExecutorDispatch = ParachainRuntimeExecutor;9091type FullClient =92	sc_service::TFullClient<Block, RuntimeApi, NativeElseWasmExecutor<ExecutorDispatch>>;93type FullBackend = sc_service::TFullBackend<Block>;94type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;9596/// Starts a `ServiceBuilder` for a full service.97///98/// Use this macro if you don't actually need the full service, but just the builder in order to99/// be able to perform chain operations.100#[allow(clippy::type_complexity)]101pub fn new_partial<BIQ>(102	config: &Configuration,103	build_import_queue: BIQ,104) -> Result<105	PartialComponents<106		FullClient,107		FullBackend,108		FullSelectChain,109		sc_consensus::DefaultImportQueue<Block, FullClient>,110		sc_transaction_pool::FullPool<Block, FullClient>,111		(112			Option<Telemetry>,113			Option<FilterPool>,114			Arc<fc_db::Backend<Block>>,115			Option<TelemetryWorkerHandle>,116			FeeHistoryCache,117		),118	>,119	sc_service::Error,120>121where122	sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,123	ExecutorDispatch: NativeExecutionDispatch + 'static,124	BIQ: FnOnce(125		Arc<FullClient>,126		&Configuration,127		Option<TelemetryHandle>,128		&TaskManager,129	) -> Result<sc_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error>,130{131	let _telemetry = config132		.telemetry_endpoints133		.clone()134		.filter(|x| !x.is_empty())135		.map(|endpoints| -> Result<_, sc_telemetry::Error> {136			let worker = TelemetryWorker::new(16)?;137			let telemetry = worker.handle().new_telemetry(endpoints);138			Ok((worker, telemetry))139		})140		.transpose()?;141142	let telemetry = config143		.telemetry_endpoints144		.clone()145		.filter(|x| !x.is_empty())146		.map(|endpoints| -> Result<_, sc_telemetry::Error> {147			let worker = TelemetryWorker::new(16)?;148			let telemetry = worker.handle().new_telemetry(endpoints);149			Ok((worker, telemetry))150		})151		.transpose()?;152153	let executor = NativeElseWasmExecutor::<ExecutorDispatch>::new(154		config.wasm_method,155		config.default_heap_pages,156		config.max_runtime_instances,157		config.runtime_cache_size,158	);159160	let (client, backend, keystore_container, task_manager) =161		sc_service::new_full_parts::<Block, RuntimeApi, _>(162			config,163			telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),164			executor,165		)?;166	let client = Arc::new(client);167168	let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());169170	let telemetry = telemetry.map(|(worker, telemetry)| {171		task_manager172			.spawn_handle()173			.spawn("telemetry", None, worker.run());174		telemetry175	});176177	let select_chain = sc_consensus::LongestChain::new(backend.clone());178179	let transaction_pool = sc_transaction_pool::BasicPool::new_full(180		config.transaction_pool.clone(),181		config.role.is_authority().into(),182		config.prometheus_registry(),183		task_manager.spawn_essential_handle(),184		client.clone(),185	);186187	let filter_pool: Option<FilterPool> = Some(Arc::new(Mutex::new(BTreeMap::new())));188189	let frontier_backend = open_frontier_backend(config)?;190191	let import_queue = build_import_queue(192		client.clone(),193		config,194		telemetry.as_ref().map(|telemetry| telemetry.handle()),195		&task_manager,196	)?;197	let fee_history_cache: FeeHistoryCache = Arc::new(Mutex::new(BTreeMap::new()));198199	let params = PartialComponents {200		backend,201		client,202		import_queue,203		keystore_container,204		task_manager,205		transaction_pool,206		select_chain,207		other: (208			telemetry,209			filter_pool,210			frontier_backend,211			telemetry_worker_handle,212			fee_history_cache,213		),214	};215216	Ok(params)217}218219/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.220///221/// This is the actual implementation that is abstract over the executor and the runtime api.222#[sc_tracing::logging::prefix_logs_with("Parachain")]223async fn start_node_impl<BIQ, BIC>(224	parachain_config: Configuration,225	polkadot_config: Configuration,226	id: ParaId,227	build_import_queue: BIQ,228	build_consensus: BIC,229) -> sc_service::error::Result<(TaskManager, Arc<FullClient>)>230where231	sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,232	ExecutorDispatch: NativeExecutionDispatch + 'static,233	BIQ: FnOnce(234		Arc<FullClient>,235		&Configuration,236		Option<TelemetryHandle>,237		&TaskManager,238	) -> Result<sc_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error>,239	BIC: FnOnce(240		Arc<FullClient>,241		Option<&Registry>,242		Option<TelemetryHandle>,243		&TaskManager,244		Arc<dyn RelayChainInterface>,245		Arc<sc_transaction_pool::FullPool<Block, FullClient>>,246		Arc<NetworkService<Block, Hash>>,247		SyncCryptoStorePtr,248		bool,249	) -> Result<Box<dyn ParachainConsensus<Block>>, sc_service::Error>,250{251	if matches!(parachain_config.role, Role::Light) {252		return Err("Light client not supported!".into());253	}254255	let parachain_config = prepare_node_config(parachain_config);256257	let params = new_partial::<BIQ>(&parachain_config, build_import_queue)?;258	let (mut telemetry, filter_pool, frontier_backend, telemetry_worker_handle, fee_history_cache) =259		params.other;260261	let client = params.client.clone();262	let backend = params.backend.clone();263	let mut task_manager = params.task_manager;264265	let (relay_chain_interface, collator_key) =266		build_relay_chain_interface(polkadot_config, telemetry_worker_handle, &mut task_manager)267			.map_err(|e| match e {268				polkadot_service::Error::Sub(x) => x,269				s => format!("{}", s).into(),270			})?;271272	let block_announce_validator = BlockAnnounceValidator::new(relay_chain_interface.clone(), id);273274	let force_authoring = parachain_config.force_authoring;275	let validator = parachain_config.role.is_authority();276	let prometheus_registry = parachain_config.prometheus_registry().cloned();277	let transaction_pool = params.transaction_pool.clone();278	let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);279280	let (network, system_rpc_tx, start_network) =281		sc_service::build_network(sc_service::BuildNetworkParams {282			config: &parachain_config,283			client: client.clone(),284			transaction_pool: transaction_pool.clone(),285			spawn_handle: task_manager.spawn_handle(),286			import_queue: import_queue.clone(),287			block_announce_validator_builder: Some(Box::new(|_| {288				Box::new(block_announce_validator)289			})),290			warp_sync: None,291		})?;292293	let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());294	let rpc_client = client.clone();295	let rpc_pool = transaction_pool.clone();296	let select_chain = params.select_chain.clone();297	let rpc_network = network.clone();298299	let rpc_frontier_backend = frontier_backend.clone();300301	let block_data_cache = Arc::new(fc_rpc::EthBlockDataCache::new(302		task_manager.spawn_handle(),303		overrides_handle(client.clone()),304		50,305		50,306	));307308	let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {309		let full_deps = unique_rpc::FullDeps {310			backend: rpc_frontier_backend.clone(),311			deny_unsafe,312			client: rpc_client.clone(),313			pool: rpc_pool.clone(),314			graph: rpc_pool.pool().clone(),315			// TODO: Unhardcode316			enable_dev_signer: false,317			filter_pool: filter_pool.clone(),318			network: rpc_network.clone(),319			select_chain: select_chain.clone(),320			is_authority: validator,321			// TODO: Unhardcode322			max_past_logs: 10000,323			block_data_cache: block_data_cache.clone(),324			fee_history_cache: fee_history_cache.clone(),325			// TODO: Unhardcode326			fee_history_limit: 2048,327		};328329		Ok(unique_rpc::create_full::<_, _, _, _, RuntimeApi, _>(330			full_deps,331			subscription_executor.clone(),332		))333	});334335	task_manager.spawn_essential_handle().spawn(336		"frontier-mapping-sync-worker",337		None,338		MappingSyncWorker::new(339			client.import_notification_stream(),340			Duration::new(6, 0),341			client.clone(),342			backend.clone(),343			frontier_backend.clone(),344			SyncStrategy::Normal,345		)346		.for_each(|()| futures::future::ready(())),347	);348349	sc_service::spawn_tasks(sc_service::SpawnTasksParams {350		rpc_extensions_builder,351		client: client.clone(),352		transaction_pool: transaction_pool.clone(),353		task_manager: &mut task_manager,354		config: parachain_config,355		keystore: params.keystore_container.sync_keystore(),356		backend: backend.clone(),357		network: network.clone(),358		system_rpc_tx,359		telemetry: telemetry.as_mut(),360	})?;361362	let announce_block = {363		let network = network.clone();364		Arc::new(move |hash, data| network.announce_block(hash, data))365	};366367	let relay_chain_slot_duration = Duration::from_secs(6);368369	if validator {370		let parachain_consensus = build_consensus(371			client.clone(),372			prometheus_registry.as_ref(),373			telemetry.as_ref().map(|t| t.handle()),374			&task_manager,375			relay_chain_interface.clone(),376			transaction_pool,377			network,378			params.keystore_container.sync_keystore(),379			force_authoring,380		)?;381382		let spawner = task_manager.spawn_handle();383384		let params = StartCollatorParams {385			para_id: id,386			block_status: client.clone(),387			announce_block,388			client: client.clone(),389			task_manager: &mut task_manager,390			spawner,391			parachain_consensus,392			import_queue,393			collator_key,394			relay_chain_interface,395			relay_chain_slot_duration,396		};397398		start_collator(params).await?;399	} else {400		let params = StartFullNodeParams {401			client: client.clone(),402			announce_block,403			task_manager: &mut task_manager,404			para_id: id,405			import_queue,406			relay_chain_interface,407			relay_chain_slot_duration,408		};409410		start_full_node(params)?;411	}412413	start_network.start_network();414415	Ok((task_manager, client))416}417418/// Build the import queue for the the parachain runtime.419pub fn parachain_build_import_queue(420	client: Arc<FullClient>,421	config: &Configuration,422	telemetry: Option<TelemetryHandle>,423	task_manager: &TaskManager,424) -> Result<sc_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error> {425	let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;426427	cumulus_client_consensus_aura::import_queue::<428		sp_consensus_aura::sr25519::AuthorityPair,429		_,430		_,431		_,432		_,433		_,434		_,435	>(cumulus_client_consensus_aura::ImportQueueParams {436		block_import: client.clone(),437		client: client.clone(),438		create_inherent_data_providers: move |_, _| async move {439			let time = sp_timestamp::InherentDataProvider::from_system_time();440441			let slot =442				sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_duration(443					*time,444					slot_duration.slot_duration(),445				);446447			Ok((time, slot))448		},449		registry: config.prometheus_registry(),450		can_author_with: sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()),451		spawner: &task_manager.spawn_essential_handle(),452		telemetry,453	})454	.map_err(Into::into)455}456457/// Start a normal parachain node.458pub async fn start_node(459	parachain_config: Configuration,460	polkadot_config: Configuration,461	id: ParaId,462) -> sc_service::error::Result<(TaskManager, Arc<FullClient>)> {463	start_node_impl::<_, _>(464		parachain_config,465		polkadot_config,466		id,467		parachain_build_import_queue,468		|client,469		 prometheus_registry,470		 telemetry,471		 task_manager,472		 relay_chain_interface,473		 transaction_pool,474		 sync_oracle,475		 keystore,476		 force_authoring| {477			let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;478479			let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(480				task_manager.spawn_handle(),481				client.clone(),482				transaction_pool,483				prometheus_registry,484				telemetry.clone(),485			);486487			Ok(AuraConsensus::build::<488				sp_consensus_aura::sr25519::AuthorityPair,489				_,490				_,491				_,492				_,493				_,494				_,495			>(BuildAuraConsensusParams {496				proposer_factory,497				create_inherent_data_providers: move |_, (relay_parent, validation_data)| {498					let relay_chain_interface = relay_chain_interface.clone();499					async move {500						let parachain_inherent =501						cumulus_primitives_parachain_inherent::ParachainInherentData::create_at(502							relay_parent,503							&relay_chain_interface,504							&validation_data,505							id,506						).await;507508						let time = sp_timestamp::InherentDataProvider::from_system_time();509510						let slot =511						sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_duration(512							*time,513							slot_duration.slot_duration(),514						);515516						let parachain_inherent = parachain_inherent.ok_or_else(|| {517							Box::<dyn std::error::Error + Send + Sync>::from(518								"Failed to create parachain inherent",519							)520						})?;521						Ok((time, slot, parachain_inherent))522					}523				},524				block_import: client.clone(),525				para_client: client,526				backoff_authoring_blocks: Option::<()>::None,527				sync_oracle,528				keystore,529				force_authoring,530				slot_duration,531				// We got around 500ms for proposing532				block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),533				telemetry,534				max_block_proposal_slot_portion: None,535			}))536		},537	)538	.await539}