123456789use std::sync::Arc;10use std::sync::Mutex;11use std::collections::BTreeMap;12use std::collections::HashMap;13use std::time::Duration;14use futures::StreamExt;151617use nft_runtime::RuntimeApi;181920use 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;272829use 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;404142use fc_rpc_core::types::FilterPool;43use fc_rpc_core::types::PendingTransactions;44use fc_mapping_sync::MappingSyncWorker;454647type 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;515253native_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>;858687888990#[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 sp_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<sp_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}199200201202203#[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<sp_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>(¶chain_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);269 let (network, system_rpc_tx, start_network) =270 sc_service::build_network(sc_service::BuildNetworkParams {271 config: ¶chain_config,272 client: client.clone(),273 transaction_pool: transaction_pool.clone(),274 spawn_handle: task_manager.spawn_handle(),275 import_queue: import_queue.clone(),276 on_demand: None,277 block_announce_validator_builder: Some(Box::new(|_| block_announce_validator)),278 })?;279280 let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());281 let rpc_client = client.clone();282 let rpc_pool = transaction_pool.clone();283 let select_chain = params.select_chain.clone();284 let is_authority = parachain_config.role.clone().is_authority();285 let rpc_network = network.clone();286287 let rpc_frontier_backend = frontier_backend.clone();288 let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {289 let full_deps = nft_rpc::FullDeps {290 backend: rpc_frontier_backend.clone(),291 deny_unsafe,292 client: rpc_client.clone(),293 pool: rpc_pool.clone(),294 295 enable_dev_signer: false,296 filter_pool: filter_pool.clone(),297 network: rpc_network.clone(),298 pending_transactions: pending_transactions.clone(),299 select_chain: select_chain.clone(),300 is_authority,301 302 max_past_logs: 10000,303 };304305 nft_rpc::create_full::<_, _, _, RuntimeApi, _>(full_deps, subscription_executor.clone())306 });307 308 task_manager.spawn_essential_handle().spawn(309 "frontier-mapping-sync-worker",310 MappingSyncWorker::new(311 client.import_notification_stream(),312 Duration::new(6, 0),313 client.clone(),314 backend.clone(),315 frontier_backend.clone(),316 ).for_each(|()| futures::future::ready(()))317 );318319 sc_service::spawn_tasks(sc_service::SpawnTasksParams {320 on_demand: None,321 remote_blockchain: None,322 rpc_extensions_builder,323 client: client.clone(),324 transaction_pool: transaction_pool.clone(),325 task_manager: &mut task_manager,326 config: parachain_config,327 keystore: params.keystore_container.sync_keystore(),328 backend: backend.clone(),329 network: network.clone(),330 system_rpc_tx,331 telemetry: telemetry.as_mut(),332 })?;333334 let announce_block = {335 let network = network.clone();336 Arc::new(move |hash, data| network.announce_block(hash, data))337 };338339 if validator {340 let parachain_consensus = build_consensus(341 client.clone(),342 prometheus_registry.as_ref(),343 telemetry.as_ref().map(|t| t.handle()),344 &task_manager,345 &relay_chain_full_node,346 transaction_pool,347 network,348 params.keystore_container.sync_keystore(),349 force_authoring,350 )?;351352 let spawner = task_manager.spawn_handle();353354 let params = StartCollatorParams {355 para_id: id,356 block_status: client.clone(),357 announce_block,358 client: client.clone(),359 task_manager: &mut task_manager,360 relay_chain_full_node,361 spawner,362 parachain_consensus,363 import_queue,364 };365366 start_collator(params).await?;367 } else {368 let params = StartFullNodeParams {369 client: client.clone(),370 announce_block,371 task_manager: &mut task_manager,372 para_id: id,373 relay_chain_full_node,374 };375376 start_full_node(params)?;377 }378379 start_network.start_network();380381 Ok((task_manager, client))382}383384385pub fn parachain_build_import_queue(386 client: Arc<FullClient>,387 config: &Configuration,388 telemetry: Option<TelemetryHandle>,389 task_manager: &TaskManager,390) -> Result<sp_consensus::DefaultImportQueue<Block, FullClient>, sc_service::Error> {391 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;392393 cumulus_client_consensus_aura::import_queue::<394 sp_consensus_aura::sr25519::AuthorityPair,395 _,396 _,397 _,398 _,399 _,400 _,401 >(cumulus_client_consensus_aura::ImportQueueParams {402 block_import: client.clone(),403 client: client.clone(),404 create_inherent_data_providers: move |_, _| async move {405 let time = sp_timestamp::InherentDataProvider::from_system_time();406407 let slot =408 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_duration(409 *time,410 slot_duration.slot_duration(),411 );412413 Ok((time, slot))414 },415 registry: config.prometheus_registry(),416 can_author_with: sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()),417 spawner: &task_manager.spawn_essential_handle(),418 telemetry,419 })420 .map_err(Into::into)421}422423424pub async fn start_node(425 parachain_config: Configuration,426 polkadot_config: Configuration,427 id: ParaId,428) -> sc_service::error::Result<(TaskManager, Arc<FullClient>)> {429 start_node_impl::<_, _>(430 parachain_config,431 polkadot_config,432 id,433 parachain_build_import_queue,434 |client,435 prometheus_registry,436 telemetry,437 task_manager,438 relay_chain_node,439 transaction_pool,440 sync_oracle,441 keystore,442 force_authoring| {443 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;444445 let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(446 task_manager.spawn_handle(),447 client.clone(),448 transaction_pool,449 prometheus_registry,450 telemetry.clone(),451 );452453 let relay_chain_backend = relay_chain_node.backend.clone();454 let relay_chain_client = relay_chain_node.client.clone();455 Ok(build_aura_consensus::<456 sp_consensus_aura::sr25519::AuthorityPair,457 _,458 _,459 _,460 _,461 _,462 _,463 _,464 _,465 _,466 >(BuildAuraConsensusParams {467 proposer_factory,468 create_inherent_data_providers: move |_, (relay_parent, validation_data)| {469 let parachain_inherent =470 cumulus_primitives_parachain_inherent::ParachainInherentData::create_at_with_client(471 relay_parent,472 &relay_chain_client,473 &*relay_chain_backend,474 &validation_data,475 id,476 );477 async move {478 let time = sp_timestamp::InherentDataProvider::from_system_time();479480 let slot =481 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_duration(482 *time,483 slot_duration.slot_duration(),484 );485486 let parachain_inherent = parachain_inherent.ok_or_else(|| {487 Box::<dyn std::error::Error + Send + Sync>::from(488 "Failed to create parachain inherent",489 )490 })?;491 Ok((time, slot, parachain_inherent))492 }493 },494 block_import: client.clone(),495 relay_chain_client: relay_chain_node.client.clone(),496 relay_chain_backend: relay_chain_node.backend.clone(),497 para_client: client,498 backoff_authoring_blocks: Option::<()>::None,499 sync_oracle,500 keystore,501 force_authoring,502 slot_duration,503 504 block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),505 telemetry,506 max_block_proposal_slot_portion: None,507 }))508 },509 )510 .await511}