difftreelog
cargo fmt
in: master
2 files changed
node/cli/src/command.rsdiffbeforeafterboth--- a/node/cli/src/command.rs
+++ b/node/cli/src/command.rs
@@ -405,9 +405,8 @@
if is_dev_service {
info!("Running Dev service");
- let autoseal_interval = AutosealInterval::new(
- Duration::from_millis(cli.idle_autoseal_interval)
- )?;
+ let autoseal_interval =
+ AutosealInterval::new(Duration::from_millis(cli.idle_autoseal_interval))?;
return start_node_using_chain_runtime! {
start_dev_node(config, autoseal_interval).map_err(Into::into)
node/cli/src/service.rsdiffbeforeafterboth1// 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//! Service and ServiceFactory implementation. Specialized wrapper over substrate service.1819// std20use std::sync::Arc;21use std::sync::Mutex;22use std::collections::BTreeMap;23use std::time::Duration;24use std::pin::Pin;25use fc_rpc_core::types::FeeHistoryCache;26use futures::Future;27use futures::{Stream, StreamExt, stream::select, task::{Context, Poll}};28use futures_timer::Delay;2930use unique_rpc::overrides_handle;3132use serde::{Serialize, Deserialize};3334// Cumulus Imports35use cumulus_client_consensus_aura::{AuraConsensus, BuildAuraConsensusParams, SlotProportion};36use cumulus_client_consensus_common::ParachainConsensus;37use cumulus_client_service::{38 prepare_node_config, start_collator, start_full_node, StartCollatorParams, StartFullNodeParams,39};40use cumulus_client_cli::CollatorOptions;41use cumulus_client_network::BlockAnnounceValidator;42use cumulus_primitives_core::ParaId;43use cumulus_relay_chain_inprocess_interface::build_inprocess_relay_chain;44use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface, RelayChainResult};45use cumulus_relay_chain_rpc_interface::RelayChainRPCInterface;4647// Substrate Imports48use sc_client_api::ExecutorProvider;49use sc_executor::NativeElseWasmExecutor;50use sc_executor::NativeExecutionDispatch;51use sc_network::NetworkService;52use sc_service::{BasePath, Configuration, PartialComponents, Role, TaskManager};53use sc_telemetry::{Telemetry, TelemetryHandle, TelemetryWorker, TelemetryWorkerHandle};54use sp_keystore::SyncCryptoStorePtr;55use sp_runtime::traits::BlakeTwo256;56use substrate_prometheus_endpoint::Registry;57use sc_client_api::BlockchainEvents;5859use polkadot_service::CollatorPair;6061// Frontier Imports62use fc_rpc_core::types::FilterPool;63use fc_mapping_sync::{MappingSyncWorker, SyncStrategy};6465use unique_runtime_common::types::{AuraId, RuntimeInstance, AccountId, Balance, Index, Hash, Block};6667/// Unique native executor instance.68#[cfg(feature = "unique-runtime")]69pub struct UniqueRuntimeExecutor;7071#[cfg(feature = "quartz-runtime")]72/// Quartz native executor instance.7374pub struct QuartzRuntimeExecutor;7576/// Opal native executor instance.77pub struct OpalRuntimeExecutor;7879#[cfg(feature = "unique-runtime")]80impl NativeExecutionDispatch for UniqueRuntimeExecutor {81 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;8283 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {84 unique_runtime::api::dispatch(method, data)85 }8687 fn native_version() -> sc_executor::NativeVersion {88 unique_runtime::native_version()89 }90}9192#[cfg(feature = "quartz-runtime")]93impl NativeExecutionDispatch for QuartzRuntimeExecutor {94 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;9596 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {97 quartz_runtime::api::dispatch(method, data)98 }99100 fn native_version() -> sc_executor::NativeVersion {101 quartz_runtime::native_version()102 }103}104105impl NativeExecutionDispatch for OpalRuntimeExecutor {106 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;107108 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {109 opal_runtime::api::dispatch(method, data)110 }111112 fn native_version() -> sc_executor::NativeVersion {113 opal_runtime::native_version()114 }115}116117pub struct AutosealInterval {118 duration: Duration,119 delay_handle: Pin<Box<Delay>>120}121122impl AutosealInterval {123 pub fn new(duration: Duration) -> Result<Self, String> {124 if duration.is_zero() {125 return Err("Invalid autoseal interval: 0 seconds".into());126 }127128 Ok(Self {129 duration,130 delay_handle: Box::pin(Delay::new(duration))131 })132 }133}134135impl Stream for AutosealInterval {136 type Item = ();137138 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {139 match self.delay_handle.as_mut().poll(cx) {140 Poll::Ready(_) => {141 let duration = self.duration;142 self.delay_handle.reset(duration);143144 Poll::Ready(Some(()))145 }146 Poll::Pending => Poll::Pending147 }148 }149}150151pub fn open_frontier_backend(config: &Configuration) -> Result<Arc<fc_db::Backend<Block>>, String> {152 let config_dir = config153 .base_path154 .as_ref()155 .map(|base_path| base_path.config_dir(config.chain_spec.id()))156 .unwrap_or_else(|| {157 BasePath::from_project("", "", "unique").config_dir(config.chain_spec.id())158 });159 let database_dir = config_dir.join("frontier").join("db");160161 Ok(Arc::new(fc_db::Backend::<Block>::new(162 &fc_db::DatabaseSettings {163 source: fc_db::DatabaseSettingsSrc::RocksDb {164 path: database_dir,165 cache_size: 0,166 },167 },168 )?))169}170171type FullClient<RuntimeApi, ExecutorDispatch> =172 sc_service::TFullClient<Block, RuntimeApi, NativeElseWasmExecutor<ExecutorDispatch>>;173type FullBackend = sc_service::TFullBackend<Block>;174type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;175176/// Starts a `ServiceBuilder` for a full service.177///178/// Use this macro if you don't actually need the full service, but just the builder in order to179/// be able to perform chain operations.180#[allow(clippy::type_complexity)]181pub fn new_partial<RuntimeApi, ExecutorDispatch, BIQ>(182 config: &Configuration,183 build_import_queue: BIQ,184) -> Result<185 PartialComponents<186 FullClient<RuntimeApi, ExecutorDispatch>,187 FullBackend,188 FullSelectChain,189 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,190 sc_transaction_pool::FullPool<Block, FullClient<RuntimeApi, ExecutorDispatch>>,191 (192 Option<Telemetry>,193 Option<FilterPool>,194 Arc<fc_db::Backend<Block>>,195 Option<TelemetryWorkerHandle>,196 FeeHistoryCache,197 ),198 >,199 sc_service::Error,200>201where202 sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,203 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>204 + Send205 + Sync206 + 'static,207 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>,208 ExecutorDispatch: NativeExecutionDispatch + 'static,209 BIQ: FnOnce(210 Arc<FullClient<RuntimeApi, ExecutorDispatch>>,211 &Configuration,212 Option<TelemetryHandle>,213 &TaskManager,214 ) -> Result<215 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,216 sc_service::Error,217 >,218{219 let _telemetry = config220 .telemetry_endpoints221 .clone()222 .filter(|x| !x.is_empty())223 .map(|endpoints| -> Result<_, sc_telemetry::Error> {224 let worker = TelemetryWorker::new(16)?;225 let telemetry = worker.handle().new_telemetry(endpoints);226 Ok((worker, telemetry))227 })228 .transpose()?;229230 let telemetry = config231 .telemetry_endpoints232 .clone()233 .filter(|x| !x.is_empty())234 .map(|endpoints| -> Result<_, sc_telemetry::Error> {235 let worker = TelemetryWorker::new(16)?;236 let telemetry = worker.handle().new_telemetry(endpoints);237 Ok((worker, telemetry))238 })239 .transpose()?;240241 let executor = NativeElseWasmExecutor::<ExecutorDispatch>::new(242 config.wasm_method,243 config.default_heap_pages,244 config.max_runtime_instances,245 config.runtime_cache_size,246 );247248 let (client, backend, keystore_container, task_manager) =249 sc_service::new_full_parts::<Block, RuntimeApi, _>(250 config,251 telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),252 executor,253 )?;254 let client = Arc::new(client);255256 let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());257258 let telemetry = telemetry.map(|(worker, telemetry)| {259 task_manager260 .spawn_handle()261 .spawn("telemetry", None, worker.run());262 telemetry263 });264265 let select_chain = sc_consensus::LongestChain::new(backend.clone());266267 let transaction_pool = sc_transaction_pool::BasicPool::new_full(268 config.transaction_pool.clone(),269 config.role.is_authority().into(),270 config.prometheus_registry(),271 task_manager.spawn_essential_handle(),272 client.clone(),273 );274275 let filter_pool: Option<FilterPool> = Some(Arc::new(Mutex::new(BTreeMap::new())));276277 let frontier_backend = open_frontier_backend(config)?;278279 let import_queue = build_import_queue(280 client.clone(),281 config,282 telemetry.as_ref().map(|telemetry| telemetry.handle()),283 &task_manager,284 )?;285 let fee_history_cache: FeeHistoryCache = Arc::new(Mutex::new(BTreeMap::new()));286287 let params = PartialComponents {288 backend,289 client,290 import_queue,291 keystore_container,292 task_manager,293 transaction_pool,294 select_chain,295 other: (296 telemetry,297 filter_pool,298 frontier_backend,299 telemetry_worker_handle,300 fee_history_cache,301 ),302 };303304 Ok(params)305}306307async fn build_relay_chain_interface(308 polkadot_config: Configuration,309 parachain_config: &Configuration,310 telemetry_worker_handle: Option<TelemetryWorkerHandle>,311 task_manager: &mut TaskManager,312 collator_options: CollatorOptions,313) -> RelayChainResult<(314 Arc<(dyn RelayChainInterface + 'static)>,315 Option<CollatorPair>,316)> {317 match collator_options.relay_chain_rpc_url {318 Some(relay_chain_url) => Ok((319 Arc::new(RelayChainRPCInterface::new(relay_chain_url).await?) as Arc<_>,320 None,321 )),322 None => build_inprocess_relay_chain(323 polkadot_config,324 parachain_config,325 telemetry_worker_handle,326 task_manager,327 ),328 }329}330331/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.332///333/// This is the actual implementation that is abstract over the executor and the runtime api.334#[sc_tracing::logging::prefix_logs_with("Parachain")]335async fn start_node_impl<Runtime, RuntimeApi, ExecutorDispatch, BIQ, BIC>(336 parachain_config: Configuration,337 polkadot_config: Configuration,338 collator_options: CollatorOptions,339 id: ParaId,340 build_import_queue: BIQ,341 build_consensus: BIC,342) -> sc_service::error::Result<(TaskManager, Arc<FullClient<RuntimeApi, ExecutorDispatch>>)>343where344 sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,345 Runtime: RuntimeInstance + Send + Sync + 'static,346 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,347 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,348 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>349 + Send350 + Sync351 + 'static,352 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>353 + fp_rpc::EthereumRuntimeRPCApi<Block>354 + sp_session::SessionKeys<Block>355 + sp_block_builder::BlockBuilder<Block>356 + pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi<Block, Balance>357 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>358 + up_rpc::UniqueApi<Block, Runtime::CrossAccountId, AccountId>359 + substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Index>360 + sp_api::Metadata<Block>361 + sp_offchain::OffchainWorkerApi<Block>362 + cumulus_primitives_core::CollectCollationInfo<Block>,363 ExecutorDispatch: NativeExecutionDispatch + 'static,364 BIQ: FnOnce(365 Arc<FullClient<RuntimeApi, ExecutorDispatch>>,366 &Configuration,367 Option<TelemetryHandle>,368 &TaskManager,369 ) -> Result<370 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,371 sc_service::Error,372 >,373 BIC: FnOnce(374 Arc<FullClient<RuntimeApi, ExecutorDispatch>>,375 Option<&Registry>,376 Option<TelemetryHandle>,377 &TaskManager,378 Arc<dyn RelayChainInterface>,379 Arc<sc_transaction_pool::FullPool<Block, FullClient<RuntimeApi, ExecutorDispatch>>>,380 Arc<NetworkService<Block, Hash>>,381 SyncCryptoStorePtr,382 bool,383 ) -> Result<Box<dyn ParachainConsensus<Block>>, sc_service::Error>,384{385 if matches!(parachain_config.role, Role::Light) {386 return Err("Light client not supported!".into());387 }388389 let parachain_config = prepare_node_config(parachain_config);390391 let params =392 new_partial::<RuntimeApi, ExecutorDispatch, BIQ>(¶chain_config, build_import_queue)?;393 let (mut telemetry, filter_pool, frontier_backend, telemetry_worker_handle, fee_history_cache) =394 params.other;395396 let client = params.client.clone();397 let backend = params.backend.clone();398 let mut task_manager = params.task_manager;399400 let (relay_chain_interface, collator_key) = build_relay_chain_interface(401 polkadot_config,402 ¶chain_config,403 telemetry_worker_handle,404 &mut task_manager,405 collator_options.clone(),406 )407 .await408 .map_err(|e| match e {409 RelayChainError::ServiceError(polkadot_service::Error::Sub(x)) => x,410 s => s.to_string().into(),411 })?;412413 let block_announce_validator = BlockAnnounceValidator::new(relay_chain_interface.clone(), id);414415 let force_authoring = parachain_config.force_authoring;416 let validator = parachain_config.role.is_authority();417 let prometheus_registry = parachain_config.prometheus_registry().cloned();418 let transaction_pool = params.transaction_pool.clone();419 let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);420421 let (network, system_rpc_tx, start_network) =422 sc_service::build_network(sc_service::BuildNetworkParams {423 config: ¶chain_config,424 client: client.clone(),425 transaction_pool: transaction_pool.clone(),426 spawn_handle: task_manager.spawn_handle(),427 import_queue: import_queue.clone(),428 block_announce_validator_builder: Some(Box::new(|_| {429 Box::new(block_announce_validator)430 })),431 warp_sync: None,432 })?;433434 let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());435 let rpc_client = client.clone();436 let rpc_pool = transaction_pool.clone();437 let select_chain = params.select_chain.clone();438 let rpc_network = network.clone();439440 let rpc_frontier_backend = frontier_backend.clone();441442 let block_data_cache = Arc::new(fc_rpc::EthBlockDataCache::new(443 task_manager.spawn_handle(),444 overrides_handle::<_, _, Runtime>(client.clone()),445 50,446 50,447 ));448449 let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {450 let full_deps = unique_rpc::FullDeps {451 backend: rpc_frontier_backend.clone(),452 deny_unsafe,453 client: rpc_client.clone(),454 pool: rpc_pool.clone(),455 graph: rpc_pool.pool().clone(),456 // TODO: Unhardcode457 enable_dev_signer: false,458 filter_pool: filter_pool.clone(),459 network: rpc_network.clone(),460 select_chain: select_chain.clone(),461 is_authority: validator,462 // TODO: Unhardcode463 max_past_logs: 10000,464 block_data_cache: block_data_cache.clone(),465 fee_history_cache: fee_history_cache.clone(),466 // TODO: Unhardcode467 fee_history_limit: 2048,468 };469470 Ok(471 unique_rpc::create_full::<_, _, _, _, Runtime, RuntimeApi, _>(472 full_deps,473 subscription_executor.clone(),474 ),475 )476 });477478 task_manager.spawn_essential_handle().spawn(479 "frontier-mapping-sync-worker",480 None,481 MappingSyncWorker::new(482 client.import_notification_stream(),483 Duration::new(6, 0),484 client.clone(),485 backend.clone(),486 frontier_backend.clone(),487 SyncStrategy::Normal,488 )489 .for_each(|()| futures::future::ready(())),490 );491492 sc_service::spawn_tasks(sc_service::SpawnTasksParams {493 rpc_extensions_builder,494 client: client.clone(),495 transaction_pool: transaction_pool.clone(),496 task_manager: &mut task_manager,497 config: parachain_config,498 keystore: params.keystore_container.sync_keystore(),499 backend: backend.clone(),500 network: network.clone(),501 system_rpc_tx,502 telemetry: telemetry.as_mut(),503 })?;504505 let announce_block = {506 let network = network.clone();507 Arc::new(move |hash, data| network.announce_block(hash, data))508 };509510 let relay_chain_slot_duration = Duration::from_secs(6);511512 if validator {513 let parachain_consensus = build_consensus(514 client.clone(),515 prometheus_registry.as_ref(),516 telemetry.as_ref().map(|t| t.handle()),517 &task_manager,518 relay_chain_interface.clone(),519 transaction_pool,520 network,521 params.keystore_container.sync_keystore(),522 force_authoring,523 )?;524525 let spawner = task_manager.spawn_handle();526527 let params = StartCollatorParams {528 para_id: id,529 block_status: client.clone(),530 announce_block,531 client: client.clone(),532 task_manager: &mut task_manager,533 spawner,534 parachain_consensus,535 import_queue,536 collator_key: collator_key.expect("Command line arguments do not allow this. qed"),537 relay_chain_interface,538 relay_chain_slot_duration,539 };540541 start_collator(params).await?;542 } else {543 let params = StartFullNodeParams {544 client: client.clone(),545 announce_block,546 task_manager: &mut task_manager,547 para_id: id,548 import_queue,549 relay_chain_interface,550 relay_chain_slot_duration,551 collator_options,552 };553554 start_full_node(params)?;555 }556557 start_network.start_network();558559 Ok((task_manager, client))560}561562/// Build the import queue for the the parachain runtime.563pub fn parachain_build_import_queue<RuntimeApi, ExecutorDispatch>(564 client: Arc<FullClient<RuntimeApi, ExecutorDispatch>>,565 config: &Configuration,566 telemetry: Option<TelemetryHandle>,567 task_manager: &TaskManager,568) -> Result<569 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,570 sc_service::Error,571>572where573 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>574 + Send575 + Sync576 + 'static,577 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>578 + sp_block_builder::BlockBuilder<Block>579 + sp_consensus_aura::AuraApi<Block, AuraId>580 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>,581 ExecutorDispatch: NativeExecutionDispatch + 'static,582{583 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;584585 cumulus_client_consensus_aura::import_queue::<586 sp_consensus_aura::sr25519::AuthorityPair,587 _,588 _,589 _,590 _,591 _,592 _,593 >(cumulus_client_consensus_aura::ImportQueueParams {594 block_import: client.clone(),595 client: client.clone(),596 create_inherent_data_providers: move |_, _| async move {597 let time = sp_timestamp::InherentDataProvider::from_system_time();598599 let slot =600 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(601 *time,602 slot_duration,603 );604605 Ok((time, slot))606 },607 registry: config.prometheus_registry(),608 can_author_with: sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()),609 spawner: &task_manager.spawn_essential_handle(),610 telemetry,611 })612 .map_err(Into::into)613}614615/// Start a normal parachain node.616pub async fn start_node<Runtime, RuntimeApi, ExecutorDispatch>(617 parachain_config: Configuration,618 polkadot_config: Configuration,619 collator_options: CollatorOptions,620 id: ParaId,621) -> sc_service::error::Result<(TaskManager, Arc<FullClient<RuntimeApi, ExecutorDispatch>>)>622where623 Runtime: RuntimeInstance + Send + Sync + 'static,624 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,625 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,626 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>627 + Send628 + Sync629 + 'static,630 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>631 + fp_rpc::EthereumRuntimeRPCApi<Block>632 + sp_session::SessionKeys<Block>633 + sp_block_builder::BlockBuilder<Block>634 + pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi<Block, Balance>635 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>636 + up_rpc::UniqueApi<Block, Runtime::CrossAccountId, AccountId>637 + substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Index>638 + sp_api::Metadata<Block>639 + sp_offchain::OffchainWorkerApi<Block>640 + cumulus_primitives_core::CollectCollationInfo<Block>641 + sp_consensus_aura::AuraApi<Block, AuraId>,642 ExecutorDispatch: NativeExecutionDispatch + 'static,643{644 start_node_impl::<Runtime, RuntimeApi, ExecutorDispatch, _, _>(645 parachain_config,646 polkadot_config,647 collator_options,648 id,649 parachain_build_import_queue,650 |client,651 prometheus_registry,652 telemetry,653 task_manager,654 relay_chain_interface,655 transaction_pool,656 sync_oracle,657 keystore,658 force_authoring| {659 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;660661 let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(662 task_manager.spawn_handle(),663 client.clone(),664 transaction_pool,665 prometheus_registry,666 telemetry.clone(),667 );668669 Ok(AuraConsensus::build::<670 sp_consensus_aura::sr25519::AuthorityPair,671 _,672 _,673 _,674 _,675 _,676 _,677 >(BuildAuraConsensusParams {678 proposer_factory,679 create_inherent_data_providers: move |_, (relay_parent, validation_data)| {680 let relay_chain_interface = relay_chain_interface.clone();681 async move {682 let parachain_inherent =683 cumulus_primitives_parachain_inherent::ParachainInherentData::create_at(684 relay_parent,685 &relay_chain_interface,686 &validation_data,687 id,688 ).await;689690 let time = sp_timestamp::InherentDataProvider::from_system_time();691692 let slot =693 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(694 *time,695 slot_duration,696 );697698 let parachain_inherent = parachain_inherent.ok_or_else(|| {699 Box::<dyn std::error::Error + Send + Sync>::from(700 "Failed to create parachain inherent",701 )702 })?;703 Ok((time, slot, parachain_inherent))704 }705 },706 block_import: client.clone(),707 para_client: client,708 backoff_authoring_blocks: Option::<()>::None,709 sync_oracle,710 keystore,711 force_authoring,712 slot_duration,713 // We got around 500ms for proposing714 block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),715 telemetry,716 max_block_proposal_slot_portion: None,717 }))718 },719 )720 .await721}722723fn dev_build_import_queue<RuntimeApi, ExecutorDispatch>(724 client: Arc<FullClient<RuntimeApi, ExecutorDispatch>>,725 config: &Configuration,726 _: Option<TelemetryHandle>,727 task_manager: &TaskManager,728) -> Result<729 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,730 sc_service::Error,731>732where733 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>734 + Send735 + Sync736 + 'static,737 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>738 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>,739 ExecutorDispatch: NativeExecutionDispatch + 'static,740{741 Ok(sc_consensus_manual_seal::import_queue(742 Box::new(client.clone()),743 &task_manager.spawn_essential_handle(),744 config.prometheus_registry(),745 ))746}747748/// Builds a new development service. This service uses instant seal, and mocks749/// the parachain inherent750pub fn start_dev_node<Runtime, RuntimeApi, ExecutorDispatch>(751 config: Configuration,752 autoseal_interval: AutosealInterval,753) -> sc_service::error::Result<TaskManager>754where755 Runtime: RuntimeInstance + Send + Sync + 'static,756 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,757 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,758 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>759 + Send760 + Sync761 + 'static,762 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>763 + fp_rpc::EthereumRuntimeRPCApi<Block>764 + sp_session::SessionKeys<Block>765 + sp_block_builder::BlockBuilder<Block>766 + pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi<Block, Balance>767 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>768 + up_rpc::UniqueApi<Block, Runtime::CrossAccountId, AccountId>769 + substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Index>770 + sp_api::Metadata<Block>771 + sp_offchain::OffchainWorkerApi<Block>772 + cumulus_primitives_core::CollectCollationInfo<Block>773 + sp_consensus_aura::AuraApi<Block, AuraId>,774 ExecutorDispatch: NativeExecutionDispatch + 'static,775{776 use sc_consensus_manual_seal::{run_manual_seal, EngineCommand, ManualSealParams};777 use fc_consensus::FrontierBlockImport;778 use sc_client_api::HeaderBackend;779780 let sc_service::PartialComponents {781 client,782 backend,783 mut task_manager,784 import_queue,785 keystore_container,786 select_chain: maybe_select_chain,787 transaction_pool,788 other:789 (telemetry, filter_pool, frontier_backend, _telemetry_worker_handle, fee_history_cache),790 } = new_partial::<RuntimeApi, ExecutorDispatch, _>(791 &config,792 dev_build_import_queue::<RuntimeApi, ExecutorDispatch>,793 )?;794795 let block_data_cache = Arc::new(fc_rpc::EthBlockDataCache::new(796 task_manager.spawn_handle(),797 overrides_handle::<_, _, Runtime>(client.clone()),798 50,799 50,800 ));801802 let (network, system_rpc_tx, network_starter) =803 sc_service::build_network(sc_service::BuildNetworkParams {804 config: &config,805 client: client.clone(),806 transaction_pool: transaction_pool.clone(),807 spawn_handle: task_manager.spawn_handle(),808 import_queue,809 block_announce_validator_builder: None,810 warp_sync: None,811 })?;812813 if config.offchain_worker.enabled {814 sc_service::build_offchain_workers(815 &config,816 task_manager.spawn_handle(),817 client.clone(),818 network.clone(),819 );820 }821822 let prometheus_registry = config.prometheus_registry().cloned();823 let collator = config.role.is_authority();824825 let select_chain = maybe_select_chain.clone();826827 if collator {828 let block_import =829 FrontierBlockImport::new(client.clone(), client.clone(), frontier_backend.clone());830831 let env = sc_basic_authorship::ProposerFactory::new(832 task_manager.spawn_handle(),833 client.clone(),834 transaction_pool.clone(),835 prometheus_registry.as_ref(),836 telemetry.as_ref().map(|x| x.handle()),837 );838839 let transactions_commands_stream: Box<dyn Stream<Item = EngineCommand<Hash>> + Send + Sync + Unpin> =840 Box::new(841 transaction_pool842 .pool()843 .validated_pool()844 .import_notification_stream()845 .map(|_| EngineCommand::SealNewBlock {846 create_empty: true,847 finalize: false,848 parent_hash: None,849 sender: None,850 }),851 );852853 let autoseal_interval = Box::pin(autoseal_interval);854 let idle_commands_stream: Box<dyn Stream<Item = EngineCommand<Hash>> + Send + Sync + Unpin> =855 Box::new(856 autoseal_interval.map(|_| EngineCommand::SealNewBlock {857 create_empty: true,858 finalize: false,859 parent_hash: None,860 sender: None,861 })862 );863864 let commands_stream = select(865 transactions_commands_stream,866 idle_commands_stream867 );868869 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;870 let client_set_aside_for_cidp = client.clone();871872 task_manager.spawn_essential_handle().spawn_blocking(873 "authorship_task",874 Some("block-authoring"),875 run_manual_seal(ManualSealParams {876 block_import,877 env,878 client: client.clone(),879 pool: transaction_pool.clone(),880 commands_stream,881 select_chain: select_chain.clone(),882 consensus_data_provider: None,883 create_inherent_data_providers: move |block: Hash, ()| {884 let current_para_block = client_set_aside_for_cidp885 .number(block)886 .expect("Header lookup should succeed")887 .expect("Header passed in as parent should be present in backend.");888889 let client_for_xcm = client_set_aside_for_cidp.clone();890 async move {891 let time = sp_timestamp::InherentDataProvider::from_system_time();892893 let mocked_parachain = cumulus_primitives_parachain_inherent::MockValidationDataInherentDataProvider {894 current_para_block,895 relay_offset: 1000,896 relay_blocks_per_para_block: 2,897 xcm_config: cumulus_primitives_parachain_inherent::MockXcmConfig::new(898 &*client_for_xcm,899 block,900 Default::default(),901 Default::default(),902 ),903 raw_downward_messages: vec![],904 raw_horizontal_messages: vec![],905 };906907 let slot =908 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(909 *time,910 slot_duration,911 );912913 Ok((time, slot, mocked_parachain))914 }915 },916 }),917 );918 }919920 task_manager.spawn_essential_handle().spawn(921 "frontier-mapping-sync-worker",922 Some("block-authoring"),923 MappingSyncWorker::new(924 client.import_notification_stream(),925 Duration::new(6, 0),926 client.clone(),927 backend.clone(),928 frontier_backend.clone(),929 SyncStrategy::Normal,930 )931 .for_each(|()| futures::future::ready(())),932 );933934 let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());935 let rpc_client = client.clone();936 let rpc_pool = transaction_pool.clone();937 let rpc_network = network.clone();938 let rpc_frontier_backend = frontier_backend.clone();939 let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {940 let full_deps = unique_rpc::FullDeps {941 backend: rpc_frontier_backend.clone(),942 deny_unsafe,943 client: rpc_client.clone(),944 pool: rpc_pool.clone(),945 graph: rpc_pool.pool().clone(),946 // TODO: Unhardcode947 enable_dev_signer: false,948 filter_pool: filter_pool.clone(),949 network: rpc_network.clone(),950 select_chain: select_chain.clone(),951 is_authority: collator,952 // TODO: Unhardcode953 max_past_logs: 10000,954 block_data_cache: block_data_cache.clone(),955 fee_history_cache: fee_history_cache.clone(),956 // TODO: Unhardcode957 fee_history_limit: 2048,958 };959960 Ok(961 unique_rpc::create_full::<_, _, _, _, Runtime, RuntimeApi, _>(962 full_deps,963 subscription_executor.clone(),964 ),965 )966 });967968 sc_service::spawn_tasks(sc_service::SpawnTasksParams {969 network,970 client,971 keystore: keystore_container.sync_keystore(),972 task_manager: &mut task_manager,973 transaction_pool,974 rpc_extensions_builder,975 backend,976 system_rpc_tx,977 config,978 telemetry: None,979 })?;980981 network_starter.start_network();982 Ok(task_manager)983}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//! Service and ServiceFactory implementation. Specialized wrapper over substrate service.1819// std20use std::sync::Arc;21use std::sync::Mutex;22use std::collections::BTreeMap;23use std::time::Duration;24use std::pin::Pin;25use fc_rpc_core::types::FeeHistoryCache;26use futures::Future;27use futures::{28 Stream, StreamExt,29 stream::select,30 task::{Context, Poll},31};32use futures_timer::Delay;3334use unique_rpc::overrides_handle;3536use serde::{Serialize, Deserialize};3738// Cumulus Imports39use cumulus_client_consensus_aura::{AuraConsensus, BuildAuraConsensusParams, SlotProportion};40use cumulus_client_consensus_common::ParachainConsensus;41use cumulus_client_service::{42 prepare_node_config, start_collator, start_full_node, StartCollatorParams, StartFullNodeParams,43};44use cumulus_client_cli::CollatorOptions;45use cumulus_client_network::BlockAnnounceValidator;46use cumulus_primitives_core::ParaId;47use cumulus_relay_chain_inprocess_interface::build_inprocess_relay_chain;48use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface, RelayChainResult};49use cumulus_relay_chain_rpc_interface::RelayChainRPCInterface;5051// Substrate Imports52use sc_client_api::ExecutorProvider;53use sc_executor::NativeElseWasmExecutor;54use sc_executor::NativeExecutionDispatch;55use sc_network::NetworkService;56use sc_service::{BasePath, Configuration, PartialComponents, Role, TaskManager};57use sc_telemetry::{Telemetry, TelemetryHandle, TelemetryWorker, TelemetryWorkerHandle};58use sp_keystore::SyncCryptoStorePtr;59use sp_runtime::traits::BlakeTwo256;60use substrate_prometheus_endpoint::Registry;61use sc_client_api::BlockchainEvents;6263use polkadot_service::CollatorPair;6465// Frontier Imports66use fc_rpc_core::types::FilterPool;67use fc_mapping_sync::{MappingSyncWorker, SyncStrategy};6869use unique_runtime_common::types::{AuraId, RuntimeInstance, AccountId, Balance, Index, Hash, Block};7071/// Unique native executor instance.72#[cfg(feature = "unique-runtime")]73pub struct UniqueRuntimeExecutor;7475#[cfg(feature = "quartz-runtime")]76/// Quartz native executor instance.7778pub struct QuartzRuntimeExecutor;7980/// Opal native executor instance.81pub struct OpalRuntimeExecutor;8283#[cfg(feature = "unique-runtime")]84impl NativeExecutionDispatch for UniqueRuntimeExecutor {85 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;8687 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {88 unique_runtime::api::dispatch(method, data)89 }9091 fn native_version() -> sc_executor::NativeVersion {92 unique_runtime::native_version()93 }94}9596#[cfg(feature = "quartz-runtime")]97impl NativeExecutionDispatch for QuartzRuntimeExecutor {98 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;99100 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {101 quartz_runtime::api::dispatch(method, data)102 }103104 fn native_version() -> sc_executor::NativeVersion {105 quartz_runtime::native_version()106 }107}108109impl NativeExecutionDispatch for OpalRuntimeExecutor {110 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;111112 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {113 opal_runtime::api::dispatch(method, data)114 }115116 fn native_version() -> sc_executor::NativeVersion {117 opal_runtime::native_version()118 }119}120121pub struct AutosealInterval {122 duration: Duration,123 delay_handle: Pin<Box<Delay>>,124}125126impl AutosealInterval {127 pub fn new(duration: Duration) -> Result<Self, String> {128 if duration.is_zero() {129 return Err("Invalid autoseal interval: 0 seconds".into());130 }131132 Ok(Self {133 duration,134 delay_handle: Box::pin(Delay::new(duration)),135 })136 }137}138139impl Stream for AutosealInterval {140 type Item = ();141142 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {143 match self.delay_handle.as_mut().poll(cx) {144 Poll::Ready(_) => {145 let duration = self.duration;146 self.delay_handle.reset(duration);147148 Poll::Ready(Some(()))149 }150 Poll::Pending => Poll::Pending,151 }152 }153}154155pub fn open_frontier_backend(config: &Configuration) -> Result<Arc<fc_db::Backend<Block>>, String> {156 let config_dir = config157 .base_path158 .as_ref()159 .map(|base_path| base_path.config_dir(config.chain_spec.id()))160 .unwrap_or_else(|| {161 BasePath::from_project("", "", "unique").config_dir(config.chain_spec.id())162 });163 let database_dir = config_dir.join("frontier").join("db");164165 Ok(Arc::new(fc_db::Backend::<Block>::new(166 &fc_db::DatabaseSettings {167 source: fc_db::DatabaseSettingsSrc::RocksDb {168 path: database_dir,169 cache_size: 0,170 },171 },172 )?))173}174175type FullClient<RuntimeApi, ExecutorDispatch> =176 sc_service::TFullClient<Block, RuntimeApi, NativeElseWasmExecutor<ExecutorDispatch>>;177type FullBackend = sc_service::TFullBackend<Block>;178type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;179180/// Starts a `ServiceBuilder` for a full service.181///182/// Use this macro if you don't actually need the full service, but just the builder in order to183/// be able to perform chain operations.184#[allow(clippy::type_complexity)]185pub fn new_partial<RuntimeApi, ExecutorDispatch, BIQ>(186 config: &Configuration,187 build_import_queue: BIQ,188) -> Result<189 PartialComponents<190 FullClient<RuntimeApi, ExecutorDispatch>,191 FullBackend,192 FullSelectChain,193 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,194 sc_transaction_pool::FullPool<Block, FullClient<RuntimeApi, ExecutorDispatch>>,195 (196 Option<Telemetry>,197 Option<FilterPool>,198 Arc<fc_db::Backend<Block>>,199 Option<TelemetryWorkerHandle>,200 FeeHistoryCache,201 ),202 >,203 sc_service::Error,204>205where206 sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,207 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>208 + Send209 + Sync210 + 'static,211 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>,212 ExecutorDispatch: NativeExecutionDispatch + 'static,213 BIQ: FnOnce(214 Arc<FullClient<RuntimeApi, ExecutorDispatch>>,215 &Configuration,216 Option<TelemetryHandle>,217 &TaskManager,218 ) -> Result<219 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,220 sc_service::Error,221 >,222{223 let _telemetry = config224 .telemetry_endpoints225 .clone()226 .filter(|x| !x.is_empty())227 .map(|endpoints| -> Result<_, sc_telemetry::Error> {228 let worker = TelemetryWorker::new(16)?;229 let telemetry = worker.handle().new_telemetry(endpoints);230 Ok((worker, telemetry))231 })232 .transpose()?;233234 let telemetry = config235 .telemetry_endpoints236 .clone()237 .filter(|x| !x.is_empty())238 .map(|endpoints| -> Result<_, sc_telemetry::Error> {239 let worker = TelemetryWorker::new(16)?;240 let telemetry = worker.handle().new_telemetry(endpoints);241 Ok((worker, telemetry))242 })243 .transpose()?;244245 let executor = NativeElseWasmExecutor::<ExecutorDispatch>::new(246 config.wasm_method,247 config.default_heap_pages,248 config.max_runtime_instances,249 config.runtime_cache_size,250 );251252 let (client, backend, keystore_container, task_manager) =253 sc_service::new_full_parts::<Block, RuntimeApi, _>(254 config,255 telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),256 executor,257 )?;258 let client = Arc::new(client);259260 let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());261262 let telemetry = telemetry.map(|(worker, telemetry)| {263 task_manager264 .spawn_handle()265 .spawn("telemetry", None, worker.run());266 telemetry267 });268269 let select_chain = sc_consensus::LongestChain::new(backend.clone());270271 let transaction_pool = sc_transaction_pool::BasicPool::new_full(272 config.transaction_pool.clone(),273 config.role.is_authority().into(),274 config.prometheus_registry(),275 task_manager.spawn_essential_handle(),276 client.clone(),277 );278279 let filter_pool: Option<FilterPool> = Some(Arc::new(Mutex::new(BTreeMap::new())));280281 let frontier_backend = open_frontier_backend(config)?;282283 let import_queue = build_import_queue(284 client.clone(),285 config,286 telemetry.as_ref().map(|telemetry| telemetry.handle()),287 &task_manager,288 )?;289 let fee_history_cache: FeeHistoryCache = Arc::new(Mutex::new(BTreeMap::new()));290291 let params = PartialComponents {292 backend,293 client,294 import_queue,295 keystore_container,296 task_manager,297 transaction_pool,298 select_chain,299 other: (300 telemetry,301 filter_pool,302 frontier_backend,303 telemetry_worker_handle,304 fee_history_cache,305 ),306 };307308 Ok(params)309}310311async fn build_relay_chain_interface(312 polkadot_config: Configuration,313 parachain_config: &Configuration,314 telemetry_worker_handle: Option<TelemetryWorkerHandle>,315 task_manager: &mut TaskManager,316 collator_options: CollatorOptions,317) -> RelayChainResult<(318 Arc<(dyn RelayChainInterface + 'static)>,319 Option<CollatorPair>,320)> {321 match collator_options.relay_chain_rpc_url {322 Some(relay_chain_url) => Ok((323 Arc::new(RelayChainRPCInterface::new(relay_chain_url).await?) as Arc<_>,324 None,325 )),326 None => build_inprocess_relay_chain(327 polkadot_config,328 parachain_config,329 telemetry_worker_handle,330 task_manager,331 ),332 }333}334335/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.336///337/// This is the actual implementation that is abstract over the executor and the runtime api.338#[sc_tracing::logging::prefix_logs_with("Parachain")]339async fn start_node_impl<Runtime, RuntimeApi, ExecutorDispatch, BIQ, BIC>(340 parachain_config: Configuration,341 polkadot_config: Configuration,342 collator_options: CollatorOptions,343 id: ParaId,344 build_import_queue: BIQ,345 build_consensus: BIC,346) -> sc_service::error::Result<(TaskManager, Arc<FullClient<RuntimeApi, ExecutorDispatch>>)>347where348 sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,349 Runtime: RuntimeInstance + Send + Sync + 'static,350 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,351 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,352 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>353 + Send354 + Sync355 + 'static,356 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>357 + fp_rpc::EthereumRuntimeRPCApi<Block>358 + sp_session::SessionKeys<Block>359 + sp_block_builder::BlockBuilder<Block>360 + pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi<Block, Balance>361 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>362 + up_rpc::UniqueApi<Block, Runtime::CrossAccountId, AccountId>363 + substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Index>364 + sp_api::Metadata<Block>365 + sp_offchain::OffchainWorkerApi<Block>366 + cumulus_primitives_core::CollectCollationInfo<Block>,367 ExecutorDispatch: NativeExecutionDispatch + 'static,368 BIQ: FnOnce(369 Arc<FullClient<RuntimeApi, ExecutorDispatch>>,370 &Configuration,371 Option<TelemetryHandle>,372 &TaskManager,373 ) -> Result<374 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,375 sc_service::Error,376 >,377 BIC: FnOnce(378 Arc<FullClient<RuntimeApi, ExecutorDispatch>>,379 Option<&Registry>,380 Option<TelemetryHandle>,381 &TaskManager,382 Arc<dyn RelayChainInterface>,383 Arc<sc_transaction_pool::FullPool<Block, FullClient<RuntimeApi, ExecutorDispatch>>>,384 Arc<NetworkService<Block, Hash>>,385 SyncCryptoStorePtr,386 bool,387 ) -> Result<Box<dyn ParachainConsensus<Block>>, sc_service::Error>,388{389 if matches!(parachain_config.role, Role::Light) {390 return Err("Light client not supported!".into());391 }392393 let parachain_config = prepare_node_config(parachain_config);394395 let params =396 new_partial::<RuntimeApi, ExecutorDispatch, BIQ>(¶chain_config, build_import_queue)?;397 let (mut telemetry, filter_pool, frontier_backend, telemetry_worker_handle, fee_history_cache) =398 params.other;399400 let client = params.client.clone();401 let backend = params.backend.clone();402 let mut task_manager = params.task_manager;403404 let (relay_chain_interface, collator_key) = build_relay_chain_interface(405 polkadot_config,406 ¶chain_config,407 telemetry_worker_handle,408 &mut task_manager,409 collator_options.clone(),410 )411 .await412 .map_err(|e| match e {413 RelayChainError::ServiceError(polkadot_service::Error::Sub(x)) => x,414 s => s.to_string().into(),415 })?;416417 let block_announce_validator = BlockAnnounceValidator::new(relay_chain_interface.clone(), id);418419 let force_authoring = parachain_config.force_authoring;420 let validator = parachain_config.role.is_authority();421 let prometheus_registry = parachain_config.prometheus_registry().cloned();422 let transaction_pool = params.transaction_pool.clone();423 let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);424425 let (network, system_rpc_tx, start_network) =426 sc_service::build_network(sc_service::BuildNetworkParams {427 config: ¶chain_config,428 client: client.clone(),429 transaction_pool: transaction_pool.clone(),430 spawn_handle: task_manager.spawn_handle(),431 import_queue: import_queue.clone(),432 block_announce_validator_builder: Some(Box::new(|_| {433 Box::new(block_announce_validator)434 })),435 warp_sync: None,436 })?;437438 let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());439 let rpc_client = client.clone();440 let rpc_pool = transaction_pool.clone();441 let select_chain = params.select_chain.clone();442 let rpc_network = network.clone();443444 let rpc_frontier_backend = frontier_backend.clone();445446 let block_data_cache = Arc::new(fc_rpc::EthBlockDataCache::new(447 task_manager.spawn_handle(),448 overrides_handle::<_, _, Runtime>(client.clone()),449 50,450 50,451 ));452453 let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {454 let full_deps = unique_rpc::FullDeps {455 backend: rpc_frontier_backend.clone(),456 deny_unsafe,457 client: rpc_client.clone(),458 pool: rpc_pool.clone(),459 graph: rpc_pool.pool().clone(),460 // TODO: Unhardcode461 enable_dev_signer: false,462 filter_pool: filter_pool.clone(),463 network: rpc_network.clone(),464 select_chain: select_chain.clone(),465 is_authority: validator,466 // TODO: Unhardcode467 max_past_logs: 10000,468 block_data_cache: block_data_cache.clone(),469 fee_history_cache: fee_history_cache.clone(),470 // TODO: Unhardcode471 fee_history_limit: 2048,472 };473474 Ok(475 unique_rpc::create_full::<_, _, _, _, Runtime, RuntimeApi, _>(476 full_deps,477 subscription_executor.clone(),478 ),479 )480 });481482 task_manager.spawn_essential_handle().spawn(483 "frontier-mapping-sync-worker",484 None,485 MappingSyncWorker::new(486 client.import_notification_stream(),487 Duration::new(6, 0),488 client.clone(),489 backend.clone(),490 frontier_backend.clone(),491 SyncStrategy::Normal,492 )493 .for_each(|()| futures::future::ready(())),494 );495496 sc_service::spawn_tasks(sc_service::SpawnTasksParams {497 rpc_extensions_builder,498 client: client.clone(),499 transaction_pool: transaction_pool.clone(),500 task_manager: &mut task_manager,501 config: parachain_config,502 keystore: params.keystore_container.sync_keystore(),503 backend: backend.clone(),504 network: network.clone(),505 system_rpc_tx,506 telemetry: telemetry.as_mut(),507 })?;508509 let announce_block = {510 let network = network.clone();511 Arc::new(move |hash, data| network.announce_block(hash, data))512 };513514 let relay_chain_slot_duration = Duration::from_secs(6);515516 if validator {517 let parachain_consensus = build_consensus(518 client.clone(),519 prometheus_registry.as_ref(),520 telemetry.as_ref().map(|t| t.handle()),521 &task_manager,522 relay_chain_interface.clone(),523 transaction_pool,524 network,525 params.keystore_container.sync_keystore(),526 force_authoring,527 )?;528529 let spawner = task_manager.spawn_handle();530531 let params = StartCollatorParams {532 para_id: id,533 block_status: client.clone(),534 announce_block,535 client: client.clone(),536 task_manager: &mut task_manager,537 spawner,538 parachain_consensus,539 import_queue,540 collator_key: collator_key.expect("Command line arguments do not allow this. qed"),541 relay_chain_interface,542 relay_chain_slot_duration,543 };544545 start_collator(params).await?;546 } else {547 let params = StartFullNodeParams {548 client: client.clone(),549 announce_block,550 task_manager: &mut task_manager,551 para_id: id,552 import_queue,553 relay_chain_interface,554 relay_chain_slot_duration,555 collator_options,556 };557558 start_full_node(params)?;559 }560561 start_network.start_network();562563 Ok((task_manager, client))564}565566/// Build the import queue for the the parachain runtime.567pub fn parachain_build_import_queue<RuntimeApi, ExecutorDispatch>(568 client: Arc<FullClient<RuntimeApi, ExecutorDispatch>>,569 config: &Configuration,570 telemetry: Option<TelemetryHandle>,571 task_manager: &TaskManager,572) -> Result<573 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,574 sc_service::Error,575>576where577 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>578 + Send579 + Sync580 + 'static,581 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>582 + sp_block_builder::BlockBuilder<Block>583 + sp_consensus_aura::AuraApi<Block, AuraId>584 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>,585 ExecutorDispatch: NativeExecutionDispatch + 'static,586{587 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;588589 cumulus_client_consensus_aura::import_queue::<590 sp_consensus_aura::sr25519::AuthorityPair,591 _,592 _,593 _,594 _,595 _,596 _,597 >(cumulus_client_consensus_aura::ImportQueueParams {598 block_import: client.clone(),599 client: client.clone(),600 create_inherent_data_providers: move |_, _| async move {601 let time = sp_timestamp::InherentDataProvider::from_system_time();602603 let slot =604 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(605 *time,606 slot_duration,607 );608609 Ok((time, slot))610 },611 registry: config.prometheus_registry(),612 can_author_with: sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()),613 spawner: &task_manager.spawn_essential_handle(),614 telemetry,615 })616 .map_err(Into::into)617}618619/// Start a normal parachain node.620pub async fn start_node<Runtime, RuntimeApi, ExecutorDispatch>(621 parachain_config: Configuration,622 polkadot_config: Configuration,623 collator_options: CollatorOptions,624 id: ParaId,625) -> sc_service::error::Result<(TaskManager, Arc<FullClient<RuntimeApi, ExecutorDispatch>>)>626where627 Runtime: RuntimeInstance + Send + Sync + 'static,628 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,629 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,630 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>631 + Send632 + Sync633 + 'static,634 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>635 + fp_rpc::EthereumRuntimeRPCApi<Block>636 + sp_session::SessionKeys<Block>637 + sp_block_builder::BlockBuilder<Block>638 + pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi<Block, Balance>639 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>640 + up_rpc::UniqueApi<Block, Runtime::CrossAccountId, AccountId>641 + substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Index>642 + sp_api::Metadata<Block>643 + sp_offchain::OffchainWorkerApi<Block>644 + cumulus_primitives_core::CollectCollationInfo<Block>645 + sp_consensus_aura::AuraApi<Block, AuraId>,646 ExecutorDispatch: NativeExecutionDispatch + 'static,647{648 start_node_impl::<Runtime, RuntimeApi, ExecutorDispatch, _, _>(649 parachain_config,650 polkadot_config,651 collator_options,652 id,653 parachain_build_import_queue,654 |client,655 prometheus_registry,656 telemetry,657 task_manager,658 relay_chain_interface,659 transaction_pool,660 sync_oracle,661 keystore,662 force_authoring| {663 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;664665 let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(666 task_manager.spawn_handle(),667 client.clone(),668 transaction_pool,669 prometheus_registry,670 telemetry.clone(),671 );672673 Ok(AuraConsensus::build::<674 sp_consensus_aura::sr25519::AuthorityPair,675 _,676 _,677 _,678 _,679 _,680 _,681 >(BuildAuraConsensusParams {682 proposer_factory,683 create_inherent_data_providers: move |_, (relay_parent, validation_data)| {684 let relay_chain_interface = relay_chain_interface.clone();685 async move {686 let parachain_inherent =687 cumulus_primitives_parachain_inherent::ParachainInherentData::create_at(688 relay_parent,689 &relay_chain_interface,690 &validation_data,691 id,692 ).await;693694 let time = sp_timestamp::InherentDataProvider::from_system_time();695696 let slot =697 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(698 *time,699 slot_duration,700 );701702 let parachain_inherent = parachain_inherent.ok_or_else(|| {703 Box::<dyn std::error::Error + Send + Sync>::from(704 "Failed to create parachain inherent",705 )706 })?;707 Ok((time, slot, parachain_inherent))708 }709 },710 block_import: client.clone(),711 para_client: client,712 backoff_authoring_blocks: Option::<()>::None,713 sync_oracle,714 keystore,715 force_authoring,716 slot_duration,717 // We got around 500ms for proposing718 block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),719 telemetry,720 max_block_proposal_slot_portion: None,721 }))722 },723 )724 .await725}726727fn dev_build_import_queue<RuntimeApi, ExecutorDispatch>(728 client: Arc<FullClient<RuntimeApi, ExecutorDispatch>>,729 config: &Configuration,730 _: Option<TelemetryHandle>,731 task_manager: &TaskManager,732) -> Result<733 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,734 sc_service::Error,735>736where737 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>738 + Send739 + Sync740 + 'static,741 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>742 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>,743 ExecutorDispatch: NativeExecutionDispatch + 'static,744{745 Ok(sc_consensus_manual_seal::import_queue(746 Box::new(client.clone()),747 &task_manager.spawn_essential_handle(),748 config.prometheus_registry(),749 ))750}751752/// Builds a new development service. This service uses instant seal, and mocks753/// the parachain inherent754pub fn start_dev_node<Runtime, RuntimeApi, ExecutorDispatch>(755 config: Configuration,756 autoseal_interval: AutosealInterval,757) -> sc_service::error::Result<TaskManager>758where759 Runtime: RuntimeInstance + Send + Sync + 'static,760 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,761 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,762 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>763 + Send764 + Sync765 + 'static,766 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>767 + fp_rpc::EthereumRuntimeRPCApi<Block>768 + sp_session::SessionKeys<Block>769 + sp_block_builder::BlockBuilder<Block>770 + pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi<Block, Balance>771 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>772 + up_rpc::UniqueApi<Block, Runtime::CrossAccountId, AccountId>773 + substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Index>774 + sp_api::Metadata<Block>775 + sp_offchain::OffchainWorkerApi<Block>776 + cumulus_primitives_core::CollectCollationInfo<Block>777 + sp_consensus_aura::AuraApi<Block, AuraId>,778 ExecutorDispatch: NativeExecutionDispatch + 'static,779{780 use sc_consensus_manual_seal::{run_manual_seal, EngineCommand, ManualSealParams};781 use fc_consensus::FrontierBlockImport;782 use sc_client_api::HeaderBackend;783784 let sc_service::PartialComponents {785 client,786 backend,787 mut task_manager,788 import_queue,789 keystore_container,790 select_chain: maybe_select_chain,791 transaction_pool,792 other:793 (telemetry, filter_pool, frontier_backend, _telemetry_worker_handle, fee_history_cache),794 } = new_partial::<RuntimeApi, ExecutorDispatch, _>(795 &config,796 dev_build_import_queue::<RuntimeApi, ExecutorDispatch>,797 )?;798799 let block_data_cache = Arc::new(fc_rpc::EthBlockDataCache::new(800 task_manager.spawn_handle(),801 overrides_handle::<_, _, Runtime>(client.clone()),802 50,803 50,804 ));805806 let (network, system_rpc_tx, network_starter) =807 sc_service::build_network(sc_service::BuildNetworkParams {808 config: &config,809 client: client.clone(),810 transaction_pool: transaction_pool.clone(),811 spawn_handle: task_manager.spawn_handle(),812 import_queue,813 block_announce_validator_builder: None,814 warp_sync: None,815 })?;816817 if config.offchain_worker.enabled {818 sc_service::build_offchain_workers(819 &config,820 task_manager.spawn_handle(),821 client.clone(),822 network.clone(),823 );824 }825826 let prometheus_registry = config.prometheus_registry().cloned();827 let collator = config.role.is_authority();828829 let select_chain = maybe_select_chain.clone();830831 if collator {832 let block_import =833 FrontierBlockImport::new(client.clone(), client.clone(), frontier_backend.clone());834835 let env = sc_basic_authorship::ProposerFactory::new(836 task_manager.spawn_handle(),837 client.clone(),838 transaction_pool.clone(),839 prometheus_registry.as_ref(),840 telemetry.as_ref().map(|x| x.handle()),841 );842843 let transactions_commands_stream: Box<844 dyn Stream<Item = EngineCommand<Hash>> + Send + Sync + Unpin,845 > = Box::new(846 transaction_pool847 .pool()848 .validated_pool()849 .import_notification_stream()850 .map(|_| EngineCommand::SealNewBlock {851 create_empty: true,852 finalize: false,853 parent_hash: None,854 sender: None,855 }),856 );857858 let autoseal_interval = Box::pin(autoseal_interval);859 let idle_commands_stream: Box<860 dyn Stream<Item = EngineCommand<Hash>> + Send + Sync + Unpin,861 > = Box::new(autoseal_interval.map(|_| EngineCommand::SealNewBlock {862 create_empty: true,863 finalize: false,864 parent_hash: None,865 sender: None,866 }));867868 let commands_stream = select(transactions_commands_stream, idle_commands_stream);869870 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;871 let client_set_aside_for_cidp = client.clone();872873 task_manager.spawn_essential_handle().spawn_blocking(874 "authorship_task",875 Some("block-authoring"),876 run_manual_seal(ManualSealParams {877 block_import,878 env,879 client: client.clone(),880 pool: transaction_pool.clone(),881 commands_stream,882 select_chain: select_chain.clone(),883 consensus_data_provider: None,884 create_inherent_data_providers: move |block: Hash, ()| {885 let current_para_block = client_set_aside_for_cidp886 .number(block)887 .expect("Header lookup should succeed")888 .expect("Header passed in as parent should be present in backend.");889890 let client_for_xcm = client_set_aside_for_cidp.clone();891 async move {892 let time = sp_timestamp::InherentDataProvider::from_system_time();893894 let mocked_parachain = cumulus_primitives_parachain_inherent::MockValidationDataInherentDataProvider {895 current_para_block,896 relay_offset: 1000,897 relay_blocks_per_para_block: 2,898 xcm_config: cumulus_primitives_parachain_inherent::MockXcmConfig::new(899 &*client_for_xcm,900 block,901 Default::default(),902 Default::default(),903 ),904 raw_downward_messages: vec![],905 raw_horizontal_messages: vec![],906 };907908 let slot =909 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(910 *time,911 slot_duration,912 );913914 Ok((time, slot, mocked_parachain))915 }916 },917 }),918 );919 }920921 task_manager.spawn_essential_handle().spawn(922 "frontier-mapping-sync-worker",923 Some("block-authoring"),924 MappingSyncWorker::new(925 client.import_notification_stream(),926 Duration::new(6, 0),927 client.clone(),928 backend.clone(),929 frontier_backend.clone(),930 SyncStrategy::Normal,931 )932 .for_each(|()| futures::future::ready(())),933 );934935 let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());936 let rpc_client = client.clone();937 let rpc_pool = transaction_pool.clone();938 let rpc_network = network.clone();939 let rpc_frontier_backend = frontier_backend.clone();940 let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {941 let full_deps = unique_rpc::FullDeps {942 backend: rpc_frontier_backend.clone(),943 deny_unsafe,944 client: rpc_client.clone(),945 pool: rpc_pool.clone(),946 graph: rpc_pool.pool().clone(),947 // TODO: Unhardcode948 enable_dev_signer: false,949 filter_pool: filter_pool.clone(),950 network: rpc_network.clone(),951 select_chain: select_chain.clone(),952 is_authority: collator,953 // TODO: Unhardcode954 max_past_logs: 10000,955 block_data_cache: block_data_cache.clone(),956 fee_history_cache: fee_history_cache.clone(),957 // TODO: Unhardcode958 fee_history_limit: 2048,959 };960961 Ok(962 unique_rpc::create_full::<_, _, _, _, Runtime, RuntimeApi, _>(963 full_deps,964 subscription_executor.clone(),965 ),966 )967 });968969 sc_service::spawn_tasks(sc_service::SpawnTasksParams {970 network,971 client,972 keystore: keystore_container.sync_keystore(),973 task_manager: &mut task_manager,974 transaction_pool,975 rpc_extensions_builder,976 backend,977 system_rpc_tx,978 config,979 telemetry: None,980 })?;981982 network_starter.start_network();983 Ok(task_manager)984}