Introduce rpc client for relay chain full node

Cargo.toml

@@ -9,6 +9,7 @@ members = [
 	"client/service",
 	"client/relay-chain-interface",
 	"client/relay-chain-local",
+	"client/relay-chain-network",
 	"pallets/aura-ext",
 	"pallets/collator-selection",
 	"pallets/dmp-queue",

client/cli/Cargo.toml

@@ -10,3 +10,4 @@ clap = { version = "3.0", features = ["derive"] }
 # Substrate dependencies
 sc-cli = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-service = { git = "https://github.com/paritytech/substrate", branch = "master" }
+url = "2.2.2"

client/cli/src/lib.rs

@@ -19,7 +19,6 @@
 #![warn(missing_docs)]
 
 use clap::Parser;
-use sc_cli;
 use sc_service::{
 	config::{PrometheusConfig, TelemetryEndpoints},
 	BasePath, TransactionPoolOptions,
@@ -29,6 +28,7 @@ use std::{
 	io::{self, Write},
 	net::SocketAddr,
 };
+use url::Url;
 
 /// The `purge-chain` command used to remove the whole chain: the parachain and the relay chain.
 #[derive(Debug, Parser)]
@@ -119,6 +119,19 @@ impl sc_cli::CliConfiguration for PurgeChainCmd {
 	}
 }
 
+fn validate_relay_chain_url(arg: &str) -> Result<(), String> {
+	let url = Url::parse(arg).map_err(|e| e.to_string())?;
+
+	if url.scheme() == "ws" {
+		Ok(())
+	} else {
+		Err(format!(
+			"'{}' URL scheme not supported. Only websocket RPC is currently supported",
+			url.scheme()
+		))
+	}
+}
+
 /// The `run` command used to run a node.
 #[derive(Debug, Parser)]
 pub struct RunCmd {
@@ -131,6 +144,23 @@ pub struct RunCmd {
 	/// Note that this is the same as running with `--validator`.
 	#[clap(long, conflicts_with = "validator")]
 	pub collator: bool,
+
+	/// EXPERIMENTAL: Specify an URL to a relay chain full node to communicate with.
+	#[clap(
+		long,
+		parse(try_from_str),
+		validator = validate_relay_chain_url,
+		conflicts_with = "collator",
+		conflicts_with = "validator"
+	)]
+	pub relay_chain_rpc_url: Option<Url>,
+}
+
+/// 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>,
 }
 
 /// A non-redundant version of the `RunCmd` that sets the `validator` field when the
@@ -150,6 +180,11 @@ impl RunCmd {
 
 		NormalizedRunCmd { base: new_base }
 	}
+
+	/// 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() }
+	}
 }
 
 impl sc_cli::CliConfiguration for NormalizedRunCmd {

client/consensus/common/src/parachain_consensus.rs

@@ -27,7 +27,7 @@ use sp_runtime::{
 	traits::{Block as BlockT, Header as HeaderT},
 };
 
-use polkadot_primitives::v1::{Block as PBlock, Id as ParaId, OccupiedCoreAssumption};
+use polkadot_primitives::v1::{Hash as PHash, Id as ParaId, OccupiedCoreAssumption};
 
 use codec::Decode;
 use futures::{select, FutureExt, Stream, StreamExt};
@@ -54,7 +54,7 @@ pub trait RelaychainClient: Clone + 'static {
 	/// Returns the parachain head for the given `para_id` at the given block id.
 	async fn parachain_head_at(
 		&self,
-		at: &BlockId<PBlock>,
+		at: &PHash,
 		para_id: ParaId,
 	) -> RelayChainResult<Option<Vec<u8>>>;
 }
@@ -402,13 +402,7 @@ where
 			.await?
 			.filter_map(move |n| {
 				let relay_chain = relay_chain.clone();
-				async move {
-					relay_chain
-						.parachain_head_at(&BlockId::hash(n.hash()), para_id)
-						.await
-						.ok()
-						.flatten()
-				}
+				async move { relay_chain.parachain_head_at(&n.hash(), para_id).await.ok().flatten() }
 			})
 			.boxed();
 		Ok(new_best_notification_stream)
@@ -422,21 +416,15 @@ where
 			.await?
 			.filter_map(move |n| {
 				let relay_chain = relay_chain.clone();
-				async move {
-					relay_chain
-						.parachain_head_at(&BlockId::hash(n.hash()), para_id)
-						.await
-						.ok()
-						.flatten()
-				}
+				async move { relay_chain.parachain_head_at(&n.hash(), para_id).await.ok().flatten() }
 			})
 			.boxed();
 		Ok(finality_notification_stream)
 	}
 
 	async fn parachain_head_at(
 		&self,
-		at: &BlockId<PBlock>,
+		at: &PHash,
 		para_id: ParaId,
 	) -> RelayChainResult<Option<Vec<u8>>> {
 		self.persisted_validation_data(at, para_id, OccupiedCoreAssumption::TimedOut)

client/consensus/common/src/tests.rs

@@ -25,7 +25,7 @@ use cumulus_test_client::{
 };
 use futures::{channel::mpsc, executor::block_on, select, FutureExt, Stream, StreamExt};
 use futures_timer::Delay;
-use polkadot_primitives::v1::{Block as PBlock, Id as ParaId};
+use polkadot_primitives::v1::Id as ParaId;
 use sc_client_api::UsageProvider;
 use sc_consensus::{BlockImport, BlockImportParams, ForkChoiceStrategy};
 use sp_blockchain::Error as ClientError;
@@ -98,11 +98,7 @@ impl crate::parachain_consensus::RelaychainClient for Relaychain {
 		Ok(Box::new(stream.map(|v| v.encode())))
 	}
 
-	async fn parachain_head_at(
-		&self,
-		_: &BlockId<PBlock>,
-		_: ParaId,
-	) -> RelayChainResult<Option<Vec<u8>>> {
+	async fn parachain_head_at(&self, _: &PHash, _: ParaId) -> RelayChainResult<Option<Vec<u8>>> {
 		unimplemented!("Not required for tests")
 	}
 }

client/network/Cargo.toml

@@ -33,6 +33,8 @@ async-trait = "0.1.52"
 
 [dev-dependencies]
 tokio = { version = "1.10", features = ["macros"] }
+portpicker = "0.1.1"
+url = "2.2.2"
 
 # Cumulus deps
 cumulus-test-service = { path = "../../test/service" }

client/network/src/lib.rs

@@ -24,17 +24,14 @@ use sp_consensus::block_validation::{
 	BlockAnnounceValidator as BlockAnnounceValidatorT, Validation,
 };
 use sp_core::traits::SpawnNamed;
-use sp_runtime::{
-	generic::BlockId,
-	traits::{Block as BlockT, Header as HeaderT},
-};
+use sp_runtime::traits::{Block as BlockT, Header as HeaderT};
 
 use cumulus_relay_chain_interface::RelayChainInterface;
 use polkadot_node_primitives::{CollationSecondedSignal, Statement};
 use polkadot_parachain::primitives::HeadData;
 use polkadot_primitives::v1::{
-	Block as PBlock, CandidateReceipt, CompactStatement, Hash as PHash, Id as ParaId,
-	OccupiedCoreAssumption, SigningContext, UncheckedSigned,
+	CandidateReceipt, CompactStatement, Hash as PHash, Id as ParaId, OccupiedCoreAssumption,
+	SigningContext, UncheckedSigned,
 };
 
 use codec::{Decode, DecodeAll, Encode};
@@ -133,17 +130,16 @@ impl BlockAnnounceData {
 	{
 		let validator_index = self.statement.unchecked_validator_index();
 
-		let runtime_api_block_id = BlockId::Hash(self.relay_parent);
 		let session_index =
-			match relay_chain_client.session_index_for_child(&runtime_api_block_id).await {
+			match relay_chain_client.session_index_for_child(&self.relay_parent).await {
 				Ok(r) => r,
 				Err(e) => return Err(BlockAnnounceError(format!("{:?}", e))),
 			};
 
 		let signing_context = SigningContext { parent_hash: self.relay_parent, session_index };
 
 		// Check that the signer is a legit validator.
-		let authorities = match relay_chain_client.validators(&runtime_api_block_id).await {
+		let authorities = match relay_chain_client.validators(&self.relay_parent).await {
 			Ok(r) => r,
 			Err(e) => return Err(BlockAnnounceError(format!("{:?}", e))),
 		};
@@ -160,7 +156,7 @@ impl BlockAnnounceData {
 		};
 
 		// Check statement is correctly signed.
-		if self.statement.try_into_checked(&signing_context, &signer).is_err() {
+		if self.statement.try_into_checked(&signing_context, signer).is_err() {
 			tracing::debug!(
 				target: LOG_TARGET,
 				"Block announcement justification signature is invalid.",
@@ -231,11 +227,7 @@ where
 {
 	/// Create a new [`BlockAnnounceValidator`].
 	pub fn new(relay_chain_interface: RCInterface, para_id: ParaId) -> Self {
-		Self {
-			phantom: Default::default(),
-			relay_chain_interface: relay_chain_interface.clone(),
-			para_id,
-		}
+		Self { phantom: Default::default(), relay_chain_interface, para_id }
 	}
 }
 
@@ -246,11 +238,11 @@ where
 	/// Get the included block of the given parachain in the relay chain.
 	async fn included_block(
 		relay_chain_interface: &RCInterface,
-		block_id: &BlockId<PBlock>,
+		hash: &PHash,
 		para_id: ParaId,
 	) -> Result<Block::Header, BoxedError> {
 		let validation_data = relay_chain_interface
-			.persisted_validation_data(block_id, para_id, OccupiedCoreAssumption::TimedOut)
+			.persisted_validation_data(hash, para_id, OccupiedCoreAssumption::TimedOut)
 			.await
 			.map_err(|e| Box::new(BlockAnnounceError(format!("{:?}", e))) as Box<_>)?
 			.ok_or_else(|| {
@@ -269,11 +261,11 @@ where
 	/// Get the backed block hash of the given parachain in the relay chain.
 	async fn backed_block_hash(
 		relay_chain_interface: &RCInterface,
-		block_id: &BlockId<PBlock>,
+		hash: &PHash,
 		para_id: ParaId,
 	) -> Result<Option<PHash>, BoxedError> {
 		let candidate_receipt = relay_chain_interface
-			.candidate_pending_availability(block_id, para_id)
+			.candidate_pending_availability(hash, para_id)
 			.await
 			.map_err(|e| Box::new(BlockAnnounceError(format!("{:?}", e))) as Box<_>)?;
 
@@ -293,14 +285,13 @@ where
 			.best_block_hash()
 			.await
 			.map_err(|e| Box::new(e) as Box<_>)?;
-		let runtime_api_block_id = BlockId::Hash(relay_chain_best_hash);
 		let block_number = header.number();
 
 		let best_head =
-			Self::included_block(&relay_chain_interface, &runtime_api_block_id, para_id).await?;
+			Self::included_block(&relay_chain_interface, &relay_chain_best_hash, para_id).await?;
 		let known_best_number = best_head.number();
 		let backed_block = || async {
-			Self::backed_block_hash(&relay_chain_interface, &runtime_api_block_id, para_id).await
+			Self::backed_block_hash(&relay_chain_interface, &relay_chain_best_hash, para_id).await
 		};
 
 		if best_head == header {
@@ -335,7 +326,7 @@ where
 		data: &[u8],
 	) -> Pin<Box<dyn Future<Output = Result<Validation, BoxedError>> + Send>> {
 		let relay_chain_interface = self.relay_chain_interface.clone();
-		let mut data = data.to_vec();
+		let data = data.to_vec();
 		let header = header.clone();
 		let header_encoded = header.encode();
 		let block_announce_validator = self.clone();
@@ -357,7 +348,7 @@ where
 				return block_announce_validator.handle_empty_block_announce_data(header).await
 			}
 
-			let block_announce_data = match BlockAnnounceData::decode_all(&mut data) {
+			let block_announce_data = match BlockAnnounceData::decode_all(&data) {
 				Ok(r) => r,
 				Err(err) =>
 					return Err(Box::new(BlockAnnounceError(format!(

client/network/src/tests.rs

@@ -77,23 +77,10 @@ impl DummyRelayChainInterface {
 
 #[async_trait]
 impl RelayChainInterface for DummyRelayChainInterface {
-	async fn validators(
-		&self,
-		_: &cumulus_primitives_core::relay_chain::BlockId,
-	) -> RelayChainResult<Vec<ValidatorId>> {
+	async fn validators(&self, _: &PHash) -> RelayChainResult<Vec<ValidatorId>> {
 		Ok(self.data.lock().validators.clone())
 	}
 
-	async fn block_status(
-		&self,
-		block_id: cumulus_primitives_core::relay_chain::BlockId,
-	) -> RelayChainResult<sp_blockchain::BlockStatus> {
-		self.relay_backend
-			.blockchain()
-			.status(block_id)
-			.map_err(RelayChainError::BlockchainError)
-	}
-
 	async fn best_block_hash(&self) -> RelayChainResult<PHash> {
 		Ok(self.relay_backend.blockchain().info().best_hash)
 	}
@@ -116,7 +103,7 @@ impl RelayChainInterface for DummyRelayChainInterface {
 
 	async fn persisted_validation_data(
 		&self,
-		_: &cumulus_primitives_core::relay_chain::BlockId,
+		_: &PHash,
 		_: ParaId,
 		_: OccupiedCoreAssumption,
 	) -> RelayChainResult<Option<PersistedValidationData>> {
@@ -128,7 +115,7 @@ impl RelayChainInterface for DummyRelayChainInterface {
 
 	async fn candidate_pending_availability(
 		&self,
-		_: &cumulus_primitives_core::relay_chain::BlockId,
+		_: &PHash,
 		_: ParaId,
 	) -> RelayChainResult<Option<CommittedCandidateReceipt>> {
 		if self.data.lock().has_pending_availability {
@@ -159,10 +146,7 @@ impl RelayChainInterface for DummyRelayChainInterface {
 		}
 	}
 
-	async fn session_index_for_child(
-		&self,
-		_: &cumulus_primitives_core::relay_chain::BlockId,
-	) -> RelayChainResult<SessionIndex> {
+	async fn session_index_for_child(&self, _: &PHash) -> RelayChainResult<SessionIndex> {
 		Ok(0)
 	}
 
@@ -196,15 +180,15 @@ impl RelayChainInterface for DummyRelayChainInterface {
 
 	async fn get_storage_by_key(
 		&self,
-		_: &polkadot_service::BlockId,
+		_: &PHash,
 		_: &[u8],
 	) -> RelayChainResult<Option<StorageValue>> {
 		unimplemented!("Not needed for test")
 	}
 
 	async fn prove_read(
 		&self,
-		_: &polkadot_service::BlockId,
+		_: &PHash,
 		_: &Vec<Vec<u8>>,
 	) -> RelayChainResult<sc_client_api::StorageProof> {
 		unimplemented!("Not needed for test")
@@ -293,10 +277,7 @@ async fn make_gossip_message_and_header(
 		Some(&Sr25519Keyring::Alice.to_seed()),
 	)
 	.unwrap();
-	let session_index = relay_chain_interface
-		.session_index_for_child(&BlockId::Hash(relay_parent))
-		.await
-		.unwrap();
+	let session_index = relay_chain_interface.session_index_for_child(&relay_parent).await.unwrap();
 	let signing_context = SigningContext { parent_hash: relay_parent, session_index };
 
 	let header = default_header();
@@ -477,10 +458,7 @@ async fn check_statement_seconded() {
 		Some(&Sr25519Keyring::Alice.to_seed()),
 	)
 	.unwrap();
-	let session_index = relay_chain_interface
-		.session_index_for_child(&BlockId::Hash(relay_parent))
-		.await
-		.unwrap();
+	let session_index = relay_chain_interface.session_index_for_child(&relay_parent).await.unwrap();
 	let signing_context = SigningContext { parent_hash: relay_parent, session_index };
 
 	let statement = Statement::Valid(Default::default());