referrerpolicy=no-referrer-when-downgrade

sc_service/
lib.rs

1// This file is part of Substrate.
2
3// Copyright (C) Parity Technologies (UK) Ltd.
4// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
5
6// This program is free software: you can redistribute it and/or modify
7// it under the terms of the GNU General Public License as published by
8// the Free Software Foundation, either version 3 of the License, or
9// (at your option) any later version.
10
11// This program is distributed in the hope that it will be useful,
12// but WITHOUT ANY WARRANTY; without even the implied warranty of
13// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14// GNU General Public License for more details.
15
16// You should have received a copy of the GNU General Public License
17// along with this program. If not, see <https://www.gnu.org/licenses/>.
18
19//! Substrate service. Starts a thread that spins up the network, client, and extrinsic pool.
20//! Manages communication between them.
21
22#![warn(missing_docs)]
23#![recursion_limit = "1024"]
24
25pub mod chain_ops;
26pub mod client;
27pub mod config;
28pub mod error;
29
30mod builder;
31mod metrics;
32mod task_manager;
33
34use crate::config::Multiaddr;
35use std::{
36	collections::HashMap,
37	net::{Ipv4Addr, Ipv6Addr, SocketAddr, SocketAddrV4, SocketAddrV6},
38};
39
40use codec::{Decode, Encode};
41use futures::{pin_mut, FutureExt, StreamExt};
42use jsonrpsee::RpcModule;
43use log::{debug, error, trace, warn};
44use sc_client_api::{blockchain::HeaderBackend, BlockBackend, BlockchainEvents, ProofProvider};
45use sc_network::{
46	config::MultiaddrWithPeerId, service::traits::NetworkService, NetworkBackend, NetworkBlock,
47	NetworkPeers, NetworkStateInfo,
48};
49use sc_network_sync::SyncingService;
50use sc_network_types::PeerId;
51use sc_rpc_server::Server;
52use sc_utils::mpsc::TracingUnboundedReceiver;
53use sp_blockchain::HeaderMetadata;
54use sp_consensus::SyncOracle;
55use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
56
57pub use self::{
58	builder::{
59		build_default_block_downloader, build_default_syncing_engine, build_network,
60		build_network_advanced, build_polkadot_syncing_strategy, gen_rpc_module, init_telemetry,
61		new_client, new_db_backend, new_full_client, new_full_parts, new_full_parts_record_import,
62		new_full_parts_with_genesis_builder, new_wasm_executor,
63		propagate_transaction_notifications, spawn_tasks, BuildNetworkAdvancedParams,
64		BuildNetworkParams, DefaultSyncingEngineConfig, KeystoreContainer, SpawnTasksParams,
65		TFullBackend, TFullCallExecutor, TFullClient,
66	},
67	client::{ClientConfig, LocalCallExecutor},
68	error::Error,
69	metrics::MetricsService,
70};
71#[allow(deprecated)]
72pub use builder::new_native_or_wasm_executor;
73
74pub use sc_chain_spec::{
75	construct_genesis_block, resolve_state_version_from_wasm, BuildGenesisBlock,
76	GenesisBlockBuilder,
77};
78
79pub use config::{
80	BasePath, BlocksPruning, Configuration, DatabaseSource, PruningMode, Role, RpcMethods, TaskType,
81};
82pub use sc_chain_spec::{
83	ChainSpec, ChainType, Extension as ChainSpecExtension, GenericChainSpec, NoExtension,
84	Properties,
85};
86
87use crate::config::RpcConfiguration;
88use prometheus_endpoint::Registry;
89pub use sc_consensus::ImportQueue;
90pub use sc_executor::NativeExecutionDispatch;
91pub use sc_network_sync::WarpSyncConfig;
92#[doc(hidden)]
93pub use sc_network_transactions::config::{TransactionImport, TransactionImportFuture};
94pub use sc_rpc::{RandomIntegerSubscriptionId, RandomStringSubscriptionId};
95pub use sc_tracing::TracingReceiver;
96pub use sc_transaction_pool::TransactionPoolOptions;
97pub use sc_transaction_pool_api::{error::IntoPoolError, InPoolTransaction, TransactionPool};
98#[doc(hidden)]
99pub use std::{ops::Deref, result::Result, sync::Arc};
100pub use task_manager::{
101	SpawnEssentialTaskHandle, SpawnTaskHandle, Task, TaskManager, TaskRegistry, DEFAULT_GROUP_NAME,
102};
103use tokio::runtime::Handle;
104
105const DEFAULT_PROTOCOL_ID: &str = "sup";
106
107/// A running RPC service that can perform in-memory RPC queries.
108#[derive(Clone)]
109pub struct RpcHandlers {
110	// This is legacy and may be removed at some point, it was for WASM stuff before smoldot was a
111	// thing. https://github.com/paritytech/polkadot-sdk/pull/5038#discussion_r1694971805
112	rpc_module: Arc<RpcModule<()>>,
113
114	// This can be used to introspect the port the RPC server is listening on. SDK consumers are
115	// depending on this and it should be supported even if in-memory query support is removed.
116	listen_addresses: Vec<Multiaddr>,
117}
118
119impl RpcHandlers {
120	/// Create PRC handlers instance.
121	pub fn new(rpc_module: Arc<RpcModule<()>>, listen_addresses: Vec<Multiaddr>) -> Self {
122		Self { rpc_module, listen_addresses }
123	}
124
125	/// Starts an RPC query.
126	///
127	/// The query is passed as a string and must be valid JSON-RPC request object.
128	///
129	/// Returns a response and a stream if the call successful, fails if the
130	/// query could not be decoded as a JSON-RPC request object.
131	///
132	/// If the request subscribes you to events, the `stream` can be used to
133	/// retrieve the events.
134	pub async fn rpc_query(
135		&self,
136		json_query: &str,
137	) -> Result<(String, tokio::sync::mpsc::Receiver<String>), serde_json::Error> {
138		// Because `tokio::sync::mpsc::channel` is used under the hood
139		// it will panic if it's set to usize::MAX.
140		//
141		// This limit is used to prevent panics and is large enough.
142		const TOKIO_MPSC_MAX_SIZE: usize = tokio::sync::Semaphore::MAX_PERMITS;
143
144		self.rpc_module.raw_json_request(json_query, TOKIO_MPSC_MAX_SIZE).await
145	}
146
147	/// Provides access to the underlying `RpcModule`
148	pub fn handle(&self) -> Arc<RpcModule<()>> {
149		self.rpc_module.clone()
150	}
151
152	/// Provides access to listen addresses
153	pub fn listen_addresses(&self) -> &[Multiaddr] {
154		&self.listen_addresses[..]
155	}
156}
157
158/// An incomplete set of chain components, but enough to run the chain ops subcommands.
159pub struct PartialComponents<Client, Backend, SelectChain, ImportQueue, TransactionPool, Other> {
160	/// A shared client instance.
161	pub client: Arc<Client>,
162	/// A shared backend instance.
163	pub backend: Arc<Backend>,
164	/// The chain task manager.
165	pub task_manager: TaskManager,
166	/// A keystore container instance.
167	pub keystore_container: KeystoreContainer,
168	/// A chain selection algorithm instance.
169	pub select_chain: SelectChain,
170	/// An import queue.
171	pub import_queue: ImportQueue,
172	/// A shared transaction pool.
173	pub transaction_pool: Arc<TransactionPool>,
174	/// Everything else that needs to be passed into the main build function.
175	pub other: Other,
176}
177
178/// Builds a future that continuously polls the network.
179async fn build_network_future<
180	B: BlockT,
181	C: BlockchainEvents<B>
182		+ HeaderBackend<B>
183		+ BlockBackend<B>
184		+ HeaderMetadata<B, Error = sp_blockchain::Error>
185		+ ProofProvider<B>
186		+ Send
187		+ Sync
188		+ 'static,
189	H: sc_network_common::ExHashT,
190	N: NetworkBackend<B, <B as BlockT>::Hash>,
191>(
192	network: N,
193	client: Arc<C>,
194	sync_service: Arc<SyncingService<B>>,
195	announce_imported_blocks: bool,
196) {
197	let mut imported_blocks_stream = client.import_notification_stream().fuse();
198
199	// Stream of finalized blocks reported by the client.
200	let mut finality_notification_stream = client.finality_notification_stream().fuse();
201
202	let network_run = network.run().fuse();
203	pin_mut!(network_run);
204
205	loop {
206		futures::select! {
207			// List of blocks that the client has imported.
208			notification = imported_blocks_stream.next() => {
209				let notification = match notification {
210					Some(n) => n,
211					// If this stream is shut down, that means the client has shut down, and the
212					// most appropriate thing to do for the network future is to shut down too.
213					None => {
214						debug!("Block import stream has terminated, shutting down the network future.");
215						return
216					},
217				};
218
219				if announce_imported_blocks {
220					sync_service.announce_block(notification.hash, None);
221				}
222
223				if notification.is_new_best {
224					sync_service.new_best_block_imported(
225						notification.hash,
226						*notification.header.number(),
227					);
228				}
229			}
230
231			// List of blocks that the client has finalized.
232			notification = finality_notification_stream.select_next_some() => {
233				sync_service.on_block_finalized(notification.hash, notification.header);
234			}
235
236			// Drive the network. Shut down the network future if `NetworkWorker` has terminated.
237			_ = network_run => {
238				debug!("`NetworkWorker` has terminated, shutting down the network future.");
239				return
240			}
241		}
242	}
243}
244
245/// Builds a future that processes system RPC requests.
246pub async fn build_system_rpc_future<
247	B: BlockT,
248	C: BlockchainEvents<B>
249		+ HeaderBackend<B>
250		+ BlockBackend<B>
251		+ HeaderMetadata<B, Error = sp_blockchain::Error>
252		+ ProofProvider<B>
253		+ Send
254		+ Sync
255		+ 'static,
256	H: sc_network_common::ExHashT,
257>(
258	role: Role,
259	network_service: Arc<dyn NetworkService>,
260	sync_service: Arc<SyncingService<B>>,
261	client: Arc<C>,
262	mut rpc_rx: TracingUnboundedReceiver<sc_rpc::system::Request<B>>,
263	should_have_peers: bool,
264) {
265	// Current best block at initialization, to report to the RPC layer.
266	let starting_block = client.info().best_number;
267
268	loop {
269		// Answer incoming RPC requests.
270		let Some(req) = rpc_rx.next().await else {
271			debug!("RPC requests stream has terminated, shutting down the system RPC future.");
272			return
273		};
274
275		match req {
276			sc_rpc::system::Request::Health(sender) => match sync_service.peers_info().await {
277				Ok(info) => {
278					let _ = sender.send(sc_rpc::system::Health {
279						peers: info.len(),
280						is_syncing: sync_service.is_major_syncing(),
281						should_have_peers,
282					});
283				},
284				Err(_) => log::error!("`SyncingEngine` shut down"),
285			},
286			sc_rpc::system::Request::LocalPeerId(sender) => {
287				let _ = sender.send(network_service.local_peer_id().to_base58());
288			},
289			sc_rpc::system::Request::LocalListenAddresses(sender) => {
290				let peer_id = (network_service.local_peer_id()).into();
291				let p2p_proto_suffix = sc_network::multiaddr::Protocol::P2p(peer_id);
292				let addresses = network_service
293					.listen_addresses()
294					.iter()
295					.map(|addr| addr.clone().with(p2p_proto_suffix.clone()).to_string())
296					.collect();
297				let _ = sender.send(addresses);
298			},
299			sc_rpc::system::Request::Peers(sender) => match sync_service.peers_info().await {
300				Ok(info) => {
301					let _ = sender.send(
302						info.into_iter()
303							.map(|(peer_id, p)| sc_rpc::system::PeerInfo {
304								peer_id: peer_id.to_base58(),
305								roles: format!("{:?}", p.roles),
306								best_hash: p.best_hash,
307								best_number: p.best_number,
308							})
309							.collect(),
310					);
311				},
312				Err(_) => log::error!("`SyncingEngine` shut down"),
313			},
314			sc_rpc::system::Request::NetworkState(sender) => {
315				let network_state = network_service.network_state().await;
316				if let Ok(network_state) = network_state {
317					if let Ok(network_state) = serde_json::to_value(network_state) {
318						let _ = sender.send(network_state);
319					}
320				} else {
321					break
322				}
323			},
324			sc_rpc::system::Request::NetworkAddReservedPeer(peer_addr, sender) => {
325				let result = match MultiaddrWithPeerId::try_from(peer_addr) {
326					Ok(peer) => network_service.add_reserved_peer(peer),
327					Err(err) => Err(err.to_string()),
328				};
329				let x = result.map_err(sc_rpc::system::error::Error::MalformattedPeerArg);
330				let _ = sender.send(x);
331			},
332			sc_rpc::system::Request::NetworkRemoveReservedPeer(peer_id, sender) => {
333				let _ = match peer_id.parse::<PeerId>() {
334					Ok(peer_id) => {
335						network_service.remove_reserved_peer(peer_id);
336						sender.send(Ok(()))
337					},
338					Err(e) => sender.send(Err(sc_rpc::system::error::Error::MalformattedPeerArg(
339						e.to_string(),
340					))),
341				};
342			},
343			sc_rpc::system::Request::NetworkReservedPeers(sender) => {
344				let Ok(reserved_peers) = network_service.reserved_peers().await else {
345					break;
346				};
347
348				let _ =
349					sender.send(reserved_peers.iter().map(|peer_id| peer_id.to_base58()).collect());
350			},
351			sc_rpc::system::Request::NodeRoles(sender) => {
352				use sc_rpc::system::NodeRole;
353
354				let node_role = match role {
355					Role::Authority { .. } => NodeRole::Authority,
356					Role::Full => NodeRole::Full,
357				};
358
359				let _ = sender.send(vec![node_role]);
360			},
361			sc_rpc::system::Request::SyncState(sender) => {
362				use sc_rpc::system::SyncState;
363
364				match sync_service.status().await.map(|status| status.best_seen_block) {
365					Ok(best_seen_block) => {
366						let best_number = client.info().best_number;
367						let _ = sender.send(SyncState {
368							starting_block,
369							current_block: best_number,
370							highest_block: best_seen_block.unwrap_or(best_number),
371						});
372					},
373					Err(_) => log::error!("`SyncingEngine` shut down"),
374				}
375			},
376		}
377	}
378
379	debug!("`NetworkWorker` has terminated, shutting down the system RPC future.");
380}
381
382/// Starts RPC servers.
383pub fn start_rpc_servers<R>(
384	rpc_configuration: &RpcConfiguration,
385	registry: Option<&Registry>,
386	tokio_handle: &Handle,
387	gen_rpc_module: R,
388	rpc_id_provider: Option<Box<dyn sc_rpc_server::SubscriptionIdProvider>>,
389) -> Result<Server, error::Error>
390where
391	R: Fn() -> Result<RpcModule<()>, Error>,
392{
393	let endpoints: Vec<sc_rpc_server::RpcEndpoint> = if let Some(endpoints) =
394		rpc_configuration.addr.as_ref()
395	{
396		endpoints.clone()
397	} else {
398		let ipv6 =
399			SocketAddr::V6(SocketAddrV6::new(Ipv6Addr::LOCALHOST, rpc_configuration.port, 0, 0));
400		let ipv4 = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, rpc_configuration.port));
401
402		vec![
403			sc_rpc_server::RpcEndpoint {
404				batch_config: rpc_configuration.batch_config,
405				cors: rpc_configuration.cors.clone(),
406				listen_addr: ipv4,
407				max_buffer_capacity_per_connection: rpc_configuration.message_buffer_capacity,
408				max_connections: rpc_configuration.max_connections,
409				max_payload_in_mb: rpc_configuration.max_request_size,
410				max_payload_out_mb: rpc_configuration.max_response_size,
411				max_subscriptions_per_connection: rpc_configuration.max_subs_per_conn,
412				rpc_methods: rpc_configuration.methods.into(),
413				rate_limit: rpc_configuration.rate_limit,
414				rate_limit_trust_proxy_headers: rpc_configuration.rate_limit_trust_proxy_headers,
415				rate_limit_whitelisted_ips: rpc_configuration.rate_limit_whitelisted_ips.clone(),
416				retry_random_port: true,
417				is_optional: false,
418			},
419			sc_rpc_server::RpcEndpoint {
420				batch_config: rpc_configuration.batch_config,
421				cors: rpc_configuration.cors.clone(),
422				listen_addr: ipv6,
423				max_buffer_capacity_per_connection: rpc_configuration.message_buffer_capacity,
424				max_connections: rpc_configuration.max_connections,
425				max_payload_in_mb: rpc_configuration.max_request_size,
426				max_payload_out_mb: rpc_configuration.max_response_size,
427				max_subscriptions_per_connection: rpc_configuration.max_subs_per_conn,
428				rpc_methods: rpc_configuration.methods.into(),
429				rate_limit: rpc_configuration.rate_limit,
430				rate_limit_trust_proxy_headers: rpc_configuration.rate_limit_trust_proxy_headers,
431				rate_limit_whitelisted_ips: rpc_configuration.rate_limit_whitelisted_ips.clone(),
432				retry_random_port: true,
433				is_optional: true,
434			},
435		]
436	};
437
438	let metrics = sc_rpc_server::RpcMetrics::new(registry)?;
439	let rpc_api = gen_rpc_module()?;
440
441	let server_config = sc_rpc_server::Config {
442		endpoints,
443		rpc_api,
444		metrics,
445		id_provider: rpc_id_provider,
446		tokio_handle: tokio_handle.clone(),
447	};
448
449	// TODO: https://github.com/paritytech/substrate/issues/13773
450	//
451	// `block_in_place` is a hack to allow callers to call `block_on` prior to
452	// calling `start_rpc_servers`.
453	match tokio::task::block_in_place(|| {
454		tokio_handle.block_on(sc_rpc_server::start_server(server_config))
455	}) {
456		Ok(server) => Ok(server),
457		Err(e) => Err(Error::Application(e)),
458	}
459}
460
461/// Transaction pool adapter.
462pub struct TransactionPoolAdapter<C, P> {
463	pool: Arc<P>,
464	client: Arc<C>,
465}
466
467impl<C, P> TransactionPoolAdapter<C, P> {
468	/// Constructs a new instance of [`TransactionPoolAdapter`].
469	pub fn new(pool: Arc<P>, client: Arc<C>) -> Self {
470		Self { pool, client }
471	}
472}
473
474/// Get transactions for propagation.
475///
476/// Function extracted to simplify the test and prevent creating `ServiceFactory`.
477fn transactions_to_propagate<Pool, B, H, E>(pool: &Pool) -> Vec<(H, Arc<B::Extrinsic>)>
478where
479	Pool: TransactionPool<Block = B, Hash = H, Error = E>,
480	B: BlockT,
481	H: std::hash::Hash + Eq + sp_runtime::traits::Member + sp_runtime::traits::MaybeSerialize,
482	E: IntoPoolError + From<sc_transaction_pool_api::error::Error>,
483{
484	pool.ready()
485		.filter(|t| t.is_propagable())
486		.map(|t| {
487			let hash = t.hash().clone();
488			let ex = t.data().clone();
489			(hash, ex)
490		})
491		.collect()
492}
493
494impl<B, H, C, Pool, E> sc_network_transactions::config::TransactionPool<H, B>
495	for TransactionPoolAdapter<C, Pool>
496where
497	C: HeaderBackend<B>
498		+ BlockBackend<B>
499		+ HeaderMetadata<B, Error = sp_blockchain::Error>
500		+ ProofProvider<B>
501		+ Send
502		+ Sync
503		+ 'static,
504	Pool: 'static + TransactionPool<Block = B, Hash = H, Error = E>,
505	B: BlockT,
506	H: std::hash::Hash + Eq + sp_runtime::traits::Member + sp_runtime::traits::MaybeSerialize,
507	E: 'static + IntoPoolError + From<sc_transaction_pool_api::error::Error>,
508{
509	fn transactions(&self) -> Vec<(H, Arc<B::Extrinsic>)> {
510		transactions_to_propagate(&*self.pool)
511	}
512
513	fn hash_of(&self, transaction: &B::Extrinsic) -> H {
514		self.pool.hash_of(transaction)
515	}
516
517	fn import(&self, transaction: B::Extrinsic) -> TransactionImportFuture {
518		let encoded = transaction.encode();
519		let uxt = match Decode::decode(&mut &encoded[..]) {
520			Ok(uxt) => uxt,
521			Err(e) => {
522				debug!(target: sc_transaction_pool::LOG_TARGET, "Transaction invalid: {:?}", e);
523				return Box::pin(futures::future::ready(TransactionImport::Bad))
524			},
525		};
526
527		let start = std::time::Instant::now();
528		let pool = self.pool.clone();
529		let client = self.client.clone();
530		Box::pin(async move {
531			match pool
532				.submit_one(
533					client.info().best_hash,
534					sc_transaction_pool_api::TransactionSource::External,
535					uxt,
536				)
537				.await
538			{
539				Ok(_) => {
540					let elapsed = start.elapsed();
541					trace!(target: sc_transaction_pool::LOG_TARGET, "import transaction: {elapsed:?}");
542					TransactionImport::NewGood
543				},
544				Err(e) => match e.into_pool_error() {
545					Ok(sc_transaction_pool_api::error::Error::AlreadyImported(_)) =>
546						TransactionImport::KnownGood,
547					Ok(_) => TransactionImport::Bad,
548					Err(_) => {
549						// it is not bad at least, just some internal node logic error, so peer is
550						// innocent.
551						TransactionImport::KnownGood
552					},
553				},
554			}
555		})
556	}
557
558	fn on_broadcasted(&self, propagations: HashMap<H, Vec<String>>) {
559		self.pool.on_broadcasted(propagations)
560	}
561
562	fn transaction(&self, hash: &H) -> Option<Arc<B::Extrinsic>> {
563		self.pool.ready_transaction(hash).and_then(
564			// Only propagable transactions should be resolved for network service.
565			|tx| tx.is_propagable().then(|| tx.data().clone()),
566		)
567	}
568}
569
570#[cfg(test)]
571mod tests {
572	use super::*;
573	use futures::executor::block_on;
574	use sc_transaction_pool::BasicPool;
575	use sp_consensus::SelectChain;
576	use substrate_test_runtime_client::{
577		prelude::*,
578		runtime::{ExtrinsicBuilder, Transfer, TransferData},
579	};
580
581	#[test]
582	fn should_not_propagate_transactions_that_are_marked_as_such() {
583		// given
584		let (client, longest_chain) = TestClientBuilder::new().build_with_longest_chain();
585		let client = Arc::new(client);
586		let spawner = sp_core::testing::TaskExecutor::new();
587		let pool = Arc::from(BasicPool::new_full(
588			Default::default(),
589			true.into(),
590			None,
591			spawner,
592			client.clone(),
593		));
594		let source = sp_runtime::transaction_validity::TransactionSource::External;
595		let best = block_on(longest_chain.best_chain()).unwrap();
596		let transaction = Transfer {
597			amount: 5,
598			nonce: 0,
599			from: Sr25519Keyring::Alice.into(),
600			to: Sr25519Keyring::Bob.into(),
601		}
602		.into_unchecked_extrinsic();
603		block_on(pool.submit_one(best.hash(), source, transaction.clone())).unwrap();
604		block_on(pool.submit_one(
605			best.hash(),
606			source,
607			ExtrinsicBuilder::new_call_do_not_propagate().nonce(1).build(),
608		))
609		.unwrap();
610		assert_eq!(pool.status().ready, 2);
611
612		// when
613		let transactions = transactions_to_propagate(&*pool);
614
615		// then
616		assert_eq!(transactions.len(), 1);
617		assert!(TransferData::try_from(&*transactions[0].1).is_ok());
618	}
619}