git.delta.rocks / unique-network / refs/commits / 13d4a7ecf461

1// Copyright 2019-2022 Unique Network (Gibraltar) Ltd.2// This file is part of Unique Network.34// Unique Network is free software: you can redistribute it and/or modify5// it under the terms of the GNU General Public License as published by6// the Free Software Foundation, either version 3 of the License, or7// (at your option) any later version.89// Unique Network is distributed in the hope that it will be useful,10// but WITHOUT ANY WARRANTY; without even the implied warranty of11// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the12// GNU General Public License for more details.1314// You should have received a copy of the GNU General Public License15// along with Unique Network. If not, see <http://www.gnu.org/licenses/>.1617// std18use std::sync::Arc;19use std::sync::Mutex;20use std::collections::BTreeMap;21use std::time::Duration;22use fc_rpc_core::types::FeeHistoryCache;23use futures::StreamExt;2425use unique_rpc::overrides_handle;26// Local Runtime Types27use unique_runtime::RuntimeApi;2829// Cumulus Imports30use cumulus_client_consensus_aura::{AuraConsensus, BuildAuraConsensusParams, SlotProportion};31use cumulus_client_consensus_common::ParachainConsensus;32use cumulus_client_service::{33	prepare_node_config, start_collator, start_full_node, StartCollatorParams, StartFullNodeParams,34};35use cumulus_client_network::BlockAnnounceValidator;36use cumulus_primitives_core::ParaId;37use cumulus_relay_chain_interface::RelayChainInterface;38use cumulus_relay_chain_local::build_relay_chain_interface;3940// Substrate Imports41use sc_client_api::ExecutorProvider;42use sc_executor::NativeElseWasmExecutor;43use sc_executor::NativeExecutionDispatch;44use sc_network::NetworkService;45use sc_service::{BasePath, Configuration, PartialComponents, Role, TaskManager};46use sc_telemetry::{Telemetry, TelemetryHandle, TelemetryWorker, TelemetryWorkerHandle};47use sp_consensus::SlotData;48use sp_keystore::SyncCryptoStorePtr;49use sp_runtime::traits::BlakeTwo256;50use substrate_prometheus_endpoint::Registry;51use sc_client_api::BlockchainEvents;5253// Frontier Imports54use fc_rpc_core::types::FilterPool;55use fc_mapping_sync::{MappingSyncWorker, SyncStrategy};5657// Runtime type overrides58type BlockNumber = u32;59type Header = sp_runtime::generic::Header<BlockNumber, sp_runtime::traits::BlakeTwo256>;60pub type Block = sp_runtime::generic::Block<Header, sp_runtime::OpaqueExtrinsic>;61type Hash = sp_core::H256;6263/// Native executor instance.64pub struct ParachainRuntimeExecutor;6566impl NativeExecutionDispatch for ParachainRuntimeExecutor {67	type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;6869	fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {70		unique_runtime::api::dispatch(method, data)71	}7273	fn native_version() -> sc_executor::NativeVersion {74		unique_runtime::native_version()75	}76}7778pub fn open_frontier_backend(config: &Configuration) -> Result<Arc<fc_db::Backend<Block>>, String> {79	let config_dir = config80		.base_path81		.as_ref()82		.map(|base_path| base_path.config_dir(config.chain_spec.id()))83		.unwrap_or_else(|| {84			BasePath::from_project("", "", "unique").config_dir(config.chain_spec.id())85		});86	let database_dir = config_dir.join("frontier").join("db");8788	Ok(Arc::new(fc_db::Backend::<Block>::new(89		&fc_db::DatabaseSettings {90			source: fc_db::DatabaseSettingsSrc::RocksDb {91				path: database_dir,92				cache_size: 0,93			},94		},95	)?))96}9798type ExecutorDispatch = ParachainRuntimeExecutor;99100type FullClient =101	sc_service::TFullClient<Block, RuntimeApi, NativeElseWasmExecutor<ExecutorDispatch>>;102type FullBackend = sc_service::TFullBackend<Block>;103type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;104105/// Starts a `ServiceBuilder` for a full service.106///107/// Use this macro if you don't actually need the full service, but just the builder in order to108/// be able to perform chain operations.109#[allow(clippy::type_complexity)]110pub fn new_partial<BIQ>(111	config: &Configuration,112	build_import_queue: BIQ,113) -> Result<114	PartialComponents<115		FullClient,116		FullBackend,117		FullSelectChain,118		sc_consensus::DefaultImportQueue<Block, FullClient>,119		sc_transaction_pool::FullPool<Block, FullClient>,120		(121			Option<Telemetry>,122			Option<FilterPool>,123			Arc<fc_db::Backend<Block>>,124			Option<TelemetryWorkerHandle>,125			FeeHistoryCache,126		),127	>,128	sc_service::Error,129>130where131	sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,132	ExecutorDispatch: NativeExecutionDispatch + 'static,133	BIQ: FnOnce(134		Arc<FullClient>,135		&Configuration,136		Option<TelemetryHandle>,137		&TaskManager,138	) -> Result<sc_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error>,139{140	let _telemetry = config141		.telemetry_endpoints142		.clone()143		.filter(|x| !x.is_empty())144		.map(|endpoints| -> Result<_, sc_telemetry::Error> {145			let worker = TelemetryWorker::new(16)?;146			let telemetry = worker.handle().new_telemetry(endpoints);147			Ok((worker, telemetry))148		})149		.transpose()?;150151	let telemetry = config152		.telemetry_endpoints153		.clone()154		.filter(|x| !x.is_empty())155		.map(|endpoints| -> Result<_, sc_telemetry::Error> {156			let worker = TelemetryWorker::new(16)?;157			let telemetry = worker.handle().new_telemetry(endpoints);158			Ok((worker, telemetry))159		})160		.transpose()?;161162	let executor = NativeElseWasmExecutor::<ExecutorDispatch>::new(163		config.wasm_method,164		config.default_heap_pages,165		config.max_runtime_instances,166		config.runtime_cache_size,167	);168169	let (client, backend, keystore_container, task_manager) =170		sc_service::new_full_parts::<Block, RuntimeApi, _>(171			config,172			telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),173			executor,174		)?;175	let client = Arc::new(client);176177	let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());178179	let telemetry = telemetry.map(|(worker, telemetry)| {180		task_manager181			.spawn_handle()182			.spawn("telemetry", None, worker.run());183		telemetry184	});185186	let select_chain = sc_consensus::LongestChain::new(backend.clone());187188	let transaction_pool = sc_transaction_pool::BasicPool::new_full(189		config.transaction_pool.clone(),190		config.role.is_authority().into(),191		config.prometheus_registry(),192		task_manager.spawn_essential_handle(),193		client.clone(),194	);195196	let filter_pool: Option<FilterPool> = Some(Arc::new(Mutex::new(BTreeMap::new())));197198	let frontier_backend = open_frontier_backend(config)?;199200	let import_queue = build_import_queue(201		client.clone(),202		config,203		telemetry.as_ref().map(|telemetry| telemetry.handle()),204		&task_manager,205	)?;206	let fee_history_cache: FeeHistoryCache = Arc::new(Mutex::new(BTreeMap::new()));207208	let params = PartialComponents {209		backend,210		client,211		import_queue,212		keystore_container,213		task_manager,214		transaction_pool,215		select_chain,216		other: (217			telemetry,218			filter_pool,219			frontier_backend,220			telemetry_worker_handle,221			fee_history_cache,222		),223	};224225	Ok(params)226}227228/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.229///230/// This is the actual implementation that is abstract over the executor and the runtime api.231#[sc_tracing::logging::prefix_logs_with("Parachain")]232async fn start_node_impl<BIQ, BIC>(233	parachain_config: Configuration,234	polkadot_config: Configuration,235	id: ParaId,236	build_import_queue: BIQ,237	build_consensus: BIC,238) -> sc_service::error::Result<(TaskManager, Arc<FullClient>)>239where240	sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,241	ExecutorDispatch: NativeExecutionDispatch + 'static,242	BIQ: FnOnce(243		Arc<FullClient>,244		&Configuration,245		Option<TelemetryHandle>,246		&TaskManager,247	) -> Result<sc_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error>,248	BIC: FnOnce(249		Arc<FullClient>,250		Option<&Registry>,251		Option<TelemetryHandle>,252		&TaskManager,253		Arc<dyn RelayChainInterface>,254		Arc<sc_transaction_pool::FullPool<Block, FullClient>>,255		Arc<NetworkService<Block, Hash>>,256		SyncCryptoStorePtr,257		bool,258	) -> Result<Box<dyn ParachainConsensus<Block>>, sc_service::Error>,259{260	if matches!(parachain_config.role, Role::Light) {261		return Err("Light client not supported!".into());262	}263264	let parachain_config = prepare_node_config(parachain_config);265266	let params = new_partial::<BIQ>(&parachain_config, build_import_queue)?;267	let (mut telemetry, filter_pool, frontier_backend, telemetry_worker_handle, fee_history_cache) =268		params.other;269270	let client = params.client.clone();271	let backend = params.backend.clone();272	let mut task_manager = params.task_manager;273274	let (relay_chain_interface, collator_key) =275		build_relay_chain_interface(polkadot_config, telemetry_worker_handle, &mut task_manager)276			.map_err(|e| match e {277				polkadot_service::Error::Sub(x) => x,278				s => format!("{}", s).into(),279			})?;280281	let block_announce_validator = BlockAnnounceValidator::new(relay_chain_interface.clone(), id);282283	let force_authoring = parachain_config.force_authoring;284	let validator = parachain_config.role.is_authority();285	let prometheus_registry = parachain_config.prometheus_registry().cloned();286	let transaction_pool = params.transaction_pool.clone();287	let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);288289	let (network, system_rpc_tx, start_network) =290		sc_service::build_network(sc_service::BuildNetworkParams {291			config: &parachain_config,292			client: client.clone(),293			transaction_pool: transaction_pool.clone(),294			spawn_handle: task_manager.spawn_handle(),295			import_queue: import_queue.clone(),296			block_announce_validator_builder: Some(Box::new(|_| {297				Box::new(block_announce_validator)298			})),299			warp_sync: None,300		})?;301302	let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());303	let rpc_client = client.clone();304	let rpc_pool = transaction_pool.clone();305	let select_chain = params.select_chain.clone();306	let rpc_network = network.clone();307308	let rpc_frontier_backend = frontier_backend.clone();309310	let block_data_cache = Arc::new(fc_rpc::EthBlockDataCache::new(311		task_manager.spawn_handle(),312		overrides_handle(client.clone()),313		50,314		50,315	));316317	let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {318		let full_deps = unique_rpc::FullDeps {319			backend: rpc_frontier_backend.clone(),320			deny_unsafe,321			client: rpc_client.clone(),322			pool: rpc_pool.clone(),323			graph: rpc_pool.pool().clone(),324			// TODO: Unhardcode325			enable_dev_signer: false,326			filter_pool: filter_pool.clone(),327			network: rpc_network.clone(),328			select_chain: select_chain.clone(),329			is_authority: validator,330			// TODO: Unhardcode331			max_past_logs: 10000,332			block_data_cache: block_data_cache.clone(),333			fee_history_cache: fee_history_cache.clone(),334			// TODO: Unhardcode335			fee_history_limit: 2048,336		};337338		Ok(unique_rpc::create_full::<_, _, _, _, RuntimeApi, _>(339			full_deps,340			subscription_executor.clone(),341		))342	});343344	task_manager.spawn_essential_handle().spawn(345		"frontier-mapping-sync-worker",346		None,347		MappingSyncWorker::new(348			client.import_notification_stream(),349			Duration::new(6, 0),350			client.clone(),351			backend.clone(),352			frontier_backend.clone(),353			SyncStrategy::Normal,354		)355		.for_each(|()| futures::future::ready(())),356	);357358	sc_service::spawn_tasks(sc_service::SpawnTasksParams {359		rpc_extensions_builder,360		client: client.clone(),361		transaction_pool: transaction_pool.clone(),362		task_manager: &mut task_manager,363		config: parachain_config,364		keystore: params.keystore_container.sync_keystore(),365		backend: backend.clone(),366		network: network.clone(),367		system_rpc_tx,368		telemetry: telemetry.as_mut(),369	})?;370371	let announce_block = {372		let network = network.clone();373		Arc::new(move |hash, data| network.announce_block(hash, data))374	};375376	let relay_chain_slot_duration = Duration::from_secs(6);377378	if validator {379		let parachain_consensus = build_consensus(380			client.clone(),381			prometheus_registry.as_ref(),382			telemetry.as_ref().map(|t| t.handle()),383			&task_manager,384			relay_chain_interface.clone(),385			transaction_pool,386			network,387			params.keystore_container.sync_keystore(),388			force_authoring,389		)?;390391		let spawner = task_manager.spawn_handle();392393		let params = StartCollatorParams {394			para_id: id,395			block_status: client.clone(),396			announce_block,397			client: client.clone(),398			task_manager: &mut task_manager,399			spawner,400			parachain_consensus,401			import_queue,402			collator_key,403			relay_chain_interface,404			relay_chain_slot_duration,405		};406407		start_collator(params).await?;408	} else {409		let params = StartFullNodeParams {410			client: client.clone(),411			announce_block,412			task_manager: &mut task_manager,413			para_id: id,414			import_queue,415			relay_chain_interface,416			relay_chain_slot_duration,417		};418419		start_full_node(params)?;420	}421422	start_network.start_network();423424	Ok((task_manager, client))425}426427/// Build the import queue for the the parachain runtime.428pub fn parachain_build_import_queue(429	client: Arc<FullClient>,430	config: &Configuration,431	telemetry: Option<TelemetryHandle>,432	task_manager: &TaskManager,433) -> Result<sc_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error> {434	let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;435436	cumulus_client_consensus_aura::import_queue::<437		sp_consensus_aura::sr25519::AuthorityPair,438		_,439		_,440		_,441		_,442		_,443		_,444	>(cumulus_client_consensus_aura::ImportQueueParams {445		block_import: client.clone(),446		client: client.clone(),447		create_inherent_data_providers: move |_, _| async move {448			let time = sp_timestamp::InherentDataProvider::from_system_time();449450			let slot =451				sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_duration(452					*time,453					slot_duration.slot_duration(),454				);455456			Ok((time, slot))457		},458		registry: config.prometheus_registry(),459		can_author_with: sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()),460		spawner: &task_manager.spawn_essential_handle(),461		telemetry,462	})463	.map_err(Into::into)464}465466/// Start a normal parachain node.467pub async fn start_node(468	parachain_config: Configuration,469	polkadot_config: Configuration,470	id: ParaId,471) -> sc_service::error::Result<(TaskManager, Arc<FullClient>)> {472	start_node_impl::<_, _>(473		parachain_config,474		polkadot_config,475		id,476		parachain_build_import_queue,477		|client,478		 prometheus_registry,479		 telemetry,480		 task_manager,481		 relay_chain_interface,482		 transaction_pool,483		 sync_oracle,484		 keystore,485		 force_authoring| {486			let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;487488			let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(489				task_manager.spawn_handle(),490				client.clone(),491				transaction_pool,492				prometheus_registry,493				telemetry.clone(),494			);495496			Ok(AuraConsensus::build::<497				sp_consensus_aura::sr25519::AuthorityPair,498				_,499				_,500				_,501				_,502				_,503				_,504			>(BuildAuraConsensusParams {505				proposer_factory,506				create_inherent_data_providers: move |_, (relay_parent, validation_data)| {507					let relay_chain_interface = relay_chain_interface.clone();508					async move {509						let parachain_inherent =510						cumulus_primitives_parachain_inherent::ParachainInherentData::create_at(511							relay_parent,512							&relay_chain_interface,513							&validation_data,514							id,515						).await;516517						let time = sp_timestamp::InherentDataProvider::from_system_time();518519						let slot =520						sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_duration(521							*time,522							slot_duration.slot_duration(),523						);524525						let parachain_inherent = parachain_inherent.ok_or_else(|| {526							Box::<dyn std::error::Error + Send + Sync>::from(527								"Failed to create parachain inherent",528							)529						})?;530						Ok((time, slot, parachain_inherent))531					}532				},533				block_import: client.clone(),534				para_client: client,535				backoff_authoring_blocks: Option::<()>::None,536				sync_oracle,537				keystore,538				force_authoring,539				slot_duration: *slot_duration,540				// We got around 500ms for proposing541				block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),542				telemetry,543				max_block_proposal_slot_portion: None,544			}))545		},546	)547	.await548}