Allow to specify multiple relay chain RPC urls for collator node

client/cli/src/lib.rs

@@ -293,9 +293,10 @@ 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 = 1..
 	)]
-	pub relay_chain_rpc_url: Option<Url>,
+	pub relay_chain_rpc_urls: Option<Vec<Url>>,
 }
 
 impl RunCmd {
@@ -310,15 +311,15 @@ 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() }
 	}
 }
 
 /// Options only relevant for collator nodes
 #[derive(Clone, Debug)]
 pub struct CollatorOptions {
 	/// Location of relay chain full node
-	pub relay_chain_rpc_url: Option<Url>,
+	pub relay_chain_rpc_urls: Option<Vec<Url>>,
 }
 
 /// A non-redundant version of the `RunCmd` that sets the `validator` field when the

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.24"
 async-trait = "0.1.58"
 thiserror = "1.0.37"

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<dyn sp_state_machine::Error>),
-	#[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),
@@ -87,6 +87,24 @@ impl From<RelayChainError> for sp_blockchain::Error {
 	}
 }
 
+impl<T> From<tokio::sync::mpsc::error::SendError<T>> for RelayChainError {
+	fn from(e: tokio::sync::mpsc::error::SendError<T>) -> Self {
+		return RelayChainError::WorkerCommunicationError(format!(
+			"Unable to send message to RPC worker: {}",
+			e.to_string()
+		))
+	}
+}
+
+impl From<futures::channel::oneshot::Canceled> for RelayChainError {
+	fn from(e: futures::channel::oneshot::Canceled) -> Self {
+		return RelayChainError::WorkerCommunicationError(format!(
+			"Unexpected channel close on RPC worker side: {}",
+			e.to_string()
+		))
+	}
+}
+
 /// Trait that provides all necessary methods for interaction between collator and relay chain.
 #[async_trait]
 pub trait RelayChainInterface: Send + Sync {

client/relay-chain-minimal-node/src/blockchain_rpc_client.rs

@@ -363,13 +363,13 @@ impl BlockChainRpcClient {
 	pub async fn import_notification_stream(
 		&self,
 	) -> RelayChainResult<Pin<Box<dyn Stream<Item = Header> + 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<Pin<Box<dyn Stream<Item = Header> + Send>>> {
-		Ok(self.rpc_client.get_finalized_heads_stream().await?.boxed())
+		Ok(self.rpc_client.get_finalized_heads_stream()?.boxed())
 	}
 }
 

client/relay-chain-minimal-node/src/collator_overseer.rs

@@ -252,8 +252,8 @@ 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())),
@@ -262,8 +262,8 @@ 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())),

client/relay-chain-minimal-node/src/lib.rs

@@ -84,7 +84,7 @@ fn build_authority_discovery_service<Block: BlockT>(
 pub async fn build_minimal_relay_chain_node(
 	polkadot_config: Configuration,
 	task_manager: &mut TaskManager,
-	relay_chain_url: Url,
+	relay_chain_url: Vec<Url>,
 ) -> RelayChainResult<(Arc<(dyn RelayChainInterface + 'static)>, Option<CollatorPair>)> {
 	let client = cumulus_relay_chain_rpc_interface::create_client_and_start_worker(
 		relay_chain_url,

client/relay-chain-rpc-interface/Cargo.toml

@@ -20,7 +20,7 @@ sp-state-machine = { git = "https://github.com/paritytech/substrate", branch = "
 sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-storage = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-rpc-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
-tokio = { version = "1.21.2", features = ["sync"] }
+tokio = { version = "1.21.2", features = ["sync", "time"] }
 
 futures = "0.3.24"
 futures-timer = "3.0.2"
@@ -29,4 +29,4 @@ jsonrpsee = { version = "0.15.1", features = ["ws-client"] }
 tracing = "0.1.37"
 async-trait = "0.1.58"
 url = "2.3.1"
-backoff = { version = "0.4.0", features = ["tokio"] }
+serde_json = "1.0.87"

client/relay-chain-rpc-interface/src/lib.rs

@@ -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};
 
@@ -105,15 +106,15 @@ impl RelayChainInterface for RelayChainRpcInterface {
 	async fn import_notification_stream(
 		&self,
 	) -> RelayChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
-		let imported_headers_stream = self.rpc_client.get_imported_heads_stream().await?;
+		let imported_headers_stream = self.rpc_client.get_imported_heads_stream()?;
 
 		Ok(imported_headers_stream.boxed())
 	}
 
 	async fn finality_notification_stream(
 		&self,
 	) -> RelayChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
-		let imported_headers_stream = self.rpc_client.get_finalized_heads_stream().await?;
+		let imported_headers_stream = self.rpc_client.get_finalized_heads_stream()?;
 
 		Ok(imported_headers_stream.boxed())
 	}
@@ -168,7 +169,7 @@ impl RelayChainInterface for RelayChainRpcInterface {
 	/// 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?;
+		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(())
@@ -192,7 +193,7 @@ impl RelayChainInterface for RelayChainRpcInterface {
 	async fn new_best_notification_stream(
 		&self,
 	) -> RelayChainResult<Pin<Box<dyn Stream<Item = PHeader> + Send>>> {
-		let imported_headers_stream = self.rpc_client.get_best_heads_stream().await?;
+		let imported_headers_stream = self.rpc_client.get_best_heads_stream()?;
 		Ok(imported_headers_stream.boxed())
 	}
 }