1
// Copyright 2024 Moonbeam foundation
2
// This file is part of Moonbeam.
3

            
4
// Moonbeam is free software: you can redistribute it and/or modify
5
// it under the terms of the GNU General Public License as published by
6
// the Free Software Foundation, either version 3 of the License, or
7
// (at your option) any later version.
8

            
9
// Moonbeam is distributed in the hope that it will be useful,
10
// but WITHOUT ANY WARRANTY; without even the implied warranty of
11
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
12
// GNU General Public License for more details.
13

            
14
// You should have received a copy of the GNU General Public License
15
// along with Moonbeam.  If not, see <http://www.gnu.org/licenses/>.
16

            
17
use crate::{
18
	lazy_loading, open_frontier_backend, rpc, set_prometheus_registry, BlockImportPipeline,
19
	ClientCustomizations, FrontierBlockImport, HostFunctions, PartialComponentsResult,
20
	PendingConsensusDataProvider, RuntimeApiCollection, SOFT_DEADLINE_PERCENT,
21
};
22
use cumulus_client_parachain_inherent::{MockValidationDataInherentDataProvider, MockXcmConfig};
23
use cumulus_primitives_core::{relay_chain, BlockT, ParaId};
24
use cumulus_primitives_parachain_inherent::ParachainInherentData;
25
use cumulus_test_relay_sproof_builder::RelayStateSproofBuilder;
26
use fc_rpc::StorageOverrideHandler;
27
use fc_rpc_core::types::{FeeHistoryCache, FilterPool};
28
use futures::{FutureExt, StreamExt};
29
use moonbeam_cli_opt::{EthApi as EthApiCmd, LazyLoadingConfig, RpcConfig};
30
use moonbeam_core_primitives::{Block, Hash};
31
use nimbus_consensus::NimbusManualSealConsensusDataProvider;
32
use nimbus_primitives::NimbusId;
33
use parity_scale_codec::Encode;
34
use polkadot_primitives::{
35
	AbridgedHostConfiguration, AsyncBackingParams, PersistedValidationData, Slot, UpgradeGoAhead,
36
};
37
use sc_chain_spec::{get_extension, BuildGenesisBlock, GenesisBlockBuilder};
38
use sc_client_api::{Backend, BadBlocks, ExecutorProvider, ForkBlocks, StorageProvider};
39
use sc_executor::{HeapAllocStrategy, RuntimeVersionOf, WasmExecutor, DEFAULT_HEAP_ALLOC_STRATEGY};
40
use sc_network::config::FullNetworkConfiguration;
41
use sc_network::NetworkBackend;
42
use sc_network_common::sync::SyncMode;
43
use sc_service::{
44
	error::Error as ServiceError, ClientConfig, Configuration, Error, KeystoreContainer,
45
	LocalCallExecutor, PartialComponents, TaskManager,
46
};
47
use sc_telemetry::{TelemetryHandle, TelemetryWorker};
48
use sc_transaction_pool_api::OffchainTransactionPoolFactory;
49
use sp_api::ConstructRuntimeApi;
50
use sp_blockchain::HeaderBackend;
51
use sp_core::traits::CodeExecutor;
52
use sp_core::{twox_128, H256};
53
use sp_runtime::traits::NumberFor;
54
use sp_storage::StorageKey;
55
use std::collections::BTreeMap;
56
use std::str::FromStr;
57
use std::sync::{Arc, Mutex};
58
use std::time::Duration;
59

            
60
pub mod backend;
61
pub mod call_executor;
62
mod client;
63
mod helpers;
64
mod lock;
65
mod manual_sealing;
66
mod state_overrides;
67

            
68
pub const LAZY_LOADING_LOG_TARGET: &'static str = "lazy-loading";
69

            
70
/// Lazy loading client type.
71
pub type TLazyLoadingClient<TBl, TRtApi, TExec> = sc_service::client::Client<
72
	TLazyLoadingBackend<TBl>,
73
	TLazyLoadingCallExecutor<TBl, TExec>,
74
	TBl,
75
	TRtApi,
76
>;
77

            
78
/// Lazy loading client backend type.
79
pub type TLazyLoadingBackend<TBl> = backend::Backend<TBl>;
80

            
81
/// Lazy loading client call executor type.
82
pub type TLazyLoadingCallExecutor<TBl, TExec> = call_executor::LazyLoadingCallExecutor<
83
	TBl,
84
	LocalCallExecutor<TBl, TLazyLoadingBackend<TBl>, TExec>,
85
>;
86

            
87
/// Lazy loading parts type.
88
pub type TLazyLoadingParts<TBl, TRtApi, TExec> = (
89
	TLazyLoadingClient<TBl, TRtApi, TExec>,
90
	Arc<TLazyLoadingBackend<TBl>>,
91
	KeystoreContainer,
92
	TaskManager,
93
);
94

            
95
type LazyLoadingClient<RuntimeApi> =
96
	TLazyLoadingClient<Block, RuntimeApi, WasmExecutor<HostFunctions>>;
97
type LazyLoadingBackend = TLazyLoadingBackend<Block>;
98

            
99
/// Create the initial parts of a lazy loading node.
100
pub fn new_lazy_loading_parts<TBl, TRtApi, TExec>(
101
	config: &mut Configuration,
102
	lazy_loading_config: &LazyLoadingConfig,
103
	telemetry: Option<TelemetryHandle>,
104
	executor: TExec,
105
) -> Result<TLazyLoadingParts<TBl, TRtApi, TExec>, Error>
106
where
107
	TBl: BlockT + sp_runtime::DeserializeOwned,
108
	TBl::Hash: From<H256>,
109
	TExec: CodeExecutor + RuntimeVersionOf + Clone,
110
{
111
	let backend = backend::new_lazy_loading_backend(config, &lazy_loading_config)?;
112

            
113
	let genesis_block_builder = GenesisBlockBuilder::new(
114
		config.chain_spec.as_storage_builder(),
115
		!config.no_genesis(),
116
		backend.clone(),
117
		executor.clone(),
118
	)?;
119

            
120
	new_lazy_loading_parts_with_genesis_builder(
121
		config,
122
		telemetry,
123
		executor,
124
		backend,
125
		genesis_block_builder,
126
	)
127
}
128

            
129
/// Create the initial parts of a lazy loading node.
130
pub fn new_lazy_loading_parts_with_genesis_builder<TBl, TRtApi, TExec, TBuildGenesisBlock>(
131
	config: &Configuration,
132
	telemetry: Option<TelemetryHandle>,
133
	executor: TExec,
134
	backend: Arc<TLazyLoadingBackend<TBl>>,
135
	genesis_block_builder: TBuildGenesisBlock,
136
) -> Result<TLazyLoadingParts<TBl, TRtApi, TExec>, Error>
137
where
138
	TBl: BlockT + sp_runtime::DeserializeOwned,
139
	TBl::Hash: From<H256>,
140
	TExec: CodeExecutor + RuntimeVersionOf + Clone,
141
	TBuildGenesisBlock:
142
		BuildGenesisBlock<
143
			TBl,
144
			BlockImportOperation = <TLazyLoadingBackend<TBl> as sc_client_api::backend::Backend<
145
				TBl,
146
			>>::BlockImportOperation,
147
		>,
148
{
149
	let keystore_container = KeystoreContainer::new(&config.keystore)?;
150

            
151
	let task_manager = {
152
		let registry = config.prometheus_config.as_ref().map(|cfg| &cfg.registry);
153
		TaskManager::new(config.tokio_handle.clone(), registry)?
154
	};
155

            
156
	let chain_spec = &config.chain_spec;
157
	let fork_blocks = get_extension::<ForkBlocks<TBl>>(chain_spec.extensions())
158
		.cloned()
159
		.unwrap_or_default();
160

            
161
	let bad_blocks = get_extension::<BadBlocks<TBl>>(chain_spec.extensions())
162
		.cloned()
163
		.unwrap_or_default();
164

            
165
	let client = {
166
		let extensions = sc_client_api::execution_extensions::ExecutionExtensions::new(
167
			None,
168
			Arc::new(executor.clone()),
169
		);
170

            
171
		let wasm_runtime_substitutes = config
172
			.chain_spec
173
			.code_substitutes()
174
			.into_iter()
175
			.map(|(n, c)| {
176
				let number = NumberFor::<TBl>::from_str(&n).map_err(|_| {
177
					Error::Application(Box::from(format!(
178
						"Failed to parse `{}` as block number for code substitutes. \
179
						 In an old version the key for code substitute was a block hash. \
180
						 Please update the chain spec to a version that is compatible with your node.",
181
						n
182
					)))
183
				})?;
184
				Ok((number, c))
185
			})
186
			.collect::<Result<std::collections::HashMap<_, _>, Error>>()?;
187

            
188
		let client = client::new_client(
189
			backend.clone(),
190
			executor,
191
			genesis_block_builder,
192
			fork_blocks,
193
			bad_blocks,
194
			extensions,
195
			Box::new(task_manager.spawn_handle()),
196
			config
197
				.prometheus_config
198
				.as_ref()
199
				.map(|config| config.registry.clone()),
200
			telemetry,
201
			ClientConfig {
202
				offchain_worker_enabled: config.offchain_worker.enabled,
203
				offchain_indexing_api: config.offchain_worker.indexing_enabled,
204
				wasmtime_precompiled: config.executor.wasmtime_precompiled.clone(),
205
				wasm_runtime_overrides: config.wasm_runtime_overrides.clone(),
206
				no_genesis: matches!(
207
					config.network.sync_mode,
208
					SyncMode::LightState { .. } | SyncMode::Warp { .. }
209
				),
210
				wasm_runtime_substitutes,
211
				enable_import_proof_recording: false,
212
			},
213
		)?;
214

            
215
		client
216
	};
217

            
218
	Ok((client, backend, keystore_container, task_manager))
219
}
220

            
221
/// Builds the PartialComponents for a lazy loading node.
222
#[allow(clippy::type_complexity)]
223
pub fn new_lazy_loading_partial<RuntimeApi, Customizations>(
224
	config: &mut Configuration,
225
	rpc_config: &RpcConfig,
226
	lazy_loading_config: &LazyLoadingConfig,
227
) -> PartialComponentsResult<LazyLoadingClient<RuntimeApi>, LazyLoadingBackend>
228
where
229
	RuntimeApi: ConstructRuntimeApi<Block, LazyLoadingClient<RuntimeApi>> + Send + Sync + 'static,
230
	RuntimeApi::RuntimeApi: RuntimeApiCollection,
231
	Customizations: ClientCustomizations + 'static,
232
{
233
	set_prometheus_registry(config, rpc_config.no_prometheus_prefix)?;
234

            
235
	// Use ethereum style for subscription ids
236
	config.rpc.id_provider = Some(Box::new(fc_rpc::EthereumSubIdProvider));
237

            
238
	let telemetry = config
239
		.telemetry_endpoints
240
		.clone()
241
		.filter(|x| !x.is_empty())
242
		.map(|endpoints| -> Result<_, sc_telemetry::Error> {
243
			let worker = TelemetryWorker::new(16)?;
244
			let telemetry = worker.handle().new_telemetry(endpoints);
245
			Ok((worker, telemetry))
246
		})
247
		.transpose()?;
248

            
249
	let heap_pages = config
250
		.executor
251
		.default_heap_pages
252
		.map_or(DEFAULT_HEAP_ALLOC_STRATEGY, |h| HeapAllocStrategy::Static {
253
			extra_pages: h as _,
254
		});
255
	let mut wasm_builder = WasmExecutor::builder()
256
		.with_execution_method(config.executor.wasm_method)
257
		.with_onchain_heap_alloc_strategy(heap_pages)
258
		.with_offchain_heap_alloc_strategy(heap_pages)
259
		.with_ignore_onchain_heap_pages(true)
260
		.with_max_runtime_instances(config.executor.max_runtime_instances)
261
		.with_runtime_cache_size(config.executor.runtime_cache_size);
262

            
263
	if let Some(ref wasmtime_precompiled_path) = config.executor.wasmtime_precompiled {
264
		wasm_builder = wasm_builder.with_wasmtime_precompiled_path(wasmtime_precompiled_path);
265
	}
266

            
267
	let executor = wasm_builder.build();
268

            
269
	let (client, backend, keystore_container, task_manager) =
270
		new_lazy_loading_parts::<Block, RuntimeApi, _>(
271
			config,
272
			lazy_loading_config,
273
			telemetry.as_ref().map(|(_, telemetry)| telemetry.handle()),
274
			executor,
275
		)?;
276

            
277
	if let Some(block_number) = Customizations::first_block_number_compatible_with_ed25519_zebra() {
278
		client
279
			.execution_extensions()
280
			.set_extensions_factory(sc_client_api::execution_extensions::ExtensionBeforeBlock::<
281
			Block,
282
			sp_io::UseDalekExt,
283
		>::new(block_number));
284
	}
285

            
286
	let client = Arc::new(client);
287

            
288
	let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());
289

            
290
	let telemetry = telemetry.map(|(worker, telemetry)| {
291
		task_manager
292
			.spawn_handle()
293
			.spawn("telemetry", None, worker.run());
294
		telemetry
295
	});
296

            
297
	let maybe_select_chain = Some(sc_consensus::LongestChain::new(backend.clone()));
298

            
299
	let transaction_pool = sc_transaction_pool::BasicPool::new_full(
300
		config.transaction_pool.clone(),
301
		config.role.is_authority().into(),
302
		config.prometheus_registry(),
303
		task_manager.spawn_essential_handle(),
304
		client.clone(),
305
	);
306

            
307
	let filter_pool: Option<FilterPool> = Some(Arc::new(Mutex::new(BTreeMap::new())));
308
	let fee_history_cache: FeeHistoryCache = Arc::new(Mutex::new(BTreeMap::new()));
309

            
310
	let frontier_backend = Arc::new(open_frontier_backend(client.clone(), config, rpc_config)?);
311
	let frontier_block_import = FrontierBlockImport::new(client.clone(), client.clone());
312

            
313
	let create_inherent_data_providers = move |_, _| async move {
314
		let time = sp_timestamp::InherentDataProvider::from_system_time();
315
		// Create a dummy parachain inherent data provider which is required to pass
316
		// the checks by the para chain system. We use dummy values because in the 'pending context'
317
		// neither do we have access to the real values nor do we need them.
318
		let (relay_parent_storage_root, relay_chain_state) =
319
			RelayStateSproofBuilder::default().into_state_root_and_proof();
320
		let vfp = PersistedValidationData {
321
			// This is a hack to make `cumulus_pallet_parachain_system::RelayNumberStrictlyIncreases`
322
			// happy. Relay parent number can't be bigger than u32::MAX.
323
			relay_parent_number: u32::MAX,
324
			relay_parent_storage_root,
325
			..Default::default()
326
		};
327
		let parachain_inherent_data = ParachainInherentData {
328
			validation_data: vfp,
329
			relay_chain_state,
330
			downward_messages: Default::default(),
331
			horizontal_messages: Default::default(),
332
		};
333
		Ok((time, parachain_inherent_data))
334
	};
335

            
336
	let import_queue = nimbus_consensus::import_queue(
337
		client.clone(),
338
		frontier_block_import.clone(),
339
		create_inherent_data_providers,
340
		&task_manager.spawn_essential_handle(),
341
		config.prometheus_registry(),
342
		false,
343
	)?;
344
	let block_import = BlockImportPipeline::Dev(frontier_block_import);
345

            
346
	Ok(PartialComponents {
347
		backend,
348
		client,
349
		import_queue,
350
		keystore_container,
351
		task_manager,
352
		transaction_pool,
353
		select_chain: maybe_select_chain,
354
		other: (
355
			block_import,
356
			filter_pool,
357
			telemetry,
358
			telemetry_worker_handle,
359
			frontier_backend,
360
			fee_history_cache,
361
		),
362
	})
363
}
364

            
365
/// Builds a new lazy loading service. This service uses manual seal, and mocks
366
/// the parachain inherent.
367
#[sc_tracing::logging::prefix_logs_with("Lazy loading 🌗")]
368
pub async fn new_lazy_loading_service<RuntimeApi, Customizations, Net>(
369
	mut config: Configuration,
370
	_author_id: Option<NimbusId>,
371
	sealing: moonbeam_cli_opt::Sealing,
372
	rpc_config: RpcConfig,
373
	lazy_loading_config: LazyLoadingConfig,
374
	hwbench: Option<sc_sysinfo::HwBench>,
375
) -> Result<TaskManager, ServiceError>
376
where
377
	RuntimeApi: ConstructRuntimeApi<Block, LazyLoadingClient<RuntimeApi>> + Send + Sync + 'static,
378
	RuntimeApi::RuntimeApi: RuntimeApiCollection,
379
	Customizations: ClientCustomizations + 'static,
380
	Net: NetworkBackend<Block, Hash>,
381
{
382
	use async_io::Timer;
383
	use futures::Stream;
384
	use sc_consensus_manual_seal::{EngineCommand, ManualSealParams};
385

            
386
	let sc_service::PartialComponents {
387
		client,
388
		backend,
389
		mut task_manager,
390
		import_queue,
391
		keystore_container,
392
		select_chain: maybe_select_chain,
393
		transaction_pool,
394
		other:
395
			(
396
				block_import_pipeline,
397
				filter_pool,
398
				mut telemetry,
399
				_telemetry_worker_handle,
400
				frontier_backend,
401
				fee_history_cache,
402
			),
403
	} = lazy_loading::new_lazy_loading_partial::<RuntimeApi, Customizations>(
404
		&mut config,
405
		&rpc_config,
406
		&lazy_loading_config,
407
	)?;
408

            
409
	let start_delay = 10;
410
	let lazy_loading_startup_disclaimer = format!(
411
		r#"
412

            
413
		You are now running the Moonbeam client in lazy loading mode, where data is retrieved
414
		from a live RPC node on demand.
415

            
416
		Using remote state from: {rpc}
417
		Forking from block: {fork_block}
418

            
419
		To ensure the client works properly, please note the following:
420

            
421
		    1. *Avoid Throttling*: Ensure that the backing RPC node is not limiting the number of
422
		    requests, as this can prevent the lazy loading client from functioning correctly;
423

            
424
		    2. *Be Patient*: As the client may take approximately 20 times longer than normal to
425
		    retrieve and process the necessary data for the requested operation.
426

            
427

            
428
		The service will start in {start_delay} seconds...
429

            
430
		"#,
431
		rpc = lazy_loading_config.state_rpc,
432
		fork_block = backend.fork_checkpoint.number
433
	);
434

            
435
	log::warn!(
436
		"{}",
437
		ansi_term::Colour::Yellow.paint(lazy_loading_startup_disclaimer)
438
	);
439
	tokio::time::sleep(Duration::from_secs(start_delay)).await;
440

            
441
	let block_import = if let BlockImportPipeline::Dev(block_import) = block_import_pipeline {
442
		block_import
443
	} else {
444
		return Err(ServiceError::Other(
445
			"Block import pipeline is not dev".to_string(),
446
		));
447
	};
448

            
449
	let prometheus_registry = config.prometheus_registry().cloned();
450
	let net_config =
451
		FullNetworkConfiguration::<_, _, Net>::new(&config.network, prometheus_registry.clone());
452

            
453
	let metrics = Net::register_notification_metrics(
454
		config.prometheus_config.as_ref().map(|cfg| &cfg.registry),
455
	);
456

            
457
	let (network, system_rpc_tx, tx_handler_controller, network_starter, sync_service) =
458
		sc_service::build_network(sc_service::BuildNetworkParams {
459
			config: &config,
460
			client: client.clone(),
461
			transaction_pool: transaction_pool.clone(),
462
			spawn_handle: task_manager.spawn_handle(),
463
			import_queue,
464
			block_announce_validator_builder: None,
465
			warp_sync_config: None,
466
			net_config,
467
			block_relay: None,
468
			metrics,
469
		})?;
470

            
471
	if config.offchain_worker.enabled {
472
		task_manager.spawn_handle().spawn(
473
			"offchain-workers-runner",
474
			"offchain-work",
475
			sc_offchain::OffchainWorkers::new(sc_offchain::OffchainWorkerOptions {
476
				runtime_api_provider: client.clone(),
477
				keystore: Some(keystore_container.keystore()),
478
				offchain_db: backend.offchain_storage(),
479
				transaction_pool: Some(OffchainTransactionPoolFactory::new(
480
					transaction_pool.clone(),
481
				)),
482
				network_provider: Arc::new(network.clone()),
483
				is_validator: config.role.is_authority(),
484
				enable_http_requests: true,
485
				custom_extensions: move |_| vec![],
486
			})
487
			.run(client.clone(), task_manager.spawn_handle())
488
			.boxed(),
489
		);
490
	}
491

            
492
	let prometheus_registry = config.prometheus_registry().cloned();
493
	let overrides = Arc::new(StorageOverrideHandler::new(client.clone()));
494
	let fee_history_limit = rpc_config.fee_history_limit;
495
	let mut command_sink = None;
496
	let mut dev_rpc_data = None;
497
	let collator = config.role.is_authority();
498

            
499
	if collator {
500
		let mut env = sc_basic_authorship::ProposerFactory::with_proof_recording(
501
			task_manager.spawn_handle(),
502
			client.clone(),
503
			transaction_pool.clone(),
504
			prometheus_registry.as_ref(),
505
			telemetry.as_ref().map(|x| x.handle()),
506
		);
507
		env.set_soft_deadline(SOFT_DEADLINE_PERCENT);
508

            
509
		let commands_stream: Box<dyn Stream<Item = EngineCommand<H256>> + Send + Sync + Unpin> =
510
			match sealing {
511
				moonbeam_cli_opt::Sealing::Instant => {
512
					Box::new(
513
						// This bit cribbed from the implementation of instant seal.
514
						transaction_pool
515
							.pool()
516
							.validated_pool()
517
							.import_notification_stream()
518
							.map(|_| EngineCommand::SealNewBlock {
519
								create_empty: false,
520
								finalize: false,
521
								parent_hash: None,
522
								sender: None,
523
							}),
524
					)
525
				}
526
				moonbeam_cli_opt::Sealing::Manual => {
527
					let (sink, stream) = futures::channel::mpsc::channel(1000);
528
					// Keep a reference to the other end of the channel. It goes to the RPC.
529
					command_sink = Some(sink);
530
					Box::new(stream)
531
				}
532
				moonbeam_cli_opt::Sealing::Interval(millis) => Box::new(StreamExt::map(
533
					Timer::interval(Duration::from_millis(millis)),
534
					|_| EngineCommand::SealNewBlock {
535
						create_empty: true,
536
						finalize: false,
537
						parent_hash: None,
538
						sender: None,
539
					},
540
				)),
541
			};
542

            
543
		let select_chain = maybe_select_chain.expect(
544
			"`new_lazy_loading_partial` builds a `LongestChainRule` when building dev service.\
545
				We specified the dev service when calling `new_partial`.\
546
				Therefore, a `LongestChainRule` is present. qed.",
547
		);
548

            
549
		let client_set_aside_for_cidp = client.clone();
550

            
551
		// Create channels for mocked XCM messages.
552
		let (downward_xcm_sender, downward_xcm_receiver) = flume::bounded::<Vec<u8>>(100);
553
		let (hrmp_xcm_sender, hrmp_xcm_receiver) = flume::bounded::<(ParaId, Vec<u8>)>(100);
554
		let additional_relay_offset = Arc::new(std::sync::atomic::AtomicU32::new(0));
555
		dev_rpc_data = Some((
556
			downward_xcm_sender,
557
			hrmp_xcm_sender,
558
			additional_relay_offset,
559
		));
560

            
561
		let client_clone = client.clone();
562
		let keystore_clone = keystore_container.keystore().clone();
563
		let maybe_provide_vrf_digest =
564
			move |nimbus_id: NimbusId, parent: Hash| -> Option<sp_runtime::generic::DigestItem> {
565
				moonbeam_vrf::vrf_pre_digest::<Block, LazyLoadingClient<RuntimeApi>>(
566
					&client_clone,
567
					&keystore_clone,
568
					nimbus_id,
569
					parent,
570
				)
571
			};
572

            
573
		let parachain_id = helpers::get_parachain_id(backend.rpc_client.clone())
574
			.unwrap_or_else(|| panic!("Could not get parachain identifier for lazy loading mode."));
575

            
576
		task_manager.spawn_essential_handle().spawn_blocking(
577
			"authorship_task",
578
			Some("block-authoring"),
579
			manual_sealing::run_manual_seal(ManualSealParams {
580
				block_import,
581
				env,
582
				client: client.clone(),
583
				pool: transaction_pool.clone(),
584
				commands_stream,
585
				select_chain,
586
				consensus_data_provider: Some(Box::new(NimbusManualSealConsensusDataProvider {
587
					keystore: keystore_container.keystore(),
588
					client: client.clone(),
589
					additional_digests_provider: maybe_provide_vrf_digest,
590
					_phantom: Default::default(),
591
				})),
592
				create_inherent_data_providers: move |block: H256, ()| {
593
					let maybe_current_para_block = client_set_aside_for_cidp.number(block);
594
					let maybe_current_para_head = client_set_aside_for_cidp.expect_header(block);
595
					let downward_xcm_receiver = downward_xcm_receiver.clone();
596
					let hrmp_xcm_receiver = hrmp_xcm_receiver.clone();
597

            
598
					let client_for_cidp = client_set_aside_for_cidp.clone();
599
					async move {
600
						let time = sp_timestamp::InherentDataProvider::from_system_time();
601

            
602
						let current_para_block = maybe_current_para_block?
603
							.ok_or(sp_blockchain::Error::UnknownBlock(block.to_string()))?;
604

            
605
						let current_para_block_head = Some(polkadot_primitives::HeadData(
606
							maybe_current_para_head?.encode(),
607
						));
608

            
609
						let mut additional_key_values = vec![
610
							(
611
								moonbeam_core_primitives::well_known_relay_keys::TIMESTAMP_NOW
612
									.to_vec(),
613
								sp_timestamp::Timestamp::current().encode(),
614
							),
615
							(
616
								relay_chain::well_known_keys::ACTIVE_CONFIG.to_vec(),
617
								AbridgedHostConfiguration {
618
									max_code_size: 3_145_728,
619
									max_head_data_size: 20_480,
620
									max_upward_queue_count: 174_762,
621
									max_upward_queue_size: 1_048_576,
622
									max_upward_message_size: 65_531,
623
									max_upward_message_num_per_candidate: 16,
624
									hrmp_max_message_num_per_candidate: 10,
625
									validation_upgrade_cooldown: 14_400,
626
									validation_upgrade_delay: 600,
627
									async_backing_params: AsyncBackingParams {
628
										max_candidate_depth: 3,
629
										allowed_ancestry_len: 2,
630
									},
631
								}
632
								.encode(),
633
							),
634
							// Override current slot number
635
							(
636
								relay_chain::well_known_keys::CURRENT_SLOT.to_vec(),
637
								Slot::from(u64::from(current_para_block)).encode(),
638
							),
639
						];
640

            
641
						// If there is a pending upgrade, lets mimic a GoAhead
642
						// signal from the relay
643

            
644
						let storage_key = [
645
							twox_128(b"ParachainSystem"),
646
							twox_128(b"PendingValidationCode"),
647
						]
648
						.concat();
649
						let has_pending_upgrade = client_for_cidp
650
							.storage(block, &StorageKey(storage_key))
651
							.map_or(false, |ok| ok.map_or(false, |some| !some.0.is_empty()));
652
						if has_pending_upgrade {
653
							additional_key_values.push((
654
								relay_chain::well_known_keys::upgrade_go_ahead_signal(ParaId::new(
655
									parachain_id,
656
								)),
657
								Some(UpgradeGoAhead::GoAhead).encode(),
658
							));
659
						}
660

            
661
						let mocked_parachain = MockValidationDataInherentDataProvider {
662
							current_para_block,
663
							para_id: ParaId::new(parachain_id),
664
							current_para_block_head,
665
							relay_offset: 1000,
666
							relay_blocks_per_para_block: 2,
667
							// TODO: Recheck
668
							para_blocks_per_relay_epoch: 10,
669
							relay_randomness_config: (),
670
							xcm_config: MockXcmConfig::new(
671
								&*client_for_cidp,
672
								block,
673
								Default::default(),
674
							),
675
							raw_downward_messages: downward_xcm_receiver.drain().collect(),
676
							raw_horizontal_messages: hrmp_xcm_receiver.drain().collect(),
677
							additional_key_values: Some(additional_key_values),
678
						};
679

            
680
						let randomness = session_keys_primitives::InherentDataProvider;
681

            
682
						Ok((time, mocked_parachain, randomness))
683
					}
684
				},
685
			}),
686
		);
687
	}
688

            
689
	// Sinks for pubsub notifications.
690
	// Everytime a new subscription is created, a new mpsc channel is added to the sink pool.
691
	// The MappingSyncWorker sends through the channel on block import and the subscription emits a
692
	// notification to the subscriber on receiving a message through this channel.
693
	// This way we avoid race conditions when using native substrate block import notification
694
	// stream.
695
	let pubsub_notification_sinks: fc_mapping_sync::EthereumBlockNotificationSinks<
696
		fc_mapping_sync::EthereumBlockNotification<Block>,
697
	> = Default::default();
698
	let pubsub_notification_sinks = Arc::new(pubsub_notification_sinks);
699

            
700
	/* TODO: only enable this when frontier backend is compatible with lazy-loading
701
	rpc::spawn_essential_tasks(
702
		rpc::SpawnTasksParams {
703
			task_manager: &task_manager,
704
			client: client.clone(),
705
			substrate_backend: backend.clone(),
706
			frontier_backend: frontier_backend.clone(),
707
			filter_pool: filter_pool.clone(),
708
			overrides: overrides.clone(),
709
			fee_history_limit,
710
			fee_history_cache: fee_history_cache.clone(),
711
		},
712
		sync_service.clone(),
713
		pubsub_notification_sinks.clone(),
714
	);
715
	*/
716

            
717
	let ethapi_cmd = rpc_config.ethapi.clone();
718
	let tracing_requesters =
719
		if ethapi_cmd.contains(&EthApiCmd::Debug) || ethapi_cmd.contains(&EthApiCmd::Trace) {
720
			rpc::tracing::spawn_tracing_tasks(
721
				&rpc_config,
722
				prometheus_registry.clone(),
723
				rpc::SpawnTasksParams {
724
					task_manager: &task_manager,
725
					client: client.clone(),
726
					substrate_backend: backend.clone(),
727
					frontier_backend: frontier_backend.clone(),
728
					filter_pool: filter_pool.clone(),
729
					overrides: overrides.clone(),
730
					fee_history_limit,
731
					fee_history_cache: fee_history_cache.clone(),
732
				},
733
			)
734
		} else {
735
			rpc::tracing::RpcRequesters {
736
				debug: None,
737
				trace: None,
738
			}
739
		};
740

            
741
	let block_data_cache = Arc::new(fc_rpc::EthBlockDataCacheTask::new(
742
		task_manager.spawn_handle(),
743
		overrides.clone(),
744
		rpc_config.eth_log_block_cache,
745
		rpc_config.eth_statuses_cache,
746
		prometheus_registry,
747
	));
748

            
749
	let rpc_builder = {
750
		let client = client.clone();
751
		let pool = transaction_pool.clone();
752
		let backend = backend.clone();
753
		let network = network.clone();
754
		let sync = sync_service.clone();
755
		let ethapi_cmd = ethapi_cmd.clone();
756
		let max_past_logs = rpc_config.max_past_logs;
757
		let overrides = overrides.clone();
758
		let fee_history_cache = fee_history_cache.clone();
759
		let block_data_cache = block_data_cache.clone();
760
		let pubsub_notification_sinks = pubsub_notification_sinks.clone();
761

            
762
		let keystore = keystore_container.keystore();
763
		let command_sink_for_task = command_sink.clone();
764
		move |subscription_task_executor| {
765
			let deps = rpc::FullDeps {
766
				backend: backend.clone(),
767
				client: client.clone(),
768
				command_sink: command_sink_for_task.clone(),
769
				ethapi_cmd: ethapi_cmd.clone(),
770
				filter_pool: filter_pool.clone(),
771
				frontier_backend: match *frontier_backend {
772
					fc_db::Backend::KeyValue(ref b) => b.clone(),
773
					fc_db::Backend::Sql(ref b) => b.clone(),
774
				},
775
				graph: pool.pool().clone(),
776
				pool: pool.clone(),
777
				is_authority: collator,
778
				max_past_logs,
779
				fee_history_limit,
780
				fee_history_cache: fee_history_cache.clone(),
781
				network: network.clone(),
782
				sync: sync.clone(),
783
				dev_rpc_data: dev_rpc_data.clone(),
784
				overrides: overrides.clone(),
785
				block_data_cache: block_data_cache.clone(),
786
				forced_parent_hashes: None,
787
			};
788

            
789
			let pending_consensus_data_provider = Box::new(PendingConsensusDataProvider::new(
790
				client.clone(),
791
				keystore.clone(),
792
			));
793
			if ethapi_cmd.contains(&EthApiCmd::Debug) || ethapi_cmd.contains(&EthApiCmd::Trace) {
794
				rpc::create_full(
795
					deps,
796
					subscription_task_executor,
797
					Some(crate::rpc::TracingConfig {
798
						tracing_requesters: tracing_requesters.clone(),
799
						trace_filter_max_count: rpc_config.ethapi_trace_max_count,
800
					}),
801
					pubsub_notification_sinks.clone(),
802
					pending_consensus_data_provider,
803
				)
804
				.map_err(Into::into)
805
			} else {
806
				rpc::create_full(
807
					deps,
808
					subscription_task_executor,
809
					None,
810
					pubsub_notification_sinks.clone(),
811
					pending_consensus_data_provider,
812
				)
813
				.map_err(Into::into)
814
			}
815
		}
816
	};
817

            
818
	let _rpc_handlers = sc_service::spawn_tasks(sc_service::SpawnTasksParams {
819
		network,
820
		client,
821
		keystore: keystore_container.keystore(),
822
		task_manager: &mut task_manager,
823
		transaction_pool,
824
		rpc_builder: Box::new(rpc_builder),
825
		backend,
826
		system_rpc_tx,
827
		sync_service: sync_service.clone(),
828
		config,
829
		tx_handler_controller,
830
		telemetry: None,
831
	})?;
832

            
833
	if let Some(hwbench) = hwbench {
834
		sc_sysinfo::print_hwbench(&hwbench);
835

            
836
		if let Some(ref mut telemetry) = telemetry {
837
			let telemetry_handle = telemetry.handle();
838
			task_manager.spawn_handle().spawn(
839
				"telemetry_hwbench",
840
				None,
841
				sc_sysinfo::initialize_hwbench_telemetry(telemetry_handle, hwbench),
842
			);
843
		}
844
	}
845

            
846
	network_starter.start_network();
847

            
848
	log::info!("Service Ready");
849

            
850
	Ok(task_manager)
851
}