difftreelog
Use tokio instead of future-timer
in: master
3 files changed
node/cli/Cargo.tomldiffbeforeafterboth--- a/node/cli/Cargo.toml
+++ b/node/cli/Cargo.toml
@@ -294,13 +294,13 @@
[dependencies]
futures = '0.3.17'
-futures-timer = '3.0.2'
log = '0.4.14'
flexi_logger = "0.15.7"
parking_lot = '0.11.2'
clap = "3.1.2"
jsonrpc-core = '18.0.0'
jsonrpc-pubsub = "18.0.0"
+tokio = { version = "1.17.0", features = ["time"] }
fc-rpc-core = { git = "https://github.com/uniquenetwork/frontier.git", branch = "unique-polkadot-v0.9.18" }
fc-consensus = { git = "https://github.com/uniquenetwork/frontier.git", branch = "unique-polkadot-v0.9.18" }
node/cli/src/command.rsdiffbeforeafterboth--- a/node/cli/src/command.rs
+++ b/node/cli/src/command.rs
@@ -35,7 +35,7 @@
use crate::{
chain_spec::{self, RuntimeId, RuntimeIdentification, ServiceId, ServiceIdentification},
cli::{Cli, RelayChainCli, Subcommand},
- service::{new_partial, start_node, start_dev_node, AutosealInterval},
+ service::{new_partial, start_node, start_dev_node},
};
#[cfg(feature = "unique-runtime")]
@@ -405,8 +405,7 @@
if is_dev_service {
info!("Running Dev service");
- let autoseal_interval =
- AutosealInterval::new(Duration::from_millis(cli.idle_autoseal_interval))?;
+ let autoseal_interval = 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::{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}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::{27 Stream, StreamExt,28 stream::select,29 task::{Context, Poll},30};31use tokio::time::Interval;3233use unique_rpc::overrides_handle;3435use serde::{Serialize, Deserialize};3637// Cumulus Imports38use cumulus_client_consensus_aura::{AuraConsensus, BuildAuraConsensusParams, SlotProportion};39use cumulus_client_consensus_common::ParachainConsensus;40use cumulus_client_service::{41 prepare_node_config, start_collator, start_full_node, StartCollatorParams, StartFullNodeParams,42};43use cumulus_client_cli::CollatorOptions;44use cumulus_client_network::BlockAnnounceValidator;45use cumulus_primitives_core::ParaId;46use cumulus_relay_chain_inprocess_interface::build_inprocess_relay_chain;47use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface, RelayChainResult};48use cumulus_relay_chain_rpc_interface::RelayChainRPCInterface;4950// Substrate Imports51use sc_client_api::ExecutorProvider;52use sc_executor::NativeElseWasmExecutor;53use sc_executor::NativeExecutionDispatch;54use sc_network::NetworkService;55use sc_service::{BasePath, Configuration, PartialComponents, Role, TaskManager};56use sc_telemetry::{Telemetry, TelemetryHandle, TelemetryWorker, TelemetryWorkerHandle};57use sp_keystore::SyncCryptoStorePtr;58use sp_runtime::traits::BlakeTwo256;59use substrate_prometheus_endpoint::Registry;60use sc_client_api::BlockchainEvents;6162use polkadot_service::CollatorPair;6364// Frontier Imports65use fc_rpc_core::types::FilterPool;66use fc_mapping_sync::{MappingSyncWorker, SyncStrategy};6768use unique_runtime_common::types::{AuraId, RuntimeInstance, AccountId, Balance, Index, Hash, Block};6970/// Unique native executor instance.71#[cfg(feature = "unique-runtime")]72pub struct UniqueRuntimeExecutor;7374#[cfg(feature = "quartz-runtime")]75/// Quartz native executor instance.7677pub struct QuartzRuntimeExecutor;7879/// Opal native executor instance.80pub struct OpalRuntimeExecutor;8182#[cfg(feature = "unique-runtime")]83impl NativeExecutionDispatch for UniqueRuntimeExecutor {84 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;8586 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {87 unique_runtime::api::dispatch(method, data)88 }8990 fn native_version() -> sc_executor::NativeVersion {91 unique_runtime::native_version()92 }93}9495#[cfg(feature = "quartz-runtime")]96impl NativeExecutionDispatch for QuartzRuntimeExecutor {97 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;9899 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {100 quartz_runtime::api::dispatch(method, data)101 }102103 fn native_version() -> sc_executor::NativeVersion {104 quartz_runtime::native_version()105 }106}107108impl NativeExecutionDispatch for OpalRuntimeExecutor {109 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;110111 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {112 opal_runtime::api::dispatch(method, data)113 }114115 fn native_version() -> sc_executor::NativeVersion {116 opal_runtime::native_version()117 }118}119120pub struct AutosealInterval {121 interval: Interval,122}123124impl AutosealInterval {125 pub fn new(config: &Configuration, interval: Duration) -> Self {126 let _tokio_runtime = config.tokio_handle.enter();127 let interval = tokio::time::interval(interval);128129 Self { interval }130 }131}132133impl Stream for AutosealInterval {134 type Item = tokio::time::Instant;135136 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {137 self.interval.poll_tick(cx).map(Some)138 }139}140141pub fn open_frontier_backend(config: &Configuration) -> Result<Arc<fc_db::Backend<Block>>, String> {142 let config_dir = config143 .base_path144 .as_ref()145 .map(|base_path| base_path.config_dir(config.chain_spec.id()))146 .unwrap_or_else(|| {147 BasePath::from_project("", "", "unique").config_dir(config.chain_spec.id())148 });149 let database_dir = config_dir.join("frontier").join("db");150151 Ok(Arc::new(fc_db::Backend::<Block>::new(152 &fc_db::DatabaseSettings {153 source: fc_db::DatabaseSettingsSrc::RocksDb {154 path: database_dir,155 cache_size: 0,156 },157 },158 )?))159}160161type FullClient<RuntimeApi, ExecutorDispatch> =162 sc_service::TFullClient<Block, RuntimeApi, NativeElseWasmExecutor<ExecutorDispatch>>;163type FullBackend = sc_service::TFullBackend<Block>;164type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;165166/// Starts a `ServiceBuilder` for a full service.167///168/// Use this macro if you don't actually need the full service, but just the builder in order to169/// be able to perform chain operations.170#[allow(clippy::type_complexity)]171pub fn new_partial<RuntimeApi, ExecutorDispatch, BIQ>(172 config: &Configuration,173 build_import_queue: BIQ,174) -> Result<175 PartialComponents<176 FullClient<RuntimeApi, ExecutorDispatch>,177 FullBackend,178 FullSelectChain,179 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,180 sc_transaction_pool::FullPool<Block, FullClient<RuntimeApi, ExecutorDispatch>>,181 (182 Option<Telemetry>,183 Option<FilterPool>,184 Arc<fc_db::Backend<Block>>,185 Option<TelemetryWorkerHandle>,186 FeeHistoryCache,187 ),188 >,189 sc_service::Error,190>191where192 sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,193 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>194 + Send195 + Sync196 + 'static,197 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>,198 ExecutorDispatch: NativeExecutionDispatch + 'static,199 BIQ: FnOnce(200 Arc<FullClient<RuntimeApi, ExecutorDispatch>>,201 &Configuration,202 Option<TelemetryHandle>,203 &TaskManager,204 ) -> Result<205 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,206 sc_service::Error,207 >,208{209 let _telemetry = config210 .telemetry_endpoints211 .clone()212 .filter(|x| !x.is_empty())213 .map(|endpoints| -> Result<_, sc_telemetry::Error> {214 let worker = TelemetryWorker::new(16)?;215 let telemetry = worker.handle().new_telemetry(endpoints);216 Ok((worker, telemetry))217 })218 .transpose()?;219220 let telemetry = config221 .telemetry_endpoints222 .clone()223 .filter(|x| !x.is_empty())224 .map(|endpoints| -> Result<_, sc_telemetry::Error> {225 let worker = TelemetryWorker::new(16)?;226 let telemetry = worker.handle().new_telemetry(endpoints);227 Ok((worker, telemetry))228 })229 .transpose()?;230231 let executor = NativeElseWasmExecutor::<ExecutorDispatch>::new(232 config.wasm_method,233 config.default_heap_pages,234 config.max_runtime_instances,235 config.runtime_cache_size,236 );237238 let (client, backend, keystore_container, task_manager) =239 sc_service::new_full_parts::<Block, RuntimeApi, _>(240 config,241 telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),242 executor,243 )?;244 let client = Arc::new(client);245246 let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());247248 let telemetry = telemetry.map(|(worker, telemetry)| {249 task_manager250 .spawn_handle()251 .spawn("telemetry", None, worker.run());252 telemetry253 });254255 let select_chain = sc_consensus::LongestChain::new(backend.clone());256257 let transaction_pool = sc_transaction_pool::BasicPool::new_full(258 config.transaction_pool.clone(),259 config.role.is_authority().into(),260 config.prometheus_registry(),261 task_manager.spawn_essential_handle(),262 client.clone(),263 );264265 let filter_pool: Option<FilterPool> = Some(Arc::new(Mutex::new(BTreeMap::new())));266267 let frontier_backend = open_frontier_backend(config)?;268269 let import_queue = build_import_queue(270 client.clone(),271 config,272 telemetry.as_ref().map(|telemetry| telemetry.handle()),273 &task_manager,274 )?;275 let fee_history_cache: FeeHistoryCache = Arc::new(Mutex::new(BTreeMap::new()));276277 let params = PartialComponents {278 backend,279 client,280 import_queue,281 keystore_container,282 task_manager,283 transaction_pool,284 select_chain,285 other: (286 telemetry,287 filter_pool,288 frontier_backend,289 telemetry_worker_handle,290 fee_history_cache,291 ),292 };293294 Ok(params)295}296297async fn build_relay_chain_interface(298 polkadot_config: Configuration,299 parachain_config: &Configuration,300 telemetry_worker_handle: Option<TelemetryWorkerHandle>,301 task_manager: &mut TaskManager,302 collator_options: CollatorOptions,303) -> RelayChainResult<(304 Arc<(dyn RelayChainInterface + 'static)>,305 Option<CollatorPair>,306)> {307 match collator_options.relay_chain_rpc_url {308 Some(relay_chain_url) => Ok((309 Arc::new(RelayChainRPCInterface::new(relay_chain_url).await?) as Arc<_>,310 None,311 )),312 None => build_inprocess_relay_chain(313 polkadot_config,314 parachain_config,315 telemetry_worker_handle,316 task_manager,317 ),318 }319}320321/// Start a node with the given parachain `Configuration` and relay chain `Configuration`.322///323/// This is the actual implementation that is abstract over the executor and the runtime api.324#[sc_tracing::logging::prefix_logs_with("Parachain")]325async fn start_node_impl<Runtime, RuntimeApi, ExecutorDispatch, BIQ, BIC>(326 parachain_config: Configuration,327 polkadot_config: Configuration,328 collator_options: CollatorOptions,329 id: ParaId,330 build_import_queue: BIQ,331 build_consensus: BIC,332) -> sc_service::error::Result<(TaskManager, Arc<FullClient<RuntimeApi, ExecutorDispatch>>)>333where334 sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,335 Runtime: RuntimeInstance + Send + Sync + 'static,336 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,337 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,338 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>339 + Send340 + Sync341 + 'static,342 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>343 + fp_rpc::EthereumRuntimeRPCApi<Block>344 + sp_session::SessionKeys<Block>345 + sp_block_builder::BlockBuilder<Block>346 + pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi<Block, Balance>347 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>348 + up_rpc::UniqueApi<Block, Runtime::CrossAccountId, AccountId>349 + substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Index>350 + sp_api::Metadata<Block>351 + sp_offchain::OffchainWorkerApi<Block>352 + cumulus_primitives_core::CollectCollationInfo<Block>,353 ExecutorDispatch: NativeExecutionDispatch + 'static,354 BIQ: FnOnce(355 Arc<FullClient<RuntimeApi, ExecutorDispatch>>,356 &Configuration,357 Option<TelemetryHandle>,358 &TaskManager,359 ) -> Result<360 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,361 sc_service::Error,362 >,363 BIC: FnOnce(364 Arc<FullClient<RuntimeApi, ExecutorDispatch>>,365 Option<&Registry>,366 Option<TelemetryHandle>,367 &TaskManager,368 Arc<dyn RelayChainInterface>,369 Arc<sc_transaction_pool::FullPool<Block, FullClient<RuntimeApi, ExecutorDispatch>>>,370 Arc<NetworkService<Block, Hash>>,371 SyncCryptoStorePtr,372 bool,373 ) -> Result<Box<dyn ParachainConsensus<Block>>, sc_service::Error>,374{375 if matches!(parachain_config.role, Role::Light) {376 return Err("Light client not supported!".into());377 }378379 let parachain_config = prepare_node_config(parachain_config);380381 let params =382 new_partial::<RuntimeApi, ExecutorDispatch, BIQ>(¶chain_config, build_import_queue)?;383 let (mut telemetry, filter_pool, frontier_backend, telemetry_worker_handle, fee_history_cache) =384 params.other;385386 let client = params.client.clone();387 let backend = params.backend.clone();388 let mut task_manager = params.task_manager;389390 let (relay_chain_interface, collator_key) = build_relay_chain_interface(391 polkadot_config,392 ¶chain_config,393 telemetry_worker_handle,394 &mut task_manager,395 collator_options.clone(),396 )397 .await398 .map_err(|e| match e {399 RelayChainError::ServiceError(polkadot_service::Error::Sub(x)) => x,400 s => s.to_string().into(),401 })?;402403 let block_announce_validator = BlockAnnounceValidator::new(relay_chain_interface.clone(), id);404405 let force_authoring = parachain_config.force_authoring;406 let validator = parachain_config.role.is_authority();407 let prometheus_registry = parachain_config.prometheus_registry().cloned();408 let transaction_pool = params.transaction_pool.clone();409 let import_queue = cumulus_client_service::SharedImportQueue::new(params.import_queue);410411 let (network, system_rpc_tx, start_network) =412 sc_service::build_network(sc_service::BuildNetworkParams {413 config: ¶chain_config,414 client: client.clone(),415 transaction_pool: transaction_pool.clone(),416 spawn_handle: task_manager.spawn_handle(),417 import_queue: import_queue.clone(),418 block_announce_validator_builder: Some(Box::new(|_| {419 Box::new(block_announce_validator)420 })),421 warp_sync: None,422 })?;423424 let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());425 let rpc_client = client.clone();426 let rpc_pool = transaction_pool.clone();427 let select_chain = params.select_chain.clone();428 let rpc_network = network.clone();429430 let rpc_frontier_backend = frontier_backend.clone();431432 let block_data_cache = Arc::new(fc_rpc::EthBlockDataCache::new(433 task_manager.spawn_handle(),434 overrides_handle::<_, _, Runtime>(client.clone()),435 50,436 50,437 ));438439 let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {440 let full_deps = unique_rpc::FullDeps {441 backend: rpc_frontier_backend.clone(),442 deny_unsafe,443 client: rpc_client.clone(),444 pool: rpc_pool.clone(),445 graph: rpc_pool.pool().clone(),446 // TODO: Unhardcode447 enable_dev_signer: false,448 filter_pool: filter_pool.clone(),449 network: rpc_network.clone(),450 select_chain: select_chain.clone(),451 is_authority: validator,452 // TODO: Unhardcode453 max_past_logs: 10000,454 block_data_cache: block_data_cache.clone(),455 fee_history_cache: fee_history_cache.clone(),456 // TODO: Unhardcode457 fee_history_limit: 2048,458 };459460 Ok(461 unique_rpc::create_full::<_, _, _, _, Runtime, RuntimeApi, _>(462 full_deps,463 subscription_executor.clone(),464 ),465 )466 });467468 task_manager.spawn_essential_handle().spawn(469 "frontier-mapping-sync-worker",470 None,471 MappingSyncWorker::new(472 client.import_notification_stream(),473 Duration::new(6, 0),474 client.clone(),475 backend.clone(),476 frontier_backend.clone(),477 SyncStrategy::Normal,478 )479 .for_each(|()| futures::future::ready(())),480 );481482 sc_service::spawn_tasks(sc_service::SpawnTasksParams {483 rpc_extensions_builder,484 client: client.clone(),485 transaction_pool: transaction_pool.clone(),486 task_manager: &mut task_manager,487 config: parachain_config,488 keystore: params.keystore_container.sync_keystore(),489 backend: backend.clone(),490 network: network.clone(),491 system_rpc_tx,492 telemetry: telemetry.as_mut(),493 })?;494495 let announce_block = {496 let network = network.clone();497 Arc::new(move |hash, data| network.announce_block(hash, data))498 };499500 let relay_chain_slot_duration = Duration::from_secs(6);501502 if validator {503 let parachain_consensus = build_consensus(504 client.clone(),505 prometheus_registry.as_ref(),506 telemetry.as_ref().map(|t| t.handle()),507 &task_manager,508 relay_chain_interface.clone(),509 transaction_pool,510 network,511 params.keystore_container.sync_keystore(),512 force_authoring,513 )?;514515 let spawner = task_manager.spawn_handle();516517 let params = StartCollatorParams {518 para_id: id,519 block_status: client.clone(),520 announce_block,521 client: client.clone(),522 task_manager: &mut task_manager,523 spawner,524 parachain_consensus,525 import_queue,526 collator_key: collator_key.expect("Command line arguments do not allow this. qed"),527 relay_chain_interface,528 relay_chain_slot_duration,529 };530531 start_collator(params).await?;532 } else {533 let params = StartFullNodeParams {534 client: client.clone(),535 announce_block,536 task_manager: &mut task_manager,537 para_id: id,538 import_queue,539 relay_chain_interface,540 relay_chain_slot_duration,541 collator_options,542 };543544 start_full_node(params)?;545 }546547 start_network.start_network();548549 Ok((task_manager, client))550}551552/// Build the import queue for the the parachain runtime.553pub fn parachain_build_import_queue<RuntimeApi, ExecutorDispatch>(554 client: Arc<FullClient<RuntimeApi, ExecutorDispatch>>,555 config: &Configuration,556 telemetry: Option<TelemetryHandle>,557 task_manager: &TaskManager,558) -> Result<559 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,560 sc_service::Error,561>562where563 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>564 + Send565 + Sync566 + 'static,567 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>568 + sp_block_builder::BlockBuilder<Block>569 + sp_consensus_aura::AuraApi<Block, AuraId>570 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>,571 ExecutorDispatch: NativeExecutionDispatch + 'static,572{573 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;574575 cumulus_client_consensus_aura::import_queue::<576 sp_consensus_aura::sr25519::AuthorityPair,577 _,578 _,579 _,580 _,581 _,582 _,583 >(cumulus_client_consensus_aura::ImportQueueParams {584 block_import: client.clone(),585 client: client.clone(),586 create_inherent_data_providers: move |_, _| async move {587 let time = sp_timestamp::InherentDataProvider::from_system_time();588589 let slot =590 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(591 *time,592 slot_duration,593 );594595 Ok((time, slot))596 },597 registry: config.prometheus_registry(),598 can_author_with: sp_consensus::CanAuthorWithNativeVersion::new(client.executor().clone()),599 spawner: &task_manager.spawn_essential_handle(),600 telemetry,601 })602 .map_err(Into::into)603}604605/// Start a normal parachain node.606pub async fn start_node<Runtime, RuntimeApi, ExecutorDispatch>(607 parachain_config: Configuration,608 polkadot_config: Configuration,609 collator_options: CollatorOptions,610 id: ParaId,611) -> sc_service::error::Result<(TaskManager, Arc<FullClient<RuntimeApi, ExecutorDispatch>>)>612where613 Runtime: RuntimeInstance + Send + Sync + 'static,614 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,615 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,616 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>617 + Send618 + Sync619 + 'static,620 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>621 + fp_rpc::EthereumRuntimeRPCApi<Block>622 + sp_session::SessionKeys<Block>623 + sp_block_builder::BlockBuilder<Block>624 + pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi<Block, Balance>625 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>626 + up_rpc::UniqueApi<Block, Runtime::CrossAccountId, AccountId>627 + substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Index>628 + sp_api::Metadata<Block>629 + sp_offchain::OffchainWorkerApi<Block>630 + cumulus_primitives_core::CollectCollationInfo<Block>631 + sp_consensus_aura::AuraApi<Block, AuraId>,632 ExecutorDispatch: NativeExecutionDispatch + 'static,633{634 start_node_impl::<Runtime, RuntimeApi, ExecutorDispatch, _, _>(635 parachain_config,636 polkadot_config,637 collator_options,638 id,639 parachain_build_import_queue,640 |client,641 prometheus_registry,642 telemetry,643 task_manager,644 relay_chain_interface,645 transaction_pool,646 sync_oracle,647 keystore,648 force_authoring| {649 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;650651 let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(652 task_manager.spawn_handle(),653 client.clone(),654 transaction_pool,655 prometheus_registry,656 telemetry.clone(),657 );658659 Ok(AuraConsensus::build::<660 sp_consensus_aura::sr25519::AuthorityPair,661 _,662 _,663 _,664 _,665 _,666 _,667 >(BuildAuraConsensusParams {668 proposer_factory,669 create_inherent_data_providers: move |_, (relay_parent, validation_data)| {670 let relay_chain_interface = relay_chain_interface.clone();671 async move {672 let parachain_inherent =673 cumulus_primitives_parachain_inherent::ParachainInherentData::create_at(674 relay_parent,675 &relay_chain_interface,676 &validation_data,677 id,678 ).await;679680 let time = sp_timestamp::InherentDataProvider::from_system_time();681682 let slot =683 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(684 *time,685 slot_duration,686 );687688 let parachain_inherent = parachain_inherent.ok_or_else(|| {689 Box::<dyn std::error::Error + Send + Sync>::from(690 "Failed to create parachain inherent",691 )692 })?;693 Ok((time, slot, parachain_inherent))694 }695 },696 block_import: client.clone(),697 para_client: client,698 backoff_authoring_blocks: Option::<()>::None,699 sync_oracle,700 keystore,701 force_authoring,702 slot_duration,703 // We got around 500ms for proposing704 block_proposal_slot_portion: SlotProportion::new(1f32 / 24f32),705 telemetry,706 max_block_proposal_slot_portion: None,707 }))708 },709 )710 .await711}712713fn dev_build_import_queue<RuntimeApi, ExecutorDispatch>(714 client: Arc<FullClient<RuntimeApi, ExecutorDispatch>>,715 config: &Configuration,716 _: Option<TelemetryHandle>,717 task_manager: &TaskManager,718) -> Result<719 sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,720 sc_service::Error,721>722where723 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>724 + Send725 + Sync726 + 'static,727 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>728 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>,729 ExecutorDispatch: NativeExecutionDispatch + 'static,730{731 Ok(sc_consensus_manual_seal::import_queue(732 Box::new(client.clone()),733 &task_manager.spawn_essential_handle(),734 config.prometheus_registry(),735 ))736}737738/// Builds a new development service. This service uses instant seal, and mocks739/// the parachain inherent740pub fn start_dev_node<Runtime, RuntimeApi, ExecutorDispatch>(741 config: Configuration,742 autoseal_interval: Duration,743) -> sc_service::error::Result<TaskManager>744where745 Runtime: RuntimeInstance + Send + Sync + 'static,746 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,747 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,748 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>749 + Send750 + Sync751 + 'static,752 RuntimeApi::RuntimeApi: sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>753 + fp_rpc::EthereumRuntimeRPCApi<Block>754 + sp_session::SessionKeys<Block>755 + sp_block_builder::BlockBuilder<Block>756 + pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi<Block, Balance>757 + sp_api::ApiExt<Block, StateBackend = sc_client_api::StateBackendFor<FullBackend, Block>>758 + up_rpc::UniqueApi<Block, Runtime::CrossAccountId, AccountId>759 + substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Index>760 + sp_api::Metadata<Block>761 + sp_offchain::OffchainWorkerApi<Block>762 + cumulus_primitives_core::CollectCollationInfo<Block>763 + sp_consensus_aura::AuraApi<Block, AuraId>,764 ExecutorDispatch: NativeExecutionDispatch + 'static,765{766 use sc_consensus_manual_seal::{run_manual_seal, EngineCommand, ManualSealParams};767 use fc_consensus::FrontierBlockImport;768 use sc_client_api::HeaderBackend;769770 let sc_service::PartialComponents {771 client,772 backend,773 mut task_manager,774 import_queue,775 keystore_container,776 select_chain: maybe_select_chain,777 transaction_pool,778 other:779 (telemetry, filter_pool, frontier_backend, _telemetry_worker_handle, fee_history_cache),780 } = new_partial::<RuntimeApi, ExecutorDispatch, _>(781 &config,782 dev_build_import_queue::<RuntimeApi, ExecutorDispatch>,783 )?;784785 let block_data_cache = Arc::new(fc_rpc::EthBlockDataCache::new(786 task_manager.spawn_handle(),787 overrides_handle::<_, _, Runtime>(client.clone()),788 50,789 50,790 ));791792 let (network, system_rpc_tx, network_starter) =793 sc_service::build_network(sc_service::BuildNetworkParams {794 config: &config,795 client: client.clone(),796 transaction_pool: transaction_pool.clone(),797 spawn_handle: task_manager.spawn_handle(),798 import_queue,799 block_announce_validator_builder: None,800 warp_sync: None,801 })?;802803 if config.offchain_worker.enabled {804 sc_service::build_offchain_workers(805 &config,806 task_manager.spawn_handle(),807 client.clone(),808 network.clone(),809 );810 }811812 let prometheus_registry = config.prometheus_registry().cloned();813 let collator = config.role.is_authority();814815 let select_chain = maybe_select_chain.clone();816817 if collator {818 let block_import =819 FrontierBlockImport::new(client.clone(), client.clone(), frontier_backend.clone());820821 let env = sc_basic_authorship::ProposerFactory::new(822 task_manager.spawn_handle(),823 client.clone(),824 transaction_pool.clone(),825 prometheus_registry.as_ref(),826 telemetry.as_ref().map(|x| x.handle()),827 );828829 let transactions_commands_stream: Box<830 dyn Stream<Item = EngineCommand<Hash>> + Send + Sync + Unpin,831 > = Box::new(832 transaction_pool833 .pool()834 .validated_pool()835 .import_notification_stream()836 .map(|_| EngineCommand::SealNewBlock {837 create_empty: true,838 finalize: false,839 parent_hash: None,840 sender: None,841 }),842 );843844 let autoseal_interval = Box::pin(AutosealInterval::new(&config, autoseal_interval));845 let idle_commands_stream: Box<846 dyn Stream<Item = EngineCommand<Hash>> + Send + Sync + Unpin,847 > = Box::new(autoseal_interval.map(|_| EngineCommand::SealNewBlock {848 create_empty: true,849 finalize: false,850 parent_hash: None,851 sender: None,852 }));853854 let commands_stream = select(transactions_commands_stream, idle_commands_stream);855856 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;857 let client_set_aside_for_cidp = client.clone();858859 task_manager.spawn_essential_handle().spawn_blocking(860 "authorship_task",861 Some("block-authoring"),862 run_manual_seal(ManualSealParams {863 block_import,864 env,865 client: client.clone(),866 pool: transaction_pool.clone(),867 commands_stream,868 select_chain: select_chain.clone(),869 consensus_data_provider: None,870 create_inherent_data_providers: move |block: Hash, ()| {871 let current_para_block = client_set_aside_for_cidp872 .number(block)873 .expect("Header lookup should succeed")874 .expect("Header passed in as parent should be present in backend.");875876 let client_for_xcm = client_set_aside_for_cidp.clone();877 async move {878 let time = sp_timestamp::InherentDataProvider::from_system_time();879880 let mocked_parachain = cumulus_primitives_parachain_inherent::MockValidationDataInherentDataProvider {881 current_para_block,882 relay_offset: 1000,883 relay_blocks_per_para_block: 2,884 xcm_config: cumulus_primitives_parachain_inherent::MockXcmConfig::new(885 &*client_for_xcm,886 block,887 Default::default(),888 Default::default(),889 ),890 raw_downward_messages: vec![],891 raw_horizontal_messages: vec![],892 };893894 let slot =895 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(896 *time,897 slot_duration,898 );899900 Ok((time, slot, mocked_parachain))901 }902 },903 }),904 );905 }906907 task_manager.spawn_essential_handle().spawn(908 "frontier-mapping-sync-worker",909 Some("block-authoring"),910 MappingSyncWorker::new(911 client.import_notification_stream(),912 Duration::new(6, 0),913 client.clone(),914 backend.clone(),915 frontier_backend.clone(),916 SyncStrategy::Normal,917 )918 .for_each(|()| futures::future::ready(())),919 );920921 let subscription_executor = sc_rpc::SubscriptionTaskExecutor::new(task_manager.spawn_handle());922 let rpc_client = client.clone();923 let rpc_pool = transaction_pool.clone();924 let rpc_network = network.clone();925 let rpc_frontier_backend = frontier_backend.clone();926 let rpc_extensions_builder = Box::new(move |deny_unsafe, _| {927 let full_deps = unique_rpc::FullDeps {928 backend: rpc_frontier_backend.clone(),929 deny_unsafe,930 client: rpc_client.clone(),931 pool: rpc_pool.clone(),932 graph: rpc_pool.pool().clone(),933 // TODO: Unhardcode934 enable_dev_signer: false,935 filter_pool: filter_pool.clone(),936 network: rpc_network.clone(),937 select_chain: select_chain.clone(),938 is_authority: collator,939 // TODO: Unhardcode940 max_past_logs: 10000,941 block_data_cache: block_data_cache.clone(),942 fee_history_cache: fee_history_cache.clone(),943 // TODO: Unhardcode944 fee_history_limit: 2048,945 };946947 Ok(948 unique_rpc::create_full::<_, _, _, _, Runtime, RuntimeApi, _>(949 full_deps,950 subscription_executor.clone(),951 ),952 )953 });954955 sc_service::spawn_tasks(sc_service::SpawnTasksParams {956 network,957 client,958 keystore: keystore_container.sync_keystore(),959 task_manager: &mut task_manager,960 transaction_pool,961 rpc_extensions_builder,962 backend,963 system_rpc_tx,964 config,965 telemetry: None,966 })?;967968 network_starter.start_network();969 Ok(task_manager)970}