1#![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#[derive(Clone)]
109pub struct RpcHandlers {
110 rpc_module: Arc<RpcModule<()>>,
113
114 listen_addresses: Vec<Multiaddr>,
117}
118
119impl RpcHandlers {
120 pub fn new(rpc_module: Arc<RpcModule<()>>, listen_addresses: Vec<Multiaddr>) -> Self {
122 Self { rpc_module, listen_addresses }
123 }
124
125 pub async fn rpc_query(
135 &self,
136 json_query: &str,
137 ) -> Result<(String, tokio::sync::mpsc::Receiver<String>), serde_json::Error> {
138 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 pub fn handle(&self) -> Arc<RpcModule<()>> {
149 self.rpc_module.clone()
150 }
151
152 pub fn listen_addresses(&self) -> &[Multiaddr] {
154 &self.listen_addresses[..]
155 }
156}
157
158pub struct PartialComponents<Client, Backend, SelectChain, ImportQueue, TransactionPool, Other> {
160 pub client: Arc<Client>,
162 pub backend: Arc<Backend>,
164 pub task_manager: TaskManager,
166 pub keystore_container: KeystoreContainer,
168 pub select_chain: SelectChain,
170 pub import_queue: ImportQueue,
172 pub transaction_pool: Arc<TransactionPool>,
174 pub other: Other,
176}
177
178async 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 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 notification = imported_blocks_stream.next() => {
209 let notification = match notification {
210 Some(n) => n,
211 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 notification = finality_notification_stream.select_next_some() => {
233 sync_service.on_block_finalized(notification.hash, notification.header);
234 }
235
236 _ = network_run => {
238 debug!("`NetworkWorker` has terminated, shutting down the network future.");
239 return
240 }
241 }
242 }
243}
244
245pub 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 let starting_block = client.info().best_number;
267
268 loop {
269 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
382pub 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 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
461pub struct TransactionPoolAdapter<C, P> {
463 pool: Arc<P>,
464 client: Arc<C>,
465}
466
467impl<C, P> TransactionPoolAdapter<C, P> {
468 pub fn new(pool: Arc<P>, client: Arc<C>) -> Self {
470 Self { pool, client }
471 }
472}
473
474fn 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 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 |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 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 let transactions = transactions_to_propagate(&*pool);
614
615 assert_eq!(transactions.len(), 1);
617 assert!(TransferData::try_from(&*transactions[0].1).is_ok());
618 }
619}