diff --git a/Cargo.lock b/Cargo.lock index 5954e32ed15..1ceafe58163 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -379,20 +379,6 @@ version = "1.1.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d468802bab17cbc0cc575e9b053f41e72aa36bfa6b7f55e3529ffa43161b97fa" -[[package]] -name = "backoff" -version = "0.4.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b62ddb9cb1ec0a098ad4bbf9344d0713fa193ae1a80af55febcff2627b6a00c1" -dependencies = [ - "futures-core", - "getrandom 0.2.3", - "instant", - "pin-project-lite 0.2.9", - "rand 0.8.5", - "tokio", -] - [[package]] name = "backtrace" version = "0.3.64" @@ -1797,12 +1783,16 @@ dependencies = [ "cumulus-client-consensus-common", "cumulus-client-pov-recovery", "cumulus-primitives-core", + "cumulus-relay-chain-inprocess-interface", "cumulus-relay-chain-interface", + "cumulus-relay-chain-minimal-node", "parking_lot 0.12.1", "polkadot-primitives", "sc-client-api", "sc-consensus", "sc-service", + "sc-sysinfo", + "sc-telemetry", "sp-api", "sp-blockchain", "sp-consensus", @@ -2082,6 +2072,7 @@ dependencies = [ "sp-blockchain", "sp-state-machine", "thiserror", + "tokio", ] [[package]] @@ -2128,21 +2119,22 @@ name = "cumulus-relay-chain-rpc-interface" version = "0.1.0" dependencies = [ "async-trait", - "backoff", "cumulus-primitives-core", "cumulus-relay-chain-interface", "futures", "futures-timer", "jsonrpsee", + "lru", "parity-scale-codec", "polkadot-service", "sc-client-api", "sc-rpc-api", + "serde", + "serde_json", "sp-api", "sp-authority-discovery", "sp-consensus-babe", "sp-core", - "sp-runtime", "sp-state-machine", "sp-storage", "tokio", @@ -6551,10 +6543,7 @@ dependencies = [ "cumulus-client-service", "cumulus-primitives-core", "cumulus-primitives-parachain-inherent", - "cumulus-relay-chain-inprocess-interface", "cumulus-relay-chain-interface", - "cumulus-relay-chain-minimal-node", - "cumulus-relay-chain-rpc-interface", "frame-benchmarking", "frame-benchmarking-cli", "jsonrpsee", @@ -7780,10 +7769,7 @@ dependencies = [ "cumulus-client-service", "cumulus-primitives-core", "cumulus-primitives-parachain-inherent", - "cumulus-relay-chain-inprocess-interface", "cumulus-relay-chain-interface", - "cumulus-relay-chain-minimal-node", - "cumulus-relay-chain-rpc-interface", "frame-benchmarking", "frame-benchmarking-cli", "futures", @@ -10584,9 +10570,9 @@ dependencies = [ [[package]] name = "serde_json" -version = "1.0.85" +version = "1.0.89" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e55a28e3aaef9d5ce0506d0a14dbba8054ddc7e499ef522dd8b26859ec9d4a44" +checksum = "020ff22c755c2ed3f8cf162dbb41a7268d934702f3ed3631656ea597e08fc3db" dependencies = [ "itoa 1.0.4", "ryu", @@ -12611,7 +12597,7 @@ checksum = "97fee6b57c6a41524a810daee9286c02d7752c4253064d0b05472833a438f675" dependencies = [ "cfg-if", "digest 0.10.3", - "rand 0.8.5", + "rand 0.7.3", "static_assertions", ] @@ -13724,9 +13710,9 @@ dependencies = [ [[package]] name = "yamux" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0c0608f53c1dc0bad505d03a34bbd49fbf2ad7b51eb036123e896365532745a1" +checksum = "e5d9ba232399af1783a58d8eb26f6b5006fbefe2dc9ef36bd283324792d03ea5" dependencies = [ "futures", "log", diff --git a/client/cli/src/lib.rs b/client/cli/src/lib.rs index ae7943f99aa..edf11d72a0b 100644 --- a/client/cli/src/lib.rs +++ b/client/cli/src/lib.rs @@ -293,9 +293,11 @@ pub struct RunCmd { /// EXPERIMENTAL: Specify an URL to a relay chain full node to communicate with. #[arg( long, - value_parser = validate_relay_chain_url + value_parser = validate_relay_chain_url, + num_args = 0.., + alias = "relay-chain-rpc-url" )] - pub relay_chain_rpc_url: Option, + pub relay_chain_rpc_urls: Vec, } impl RunCmd { @@ -310,7 +312,7 @@ impl RunCmd { /// Create [`CollatorOptions`] representing options only relevant to parachain collator nodes pub fn collator_options(&self) -> CollatorOptions { - CollatorOptions { relay_chain_rpc_url: self.relay_chain_rpc_url.clone() } + CollatorOptions { relay_chain_rpc_urls: self.relay_chain_rpc_urls.clone() } } } @@ -318,7 +320,7 @@ impl RunCmd { #[derive(Clone, Debug)] pub struct CollatorOptions { /// Location of relay chain full node - pub relay_chain_rpc_url: Option, + pub relay_chain_rpc_urls: Vec, } /// A non-redundant version of the `RunCmd` that sets the `validator` field when the diff --git a/client/relay-chain-interface/Cargo.toml b/client/relay-chain-interface/Cargo.toml index c59d3a528a7..08a751cd961 100644 --- a/client/relay-chain-interface/Cargo.toml +++ b/client/relay-chain-interface/Cargo.toml @@ -15,6 +15,7 @@ sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "mas sp-state-machine = { git = "https://github.com/paritytech/substrate", branch = "master" } sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" } +tokio = { version = "1.21.2", features = ["sync"] } futures = "0.3.25" async-trait = "0.1.59" thiserror = "1.0.37" diff --git a/client/relay-chain-interface/src/lib.rs b/client/relay-chain-interface/src/lib.rs index 4505ac70973..56de750b687 100644 --- a/client/relay-chain-interface/src/lib.rs +++ b/client/relay-chain-interface/src/lib.rs @@ -51,11 +51,11 @@ pub enum RelayChainError { BlockchainError(#[from] sp_blockchain::Error), #[error("State machine error occured: {0}")] StateMachineError(Box), - #[error("Unable to call RPC method '{0}' due to error: {1}")] - RpcCallError(String, JsonRpcError), + #[error("Unable to call RPC method '{0}'")] + RpcCallError(String), #[error("RPC Error: '{0}'")] JsonRpcError(#[from] JsonRpcError), - #[error("Unable to reach RpcStreamWorker: {0}")] + #[error("Unable to communicate with RPC worker: {0}")] WorkerCommunicationError(String), #[error("Scale codec deserialization error: {0}")] DeserializationError(CodecError), diff --git a/client/relay-chain-minimal-node/src/blockchain_rpc_client.rs b/client/relay-chain-minimal-node/src/blockchain_rpc_client.rs index bf1a3c9a38c..1ed896533a3 100644 --- a/client/relay-chain-minimal-node/src/blockchain_rpc_client.rs +++ b/client/relay-chain-minimal-node/src/blockchain_rpc_client.rs @@ -363,13 +363,13 @@ impl BlockChainRpcClient { pub async fn import_notification_stream( &self, ) -> RelayChainResult + Send>>> { - Ok(self.rpc_client.get_imported_heads_stream().await?.boxed()) + Ok(self.rpc_client.get_imported_heads_stream()?.boxed()) } pub async fn finality_notification_stream( &self, ) -> RelayChainResult + Send>>> { - Ok(self.rpc_client.get_finalized_heads_stream().await?.boxed()) + Ok(self.rpc_client.get_finalized_heads_stream()?.boxed()) } } diff --git a/client/relay-chain-minimal-node/src/collator_overseer.rs b/client/relay-chain-minimal-node/src/collator_overseer.rs index 2d3f1224ef3..7ed01193026 100644 --- a/client/relay-chain-minimal-node/src/collator_overseer.rs +++ b/client/relay-chain-minimal-node/src/collator_overseer.rs @@ -223,8 +223,14 @@ async fn forward_collator_events( f = finality.next() => { match f { Some(header) => { - tracing::info!(target: "minimal-polkadot-node", "Received finalized block via RPC: #{} ({})", header.number, header.hash()); - let block_info = BlockInfo { hash: header.hash(), parent_hash: header.parent_hash, number: header.number }; + tracing::info!( + target: "minimal-polkadot-node", + "Received finalized block via RPC: #{} ({} -> {})", + header.number, + header.parent_hash, + header.hash() + ); + let block_info = BlockInfo { hash: header.hash(), parent_hash: header.parent_hash, number: header.number }; handle.block_finalized(block_info).await; } None => return Err(RelayChainError::GenericError("Relay chain finality stream ended.".to_string())), @@ -233,8 +239,14 @@ async fn forward_collator_events( i = imports.next() => { match i { Some(header) => { - tracing::info!(target: "minimal-polkadot-node", "Received imported block via RPC: #{} ({})", header.number, header.hash()); - let block_info = BlockInfo { hash: header.hash(), parent_hash: header.parent_hash, number: header.number }; + tracing::info!( + target: "minimal-polkadot-node", + "Received imported block via RPC: #{} ({} -> {})", + header.number, + header.parent_hash, + header.hash() + ); + let block_info = BlockInfo { hash: header.hash(), parent_hash: header.parent_hash, number: header.number }; handle.block_imported(block_info).await; } None => return Err(RelayChainError::GenericError("Relay chain import stream ended.".to_string())), diff --git a/client/relay-chain-minimal-node/src/lib.rs b/client/relay-chain-minimal-node/src/lib.rs index 2aa5ce3d066..1c61a7998bb 100644 --- a/client/relay-chain-minimal-node/src/lib.rs +++ b/client/relay-chain-minimal-node/src/lib.rs @@ -85,7 +85,7 @@ fn build_authority_discovery_service( pub async fn build_minimal_relay_chain_node( polkadot_config: Configuration, task_manager: &mut TaskManager, - relay_chain_url: Url, + relay_chain_url: Vec, ) -> RelayChainResult<(Arc<(dyn RelayChainInterface + 'static)>, Option)> { let client = cumulus_relay_chain_rpc_interface::create_client_and_start_worker( relay_chain_url, diff --git a/client/relay-chain-rpc-interface/Cargo.toml b/client/relay-chain-rpc-interface/Cargo.toml index 4072c23b245..ed281690e0a 100644 --- a/client/relay-chain-rpc-interface/Cargo.toml +++ b/client/relay-chain-rpc-interface/Cargo.toml @@ -13,12 +13,11 @@ cumulus-relay-chain-interface = { path = "../relay-chain-interface" } sp-api = { git = "https://github.com/paritytech/substrate", branch = "master" } sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" } -sp-runtime = { git = "https://github.com/paritytech/substrate", branch = "master" } sp-consensus-babe = { git = "https://github.com/paritytech/substrate", branch = "master" } sp-authority-discovery = { git = "https://github.com/paritytech/substrate", branch = "master" } sp-state-machine = { git = "https://github.com/paritytech/substrate", branch = "master" } -sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" } sp-storage = { git = "https://github.com/paritytech/substrate", branch = "master" } +sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" } sc-rpc-api = { git = "https://github.com/paritytech/substrate", branch = "master" } tokio = { version = "1.23.0", features = ["sync"] } @@ -29,4 +28,6 @@ jsonrpsee = { version = "0.16.2", features = ["ws-client"] } tracing = "0.1.37" async-trait = "0.1.59" url = "2.3.1" -backoff = { version = "0.4.0", features = ["tokio"] } +serde_json = "1.0.87" +serde = "1.0.147" +lru = "0.8.1" diff --git a/client/relay-chain-rpc-interface/src/lib.rs b/client/relay-chain-rpc-interface/src/lib.rs index 1d35ec4e747..c8a1c4e46d9 100644 --- a/client/relay-chain-rpc-interface/src/lib.rs +++ b/client/relay-chain-rpc-interface/src/lib.rs @@ -19,7 +19,7 @@ use core::time::Duration; use cumulus_primitives_core::{ relay_chain::{ v2::{CommittedCandidateReceipt, OccupiedCoreAssumption, SessionIndex, ValidatorId}, - Hash as PHash, Header as PHeader, InboundHrmpMessage, + Hash as RelayHash, Header as RelayHeader, InboundHrmpMessage, }, InboundDownwardMessage, ParaId, PersistedValidationData, }; @@ -34,6 +34,7 @@ use std::pin::Pin; pub use url::Url; +mod reconnecting_ws_client; mod rpc_client; pub use rpc_client::{create_client_and_start_worker, RelayChainRpcClient}; @@ -58,7 +59,7 @@ impl RelayChainInterface for RelayChainRpcInterface { async fn retrieve_dmq_contents( &self, para_id: ParaId, - relay_parent: PHash, + relay_parent: RelayHash, ) -> RelayChainResult> { self.rpc_client.parachain_host_dmq_contents(para_id, relay_parent).await } @@ -66,7 +67,7 @@ impl RelayChainInterface for RelayChainRpcInterface { async fn retrieve_all_inbound_hrmp_channel_contents( &self, para_id: ParaId, - relay_parent: PHash, + relay_parent: RelayHash, ) -> RelayChainResult>> { self.rpc_client .parachain_host_inbound_hrmp_channels_contents(para_id, relay_parent) @@ -75,7 +76,7 @@ impl RelayChainInterface for RelayChainRpcInterface { async fn persisted_validation_data( &self, - hash: PHash, + hash: RelayHash, para_id: ParaId, occupied_core_assumption: OccupiedCoreAssumption, ) -> RelayChainResult> { @@ -86,7 +87,7 @@ impl RelayChainInterface for RelayChainRpcInterface { async fn candidate_pending_availability( &self, - hash: PHash, + hash: RelayHash, para_id: ParaId, ) -> RelayChainResult> { self.rpc_client @@ -94,31 +95,31 @@ impl RelayChainInterface for RelayChainRpcInterface { .await } - async fn session_index_for_child(&self, hash: PHash) -> RelayChainResult { + async fn session_index_for_child(&self, hash: RelayHash) -> RelayChainResult { self.rpc_client.parachain_host_session_index_for_child(hash).await } - async fn validators(&self, block_id: PHash) -> RelayChainResult> { + async fn validators(&self, block_id: RelayHash) -> RelayChainResult> { self.rpc_client.parachain_host_validators(block_id).await } async fn import_notification_stream( &self, - ) -> RelayChainResult + Send>>> { - let imported_headers_stream = self.rpc_client.get_imported_heads_stream().await?; + ) -> RelayChainResult + Send>>> { + let imported_headers_stream = self.rpc_client.get_imported_heads_stream()?; Ok(imported_headers_stream.boxed()) } async fn finality_notification_stream( &self, - ) -> RelayChainResult + Send>>> { - let imported_headers_stream = self.rpc_client.get_finalized_heads_stream().await?; + ) -> RelayChainResult + Send>>> { + let imported_headers_stream = self.rpc_client.get_finalized_heads_stream()?; Ok(imported_headers_stream.boxed()) } - async fn best_block_hash(&self) -> RelayChainResult { + async fn best_block_hash(&self) -> RelayChainResult { self.rpc_client.chain_get_head(None).await } @@ -132,7 +133,7 @@ impl RelayChainInterface for RelayChainRpcInterface { async fn get_storage_by_key( &self, - relay_parent: PHash, + relay_parent: RelayHash, key: &[u8], ) -> RelayChainResult> { let storage_key = StorageKey(key.to_vec()); @@ -144,7 +145,7 @@ impl RelayChainInterface for RelayChainRpcInterface { async fn prove_read( &self, - relay_parent: PHash, + relay_parent: RelayHash, relevant_keys: &Vec>, ) -> RelayChainResult { let cloned = relevant_keys.clone(); @@ -167,8 +168,8 @@ impl RelayChainInterface for RelayChainRpcInterface { /// 2. Check if the block is already in chain. If yes, succeed early. /// 3. Wait for the block to be imported via subscription. /// 4. If timeout is reached, we return an error. - async fn wait_for_block(&self, wait_for_hash: PHash) -> RelayChainResult<()> { - let mut head_stream = self.rpc_client.get_imported_heads_stream().await?; + async fn wait_for_block(&self, wait_for_hash: RelayHash) -> RelayChainResult<()> { + let mut head_stream = self.rpc_client.get_imported_heads_stream()?; if self.rpc_client.chain_get_header(Some(wait_for_hash)).await?.is_some() { return Ok(()) @@ -191,8 +192,8 @@ impl RelayChainInterface for RelayChainRpcInterface { async fn new_best_notification_stream( &self, - ) -> RelayChainResult + Send>>> { - let imported_headers_stream = self.rpc_client.get_best_heads_stream().await?; + ) -> RelayChainResult + Send>>> { + let imported_headers_stream = self.rpc_client.get_best_heads_stream()?; Ok(imported_headers_stream.boxed()) } } diff --git a/client/relay-chain-rpc-interface/src/reconnecting_ws_client.rs b/client/relay-chain-rpc-interface/src/reconnecting_ws_client.rs new file mode 100644 index 00000000000..3414dd652c6 --- /dev/null +++ b/client/relay-chain-rpc-interface/src/reconnecting_ws_client.rs @@ -0,0 +1,520 @@ +// Copyright 2022 Parity Technologies (UK) Ltd. +// This file is part of Cumulus. + +// Cumulus is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// Cumulus is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with Cumulus. If not, see . + +use cumulus_primitives_core::relay_chain::{ + BlockNumber as RelayBlockNumber, Header as RelayHeader, +}; +use cumulus_relay_chain_interface::{RelayChainError, RelayChainResult}; +use futures::{ + channel::{ + mpsc::{Receiver, Sender}, + oneshot::Sender as OneshotSender, + }, + future::BoxFuture, + stream::FuturesUnordered, + FutureExt, StreamExt, +}; +use jsonrpsee::{ + core::{ + client::{Client as JsonRpcClient, ClientT, Subscription, SubscriptionClientT}, + params::ArrayParams, + Error as JsonRpseeError, JsonValue, + }, + rpc_params, + ws_client::WsClientBuilder, +}; +use lru::LruCache; +use polkadot_service::TaskManager; +use std::{num::NonZeroUsize, sync::Arc}; +use tokio::sync::mpsc::{ + channel as tokio_channel, Receiver as TokioReceiver, Sender as TokioSender, +}; +use url::Url; + +const NOTIFICATION_CHANNEL_SIZE_LIMIT: usize = 20; +const LOG_TARGET: &str = "reconnecting-websocket-client"; + +/// Messages for communication between [`ReconnectingWsClient`] and [`ReconnectingWebsocketWorker`]. +#[derive(Debug)] +enum RpcDispatcherMessage { + RegisterBestHeadListener(Sender), + RegisterImportListener(Sender), + RegisterFinalizationListener(Sender), + Request(String, ArrayParams, OneshotSender>), +} + +/// Frontend for performing websocket requests. +/// Requests and stream requests are forwarded to [`ReconnectingWebsocketWorker`]. +#[derive(Debug, Clone)] +pub struct ReconnectingWsClient { + /// Channel to communicate with the RPC worker + to_worker_channel: TokioSender, +} + +impl ReconnectingWsClient { + /// Create a new websocket client frontend. + pub async fn new(urls: Vec, task_manager: &mut TaskManager) -> RelayChainResult { + tracing::debug!(target: LOG_TARGET, "Instantiating reconnecting websocket client"); + + let (worker, sender) = ReconnectingWebsocketWorker::new(urls).await; + + task_manager + .spawn_essential_handle() + .spawn("relay-chain-rpc-worker", None, worker.run()); + + Ok(Self { to_worker_channel: sender }) + } +} + +impl ReconnectingWsClient { + /// Perform a request via websocket connection. + pub async fn request(&self, method: &str, params: ArrayParams) -> Result + where + R: serde::de::DeserializeOwned, + { + let (tx, rx) = futures::channel::oneshot::channel(); + + let message = RpcDispatcherMessage::Request(method.into(), params, tx); + self.to_worker_channel.send(message).await.map_err(|err| { + RelayChainError::WorkerCommunicationError(format!( + "Unable to send message to RPC worker: {}", + err + )) + })?; + + let value = rx.await.map_err(|err| { + RelayChainError::WorkerCommunicationError(format!( + "Unexpected channel close on RPC worker side: {}", + err + )) + })??; + + serde_json::from_value(value) + .map_err(|_| RelayChainError::GenericError("Unable to deserialize value".to_string())) + } + + /// Get a stream of new best relay chain headers + pub fn get_best_heads_stream(&self) -> Result, RelayChainError> { + let (tx, rx) = + futures::channel::mpsc::channel::(NOTIFICATION_CHANNEL_SIZE_LIMIT); + self.send_register_message_to_worker(RpcDispatcherMessage::RegisterBestHeadListener(tx))?; + Ok(rx) + } + + /// Get a stream of finalized relay chain headers + pub fn get_finalized_heads_stream(&self) -> Result, RelayChainError> { + let (tx, rx) = + futures::channel::mpsc::channel::(NOTIFICATION_CHANNEL_SIZE_LIMIT); + self.send_register_message_to_worker(RpcDispatcherMessage::RegisterFinalizationListener( + tx, + ))?; + Ok(rx) + } + + /// Get a stream of all imported relay chain headers + pub fn get_imported_heads_stream(&self) -> Result, RelayChainError> { + let (tx, rx) = + futures::channel::mpsc::channel::(NOTIFICATION_CHANNEL_SIZE_LIMIT); + self.send_register_message_to_worker(RpcDispatcherMessage::RegisterImportListener(tx))?; + Ok(rx) + } + + fn send_register_message_to_worker( + &self, + message: RpcDispatcherMessage, + ) -> Result<(), RelayChainError> { + self.to_worker_channel + .try_send(message) + .map_err(|e| RelayChainError::WorkerCommunicationError(e.to_string())) + } +} + +/// Worker that should be used in combination with [`RelayChainRpcClient`]. Must be polled to distribute header notifications to listeners. +struct ReconnectingWebsocketWorker { + ws_urls: Vec, + /// Communication channel with the RPC client + client_receiver: TokioReceiver, + + /// Senders to distribute incoming header notifications to + imported_header_listeners: Vec>, + finalized_header_listeners: Vec>, + best_header_listeners: Vec>, +} + +fn distribute_header(header: RelayHeader, senders: &mut Vec>) { + senders.retain_mut(|e| { + match e.try_send(header.clone()) { + // Receiver has been dropped, remove Sender from list. + Err(error) if error.is_disconnected() => false, + // Channel is full. This should not happen. + // TODO: Improve error handling here + // https://github.com/paritytech/cumulus/issues/1482 + Err(error) => { + tracing::error!(target: LOG_TARGET, ?error, "Event distribution channel has reached its limit. This can lead to missed notifications."); + true + }, + _ => true, + } + }); +} + +/// Manages the active websocket client. +/// Responsible for creating request futures, subscription streams +/// and reconnections. +#[derive(Debug)] +struct ClientManager { + urls: Vec, + active_client: Arc, + active_index: usize, +} + +struct RelayChainSubscriptions { + import_subscription: Subscription, + finalized_subscription: Subscription, + best_subscription: Subscription, +} + +/// Try to find a new RPC server to connect to. +async fn connect_next_available_rpc_server( + urls: &Vec, + starting_position: usize, +) -> Result<(usize, Arc), ()> { + tracing::debug!(target: LOG_TARGET, starting_position, "Connecting to RPC server."); + for (counter, url) in urls.iter().cycle().skip(starting_position).take(urls.len()).enumerate() { + let index = (starting_position + counter) % urls.len(); + tracing::info!( + target: LOG_TARGET, + index, + ?url, + "Trying to connect to next external relaychain node.", + ); + if let Ok(ws_client) = WsClientBuilder::default().build(url).await { + return Ok((index, Arc::new(ws_client))) + }; + } + Err(()) +} + +impl ClientManager { + pub async fn new(urls: Vec) -> Result { + if urls.is_empty() { + return Err(()) + } + let active_client = connect_next_available_rpc_server(&urls, 0).await?; + Ok(Self { urls, active_client: active_client.1, active_index: active_client.0 }) + } + + pub async fn connect_to_new_rpc_server(&mut self) -> Result<(), ()> { + let new_active = + connect_next_available_rpc_server(&self.urls, self.active_index + 1).await?; + self.active_client = new_active.1; + self.active_index = new_active.0; + Ok(()) + } + + async fn get_subscriptions(&self) -> Result { + let import_subscription = self + .active_client + .subscribe::( + "chain_subscribeAllHeads", + rpc_params![], + "chain_unsubscribeAllHeads", + ) + .await + .map_err(|e| { + tracing::error!( + target: LOG_TARGET, + ?e, + "Unable to open `chain_subscribeAllHeads` subscription." + ); + e + })?; + let best_subscription = self + .active_client + .subscribe::( + "chain_subscribeNewHeads", + rpc_params![], + "chain_unsubscribeNewHeads", + ) + .await + .map_err(|e| { + tracing::error!( + target: LOG_TARGET, + ?e, + "Unable to open `chain_subscribeNewHeads` subscription." + ); + e + })?; + let finalized_subscription = self + .active_client + .subscribe::( + "chain_subscribeFinalizedHeads", + rpc_params![], + "chain_unsubscribeFinalizedHeads", + ) + .await + .map_err(|e| { + tracing::error!( + target: LOG_TARGET, + ?e, + "Unable to open `chain_subscribeFinalizedHeads` subscription." + ); + e + })?; + + Ok(RelayChainSubscriptions { + import_subscription, + best_subscription, + finalized_subscription, + }) + } + + /// Create a request future that performs an RPC request and sends the results to the caller. + /// In case of a dead websocket connection, it returns the original request parameters to + /// enable retries. + fn create_request( + &self, + method: String, + params: ArrayParams, + response_sender: OneshotSender>, + ) -> BoxFuture<'static, Result<(), RpcDispatcherMessage>> { + let future_client = self.active_client.clone(); + async move { + let resp = future_client.request(&method, params.clone()).await; + + // We should only return the original request in case + // the websocket connection is dead and requires a restart. + // Other errors should be forwarded to the request caller. + if let Err(JsonRpseeError::RestartNeeded(_)) = resp { + return Err(RpcDispatcherMessage::Request(method, params, response_sender)) + } + + if let Err(err) = response_sender.send(resp) { + tracing::debug!( + target: LOG_TARGET, + ?err, + "Recipient no longer interested in request result" + ); + } + Ok(()) + } + .boxed() + } +} + +enum ConnectionStatus { + Connected, + ReconnectRequired(Option), +} + +impl ReconnectingWebsocketWorker { + /// Create new worker. Returns the worker and a channel to register new listeners. + async fn new( + urls: Vec, + ) -> (ReconnectingWebsocketWorker, TokioSender) { + let (tx, rx) = tokio_channel(100); + let worker = ReconnectingWebsocketWorker { + ws_urls: urls, + client_receiver: rx, + imported_header_listeners: Vec::new(), + finalized_header_listeners: Vec::new(), + best_header_listeners: Vec::new(), + }; + (worker, tx) + } + + /// Reconnect via [`ClientManager`] and provide new notification streams. + async fn handle_reconnect( + &mut self, + client_manager: &mut ClientManager, + pending_requests: &mut FuturesUnordered< + BoxFuture<'static, Result<(), RpcDispatcherMessage>>, + >, + first_failed_request: Option, + ) -> Result { + let mut requests_to_retry = Vec::new(); + if let Some(req @ RpcDispatcherMessage::Request(_, _, _)) = first_failed_request { + requests_to_retry.push(req); + } + + // At this point, all pending requests will return an error since the + // websocket connection is dead. So draining the pending requests should be fast. + while !pending_requests.is_empty() { + if let Some(Err(req)) = pending_requests.next().await { + requests_to_retry.push(req); + } + } + + if client_manager.connect_to_new_rpc_server().await.is_err() { + return Err(format!("Unable to find valid external RPC server, shutting down.")) + }; + + for item in requests_to_retry.into_iter() { + if let RpcDispatcherMessage::Request(method, params, response_sender) = item { + pending_requests.push(client_manager.create_request( + method, + params, + response_sender, + )); + }; + } + + client_manager.get_subscriptions().await.map_err(|e| { + format!("Not able to create streams from newly connected RPC server, shutting down. err: {:?}", e) + }) + } + + /// Run this worker to drive notification streams. + /// The worker does the following: + /// - Listen for [`RpcDispatcherMessage`], perform requests and register new listeners for the notification streams + /// - Distribute incoming import, best head and finalization notifications to registered listeners. + /// If an error occurs during sending, the receiver has been closed and we remove the sender from the list. + /// - Find a new valid RPC server to connect to in case the websocket connection is terminated. + /// If the worker is not able to connec to an RPC server from the list, the worker shuts down. + async fn run(mut self) { + let mut pending_requests = FuturesUnordered::new(); + + let urls = std::mem::take(&mut self.ws_urls); + let Ok(mut client_manager) = ClientManager::new(urls).await else { + tracing::error!(target: LOG_TARGET, "No valid RPC url found. Stopping RPC worker."); + return; + }; + let Ok(mut subscriptions) = client_manager.get_subscriptions().await else { + tracing::error!(target: LOG_TARGET, "Unable to fetch subscriptions on initial connection."); + return; + }; + + let mut imported_blocks_cache = + LruCache::new(NonZeroUsize::new(40).expect("40 is nonzero; qed.")); + let mut should_reconnect = ConnectionStatus::Connected; + let mut last_seen_finalized_num: RelayBlockNumber = 0; + loop { + // This branch is taken if the websocket connection to the current RPC server is closed. + if let ConnectionStatus::ReconnectRequired(maybe_failed_request) = should_reconnect { + match self + .handle_reconnect( + &mut client_manager, + &mut pending_requests, + maybe_failed_request, + ) + .await + { + Ok(new_subscriptions) => { + subscriptions = new_subscriptions; + }, + Err(message) => { + tracing::error!( + target: LOG_TARGET, + message, + "Unable to reconnect, stopping worker." + ); + return + }, + } + should_reconnect = ConnectionStatus::Connected; + } + + tokio::select! { + evt = self.client_receiver.recv() => match evt { + Some(RpcDispatcherMessage::RegisterBestHeadListener(tx)) => { + self.best_header_listeners.push(tx); + }, + Some(RpcDispatcherMessage::RegisterImportListener(tx)) => { + self.imported_header_listeners.push(tx) + }, + Some(RpcDispatcherMessage::RegisterFinalizationListener(tx)) => { + self.finalized_header_listeners.push(tx) + }, + Some(RpcDispatcherMessage::Request(method, params, response_sender)) => { + pending_requests.push(client_manager.create_request(method, params, response_sender)); + }, + None => { + tracing::error!(target: LOG_TARGET, "RPC client receiver closed. Stopping RPC Worker."); + return; + } + }, + should_retry = pending_requests.next(), if !pending_requests.is_empty() => { + if let Some(Err(req)) = should_retry { + should_reconnect = ConnectionStatus::ReconnectRequired(Some(req)); + } + }, + import_event = subscriptions.import_subscription.next() => { + match import_event { + Some(Ok(header)) => { + let hash = header.hash(); + if imported_blocks_cache.contains(&hash) { + tracing::debug!( + target: LOG_TARGET, + number = header.number, + ?hash, + "Duplicate imported block header. This might happen after switching to a new RPC node. Skipping distribution." + ); + continue; + } + imported_blocks_cache.put(hash, ()); + distribute_header(header, &mut self.imported_header_listeners); + }, + None => { + tracing::error!(target: LOG_TARGET, "Subscription closed."); + should_reconnect = ConnectionStatus::ReconnectRequired(None); + }, + Some(Err(error)) => { + tracing::error!(target: LOG_TARGET, ?error, "Error in RPC subscription."); + should_reconnect = ConnectionStatus::ReconnectRequired(None); + }, + } + }, + best_header_event = subscriptions.best_subscription.next() => { + match best_header_event { + Some(Ok(header)) => distribute_header(header, &mut self.best_header_listeners), + None => { + tracing::error!(target: LOG_TARGET, "Subscription closed."); + should_reconnect = ConnectionStatus::ReconnectRequired(None); + }, + Some(Err(error)) => { + tracing::error!(target: LOG_TARGET, ?error, "Error in RPC subscription."); + should_reconnect = ConnectionStatus::ReconnectRequired(None); + }, + } + } + finalized_event = subscriptions.finalized_subscription.next() => { + match finalized_event { + Some(Ok(header)) if header.number > last_seen_finalized_num => { + last_seen_finalized_num = header.number; + distribute_header(header, &mut self.finalized_header_listeners); + }, + Some(Ok(header)) => { + tracing::debug!( + target: LOG_TARGET, + number = header.number, + last_seen_finalized_num, + "Duplicate finalized block header. This might happen after switching to a new RPC node. Skipping distribution." + ); + }, + None => { + tracing::error!(target: LOG_TARGET, "Subscription closed."); + should_reconnect = ConnectionStatus::ReconnectRequired(None); + }, + Some(Err(error)) => { + tracing::error!(target: LOG_TARGET, ?error, "Error in RPC subscription."); + should_reconnect = ConnectionStatus::ReconnectRequired(None); + }, + } + } + } + } + } +} diff --git a/client/relay-chain-rpc-interface/src/rpc_client.rs b/client/relay-chain-rpc-interface/src/rpc_client.rs index c7172018505..d8ef632b693 100644 --- a/client/relay-chain-rpc-interface/src/rpc_client.rs +++ b/client/relay-chain-rpc-interface/src/rpc_client.rs @@ -14,7 +14,7 @@ // You should have received a copy of the GNU General Public License // along with Cumulus. If not, see . -use backoff::{future::retry_notify, ExponentialBackoff}; +use crate::reconnecting_ws_client::ReconnectingWsClient; use cumulus_primitives_core::{ relay_chain::{ v2::{ @@ -23,211 +23,49 @@ use cumulus_primitives_core::{ PvfCheckStatement, ScrapedOnChainVotes, SessionIndex, SessionInfo, ValidationCode, ValidationCodeHash, ValidatorId, ValidatorIndex, ValidatorSignature, }, - CandidateHash, Hash as PHash, Header as PHeader, InboundHrmpMessage, + CandidateHash, Hash as RelayHash, Header as RelayHeader, InboundHrmpMessage, }, InboundDownwardMessage, ParaId, PersistedValidationData, }; use cumulus_relay_chain_interface::{RelayChainError, RelayChainResult}; -use futures::{ - channel::mpsc::{Receiver, Sender}, - StreamExt, -}; -use jsonrpsee::{ - core::{ - client::{Client as JsonRpcClient, ClientT, Subscription, SubscriptionClientT}, - params::ArrayParams, - Error as JsonRpseeError, - }, - rpc_params, - ws_client::WsClientBuilder, -}; +use futures::channel::mpsc::Receiver; +use jsonrpsee::{core::params::ArrayParams, rpc_params}; use parity_scale_codec::{Decode, Encode}; use polkadot_service::{BlockNumber, TaskManager}; use sc_client_api::StorageData; use sc_rpc_api::{state::ReadProof, system::Health}; +use serde::de::DeserializeOwned; use sp_api::RuntimeVersion; use sp_consensus_babe::Epoch; use sp_core::sp_std::collections::btree_map::BTreeMap; -use sp_runtime::DeserializeOwned; use sp_storage::StorageKey; -use std::sync::Arc; -use tokio::sync::mpsc::{ - channel as tokio_channel, Receiver as TokioReceiver, Sender as TokioSender, -}; - pub use url::Url; const LOG_TARGET: &str = "relay-chain-rpc-client"; -const NOTIFICATION_CHANNEL_SIZE_LIMIT: usize = 20; - /// Client that maps RPC methods and deserializes results #[derive(Clone)] pub struct RelayChainRpcClient { /// Websocket client to make calls - ws_client: Arc, - - /// Retry strategy that should be used for requests and subscriptions - retry_strategy: ExponentialBackoff, - - /// Channel to communicate with the RPC worker - to_worker_channel: TokioSender, -} - -/// Worker messages to register new notification listeners -#[derive(Clone, Debug)] -pub enum NotificationRegisterMessage { - RegisterBestHeadListener(Sender), - RegisterImportListener(Sender), - RegisterFinalizationListener(Sender), -} - -/// Worker that should be used in combination with [`RelayChainRpcClient`]. Must be polled to distribute header notifications to listeners. -struct RpcStreamWorker { - // Communication channel with the RPC client - client_receiver: TokioReceiver, - - // Senders to distribute incoming header notifications to - imported_header_listeners: Vec>, - finalized_header_listeners: Vec>, - best_header_listeners: Vec>, - - // Incoming notification subscriptions - rpc_imported_header_subscription: Subscription, - rpc_finalized_header_subscription: Subscription, - rpc_best_header_subscription: Subscription, + ws_client: ReconnectingWsClient, } /// Entry point to create [`RelayChainRpcClient`] and start a worker that distributes notifications. pub async fn create_client_and_start_worker( - url: Url, + urls: Vec, task_manager: &mut TaskManager, ) -> RelayChainResult { - tracing::info!(target: LOG_TARGET, url = %url.to_string(), "Initializing RPC Client"); - let ws_client = WsClientBuilder::default().build(url.as_str()).await?; - - let best_head_stream = RelayChainRpcClient::subscribe_new_best_heads(&ws_client).await?; - let finalized_head_stream = RelayChainRpcClient::subscribe_finalized_heads(&ws_client).await?; - let imported_head_stream = RelayChainRpcClient::subscribe_imported_heads(&ws_client).await?; - - let (worker, sender) = - RpcStreamWorker::new(imported_head_stream, best_head_stream, finalized_head_stream); - let client = RelayChainRpcClient::new(ws_client, sender).await?; + let ws_client = ReconnectingWsClient::new(urls, task_manager).await?; - task_manager - .spawn_essential_handle() - .spawn("relay-chain-rpc-worker", None, worker.run()); + let client = RelayChainRpcClient::new(ws_client).await?; Ok(client) } -fn handle_event_distribution( - event: Option>, - senders: &mut Vec>, -) -> Result<(), String> { - match event { - Some(Ok(header)) => { - senders.retain_mut(|e| { - match e.try_send(header.clone()) { - // Receiver has been dropped, remove Sender from list. - Err(error) if error.is_disconnected() => false, - // Channel is full. This should not happen. - // TODO: Improve error handling here - // https://github.com/paritytech/cumulus/issues/1482 - Err(error) => { - tracing::error!(target: LOG_TARGET, ?error, "Event distribution channel has reached its limit. This can lead to missed notifications."); - true - }, - _ => true, - } - }); - Ok(()) - }, - None => Err("RPC Subscription closed.".to_string()), - Some(Err(err)) => Err(format!("Error in RPC subscription: {}", err)), - } -} - -impl RpcStreamWorker { - /// Create new worker. Returns the worker and a channel to register new listeners. - fn new( - import_sub: Subscription, - best_sub: Subscription, - finalized_sub: Subscription, - ) -> (RpcStreamWorker, TokioSender) { - let (tx, rx) = tokio_channel(100); - let worker = RpcStreamWorker { - client_receiver: rx, - imported_header_listeners: Vec::new(), - finalized_header_listeners: Vec::new(), - best_header_listeners: Vec::new(), - rpc_imported_header_subscription: import_sub, - rpc_best_header_subscription: best_sub, - rpc_finalized_header_subscription: finalized_sub, - }; - (worker, tx) - } - - /// Run this worker to drive notification streams. - /// The worker does two things: - /// 1. Listen for `NotificationRegisterMessage` and register new listeners for the notification streams - /// 2. Distribute incoming import, best head and finalization notifications to registered listeners. - /// If an error occurs during sending, the receiver has been closed and we remove the sender from the list. - pub async fn run(mut self) { - let mut import_sub = self.rpc_imported_header_subscription.fuse(); - let mut best_head_sub = self.rpc_best_header_subscription.fuse(); - let mut finalized_sub = self.rpc_finalized_header_subscription.fuse(); - loop { - tokio::select! { - evt = self.client_receiver.recv() => match evt { - Some(NotificationRegisterMessage::RegisterBestHeadListener(tx)) => { - self.best_header_listeners.push(tx); - }, - Some(NotificationRegisterMessage::RegisterImportListener(tx)) => { - self.imported_header_listeners.push(tx) - }, - Some(NotificationRegisterMessage::RegisterFinalizationListener(tx)) => { - self.finalized_header_listeners.push(tx) - }, - None => { - tracing::error!(target: LOG_TARGET, "RPC client receiver closed. Stopping RPC Worker."); - return; - } - }, - import_event = import_sub.next() => { - if let Err(err) = handle_event_distribution(import_event, &mut self.imported_header_listeners) { - tracing::error!(target: LOG_TARGET, err, "Encountered error while processing imported header notification. Stopping RPC Worker."); - return; - } - }, - best_header_event = best_head_sub.next() => { - if let Err(err) = handle_event_distribution(best_header_event, &mut self.best_header_listeners) { - tracing::error!(target: LOG_TARGET, err, "Encountered error while processing best header notification. Stopping RPC Worker."); - return; - } - } - finalized_event = finalized_sub.next() => { - if let Err(err) = handle_event_distribution(finalized_event, &mut self.finalized_header_listeners) { - tracing::error!(target: LOG_TARGET, err, "Encountered error while processing finalized header notification. Stopping RPC Worker."); - return; - } - } - } - } - } -} - impl RelayChainRpcClient { /// Initialize new RPC Client. - async fn new( - ws_client: JsonRpcClient, - sender: TokioSender, - ) -> RelayChainResult { - let client = RelayChainRpcClient { - to_worker_channel: sender, - ws_client: Arc::new(ws_client), - retry_strategy: ExponentialBackoff::default(), - }; + async fn new(ws_client: ReconnectingWsClient) -> RelayChainResult { + let client = RelayChainRpcClient { ws_client }; Ok(client) } @@ -236,7 +74,7 @@ impl RelayChainRpcClient { pub async fn call_remote_runtime_function( &self, method_name: &str, - hash: PHash, + hash: RelayHash, payload: Option, ) -> RelayChainResult { let payload_bytes = @@ -286,34 +124,23 @@ impl RelayChainRpcClient { ) -> Result where R: DeserializeOwned + std::fmt::Debug, - OR: Fn(&jsonrpsee::core::Error), + OR: Fn(&RelayChainError), { - retry_notify( - self.retry_strategy.clone(), - || async { - self.ws_client.request(method, params.clone()).await.map_err(|err| match err { - JsonRpseeError::Transport(_) => - backoff::Error::Transient { err, retry_after: None }, - _ => backoff::Error::Permanent(err), - }) - }, - |error, dur| tracing::trace!(target: LOG_TARGET, %error, ?dur, "Encountered transport error, retrying."), - ) - .await - .map_err(|err| { + self.ws_client.request(method, params).await.map_err(|err| { trace_error(&err); - RelayChainError::RpcCallError(method.to_string(), err)}) + RelayChainError::RpcCallError(method.to_string()) + }) } /// Returns information regarding the current epoch. - pub async fn babe_api_current_epoch(&self, at: PHash) -> Result { + pub async fn babe_api_current_epoch(&self, at: RelayHash) -> Result { self.call_remote_runtime_function("BabeApi_current_epoch", at, None::<()>).await } /// Old method to fetch v1 session info. pub async fn parachain_host_session_info_before_version_2( &self, - at: PHash, + at: RelayHash, index: SessionIndex, ) -> Result, RelayChainError> { self.call_remote_runtime_function( @@ -327,8 +154,8 @@ impl RelayChainRpcClient { /// Scrape dispute relevant from on-chain, backing votes and resolved disputes. pub async fn parachain_host_on_chain_votes( &self, - at: PHash, - ) -> Result>, RelayChainError> { + at: RelayHash, + ) -> Result>, RelayChainError> { self.call_remote_runtime_function("ParachainHost_on_chain_votes", at, None::<()>) .await } @@ -336,7 +163,7 @@ impl RelayChainRpcClient { /// Returns code hashes of PVFs that require pre-checking by validators in the active set. pub async fn parachain_host_pvfs_require_precheck( &self, - at: PHash, + at: RelayHash, ) -> Result, RelayChainError> { self.call_remote_runtime_function("ParachainHost_pvfs_require_precheck", at, None::<()>) .await @@ -345,7 +172,7 @@ impl RelayChainRpcClient { /// Submits a PVF pre-checking statement into the transaction pool. pub async fn parachain_host_submit_pvf_check_statement( &self, - at: PHash, + at: RelayHash, stmt: PvfCheckStatement, signature: ValidatorSignature, ) -> Result<(), RelayChainError> { @@ -371,9 +198,9 @@ impl RelayChainRpcClient { pub async fn state_get_read_proof( &self, storage_keys: Vec, - at: Option, - ) -> Result, RelayChainError> { - let params = rpc_params!(storage_keys, at); + at: Option, + ) -> Result, RelayChainError> { + let params = rpc_params![storage_keys, at]; self.request("state_getReadProof", params).await } @@ -381,17 +208,17 @@ impl RelayChainRpcClient { pub async fn state_get_storage( &self, storage_key: StorageKey, - at: Option, + at: Option, ) -> Result, RelayChainError> { - let params = rpc_params!(storage_key, at); + let params = rpc_params![storage_key, at]; self.request("state_getStorage", params).await } /// Get hash of the n-th block in the canon chain. /// /// By default returns latest block hash. - pub async fn chain_get_head(&self, at: Option) -> Result { - let params = rpc_params!(at); + pub async fn chain_get_head(&self, at: Option) -> Result { + let params = rpc_params![at]; self.request("chain_getHead", params).await } @@ -400,7 +227,7 @@ impl RelayChainRpcClient { /// should be the successor of the number of the block. pub async fn parachain_host_validator_groups( &self, - at: PHash, + at: RelayHash, ) -> Result<(Vec>, GroupRotationInfo), RelayChainError> { self.call_remote_runtime_function("ParachainHost_validator_groups", at, None::<()>) .await @@ -409,7 +236,7 @@ impl RelayChainRpcClient { /// Get a vector of events concerning candidates that occurred within a block. pub async fn parachain_host_candidate_events( &self, - at: PHash, + at: RelayHash, ) -> Result, RelayChainError> { self.call_remote_runtime_function("ParachainHost_candidate_events", at, None::<()>) .await @@ -418,7 +245,7 @@ impl RelayChainRpcClient { /// Checks if the given validation outputs pass the acceptance criteria. pub async fn parachain_host_check_validation_outputs( &self, - at: PHash, + at: RelayHash, para_id: ParaId, outputs: CandidateCommitments, ) -> Result { @@ -435,9 +262,9 @@ impl RelayChainRpcClient { /// data hash against an expected one and yields `None` if they're not equal. pub async fn parachain_host_assumed_validation_data( &self, - at: PHash, + at: RelayHash, para_id: ParaId, - expected_hash: PHash, + expected_hash: RelayHash, ) -> Result, RelayChainError> { self.call_remote_runtime_function( "ParachainHost_persisted_assumed_validation_data", @@ -448,7 +275,7 @@ impl RelayChainRpcClient { } /// Get hash of last finalized block. - pub async fn chain_get_finalized_head(&self) -> Result { + pub async fn chain_get_finalized_head(&self) -> Result { self.request("chain_getFinalizedHead", rpc_params![]).await } @@ -456,8 +283,8 @@ impl RelayChainRpcClient { pub async fn chain_get_block_hash( &self, block_number: Option, - ) -> Result, RelayChainError> { - let params = rpc_params!(block_number); + ) -> Result, RelayChainError> { + let params = rpc_params![block_number]; self.request("chain_getBlockHash", params).await } @@ -468,7 +295,7 @@ impl RelayChainRpcClient { /// and the para already occupies a core. pub async fn parachain_host_persisted_validation_data( &self, - at: PHash, + at: RelayHash, para_id: ParaId, occupied_core_assumption: OccupiedCoreAssumption, ) -> Result, RelayChainError> { @@ -483,7 +310,7 @@ impl RelayChainRpcClient { /// Get the validation code from its hash. pub async fn parachain_host_validation_code_by_hash( &self, - at: PHash, + at: RelayHash, validation_code_hash: ValidationCodeHash, ) -> Result, RelayChainError> { self.call_remote_runtime_function( @@ -498,15 +325,15 @@ impl RelayChainRpcClient { /// Cores are either free or occupied. Free cores can have paras assigned to them. pub async fn parachain_host_availability_cores( &self, - at: PHash, - ) -> Result>, RelayChainError> { + at: RelayHash, + ) -> Result>, RelayChainError> { self.call_remote_runtime_function("ParachainHost_availability_cores", at, None::<()>) .await } /// Get runtime version - pub async fn runtime_version(&self, at: PHash) -> Result { - let params = rpc_params!(at); + pub async fn runtime_version(&self, at: RelayHash) -> Result { + let params = rpc_params![at]; self.request("state_getRuntimeVersion", params).await } @@ -514,7 +341,7 @@ impl RelayChainRpcClient { /// This is a staging method! Do not use on production runtimes! pub async fn parachain_host_staging_get_disputes( &self, - at: PHash, + at: RelayHash, ) -> Result)>, RelayChainError> { self.call_remote_runtime_function("ParachainHost_staging_get_disputes", at, None::<()>) .await @@ -522,7 +349,7 @@ impl RelayChainRpcClient { pub async fn authority_discovery_authorities( &self, - at: PHash, + at: RelayHash, ) -> Result, RelayChainError> { self.call_remote_runtime_function("AuthorityDiscoveryApi_authorities", at, None::<()>) .await @@ -534,7 +361,7 @@ impl RelayChainRpcClient { /// and the para already occupies a core. pub async fn parachain_host_validation_code( &self, - at: PHash, + at: RelayHash, para_id: ParaId, occupied_core_assumption: OccupiedCoreAssumption, ) -> Result, RelayChainError> { @@ -549,7 +376,7 @@ impl RelayChainRpcClient { /// Fetch the hash of the validation code used by a para, making the given `OccupiedCoreAssumption`. pub async fn parachain_host_validation_code_hash( &self, - at: PHash, + at: RelayHash, para_id: ParaId, occupied_core_assumption: OccupiedCoreAssumption, ) -> Result, RelayChainError> { @@ -564,7 +391,7 @@ impl RelayChainRpcClient { /// Get the session info for the given session, if stored. pub async fn parachain_host_session_info( &self, - at: PHash, + at: RelayHash, index: SessionIndex, ) -> Result, RelayChainError> { self.call_remote_runtime_function("ParachainHost_session_info", at, Some(index)) @@ -574,9 +401,9 @@ impl RelayChainRpcClient { /// Get header at specified hash pub async fn chain_get_header( &self, - hash: Option, - ) -> Result, RelayChainError> { - let params = rpc_params!(hash); + hash: Option, + ) -> Result, RelayChainError> { + let params = rpc_params![hash]; self.request("chain_getHeader", params).await } @@ -584,7 +411,7 @@ impl RelayChainRpcClient { /// assigned to occupied cores in `availability_cores` and `None` otherwise. pub async fn parachain_host_candidate_pending_availability( &self, - at: PHash, + at: RelayHash, para_id: ParaId, ) -> Result, RelayChainError> { self.call_remote_runtime_function( @@ -600,7 +427,7 @@ impl RelayChainRpcClient { /// This can be used to instantiate a `SigningContext`. pub async fn parachain_host_session_index_for_child( &self, - at: PHash, + at: RelayHash, ) -> Result { self.call_remote_runtime_function("ParachainHost_session_index_for_child", at, None::<()>) .await @@ -609,7 +436,7 @@ impl RelayChainRpcClient { /// Get the current validators. pub async fn parachain_host_validators( &self, - at: PHash, + at: RelayHash, ) -> Result, RelayChainError> { self.call_remote_runtime_function("ParachainHost_validators", at, None::<()>) .await @@ -620,7 +447,7 @@ impl RelayChainRpcClient { pub async fn parachain_host_inbound_hrmp_channels_contents( &self, para_id: ParaId, - at: PHash, + at: RelayHash, ) -> Result>, RelayChainError> { self.call_remote_runtime_function( "ParachainHost_inbound_hrmp_channels_contents", @@ -634,81 +461,24 @@ impl RelayChainRpcClient { pub async fn parachain_host_dmq_contents( &self, para_id: ParaId, - at: PHash, + at: RelayHash, ) -> Result, RelayChainError> { self.call_remote_runtime_function("ParachainHost_dmq_contents", at, Some(para_id)) .await } /// Get a stream of all imported relay chain headers - pub async fn get_imported_heads_stream(&self) -> Result, RelayChainError> { - let (tx, rx) = futures::channel::mpsc::channel::(NOTIFICATION_CHANNEL_SIZE_LIMIT); - self.send_register_message_to_worker(NotificationRegisterMessage::RegisterImportListener( - tx, - ))?; - Ok(rx) + pub fn get_imported_heads_stream(&self) -> Result, RelayChainError> { + self.ws_client.get_imported_heads_stream() } /// Get a stream of new best relay chain headers - pub async fn get_best_heads_stream(&self) -> Result, RelayChainError> { - let (tx, rx) = futures::channel::mpsc::channel::(NOTIFICATION_CHANNEL_SIZE_LIMIT); - self.send_register_message_to_worker( - NotificationRegisterMessage::RegisterBestHeadListener(tx), - )?; - Ok(rx) + pub fn get_best_heads_stream(&self) -> Result, RelayChainError> { + self.ws_client.get_best_heads_stream() } /// Get a stream of finalized relay chain headers - pub async fn get_finalized_heads_stream(&self) -> Result, RelayChainError> { - let (tx, rx) = futures::channel::mpsc::channel::(NOTIFICATION_CHANNEL_SIZE_LIMIT); - self.send_register_message_to_worker( - NotificationRegisterMessage::RegisterFinalizationListener(tx), - )?; - Ok(rx) - } - - fn send_register_message_to_worker( - &self, - message: NotificationRegisterMessage, - ) -> Result<(), RelayChainError> { - self.to_worker_channel - .try_send(message) - .map_err(|e| RelayChainError::WorkerCommunicationError(e.to_string())) - } - - async fn subscribe_imported_heads( - ws_client: &JsonRpcClient, - ) -> Result, RelayChainError> { - Ok(ws_client - .subscribe::( - "chain_subscribeAllHeads", - rpc_params![], - "chain_unsubscribeAllHeads", - ) - .await?) - } - - async fn subscribe_finalized_heads( - ws_client: &JsonRpcClient, - ) -> Result, RelayChainError> { - Ok(ws_client - .subscribe::( - "chain_subscribeFinalizedHeads", - rpc_params![], - "chain_unsubscribeFinalizedHeads", - ) - .await?) - } - - async fn subscribe_new_best_heads( - ws_client: &JsonRpcClient, - ) -> Result, RelayChainError> { - Ok(ws_client - .subscribe::( - "chain_subscribeNewHeads", - rpc_params![], - "chain_unsubscribeFinalizedHeads", - ) - .await?) + pub fn get_finalized_heads_stream(&self) -> Result, RelayChainError> { + self.ws_client.get_finalized_heads_stream() } } diff --git a/client/service/Cargo.toml b/client/service/Cargo.toml index e705364a1c8..6642deef06e 100644 --- a/client/service/Cargo.toml +++ b/client/service/Cargo.toml @@ -11,6 +11,8 @@ parking_lot = "0.12.1" sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" } sc-consensus = { git = "https://github.com/paritytech/substrate", branch = "master" } sc-service = { git = "https://github.com/paritytech/substrate", branch = "master" } +sc-sysinfo = { git = "https://github.com/paritytech/substrate", branch = "master" } +sc-telemetry = { git = "https://github.com/paritytech/substrate", branch = "master" } sp-api = { git = "https://github.com/paritytech/substrate", branch = "master" } sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "master" } sp-consensus = { git = "https://github.com/paritytech/substrate", branch = "master" } @@ -27,3 +29,5 @@ cumulus-client-consensus-common = { path = "../consensus/common" } cumulus-client-pov-recovery = { path = "../pov-recovery" } cumulus-primitives-core = { path = "../../primitives/core" } cumulus-relay-chain-interface = { path = "../relay-chain-interface" } +cumulus-relay-chain-inprocess-interface = { path = "../relay-chain-inprocess-interface" } +cumulus-relay-chain-minimal-node = { path = "../relay-chain-minimal-node" } diff --git a/client/service/src/lib.rs b/client/service/src/lib.rs index 53277ca5817..82ce3ce4872 100644 --- a/client/service/src/lib.rs +++ b/client/service/src/lib.rs @@ -18,15 +18,19 @@ //! //! Provides functions for starting a collator node or a normal full node. +use cumulus_client_cli::CollatorOptions; use cumulus_client_consensus_common::ParachainConsensus; use cumulus_primitives_core::{CollectCollationInfo, ParaId}; -use cumulus_relay_chain_interface::RelayChainInterface; +use cumulus_relay_chain_inprocess_interface::build_inprocess_relay_chain; +use cumulus_relay_chain_interface::{RelayChainInterface, RelayChainResult}; +use cumulus_relay_chain_minimal_node::build_minimal_relay_chain_node; use polkadot_primitives::v2::CollatorPair; use sc_client_api::{ Backend as BackendT, BlockBackend, BlockchainEvents, Finalizer, UsageProvider, }; use sc_consensus::{import_queue::ImportQueueService, BlockImport}; use sc_service::{Configuration, TaskManager}; +use sc_telemetry::TelemetryWorkerHandle; use sp_api::ProvideRuntimeApi; use sp_blockchain::HeaderBackend; use sp_core::traits::SpawnNamed; @@ -217,3 +221,32 @@ pub fn prepare_node_config(mut parachain_config: Configuration) -> Configuration parachain_config } + +/// Build a relay chain interface. +/// Will return a minimal relay chain node with RPC +/// client or an inprocess node, based on the [`CollatorOptions`] passed in. +pub async fn build_relay_chain_interface( + polkadot_config: Configuration, + parachain_config: &Configuration, + telemetry_worker_handle: Option, + task_manager: &mut TaskManager, + collator_options: CollatorOptions, + hwbench: Option, +) -> RelayChainResult<(Arc<(dyn RelayChainInterface + 'static)>, Option)> { + if !collator_options.relay_chain_rpc_urls.is_empty() { + build_minimal_relay_chain_node( + polkadot_config, + task_manager, + collator_options.relay_chain_rpc_urls, + ) + .await + } else { + build_inprocess_relay_chain( + polkadot_config, + parachain_config, + telemetry_worker_handle, + task_manager, + hwbench, + ) + } +} diff --git a/parachain-template/node/Cargo.toml b/parachain-template/node/Cargo.toml index b3085541b56..e26d4b00efc 100644 --- a/parachain-template/node/Cargo.toml +++ b/parachain-template/node/Cargo.toml @@ -67,10 +67,7 @@ cumulus-client-network = { path = "../../client/network" } cumulus-client-service = { path = "../../client/service" } cumulus-primitives-core = { path = "../../primitives/core" } cumulus-primitives-parachain-inherent = { path = "../../primitives/parachain-inherent" } -cumulus-relay-chain-inprocess-interface = { path = "../../client/relay-chain-inprocess-interface" } cumulus-relay-chain-interface = { path = "../../client/relay-chain-interface" } -cumulus-relay-chain-rpc-interface = { path = "../../client/relay-chain-rpc-interface" } -cumulus-relay-chain-minimal-node = { path = "../../client/relay-chain-minimal-node" } [build-dependencies] substrate-build-script-utils = { git = "https://github.com/paritytech/substrate", branch = "master" } diff --git a/parachain-template/node/src/command.rs b/parachain-template/node/src/command.rs index 14a7e046bd1..b8a079e3b2e 100644 --- a/parachain-template/node/src/command.rs +++ b/parachain-template/node/src/command.rs @@ -288,7 +288,7 @@ pub fn run() -> Result<()> { info!("Parachain genesis state: {}", genesis_state); info!("Is collating: {}", if config.role.is_authority() { "yes" } else { "no" }); - if collator_options.relay_chain_rpc_url.is_some() && cli.relay_chain_args.len() > 0 { + if !collator_options.relay_chain_rpc_urls.is_empty() && cli.relay_chain_args.len() > 0 { warn!("Detected relay chain node arguments together with --relay-chain-rpc-url. This command starts a minimal Polkadot node that only uses a network-related subset of all relay chain CLI options."); } diff --git a/parachain-template/node/src/service.rs b/parachain-template/node/src/service.rs index 36ccfe2f75c..2df17e2eb89 100644 --- a/parachain-template/node/src/service.rs +++ b/parachain-template/node/src/service.rs @@ -14,12 +14,11 @@ use cumulus_client_consensus_common::{ }; use cumulus_client_network::BlockAnnounceValidator; use cumulus_client_service::{ - prepare_node_config, start_collator, start_full_node, StartCollatorParams, StartFullNodeParams, + build_relay_chain_interface, prepare_node_config, start_collator, start_full_node, + StartCollatorParams, StartFullNodeParams, }; use cumulus_primitives_core::ParaId; -use cumulus_relay_chain_inprocess_interface::build_inprocess_relay_chain; -use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface, RelayChainResult}; -use cumulus_relay_chain_minimal_node::build_minimal_relay_chain_node; +use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface}; // Substrate Imports use sc_consensus::ImportQueue; @@ -31,8 +30,6 @@ use sc_telemetry::{Telemetry, TelemetryHandle, TelemetryWorker, TelemetryWorkerH use sp_keystore::SyncCryptoStorePtr; use substrate_prometheus_endpoint::Registry; -use polkadot_service::CollatorPair; - /// Native executor type. pub struct ParachainNativeExecutor; @@ -136,27 +133,6 @@ pub fn new_partial( }) } -async fn build_relay_chain_interface( - polkadot_config: Configuration, - parachain_config: &Configuration, - telemetry_worker_handle: Option, - task_manager: &mut TaskManager, - collator_options: CollatorOptions, - hwbench: Option, -) -> RelayChainResult<(Arc<(dyn RelayChainInterface + 'static)>, Option)> { - match collator_options.relay_chain_rpc_url { - Some(relay_chain_url) => - build_minimal_relay_chain_node(polkadot_config, task_manager, relay_chain_url).await, - None => build_inprocess_relay_chain( - polkadot_config, - parachain_config, - telemetry_worker_handle, - task_manager, - hwbench, - ), - } -} - /// Start a node with the given parachain `Configuration` and relay chain `Configuration`. /// /// This is the actual implementation that is abstract over the executor and the runtime api. diff --git a/polkadot-parachain/Cargo.toml b/polkadot-parachain/Cargo.toml index f9b7943b4d8..d879c468403 100644 --- a/polkadot-parachain/Cargo.toml +++ b/polkadot-parachain/Cargo.toml @@ -88,9 +88,6 @@ cumulus-client-network = { path = "../client/network" } cumulus-primitives-core = { path = "../primitives/core" } cumulus-primitives-parachain-inherent = { path = "../primitives/parachain-inherent" } cumulus-relay-chain-interface = { path = "../client/relay-chain-interface" } -cumulus-relay-chain-inprocess-interface = { path = "../client/relay-chain-inprocess-interface" } -cumulus-relay-chain-rpc-interface = { path = "../client/relay-chain-rpc-interface" } -cumulus-relay-chain-minimal-node = { path = "../client/relay-chain-minimal-node" } [build-dependencies] substrate-build-script-utils = { git = "https://github.com/paritytech/substrate", branch = "master" } diff --git a/polkadot-parachain/src/command.rs b/polkadot-parachain/src/command.rs index 6f2b5488648..2c35cb48a9f 100644 --- a/polkadot-parachain/src/command.rs +++ b/polkadot-parachain/src/command.rs @@ -791,7 +791,7 @@ pub fn run() -> Result<()> { info!("Parachain genesis state: {}", genesis_state); info!("Is collating: {}", if config.role.is_authority() { "yes" } else { "no" }); - if collator_options.relay_chain_rpc_url.is_some() && cli.relaychain_args.len() > 0 { + if !collator_options.relay_chain_rpc_urls.is_empty() && cli.relaychain_args.len() > 0 { warn!("Detected relay chain node arguments together with --relay-chain-rpc-url. This command starts a minimal Polkadot node that only uses a network-related subset of all relay chain CLI options."); } diff --git a/polkadot-parachain/src/service.rs b/polkadot-parachain/src/service.rs index a40b290524d..b38858c8f3e 100644 --- a/polkadot-parachain/src/service.rs +++ b/polkadot-parachain/src/service.rs @@ -22,16 +22,14 @@ use cumulus_client_consensus_common::{ }; use cumulus_client_network::BlockAnnounceValidator; use cumulus_client_service::{ - prepare_node_config, start_collator, start_full_node, StartCollatorParams, StartFullNodeParams, + build_relay_chain_interface, prepare_node_config, start_collator, start_full_node, + StartCollatorParams, StartFullNodeParams, }; use cumulus_primitives_core::{ relay_chain::v2::{Hash as PHash, PersistedValidationData}, ParaId, }; -use cumulus_relay_chain_inprocess_interface::build_inprocess_relay_chain; -use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface, RelayChainResult}; -use cumulus_relay_chain_minimal_node::build_minimal_relay_chain_node; -use polkadot_service::CollatorPair; +use cumulus_relay_chain_interface::{RelayChainError, RelayChainInterface}; use sp_core::Pair; use jsonrpsee::RpcModule; @@ -301,27 +299,6 @@ where Ok(params) } -async fn build_relay_chain_interface( - polkadot_config: Configuration, - parachain_config: &Configuration, - telemetry_worker_handle: Option, - task_manager: &mut TaskManager, - collator_options: CollatorOptions, - hwbench: Option, -) -> RelayChainResult<(Arc<(dyn RelayChainInterface + 'static)>, Option)> { - match collator_options.relay_chain_rpc_url { - Some(relay_chain_url) => - build_minimal_relay_chain_node(polkadot_config, task_manager, relay_chain_url).await, - None => build_inprocess_relay_chain( - polkadot_config, - parachain_config, - telemetry_worker_handle, - task_manager, - hwbench, - ), - } -} - /// Start a shell node with the given parachain `Configuration` and relay chain `Configuration`. /// /// This is the actual implementation that is abstract over the executor and the runtime api for shell nodes. diff --git a/test/service/src/lib.rs b/test/service/src/lib.rs index 486184ff6b7..76d568ffeba 100644 --- a/test/service/src/lib.rs +++ b/test/service/src/lib.rs @@ -190,10 +190,14 @@ async fn build_relay_chain_interface( collator_options: CollatorOptions, task_manager: &mut TaskManager, ) -> RelayChainResult> { - if let Some(relay_chain_url) = collator_options.relay_chain_rpc_url { - return build_minimal_relay_chain_node(relay_chain_config, task_manager, relay_chain_url) - .await - .map(|r| r.0) + if !collator_options.relay_chain_rpc_urls.is_empty() { + return build_minimal_relay_chain_node( + relay_chain_config, + task_manager, + collator_options.relay_chain_rpc_urls, + ) + .await + .map(|r| r.0) } let relay_chain_full_node = polkadot_test_service::new_full( @@ -429,7 +433,7 @@ pub struct TestNodeBuilder { storage_update_func_parachain: Option>, storage_update_func_relay_chain: Option>, consensus: Consensus, - relay_chain_full_node_url: Option, + relay_chain_full_node_url: Vec, } impl TestNodeBuilder { @@ -451,7 +455,7 @@ impl TestNodeBuilder { storage_update_func_parachain: None, storage_update_func_relay_chain: None, consensus: Consensus::RelayChain, - relay_chain_full_node_url: None, + relay_chain_full_node_url: vec![], } } @@ -545,7 +549,7 @@ impl TestNodeBuilder { /// Connect to full node via RPC. pub fn use_external_relay_chain_node_at_url(mut self, network_address: Url) -> Self { - self.relay_chain_full_node_url = Some(network_address); + self.relay_chain_full_node_url = vec![network_address]; self } @@ -554,7 +558,7 @@ impl TestNodeBuilder { let mut localhost_url = Url::parse("ws://localhost").expect("Should be able to parse localhost Url"); localhost_url.set_port(Some(port)).expect("Should be able to set port"); - self.relay_chain_full_node_url = Some(localhost_url); + self.relay_chain_full_node_url = vec![localhost_url]; self } @@ -580,7 +584,7 @@ impl TestNodeBuilder { ); let collator_options = - CollatorOptions { relay_chain_rpc_url: self.relay_chain_full_node_url }; + CollatorOptions { relay_chain_rpc_urls: self.relay_chain_full_node_url }; relay_chain_config.network.node_name = format!("{} (relay chain)", relay_chain_config.network.node_name); diff --git a/zombienet/tests/0006-rpc_collator_builds_blocks.feature b/zombienet/tests/0006-rpc_collator_builds_blocks.feature index 558e65f96db..78886b7020b 100644 --- a/zombienet/tests/0006-rpc_collator_builds_blocks.feature +++ b/zombienet/tests/0006-rpc_collator_builds_blocks.feature @@ -7,11 +7,18 @@ bob: is up charlie: is up one: is up two: is up +three: is up dave: is up eve: is up alice: parachain 2000 is registered within 225 seconds alice: parachain 2000 block height is at least 10 within 250 seconds -dave: reports block height is at least 12 within 250 seconds eve: reports block height is at least 12 within 250 seconds +dave: reports block height is at least 12 within 250 seconds +one: restart after 1 seconds +dave: reports block height is at least 20 within 200 seconds +two: restart after 1 seconds +three: restart after 20 seconds +dave: is up +dave: reports block height is at least 30 within 200 seconds diff --git a/zombienet/tests/0006-rpc_collator_builds_blocks.toml b/zombienet/tests/0006-rpc_collator_builds_blocks.toml index 9414532682a..963b6f5d9e1 100644 --- a/zombienet/tests/0006-rpc_collator_builds_blocks.toml +++ b/zombienet/tests/0006-rpc_collator_builds_blocks.toml @@ -25,6 +25,10 @@ chain = "rococo-local" name = "two" validator = false + [[relaychain.nodes]] + name = "three" + validator = false + [[parachains]] id = 2000 cumulus_based = true @@ -35,7 +39,7 @@ cumulus_based = true validator = true image = "{{COL_IMAGE}}" command = "test-parachain" - args = ["-lparachain=debug,blockchain-rpc-client=debug", "--relay-chain-rpc-url {{'one'|zombie('wsUri')}}", "-- --bootnodes {{'one'|zombie('multiAddress')}}"] + args = ["-lparachain=trace,blockchain-rpc-client=debug", "--relay-chain-rpc-urls {{'one'|zombie('wsUri')}} {{'two'|zombie('wsUri')}} {{'three'|zombie('wsUri')}}", "-- --bootnodes {{'one'|zombie('multiAddress')}} {{'two'|zombie('multiAddress')}} {{'three'|zombie('multiAddress')}}"] # run eve as parachain full node [[parachains.collators]] @@ -43,4 +47,4 @@ cumulus_based = true validator = true image = "{{COL_IMAGE}}" command = "test-parachain" - args = ["-lparachain=debug,blockchain-rpc-client=debug", "--relay-chain-rpc-url {{'two'|zombie('wsUri')}}", "-- --bootnodes {{'two'|zombie('multiAddress')}}"] + args = ["-lparachain=trace,blockchain-rpc-client=debug", "--relay-chain-rpc-urls {{'one'|zombie('wsUri')}} {{'two'|zombie('wsUri')}} {{'three'|zombie('wsUri')}}", "-- --bootnodes {{'one'|zombie('multiAddress')}} {{'two'|zombie('multiAddress')}} {{'three'|zombie('multiAddress')}}"]