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

            
61
pub mod call_executor;
62
mod client;
63
pub mod frontier_backend;
64
mod helpers;
65
mod lock;
66
mod manual_sealing;
67
mod rpc_client;
68
mod state_overrides;
69
pub mod substrate_backend;
70

            
71
pub const LAZY_LOADING_LOG_TARGET: &'static str = "lazy-loading";
72

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

            
81
/// Lazy loading client backend type.
82
pub type TLazyLoadingBackend<TBl> = substrate_backend::Backend<TBl>;
83

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

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

            
98
type LazyLoadingClient<RuntimeApi> =
99
	TLazyLoadingClient<Block, RuntimeApi, WasmExecutor<HostFunctions>>;
100
type LazyLoadingBackend = TLazyLoadingBackend<Block>;
101

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

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

            
123
	new_lazy_loading_parts_with_genesis_builder(
124
		config,
125
		telemetry,
126
		executor,
127
		backend,
128
		genesis_block_builder,
129
	)
130
}
131

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

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

            
159
	let chain_spec = &config.chain_spec;
160
	let fork_blocks = get_extension::<ForkBlocks<TBl>>(chain_spec.extensions())
161
		.cloned()
162
		.unwrap_or_default();
163

            
164
	let bad_blocks = get_extension::<BadBlocks<TBl>>(chain_spec.extensions())
165
		.cloned()
166
		.unwrap_or_default();
167

            
168
	let client = {
169
		let extensions = sc_client_api::execution_extensions::ExecutionExtensions::new(
170
			None,
171
			Arc::new(executor.clone()),
172
		);
173

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

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

            
218
		client
219
	};
220

            
221
	Ok((client, backend, keystore_container, task_manager))
222
}
223

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

            
238
	// Use ethereum style for subscription ids
239
	config.rpc.id_provider = Some(Box::new(fc_rpc::EthereumSubIdProvider));
240

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

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

            
266
	if let Some(ref wasmtime_precompiled_path) = config.executor.wasmtime_precompiled {
267
		wasm_builder = wasm_builder.with_wasmtime_precompiled_path(wasmtime_precompiled_path);
268
	}
269

            
270
	let executor = wasm_builder.build();
271

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

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

            
289
	let client = Arc::new(client);
290

            
291
	let telemetry_worker_handle = telemetry.as_ref().map(|(worker, _)| worker.handle());
292

            
293
	let telemetry = telemetry.map(|(worker, telemetry)| {
294
		task_manager
295
			.spawn_handle()
296
			.spawn("telemetry", None, worker.run());
297
		telemetry
298
	});
299

            
300
	let maybe_select_chain = Some(sc_consensus::LongestChain::new(backend.clone()));
301

            
302
	let transaction_pool = sc_transaction_pool::Builder::new(
303
		task_manager.spawn_essential_handle(),
304
		client.clone(),
305
		config.role.is_authority().into(),
306
	)
307
	.with_options(config.transaction_pool.clone())
308
	.with_prometheus(config.prometheus_registry())
309
	.build();
310

            
311
	let filter_pool: Option<FilterPool> = Some(Arc::new(Mutex::new(BTreeMap::new())));
312
	let fee_history_cache: FeeHistoryCache = Arc::new(Mutex::new(BTreeMap::new()));
313

            
314
	let frontier_backend = Arc::new(open_frontier_backend(client.clone(), config, rpc_config)?);
315
	let frontier_block_import = FrontierBlockImport::new(client.clone(), client.clone());
316

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

            
340
	let import_queue = nimbus_consensus::import_queue(
341
		client.clone(),
342
		frontier_block_import.clone(),
343
		create_inherent_data_providers,
344
		&task_manager.spawn_essential_handle(),
345
		config.prometheus_registry(),
346
		false,
347
	)?;
348
	let block_import = BlockImportPipeline::Dev(frontier_block_import);
349

            
350
	Ok(PartialComponents {
351
		backend,
352
		client,
353
		import_queue,
354
		keystore_container,
355
		task_manager,
356
		transaction_pool: transaction_pool.into(),
357
		select_chain: maybe_select_chain,
358
		other: (
359
			block_import,
360
			filter_pool,
361
			telemetry,
362
			telemetry_worker_handle,
363
			frontier_backend,
364
			fee_history_cache,
365
		),
366
	})
367
}
368

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

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

            
413
	let start_delay = 10;
414
	let lazy_loading_startup_disclaimer = format!(
415
		r#"
416

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

            
420
		Using remote state from: {rpc}
421
		Forking from block: {fork_block}
422

            
423
		To ensure the client works properly, please note the following:
424

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

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

            
431

            
432
		The service will start in {start_delay} seconds...
433

            
434
		"#,
435
		rpc = lazy_loading_config.state_rpc,
436
		fork_block = backend.fork_checkpoint.number
437
	);
438

            
439
	log::warn!(
440
		"{}",
441
		ansi_term::Colour::Yellow.paint(lazy_loading_startup_disclaimer)
442
	);
443
	tokio::time::sleep(Duration::from_secs(start_delay)).await;
444

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

            
453
	let prometheus_registry = config.prometheus_registry().cloned();
454
	let net_config =
455
		FullNetworkConfiguration::<_, _, Net>::new(&config.network, prometheus_registry.clone());
456

            
457
	let metrics = Net::register_notification_metrics(
458
		config.prometheus_config.as_ref().map(|cfg| &cfg.registry),
459
	);
460

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

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

            
496
	let prometheus_registry = config.prometheus_registry().cloned();
497
	let overrides = Arc::new(StorageOverrideHandler::new(client.clone()));
498
	let fee_history_limit = rpc_config.fee_history_limit;
499
	let mut command_sink = None;
500
	let mut dev_rpc_data = None;
501
	let collator = config.role.is_authority();
502

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

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

            
545
		let select_chain = maybe_select_chain.expect(
546
			"`new_lazy_loading_partial` builds a `LongestChainRule` when building dev service.\
547
				We specified the dev service when calling `new_partial`.\
548
				Therefore, a `LongestChainRule` is present. qed.",
549
		);
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
		// Need to clone it and store here to avoid moving of `client`
562
		// variable in closure below.
563
		let client_vrf = client.clone();
564

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

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

            
579
		// Need to clone it and store here to avoid moving of `client`
580
		// variable in closure below.
581
		let client_for_cidp = client.clone();
582

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

            
605
					// Need to clone it and store here to avoid moving of `client`
606
					// variable in closure below.
607
					let client_for_xcm = client_for_cidp.clone();
608

            
609
					async move {
610
						let time = sp_timestamp::InherentDataProvider::from_system_time();
611

            
612
						let current_para_block = maybe_current_para_block?
613
							.ok_or(sp_blockchain::Error::UnknownBlock(block.to_string()))?;
614

            
615
						let current_para_block_head = Some(polkadot_primitives::HeadData(
616
							maybe_current_para_head?.encode(),
617
						));
618

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

            
651
						let current_para_head = client_for_xcm
652
							.header(block)
653
							.expect("Header lookup should succeed")
654
							.expect("Header passed in as parent should be present in backend.");
655

            
656
						let should_send_go_ahead = match client_for_xcm
657
							.runtime_api()
658
							.collect_collation_info(block, &current_para_head)
659
						{
660
							Ok(info) => info.new_validation_code.is_some(),
661
							Err(e) => {
662
								log::error!("Failed to collect collation info: {:?}", e);
663
								false
664
							}
665
						};
666

            
667
						let mocked_parachain = MockValidationDataInherentDataProvider {
668
							current_para_block,
669
							para_id: ParaId::new(parachain_id),
670
							upgrade_go_ahead: should_send_go_ahead.then(|| {
671
								log::info!(
672
									"Detected pending validation code, sending go-ahead signal."
673
								);
674
								UpgradeGoAhead::GoAhead
675
							}),
676
							current_para_block_head,
677
							relay_offset: 1000,
678
							relay_blocks_per_para_block: 2,
679
							// TODO: Recheck
680
							para_blocks_per_relay_epoch: 10,
681
							relay_randomness_config: (),
682
							xcm_config: MockXcmConfig::new(
683
								&*client_for_xcm,
684
								block,
685
								Default::default(),
686
							),
687
							raw_downward_messages: downward_xcm_receiver.drain().collect(),
688
							raw_horizontal_messages: hrmp_xcm_receiver.drain().collect(),
689
							additional_key_values: Some(additional_key_values),
690
						};
691

            
692
						let randomness = session_keys_primitives::InherentDataProvider;
693

            
694
						Ok((time, mocked_parachain, randomness))
695
					}
696
				},
697
			}),
698
		);
699
	}
700

            
701
	// Sinks for pubsub notifications.
702
	// Everytime a new subscription is created, a new mpsc channel is added to the sink pool.
703
	// The MappingSyncWorker sends through the channel on block import and the subscription emits a
704
	// notification to the subscriber on receiving a message through this channel.
705
	// This way we avoid race conditions when using native substrate block import notification
706
	// stream.
707
	let pubsub_notification_sinks: fc_mapping_sync::EthereumBlockNotificationSinks<
708
		fc_mapping_sync::EthereumBlockNotification<Block>,
709
	> = Default::default();
710
	let pubsub_notification_sinks = Arc::new(pubsub_notification_sinks);
711

            
712
	/* TODO: only enable this when frontier backend is compatible with lazy-loading
713
	rpc::spawn_essential_tasks(
714
		rpc::SpawnTasksParams {
715
			task_manager: &task_manager,
716
			client: client.clone(),
717
			substrate_backend: backend.clone(),
718
			frontier_backend: frontier_backend.clone(),
719
			filter_pool: filter_pool.clone(),
720
			overrides: overrides.clone(),
721
			fee_history_limit,
722
			fee_history_cache: fee_history_cache.clone(),
723
		},
724
		sync_service.clone(),
725
		pubsub_notification_sinks.clone(),
726
	);
727
	*/
728

            
729
	let ethapi_cmd = rpc_config.ethapi.clone();
730
	let tracing_requesters =
731
		if ethapi_cmd.contains(&EthApiCmd::Debug) || ethapi_cmd.contains(&EthApiCmd::Trace) {
732
			rpc::tracing::spawn_tracing_tasks(
733
				&rpc_config,
734
				prometheus_registry.clone(),
735
				rpc::SpawnTasksParams {
736
					task_manager: &task_manager,
737
					client: client.clone(),
738
					substrate_backend: backend.clone(),
739
					frontier_backend: frontier_backend.clone(),
740
					filter_pool: filter_pool.clone(),
741
					overrides: overrides.clone(),
742
					fee_history_limit,
743
					fee_history_cache: fee_history_cache.clone(),
744
				},
745
			)
746
		} else {
747
			rpc::tracing::RpcRequesters {
748
				debug: None,
749
				trace: None,
750
			}
751
		};
752

            
753
	let block_data_cache = Arc::new(fc_rpc::EthBlockDataCacheTask::new(
754
		task_manager.spawn_handle(),
755
		overrides.clone(),
756
		rpc_config.eth_log_block_cache,
757
		rpc_config.eth_statuses_cache,
758
		prometheus_registry,
759
	));
760

            
761
	let rpc_builder = {
762
		let client = client.clone();
763
		let pool = transaction_pool.clone();
764
		let backend = backend.clone();
765
		let network = network.clone();
766
		let sync = sync_service.clone();
767
		let ethapi_cmd = ethapi_cmd.clone();
768
		let max_past_logs = rpc_config.max_past_logs;
769
		let max_block_range = rpc_config.max_block_range;
770
		let overrides = overrides.clone();
771
		let fee_history_cache = fee_history_cache.clone();
772
		let block_data_cache = block_data_cache.clone();
773
		let pubsub_notification_sinks = pubsub_notification_sinks.clone();
774

            
775
		let keystore = keystore_container.keystore();
776
		let command_sink_for_task = command_sink.clone();
777
		move |subscription_task_executor| {
778
			let deps = rpc::FullDeps {
779
				backend: backend.clone(),
780
				client: client.clone(),
781
				command_sink: command_sink_for_task.clone(),
782
				ethapi_cmd: ethapi_cmd.clone(),
783
				filter_pool: filter_pool.clone(),
784
				frontier_backend: Arc::new(LazyLoadingFrontierBackend {
785
					rpc_client: backend.clone().rpc_client.clone(),
786
					frontier_backend: match *frontier_backend {
787
						fc_db::Backend::KeyValue(ref b) => b.clone(),
788
						fc_db::Backend::Sql(ref b) => b.clone(),
789
					},
790
				}),
791
				graph: pool.clone(),
792
				pool: pool.clone(),
793
				is_authority: collator,
794
				max_past_logs,
795
				max_block_range,
796
				fee_history_limit,
797
				fee_history_cache: fee_history_cache.clone(),
798
				network: network.clone(),
799
				sync: sync.clone(),
800
				dev_rpc_data: dev_rpc_data.clone(),
801
				overrides: overrides.clone(),
802
				block_data_cache: block_data_cache.clone(),
803
				forced_parent_hashes: None,
804
			};
805

            
806
			let pending_consensus_data_provider = Box::new(PendingConsensusDataProvider::new(
807
				client.clone(),
808
				keystore.clone(),
809
			));
810
			if ethapi_cmd.contains(&EthApiCmd::Debug) || ethapi_cmd.contains(&EthApiCmd::Trace) {
811
				rpc::create_full(
812
					deps,
813
					subscription_task_executor,
814
					Some(crate::rpc::TracingConfig {
815
						tracing_requesters: tracing_requesters.clone(),
816
						trace_filter_max_count: rpc_config.ethapi_trace_max_count,
817
					}),
818
					pubsub_notification_sinks.clone(),
819
					pending_consensus_data_provider,
820
				)
821
				.map_err(Into::into)
822
			} else {
823
				rpc::create_full(
824
					deps,
825
					subscription_task_executor,
826
					None,
827
					pubsub_notification_sinks.clone(),
828
					pending_consensus_data_provider,
829
				)
830
				.map_err(Into::into)
831
			}
832
		}
833
	};
834

            
835
	let _rpc_handlers = sc_service::spawn_tasks(sc_service::SpawnTasksParams {
836
		network,
837
		client,
838
		keystore: keystore_container.keystore(),
839
		task_manager: &mut task_manager,
840
		transaction_pool,
841
		rpc_builder: Box::new(rpc_builder),
842
		backend,
843
		system_rpc_tx,
844
		sync_service: sync_service.clone(),
845
		config,
846
		tx_handler_controller,
847
		telemetry: None,
848
	})?;
849

            
850
	if let Some(hwbench) = hwbench {
851
		sc_sysinfo::print_hwbench(&hwbench);
852

            
853
		if let Some(ref mut telemetry) = telemetry {
854
			let telemetry_handle = telemetry.handle();
855
			task_manager.spawn_handle().spawn(
856
				"telemetry_hwbench",
857
				None,
858
				sc_sysinfo::initialize_hwbench_telemetry(telemetry_handle, hwbench),
859
			);
860
		}
861
	}
862

            
863
	network_starter.start_network();
864

            
865
	log::info!("Service Ready");
866

            
867
	Ok(task_manager)
868
}
869

            
870
pub fn spec_builder() -> sc_chain_spec::ChainSpecBuilder<Extensions> {
871
	crate::chain_spec::moonbeam::ChainSpec::builder(
872
		moonbeam_runtime::WASM_BINARY.expect("WASM binary was not build, please build it!"),
873
		Default::default(),
874
	)
875
	.with_name("Lazy Loading")
876
	.with_id("lazy_loading")
877
	.with_chain_type(ChainType::Development)
878
	.with_properties(
879
		serde_json::from_str(
880
			"{\"tokenDecimals\": 18, \"tokenSymbol\": \"GLMR\", \"SS58Prefix\": 1284}",
881
		)
882
		.expect("Provided valid json map"),
883
	)
884
	.with_genesis_config_preset_name(sp_genesis_builder::DEV_RUNTIME_PRESET)
885
}