123456789101112131415161718use std::{19 collections::BTreeMap,20 marker::PhantomData,21 pin::Pin,22 sync::{Arc, Mutex},23 time::Duration,24};2526use cumulus_client_cli::CollatorOptions;27use cumulus_client_collator::service::CollatorService;28#[cfg(not(feature = "lookahead"))]29use cumulus_client_consensus_aura::collators::basic::{30 run as run_aura, Params as BuildAuraConsensusParams,31};32#[cfg(feature = "lookahead")]33use cumulus_client_consensus_aura::collators::lookahead::{34 run as run_aura, Params as BuildAuraConsensusParams,35};36use cumulus_client_consensus_common::ParachainBlockImport as TParachainBlockImport;37use cumulus_client_consensus_proposer::Proposer;38use cumulus_client_network::RequireSecondedInBlockAnnounce;39use cumulus_client_service::{40 build_relay_chain_interface, prepare_node_config, start_relay_chain_tasks, DARecoveryProfile,41 StartRelayChainTasksParams,42};43use cumulus_primitives_core::ParaId;44use cumulus_relay_chain_interface::{OverseerHandle, RelayChainInterface};45use fc_mapping_sync::{kv::MappingSyncWorker, EthereumBlockNotificationSinks, SyncStrategy};46use fc_rpc::{47 frontier_backend_client::SystemAccountId32StorageOverride, EthBlockDataCacheTask, EthConfig,48 EthTask, OverrideHandle, RuntimeApiStorageOverride, SchemaV1Override, SchemaV2Override,49 SchemaV3Override, StorageOverride,50};51use fc_rpc_core::types::{FeeHistoryCache, FilterPool};52use fp_rpc::EthereumRuntimeRPCApi;53use fp_storage::EthereumStorageSchema;54use futures::{55 stream::select,56 task::{Context, Poll},57 Stream, StreamExt,58};59use jsonrpsee::RpcModule;60use polkadot_service::CollatorPair;61use sc_client_api::{AuxStore, Backend, BlockOf, BlockchainEvents, StorageProvider};62use sc_consensus::ImportQueue;63use sc_executor::{NativeElseWasmExecutor, NativeExecutionDispatch};64use sc_network::NetworkBlock;65use sc_network_sync::SyncingService;66use sc_rpc::SubscriptionTaskExecutor;67use sc_service::{Configuration, PartialComponents, TaskManager};68use sc_telemetry::{Telemetry, TelemetryHandle, TelemetryWorker, TelemetryWorkerHandle};69use serde::{Deserialize, Serialize};70use sp_api::{ProvideRuntimeApi, StateBackend};71use sp_block_builder::BlockBuilder;72use sp_blockchain::{Error as BlockChainError, HeaderBackend, HeaderMetadata};73use sp_consensus_aura::sr25519::AuthorityPair as AuraAuthorityPair;74use sp_keystore::KeystorePtr;75use sp_runtime::traits::BlakeTwo256;76use substrate_prometheus_endpoint::Registry;77use tokio::time::Interval;78use up_common::types::{opaque::*, Nonce};7980use crate::{81 chain_spec::RuntimeIdentification,82 rpc::{create_eth, create_full, EthDeps, FullDeps},83};848586#[cfg(feature = "unique-runtime")]87pub struct UniqueRuntimeExecutor;8889#[cfg(feature = "quartz-runtime")]9091pub struct QuartzRuntimeExecutor;929394pub struct OpalRuntimeExecutor;9596#[cfg(feature = "unique-runtime")]97impl NativeExecutionDispatch for UniqueRuntimeExecutor {98 99 #[cfg(feature = "runtime-benchmarks")]100 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;101 102 #[cfg(not(feature = "runtime-benchmarks"))]103 type ExtendHostFunctions = ();104105 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {106 unique_runtime::api::dispatch(method, data)107 }108109 fn native_version() -> sc_executor::NativeVersion {110 unique_runtime::native_version()111 }112}113114#[cfg(feature = "quartz-runtime")]115impl NativeExecutionDispatch for QuartzRuntimeExecutor {116 117 #[cfg(feature = "runtime-benchmarks")]118 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;119 120 #[cfg(not(feature = "runtime-benchmarks"))]121 type ExtendHostFunctions = ();122123 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {124 quartz_runtime::api::dispatch(method, data)125 }126127 fn native_version() -> sc_executor::NativeVersion {128 quartz_runtime::native_version()129 }130}131132impl NativeExecutionDispatch for OpalRuntimeExecutor {133 134 #[cfg(feature = "runtime-benchmarks")]135 type ExtendHostFunctions = frame_benchmarking::benchmarking::HostFunctions;136 137 #[cfg(not(feature = "runtime-benchmarks"))]138 type ExtendHostFunctions = ();139140 fn dispatch(method: &str, data: &[u8]) -> Option<Vec<u8>> {141 opal_runtime::api::dispatch(method, data)142 }143144 fn native_version() -> sc_executor::NativeVersion {145 opal_runtime::native_version()146 }147}148149pub struct AutosealInterval {150 interval: Interval,151}152153impl AutosealInterval {154 pub fn new(config: &Configuration, interval: u64) -> Self {155 let _tokio_runtime = config.tokio_handle.enter();156 let interval = tokio::time::interval(Duration::from_millis(interval));157158 Self { interval }159 }160}161162impl Stream for AutosealInterval {163 type Item = tokio::time::Instant;164165 fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {166 self.interval.poll_tick(cx).map(Some)167 }168}169170pub fn open_frontier_backend<C: HeaderBackend<Block>>(171 client: Arc<C>,172 config: &Configuration,173) -> Result<Arc<fc_db::kv::Backend<Block>>, String> {174 let config_dir = config.base_path.config_dir(config.chain_spec.id());175 let database_dir = config_dir.join("frontier").join("db");176177 Ok(Arc::new(fc_db::kv::Backend::<Block>::new(178 client,179 &fc_db::kv::DatabaseSettings {180 source: fc_db::DatabaseSource::RocksDb {181 path: database_dir,182 cache_size: 0,183 },184 },185 )?))186}187188type FullClient<RuntimeApi, ExecutorDispatch> =189 sc_service::TFullClient<Block, RuntimeApi, NativeElseWasmExecutor<ExecutorDispatch>>;190type FullBackend = sc_service::TFullBackend<Block>;191type FullSelectChain = sc_consensus::LongestChain<FullBackend, Block>;192type ParachainBlockImport<RuntimeApi, ExecutorDispatch> =193 TParachainBlockImport<Block, Arc<FullClient<RuntimeApi, ExecutorDispatch>>, FullBackend>;194195196macro_rules! ez_bounds {197 ($vis:vis trait $name:ident$(<$($gen:ident $(: $($(+)? $bound:path)*)?),* $(,)?>)? $(:)? $($(+)? $super:path)* {}) => {198 $vis trait $name $(<$($gen $(: $($bound+)*)?,)*>)?: $($super +)* {}199 impl<T, $($($gen $(: $($bound+)*)?,)*)?> $name$(<$($gen,)*>)? for T200 where T: $($super +)* {}201 }202}203ez_bounds!(204 pub trait RuntimeApiDep<Runtime: RuntimeInstance>:205 sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block>206 + sp_consensus_aura::AuraApi<Block, AuraId>207 + fp_rpc::EthereumRuntimeRPCApi<Block>208 + sp_session::SessionKeys<Block>209 + sp_block_builder::BlockBuilder<Block>210 + pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi<Block, Balance>211 + sp_api::ApiExt<Block>212 + up_rpc::UniqueApi<Block, Runtime::CrossAccountId, AccountId>213 + app_promotion_rpc::AppPromotionApi<Block, BlockNumber, Runtime::CrossAccountId, AccountId>214 + up_pov_estimate_rpc::PovEstimateApi<Block>215 + substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Nonce>216 + sp_api::Metadata<Block>217 + sp_offchain::OffchainWorkerApi<Block>218 + cumulus_primitives_core::CollectCollationInfo<Block>219 220 + fp_rpc::ConvertTransactionRuntimeApi<Block>221 {222 }223);224225226227228229#[allow(clippy::type_complexity)]230pub fn new_partial<Runtime, RuntimeApi, ExecutorDispatch, BIQ>(231 config: &Configuration,232 build_import_queue: BIQ,233) -> Result<234 PartialComponents<235 FullClient<RuntimeApi, ExecutorDispatch>,236 FullBackend,237 FullSelectChain,238 sc_consensus::DefaultImportQueue<Block>,239 sc_transaction_pool::FullPool<Block, FullClient<RuntimeApi, ExecutorDispatch>>,240 OtherPartial,241 >,242 sc_service::Error,243>244where245 sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,246 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>247 + Send248 + Sync249 + 'static,250 RuntimeApi::RuntimeApi: RuntimeApiDep<Runtime> + 'static,251 Runtime: RuntimeInstance,252 ExecutorDispatch: NativeExecutionDispatch + 'static,253 BIQ: FnOnce(254 Arc<FullClient<RuntimeApi, ExecutorDispatch>>,255 Arc<FullBackend>,256 &Configuration,257 Option<TelemetryHandle>,258 &TaskManager,259 ) -> Result<sc_consensus::DefaultImportQueue<Block>, sc_service::Error>,260{261 let telemetry = config262 .telemetry_endpoints263 .clone()264 .filter(|x| !x.is_empty())265 .map(|endpoints| -> Result<_, sc_telemetry::Error> {266 let worker = TelemetryWorker::new(16)?;267 let telemetry = worker.handle().new_telemetry(endpoints);268 Ok((worker, telemetry))269 })270 .transpose()?;271272 let executor = sc_service::new_native_or_wasm_executor(config);273274 let (client, backend, keystore_container, task_manager) =275 sc_service::new_full_parts::<Block, RuntimeApi, _>(276 config,277 telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),278 executor,279 )?;280 let client = Arc::new(client);281282 let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());283284 let telemetry = telemetry.map(|(worker, telemetry)| {285 task_manager286 .spawn_handle()287 .spawn("telemetry", None, worker.run());288 telemetry289 });290291 let select_chain = sc_consensus::LongestChain::new(backend.clone());292293 let transaction_pool = sc_transaction_pool::BasicPool::new_full(294 config.transaction_pool.clone(),295 config.role.is_authority().into(),296 config.prometheus_registry(),297 task_manager.spawn_essential_handle(),298 client.clone(),299 );300301 let eth_filter_pool: Option<FilterPool> = Some(Arc::new(Mutex::new(BTreeMap::new())));302303 let eth_backend = open_frontier_backend(client.clone(), config)?;304305 let import_queue = build_import_queue(306 client.clone(),307 backend.clone(),308 config,309 telemetry.as_ref().map(|telemetry| telemetry.handle()),310 &task_manager,311 )?;312313 let params = PartialComponents {314 backend,315 client,316 import_queue,317 keystore_container,318 task_manager,319 transaction_pool,320 select_chain,321 other: OtherPartial {322 telemetry,323 eth_filter_pool,324 eth_backend,325 telemetry_worker_handle,326 },327 };328329 Ok(params)330}331332macro_rules! clone {333 ($($i:ident),* $(,)?) => {334 $(335 let $i = $i.clone();336 )*337 };338}339340341342343#[sc_tracing::logging::prefix_logs_with("Parachain")]344pub async fn start_node<Runtime, RuntimeApi, ExecutorDispatch>(345 parachain_config: Configuration,346 polkadot_config: Configuration,347 collator_options: CollatorOptions,348 para_id: ParaId,349 hwbench: Option<sc_sysinfo::HwBench>,350) -> sc_service::error::Result<(TaskManager, Arc<FullClient<RuntimeApi, ExecutorDispatch>>)>351where352 sc_client_api::StateBackendFor<FullBackend, Block>: sp_api::StateBackend<BlakeTwo256>,353 Runtime: RuntimeInstance + Send + Sync + 'static,354 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,355 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,356 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>357 + Send358 + Sync359 + 'static,360 RuntimeApi::RuntimeApi: RuntimeApiDep<Runtime> + 'static,361 Runtime: RuntimeInstance,362 ExecutorDispatch: NativeExecutionDispatch + 'static,363{364 let parachain_config = prepare_node_config(parachain_config);365366 let params = new_partial::<Runtime, RuntimeApi, ExecutorDispatch, _>(367 ¶chain_config,368 parachain_build_import_queue,369 )?;370 let OtherPartial {371 mut telemetry,372 telemetry_worker_handle,373 eth_filter_pool,374 eth_backend,375 } = params.other;376 let net_config = sc_network::config::FullNetworkConfiguration::new(¶chain_config.network);377378 let client = params.client.clone();379 let backend = params.backend.clone();380 let mut task_manager = params.task_manager;381382 let (relay_chain_interface, collator_key) = build_relay_chain_interface(383 polkadot_config,384 ¶chain_config,385 telemetry_worker_handle,386 &mut task_manager,387 collator_options.clone(),388 hwbench.clone(),389 )390 .await391 .map_err(|e| sc_service::Error::Application(Box::new(e) as Box<_>))?;392393 let block_announce_validator =394 RequireSecondedInBlockAnnounce::new(relay_chain_interface.clone(), para_id);395396 let validator = parachain_config.role.is_authority();397 let prometheus_registry = parachain_config.prometheus_registry().cloned();398 let transaction_pool = params.transaction_pool.clone();399 let import_queue_service = params.import_queue.service();400401 let (network, system_rpc_tx, tx_handler_controller, start_network, sync_service) =402 sc_service::build_network(sc_service::BuildNetworkParams {403 config: ¶chain_config,404 net_config,405 client: client.clone(),406 transaction_pool: transaction_pool.clone(),407 spawn_handle: task_manager.spawn_handle(),408 import_queue: params.import_queue,409 block_announce_validator_builder: Some(Box::new(|_| {410 Box::new(block_announce_validator)411 })),412 warp_sync_params: None,413 })?;414415 let select_chain = params.select_chain.clone();416417 let runtime_id = parachain_config.chain_spec.runtime_id();418419 420 let fee_history_cache: FeeHistoryCache = Arc::new(Mutex::new(BTreeMap::new()));421 let fee_history_limit = 2048;422423 let eth_pubsub_notification_sinks: Arc<424 EthereumBlockNotificationSinks<fc_mapping_sync::EthereumBlockNotification<Block>>,425 > = Default::default();426427 let overrides = overrides_handle(client.clone());428 let eth_block_data_cache = spawn_frontier_tasks(429 FrontierTaskParams {430 client: client.clone(),431 substrate_backend: backend.clone(),432 eth_filter_pool: eth_filter_pool.clone(),433 eth_backend: eth_backend.clone(),434 fee_history_limit,435 fee_history_cache: fee_history_cache.clone(),436 task_manager: &task_manager,437 prometheus_registry: prometheus_registry.clone(),438 overrides: overrides.clone(),439 sync_strategy: SyncStrategy::Parachain,440 },441 sync_service.clone(),442 eth_pubsub_notification_sinks.clone(),443 );444445 446 let rpc_builder = Box::new({447 clone!(448 client,449 backend,450 eth_backend,451 eth_pubsub_notification_sinks,452 fee_history_cache,453 eth_block_data_cache,454 overrides,455 transaction_pool,456 network,457 sync_service,458 );459 move |deny_unsafe, subscription_task_executor: SubscriptionTaskExecutor| {460 clone!(461 backend,462 eth_block_data_cache,463 client,464 eth_backend,465 eth_filter_pool,466 eth_pubsub_notification_sinks,467 fee_history_cache,468 eth_block_data_cache,469 network,470 runtime_id,471 transaction_pool,472 select_chain,473 overrides,474 );475476 #[cfg(not(feature = "pov-estimate"))]477 let _ = backend;478479 let mut rpc_handle = RpcModule::new(());480481 let full_deps = FullDeps {482 client: client.clone(),483 runtime_id,484485 #[cfg(feature = "pov-estimate")]486 exec_params: uc_rpc::pov_estimate::ExecutorParams {487 wasm_method: parachain_config.wasm_method,488 default_heap_pages: parachain_config.default_heap_pages,489 max_runtime_instances: parachain_config.max_runtime_instances,490 runtime_cache_size: parachain_config.runtime_cache_size,491 },492493 #[cfg(feature = "pov-estimate")]494 backend,495496 deny_unsafe,497 pool: transaction_pool.clone(),498 select_chain,499 };500501 create_full::<_, _, _, Runtime, RuntimeApi, _>(&mut rpc_handle, full_deps)?;502503 let eth_deps = EthDeps {504 client,505 graph: transaction_pool.pool().clone(),506 pool: transaction_pool,507 is_authority: validator,508 network,509 eth_backend,510 511 max_past_logs: 10000,512 fee_history_limit,513 fee_history_cache,514 eth_block_data_cache,515 516 enable_dev_signer: false,517 eth_filter_pool,518 eth_pubsub_notification_sinks,519 overrides,520 sync: sync_service.clone(),521 pending_create_inherent_data_providers: |_, ()| async move { Ok(()) },522 };523524 create_eth::<525 _,526 _,527 _,528 _,529 _,530 _,531 DefaultEthConfig<FullClient<RuntimeApi, ExecutorDispatch>>,532 >(533 &mut rpc_handle,534 eth_deps,535 subscription_task_executor.clone(),536 )?;537538 Ok(rpc_handle)539 }540 });541542 sc_service::spawn_tasks(sc_service::SpawnTasksParams {543 rpc_builder,544 client: client.clone(),545 transaction_pool: transaction_pool.clone(),546 task_manager: &mut task_manager,547 config: parachain_config,548 keystore: params.keystore_container.keystore(),549 backend: backend.clone(),550 network: network.clone(),551 sync_service: sync_service.clone(),552 system_rpc_tx,553 telemetry: telemetry.as_mut(),554 tx_handler_controller,555 })?;556557 if let Some(hwbench) = hwbench {558 sc_sysinfo::print_hwbench(&hwbench);559560 if let Some(ref mut telemetry) = telemetry {561 let telemetry_handle = telemetry.handle();562 task_manager.spawn_handle().spawn(563 "telemetry_hwbench",564 None,565 sc_sysinfo::initialize_hwbench_telemetry(telemetry_handle, hwbench),566 );567 }568 }569570 let announce_block = {571 let sync_service = sync_service.clone();572 Arc::new(Box::new(move |hash, data| {573 sync_service.announce_block(hash, data)574 }))575 };576577 let relay_chain_slot_duration = Duration::from_secs(6);578579 let overseer_handle = relay_chain_interface580 .overseer_handle()581 .map_err(|e| sc_service::Error::Application(Box::new(e)))?;582583 start_relay_chain_tasks(StartRelayChainTasksParams {584 client: client.clone(),585 announce_block: announce_block.clone(),586 para_id,587 relay_chain_interface: relay_chain_interface.clone(),588 task_manager: &mut task_manager,589 da_recovery_profile: if validator {590 DARecoveryProfile::Collator591 } else {592 DARecoveryProfile::FullNode593 },594 import_queue: import_queue_service,595 relay_chain_slot_duration,596 recovery_handle: Box::new(overseer_handle.clone()),597 sync_service: sync_service.clone(),598 })?;599600 if validator {601 start_consensus(602 client.clone(),603 backend.clone(),604 prometheus_registry.as_ref(),605 telemetry.as_ref().map(|t| t.handle()),606 &task_manager,607 relay_chain_interface.clone(),608 transaction_pool,609 sync_service.clone(),610 params.keystore_container.keystore(),611 overseer_handle,612 relay_chain_slot_duration,613 para_id,614 collator_key.expect("cli args do not allow this"),615 announce_block,616 )?;617 }618619 start_network.start_network();620621 Ok((task_manager, client))622}623624625pub fn parachain_build_import_queue<Runtime, RuntimeApi, ExecutorDispatch>(626 client: Arc<FullClient<RuntimeApi, ExecutorDispatch>>,627 backend: Arc<FullBackend>,628 config: &Configuration,629 telemetry: Option<TelemetryHandle>,630 task_manager: &TaskManager,631) -> Result<sc_consensus::DefaultImportQueue<Block>, sc_service::Error>632where633 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>634 + Send635 + Sync636 + 'static,637 RuntimeApi::RuntimeApi: RuntimeApiDep<Runtime> + 'static,638 Runtime: RuntimeInstance,639 ExecutorDispatch: NativeExecutionDispatch + 'static,640{641 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;642643 let block_import = ParachainBlockImport::new(client.clone(), backend);644645 cumulus_client_consensus_aura::import_queue::<646 sp_consensus_aura::sr25519::AuthorityPair,647 _,648 _,649 _,650 _,651 _,652 >(cumulus_client_consensus_aura::ImportQueueParams {653 block_import,654 client,655 create_inherent_data_providers: move |_, _| async move {656 let time = sp_timestamp::InherentDataProvider::from_system_time();657658 let slot =659 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(660 *time,661 slot_duration,662 );663664 Ok((slot, time))665 },666 registry: config.prometheus_registry(),667 spawner: &task_manager.spawn_essential_handle(),668 telemetry,669 })670 .map_err(Into::into)671}672673pub fn start_consensus<ExecutorDispatch, RuntimeApi, Runtime>(674 client: Arc<FullClient<RuntimeApi, ExecutorDispatch>>,675 backend: Arc<FullBackend>,676 prometheus_registry: Option<&Registry>,677 telemetry: Option<TelemetryHandle>,678 task_manager: &TaskManager,679 relay_chain_interface: Arc<dyn RelayChainInterface>,680 transaction_pool: Arc<681 sc_transaction_pool::FullPool<Block, FullClient<RuntimeApi, ExecutorDispatch>>,682 >,683 sync_oracle: Arc<SyncingService<Block>>,684 keystore: KeystorePtr,685 overseer_handle: OverseerHandle,686 relay_chain_slot_duration: Duration,687 para_id: ParaId,688 collator_key: CollatorPair,689 announce_block: Arc<dyn Fn(Hash, Option<Vec<u8>>) + Send + Sync>,690) -> Result<(), sc_service::Error>691where692 ExecutorDispatch: NativeExecutionDispatch + 'static,693 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>694 + Send695 + Sync696 + 'static,697 RuntimeApi::RuntimeApi: RuntimeApiDep<Runtime> + 'static,698 Runtime: RuntimeInstance,699{700 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;701702 let proposer_factory = sc_basic_authorship::ProposerFactory::with_proof_recording(703 task_manager.spawn_handle(),704 client.clone(),705 transaction_pool,706 prometheus_registry,707 telemetry.clone(),708 );709 let proposer = Proposer::new(proposer_factory);710711 let collator_service = CollatorService::new(712 client.clone(),713 Arc::new(task_manager.spawn_handle()),714 announce_block,715 client.clone(),716 );717718 let block_import = ParachainBlockImport::new(client.clone(), backend);719720 let params = BuildAuraConsensusParams {721 create_inherent_data_providers: move |_, ()| async move { Ok(()) },722 block_import,723 para_client: client,724 #[cfg(feature = "lookahead")]725 para_backend: backend,726 para_id,727 relay_client: relay_chain_interface,728 sync_oracle,729 keystore,730 slot_duration,731 proposer,732 collator_service,733 734 authoring_duration: Duration::from_millis(500),735 overseer_handle,736 #[cfg(feature = "lookahead")]737 code_hash_provider: || {},738 collator_key,739 relay_chain_slot_duration,740 };741742 task_manager.spawn_essential_handle().spawn(743 "aura",744 None,745 run_aura::<_, AuraAuthorityPair, _, _, _, _, _, _, _>(params),746 );747 Ok(())748}749750fn dev_build_import_queue<RuntimeApi, ExecutorDispatch>(751 client: Arc<FullClient<RuntimeApi, ExecutorDispatch>>,752 _: Arc<FullBackend>,753 config: &Configuration,754 _: Option<TelemetryHandle>,755 task_manager: &TaskManager,756) -> Result<sc_consensus::DefaultImportQueue<Block>, sc_service::Error>757where758 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>759 + Send760 + Sync761 + 'static,762 RuntimeApi::RuntimeApi:763 sp_transaction_pool::runtime_api::TaggedTransactionQueue<Block> + sp_api::ApiExt<Block>,764 ExecutorDispatch: NativeExecutionDispatch + 'static,765{766 Ok(sc_consensus_manual_seal::import_queue(767 Box::new(client),768 &task_manager.spawn_essential_handle(),769 config.prometheus_registry(),770 ))771}772773pub struct OtherPartial {774 pub telemetry: Option<Telemetry>,775 pub telemetry_worker_handle: Option<TelemetryWorkerHandle>,776 pub eth_filter_pool: Option<FilterPool>,777 pub eth_backend: Arc<fc_db::kv::Backend<Block>>,778}779780struct DefaultEthConfig<C>(PhantomData<C>);781impl<C> EthConfig<Block, C> for DefaultEthConfig<C>782where783 C: StorageProvider<Block, FullBackend> + Sync + Send + 'static,784{785 type EstimateGasAdapter = ();786 type RuntimeStorageOverride = SystemAccountId32StorageOverride<Block, C, FullBackend>;787}788789790791pub fn start_dev_node<Runtime, RuntimeApi, ExecutorDispatch>(792 config: Configuration,793 autoseal_interval: u64,794 autoseal_finalize_delay: Option<u64>,795 disable_autoseal_on_tx: bool,796) -> sc_service::error::Result<TaskManager>797where798 Runtime: RuntimeInstance + Send + Sync + 'static,799 <Runtime as RuntimeInstance>::CrossAccountId: Serialize,800 for<'de> <Runtime as RuntimeInstance>::CrossAccountId: Deserialize<'de>,801 RuntimeApi: sp_api::ConstructRuntimeApi<Block, FullClient<RuntimeApi, ExecutorDispatch>>802 + Send803 + Sync804 + 'static,805 RuntimeApi::RuntimeApi: RuntimeApiDep<Runtime> + 'static,806 ExecutorDispatch: NativeExecutionDispatch + 'static,807{808 use fc_consensus::FrontierBlockImport;809 use sc_consensus_manual_seal::{810 run_delayed_finalize, run_manual_seal, DelayedFinalizeParams, EngineCommand,811 ManualSealParams,812 };813814 let sc_service::PartialComponents {815 client,816 backend,817 mut task_manager,818 import_queue,819 keystore_container,820 select_chain: maybe_select_chain,821 transaction_pool,822 other:823 OtherPartial {824 telemetry,825 eth_filter_pool,826 eth_backend,827 telemetry_worker_handle: _,828 },829 } = new_partial::<Runtime, RuntimeApi, ExecutorDispatch, _>(830 &config,831 dev_build_import_queue::<RuntimeApi, ExecutorDispatch>,832 )?;833 let net_config = sc_network::config::FullNetworkConfiguration::new(&config.network);834 let prometheus_registry = config.prometheus_registry().cloned();835836 let (network, system_rpc_tx, tx_handler_controller, network_starter, sync_service) =837 sc_service::build_network(sc_service::BuildNetworkParams {838 config: &config,839 net_config,840 client: client.clone(),841 transaction_pool: transaction_pool.clone(),842 spawn_handle: task_manager.spawn_handle(),843 import_queue,844 block_announce_validator_builder: None,845 warp_sync_params: None,846 })?;847848 let collator = config.role.is_authority();849850 let select_chain = maybe_select_chain;851852 if collator {853 let block_import = FrontierBlockImport::new(client.clone(), client.clone());854855 let env = sc_basic_authorship::ProposerFactory::new(856 task_manager.spawn_handle(),857 client.clone(),858 transaction_pool.clone(),859 prometheus_registry.as_ref(),860 telemetry.as_ref().map(|x| x.handle()),861 );862863 let transactions_commands_stream: Box<864 dyn Stream<Item = EngineCommand<Hash>> + Send + Sync + Unpin,865 > = Box::new(866 transaction_pool867 .pool()868 .validated_pool()869 .import_notification_stream()870 .filter(move |_| futures::future::ready(!disable_autoseal_on_tx))871 .map(|_| EngineCommand::SealNewBlock {872 create_empty: true,873 finalize: false,874 parent_hash: None,875 sender: None,876 }),877 );878879 let autoseal_interval = Box::pin(AutosealInterval::new(&config, autoseal_interval));880881 let idle_commands_stream: Box<882 dyn Stream<Item = EngineCommand<Hash>> + Send + Sync + Unpin,883 > = Box::new(autoseal_interval.map(|_| EngineCommand::SealNewBlock {884 create_empty: true,885 finalize: false,886 parent_hash: None,887 sender: None,888 }));889890 let commands_stream = select(transactions_commands_stream, idle_commands_stream);891892 let slot_duration = cumulus_client_consensus_aura::slot_duration(&*client)?;893 let client_set_aside_for_cidp = client.clone();894895 if let Some(delay_sec) = autoseal_finalize_delay {896 let spawn_handle = task_manager.spawn_handle();897898 task_manager.spawn_essential_handle().spawn_blocking(899 "finalization_task",900 Some("block-authoring"),901 run_delayed_finalize(DelayedFinalizeParams {902 client: client.clone(),903 delay_sec,904 spawn_handle,905 }),906 );907 }908909 task_manager.spawn_essential_handle().spawn_blocking(910 "authorship_task",911 Some("block-authoring"),912 run_manual_seal(ManualSealParams {913 block_import,914 env,915 client: client.clone(),916 pool: transaction_pool.clone(),917 commands_stream,918 select_chain: select_chain.clone(),919 consensus_data_provider: None,920 create_inherent_data_providers: move |block: Hash, ()| {921 let current_para_block = client_set_aside_for_cidp922 .number(block)923 .expect("Header lookup should succeed")924 .expect("Header passed in as parent should be present in backend.");925926 let client_for_xcm = client_set_aside_for_cidp.clone();927 async move {928 let time = sp_timestamp::InherentDataProvider::from_system_time();929930 let mocked_parachain = cumulus_primitives_parachain_inherent::MockValidationDataInherentDataProvider {931 current_para_block,932 relay_offset: 1000,933 relay_blocks_per_para_block: 2,934 para_blocks_per_relay_epoch: 0,935 xcm_config: cumulus_primitives_parachain_inherent::MockXcmConfig::new(936 &*client_for_xcm,937 block,938 Default::default(),939 Default::default(),940 ),941 relay_randomness_config: (),942 raw_downward_messages: vec![],943 raw_horizontal_messages: vec![],944 };945946 let slot =947 sp_consensus_aura::inherents::InherentDataProvider::from_timestamp_and_slot_duration(948 *time,949 slot_duration,950 );951952 Ok((time, slot, mocked_parachain))953 }954 },955 }),956 );957 }958959 #[cfg(feature = "pov-estimate")]960 let rpc_backend = backend.clone();961962 let runtime_id = config.chain_spec.runtime_id();963964 965 let fee_history_cache: FeeHistoryCache = Arc::new(Mutex::new(BTreeMap::new()));966 let fee_history_limit = 2048;967968 let eth_pubsub_notification_sinks: Arc<969 EthereumBlockNotificationSinks<fc_mapping_sync::EthereumBlockNotification<Block>>,970 > = Default::default();971972 let overrides = overrides_handle(client.clone());973 let eth_block_data_cache = spawn_frontier_tasks(974 FrontierTaskParams {975 client: client.clone(),976 substrate_backend: backend.clone(),977 eth_filter_pool: eth_filter_pool.clone(),978 eth_backend: eth_backend.clone(),979 fee_history_limit,980 fee_history_cache: fee_history_cache.clone(),981 task_manager: &task_manager,982 prometheus_registry,983 overrides: overrides.clone(),984 sync_strategy: SyncStrategy::Normal,985 },986 sync_service.clone(),987 eth_pubsub_notification_sinks.clone(),988 );989990 991 let rpc_builder = Box::new({992 clone!(993 client,994 backend,995 eth_backend,996 eth_pubsub_notification_sinks,997 fee_history_cache,998 eth_block_data_cache,999 overrides,1000 transaction_pool,1001 network,1002 sync_service,1003 );1004 move |deny_unsafe, subscription_task_executor: SubscriptionTaskExecutor| {1005 clone!(1006 backend,1007 eth_block_data_cache,1008 client,1009 eth_backend,1010 eth_filter_pool,1011 eth_pubsub_notification_sinks,1012 fee_history_cache,1013 eth_block_data_cache,1014 network,1015 runtime_id,1016 transaction_pool,1017 select_chain,1018 overrides,1019 );10201021 #[cfg(not(feature = "pov-estimate"))]1022 let _ = backend;10231024 let mut rpc_module = RpcModule::new(());10251026 let full_deps = FullDeps {1027 runtime_id,10281029 #[cfg(feature = "pov-estimate")]1030 exec_params: uc_rpc::pov_estimate::ExecutorParams {1031 wasm_method: config.wasm_method,1032 default_heap_pages: config.default_heap_pages,1033 max_runtime_instances: config.max_runtime_instances,1034 runtime_cache_size: config.runtime_cache_size,1035 },10361037 #[cfg(feature = "pov-estimate")]1038 backend,1039 1040 deny_unsafe,1041 client: client.clone(),1042 pool: transaction_pool.clone(),1043 select_chain,1044 };10451046 create_full::<_, _, _, Runtime, RuntimeApi, _>(&mut rpc_module, full_deps)?;10471048 let eth_deps = EthDeps {1049 client,1050 graph: transaction_pool.pool().clone(),1051 pool: transaction_pool,1052 is_authority: true,1053 network,1054 eth_backend,1055 1056 max_past_logs: 10000,1057 fee_history_limit,1058 fee_history_cache,1059 eth_block_data_cache,1060 1061 enable_dev_signer: false,1062 eth_filter_pool,1063 eth_pubsub_notification_sinks,1064 overrides,1065 sync: sync_service.clone(),1066 1067 pending_create_inherent_data_providers: |_, ()| async move { Ok(()) },1068 };10691070 create_eth::<1071 _,1072 _,1073 _,1074 _,1075 _,1076 _,1077 DefaultEthConfig<FullClient<RuntimeApi, ExecutorDispatch>>,1078 >(1079 &mut rpc_module,1080 eth_deps,1081 subscription_task_executor.clone(),1082 )?;10831084 Ok(rpc_module)1085 }1086 });10871088 sc_service::spawn_tasks(sc_service::SpawnTasksParams {1089 network,1090 sync_service,1091 client,1092 keystore: keystore_container.keystore(),1093 task_manager: &mut task_manager,1094 transaction_pool,1095 rpc_builder,1096 backend,1097 system_rpc_tx,1098 config,1099 telemetry: None,1100 tx_handler_controller,1101 })?;11021103 network_starter.start_network();1104 Ok(task_manager)1105}11061107fn overrides_handle<C, BE>(client: Arc<C>) -> Arc<OverrideHandle<Block>>1108where1109 C: ProvideRuntimeApi<Block> + StorageProvider<Block, BE> + AuxStore,1110 C: HeaderBackend<Block> + HeaderMetadata<Block, Error = BlockChainError>,1111 C: Send + Sync + 'static,1112 C::Api: fp_rpc::EthereumRuntimeRPCApi<Block>,1113 BE: Backend<Block> + 'static,1114 BE::State: StateBackend<BlakeTwo256>,1115{1116 let mut overrides_map = BTreeMap::new();1117 overrides_map.insert(1118 EthereumStorageSchema::V1,1119 Box::new(SchemaV1Override::new(client.clone())) as Box<dyn StorageOverride<_> + 'static>,1120 );1121 overrides_map.insert(1122 EthereumStorageSchema::V2,1123 Box::new(SchemaV2Override::new(client.clone())) as Box<dyn StorageOverride<_> + 'static>,1124 );1125 overrides_map.insert(1126 EthereumStorageSchema::V3,1127 Box::new(SchemaV3Override::new(client.clone())) as Box<dyn StorageOverride<_> + 'static>,1128 );11291130 Arc::new(OverrideHandle {1131 schemas: overrides_map,1132 fallback: Box::new(RuntimeApiStorageOverride::new(client)),1133 })1134}11351136pub struct FrontierTaskParams<'a, C, B> {1137 pub task_manager: &'a TaskManager,1138 pub client: Arc<C>,1139 pub substrate_backend: Arc<B>,1140 pub eth_backend: Arc<fc_db::kv::Backend<Block>>,1141 pub eth_filter_pool: Option<FilterPool>,1142 pub overrides: Arc<OverrideHandle<Block>>,1143 pub fee_history_limit: u64,1144 pub fee_history_cache: FeeHistoryCache,1145 pub sync_strategy: SyncStrategy,1146 pub prometheus_registry: Option<Registry>,1147}11481149pub fn spawn_frontier_tasks<C, B>(1150 params: FrontierTaskParams<C, B>,1151 sync: Arc<SyncingService<Block>>,1152 pubsub_notification_sinks: Arc<1153 EthereumBlockNotificationSinks<fc_mapping_sync::EthereumBlockNotification<Block>>,1154 >,1155) -> Arc<EthBlockDataCacheTask<Block>>1156where1157 C: ProvideRuntimeApi<Block> + BlockOf,1158 C: HeaderBackend<Block> + HeaderMetadata<Block, Error = BlockChainError> + 'static,1159 C: BlockchainEvents<Block> + StorageProvider<Block, B>,1160 C: Send + Sync + 'static,1161 C::Api: EthereumRuntimeRPCApi<Block>,1162 C::Api: BlockBuilder<Block>,1163 B: Backend<Block> + 'static,1164 B::State: StateBackend<BlakeTwo256>,1165{1166 let FrontierTaskParams {1167 task_manager,1168 client,1169 substrate_backend,1170 eth_backend,1171 eth_filter_pool,1172 overrides,1173 fee_history_limit,1174 fee_history_cache,1175 sync_strategy,1176 prometheus_registry,1177 } = params;1178 1179 1180 params.task_manager.spawn_essential_handle().spawn(1181 "frontier-mapping-sync-worker",1182 Some("frontier"),1183 MappingSyncWorker::new(1184 client.import_notification_stream(),1185 Duration::new(6, 0),1186 client.clone(),1187 substrate_backend,1188 overrides.clone(),1189 eth_backend,1190 3,1191 0,1192 sync_strategy,1193 sync,1194 pubsub_notification_sinks,1195 )1196 .for_each(|()| futures::future::ready(())),1197 );11981199 1200 1201 if let Some(eth_filter_pool) = eth_filter_pool {1202 1203 const FILTER_RETAIN_THRESHOLD: u64 = 100;1204 params.task_manager.spawn_essential_handle().spawn(1205 "frontier-filter-pool",1206 Some("frontier"),1207 EthTask::filter_pool_task(client.clone(), eth_filter_pool, FILTER_RETAIN_THRESHOLD),1208 );1209 }12101211 1212 params.task_manager.spawn_essential_handle().spawn(1213 "frontier-fee-history",1214 Some("frontier"),1215 EthTask::fee_history_task(1216 client,1217 overrides.clone(),1218 fee_history_cache,1219 fee_history_limit,1220 ),1221 );12221223 Arc::new(EthBlockDataCacheTask::new(1224 task_manager.spawn_handle(),1225 overrides,1226 50,1227 50,1228 prometheus_registry,1229 ))1230}