Enable collation via RPC relay chain node

.gitlab-ci.yml

@@ -635,6 +635,35 @@ zombienet-0005-migrate_solo_to_para:
   tags:
     - zombienet-polkadot-integration-test
 
+0006-rpc_collator_builds_blocks:
+  stage:                           integration-test
+  image:                           "${ZOMBIENET_IMAGE}"
+  <<:                              *zombienet-refs
+  needs:
+    - job:                         build-push-image-test-parachain
+  variables:
+    POLKADOT_IMAGE:                "docker.io/paritypr/polkadot-debug:master"
+    GH_DIR:                        "https://github.com/paritytech/cumulus/tree/${CI_COMMIT_SHORT_SHA}/zombienet_tests"
+    COL_IMAGE:                     "docker.io/paritypr/test-parachain:${CI_COMMIT_REF_NAME}-${CI_COMMIT_SHORT_SHA}"
+  before_script:
+    - echo "Zombie-net Tests Config"
+    - echo "${ZOMBIENET_IMAGE}"
+    - echo "${RELAY_IMAGE}"
+    - echo "${COL_IMAGE}"
+    - echo "${GH_DIR}"
+    - export DEBUG=zombie
+    - export RELAY_IMAGE=${POLKADOT_IMAGE}
+    - export COL_IMAGE=${COL_IMAGE}
+  script:
+    - /home/nonroot/zombie-net/scripts/ci/run-test-env-manager.sh
+        --github-remote-dir="${GH_DIR}"
+        --concurrency=1
+        --test="0006-rpc_collator_builds_blocks.feature"
+  allow_failure:                   true
+  retry: 2
+  tags:
+    - zombienet-polkadot-integration-test
+
 #### stage:                        .post
 
 # This job cancels the whole pipeline if any of provided jobs fail.

Cargo.toml

@@ -10,6 +10,7 @@ members = [
 	"client/relay-chain-interface",
 	"client/relay-chain-inprocess-interface",
 	"client/relay-chain-rpc-interface",
+	"client/relay-chain-minimal-node",
 	"pallets/aura-ext",
 	"pallets/collator-selection",
 	"pallets/dmp-queue",

client/cli/src/lib.rs

@@ -264,7 +264,8 @@ impl sc_cli::CliConfiguration for ExportGenesisWasmCommand {
 fn validate_relay_chain_url(arg: &str) -> Result<Url, String> {
 	let url = Url::parse(arg).map_err(|e| e.to_string())?;
 
-	if url.scheme() == "ws" {
+	let scheme = url.scheme();
+	if scheme == "ws" || scheme == "wss" {
 		Ok(url)
 	} else {
 		Err(format!(
@@ -290,9 +291,8 @@ pub struct RunCmd {
 	/// EXPERIMENTAL: Specify an URL to a relay chain full node to communicate with.
 	#[clap(
 		long,
-		value_parser = validate_relay_chain_url,
-		conflicts_with_all = &["alice", "bob", "charlie", "dave", "eve", "ferdie", "one", "two"]	)
-	]
+		value_parser = validate_relay_chain_url
+	)]
 	pub relay_chain_rpc_url: Option<Url>,
 }
 

client/network/src/tests.rs

@@ -174,7 +174,7 @@ impl RelayChainInterface for DummyRelayChainInterface {
 		Ok(false)
 	}
 
-	fn overseer_handle(&self) -> RelayChainResult<Option<Handle>> {
+	fn overseer_handle(&self) -> RelayChainResult<Handle> {
 		unimplemented!("Not needed for test")
 	}
 

client/pov-recovery/Cargo.toml

@@ -31,6 +31,7 @@ cumulus-relay-chain-interface = {path = "../relay-chain-interface"}
 
 [dev-dependencies]
 tokio = { version = "1.21.1", features = ["macros"] }
+portpicker = "0.1.1"
 
 # Cumulus
 cumulus-test-service = { path = "../../test/service" }

client/pov-recovery/src/lib.rs

@@ -181,7 +181,7 @@ where
 			Ok(_) => return,
 			Err(e) => {
 				tracing::debug!(
-					target: "cumulus-consensus",
+					target: LOG_TARGET,
 					error = ?e,
 					block_hash = ?hash,
 					"Failed to get block status",
@@ -190,6 +190,7 @@ where
 			},
 		}
 
+		tracing::debug!(target: LOG_TARGET, ?hash, "Adding pending candidate");
 		if self
 			.pending_candidates
 			.insert(
@@ -233,6 +234,7 @@ where
 			None => return,
 		};
 
+		tracing::debug!(target: LOG_TARGET, ?block_hash, "Issuing recovery request");
 		self.active_candidate_recovery
 			.recover_candidate(block_hash, pending_candidate)
 			.await;
@@ -301,7 +303,7 @@ where
 			Ok(BlockStatus::Unknown) => {
 				if self.active_candidate_recovery.is_being_recovered(&parent) {
 					tracing::debug!(
-						target: "cumulus-consensus",
+						target: LOG_TARGET,
 						?block_hash,
 						parent_hash = ?parent,
 						"Parent is still being recovered, waiting.",
@@ -311,7 +313,7 @@ where
 					return
 				} else {
 					tracing::debug!(
-						target: "cumulus-consensus",
+						target: LOG_TARGET,
 						?block_hash,
 						parent_hash = ?parent,
 						"Parent not found while trying to import recovered block.",
@@ -324,7 +326,7 @@ where
 			},
 			Err(error) => {
 				tracing::debug!(
-					target: "cumulus-consensus",
+					target: LOG_TARGET,
 					block_hash = ?parent,
 					?error,
 					"Error while checking block status",
@@ -346,6 +348,8 @@ where
 	/// This will also recursivley drain `waiting_for_parent` and import them as well.
 	async fn import_block(&mut self, block: Block) {
 		let mut blocks = VecDeque::new();
+
+		tracing::debug!(target: LOG_TARGET, hash = ?block.hash(), "Importing block retrieved using pov_recovery");
 		blocks.push_back(block);
 
 		let mut incoming_blocks = Vec::new();

client/pov-recovery/tests/pov_recovery.rs

@@ -16,6 +16,7 @@
 
 use cumulus_primitives_core::ParaId;
 use cumulus_test_service::{initial_head_data, Keyring::*};
+use futures::join;
 use std::sync::Arc;
 
 /// Tests the PoV recovery.
@@ -34,12 +35,13 @@ async fn pov_recovery() {
 	let tokio_handle = tokio::runtime::Handle::current();
 
 	// Start alice
+	let ws_port = portpicker::pick_unused_port().expect("No free ports");
 	let alice = cumulus_test_service::run_relay_chain_validator_node(
 		tokio_handle.clone(),
 		Alice,
 		|| {},
 		Vec::new(),
-		None,
+		Some(ws_port),
 	);
 
 	// Start bob
@@ -90,16 +92,49 @@ async fn pov_recovery() {
 		.build()
 		.await;
 
-	let eve = cumulus_test_service::TestNodeBuilder::new(para_id, tokio_handle, Eve)
+	let eve = cumulus_test_service::TestNodeBuilder::new(para_id, tokio_handle.clone(), Eve)
+		.use_null_consensus()
+		.connect_to_parachain_node(&charlie)
+		.connect_to_relay_chain_nodes(vec![&alice, &bob])
+		.wrap_announce_block(|_| {
+			// Never announce any block
+			Arc::new(|_, _| {})
+		})
+		.build()
+		.await;
+
+	// Run ferdie as parachain RPC collator and one as parachain RPC full node
+	//
+	// They will need to recover the pov blocks through availability recovery.
+	let ferdie = cumulus_test_service::TestNodeBuilder::new(para_id, tokio_handle.clone(), Ferdie)
 		.use_null_consensus()
 		.connect_to_parachain_node(&charlie)
 		.connect_to_relay_chain_nodes(vec![&alice, &bob])
+		.use_external_relay_chain_node_at_port(ws_port)
 		.wrap_announce_block(|_| {
 			// Never announce any block
 			Arc::new(|_, _| {})
 		})
 		.build()
 		.await;
 
-	futures::future::join(dave.wait_for_blocks(7), eve.wait_for_blocks(7)).await;
+	let one = cumulus_test_service::TestNodeBuilder::new(para_id, tokio_handle, One)
+		.enable_collator()
+		.use_null_consensus()
+		.connect_to_parachain_node(&charlie)
+		.connect_to_relay_chain_nodes(vec![&alice, &bob])
+		.use_external_relay_chain_node_at_port(ws_port)
+		.wrap_announce_block(|_| {
+			// Never announce any block
+			Arc::new(|_, _| {})
+		})
+		.build()
+		.await;
+
+	join!(
+		dave.wait_for_blocks(7),
+		eve.wait_for_blocks(7),
+		ferdie.wait_for_blocks(7),
+		one.wait_for_blocks(7)
+	);
 }

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

@@ -38,3 +38,7 @@ sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master
 # Polkadot
 polkadot-primitives = { git = "https://github.com/paritytech/polkadot", branch = "master" }
 polkadot-test-client = { git = "https://github.com/paritytech/polkadot", branch = "master" }
+metered = { package = "prioritized-metered-channel", version = "0.2.0" }
+
+# Cumulus
+cumulus-test-service = { path = "../../test/service" }

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

@@ -50,7 +50,7 @@ pub struct RelayChainInProcessInterface<Client> {
 	full_client: Arc<Client>,
 	backend: Arc<FullBackend>,
 	sync_oracle: Arc<dyn SyncOracle + Send + Sync>,
-	overseer_handle: Option<Handle>,
+	overseer_handle: Handle,
 }
 
 impl<Client> RelayChainInProcessInterface<Client> {
@@ -59,7 +59,7 @@ impl<Client> RelayChainInProcessInterface<Client> {
 		full_client: Arc<Client>,
 		backend: Arc<FullBackend>,
 		sync_oracle: Arc<dyn SyncOracle + Send + Sync>,
-		overseer_handle: Option<Handle>,
+		overseer_handle: Handle,
 	) -> Self {
 		Self { full_client, backend, sync_oracle, overseer_handle }
 	}
@@ -171,7 +171,7 @@ where
 		Ok(self.sync_oracle.is_major_syncing())
 	}
 
-	fn overseer_handle(&self) -> RelayChainResult<Option<Handle>> {
+	fn overseer_handle(&self) -> RelayChainResult<Handle> {
 		Ok(self.overseer_handle.clone())
 	}
 
@@ -288,7 +288,7 @@ struct RelayChainInProcessInterfaceBuilder {
 	polkadot_client: polkadot_client::Client,
 	backend: Arc<FullBackend>,
 	sync_oracle: Arc<dyn SyncOracle + Send + Sync>,
-	overseer_handle: Option<Handle>,
+	overseer_handle: Handle,
 }
 
 impl RelayChainInProcessInterfaceBuilder {
@@ -378,7 +378,9 @@ pub fn build_inprocess_relay_chain(
 		polkadot_client: full_node.client.clone(),
 		backend: full_node.backend.clone(),
 		sync_oracle,
-		overseer_handle: full_node.overseer_handle.clone(),
+		overseer_handle: full_node.overseer_handle.clone().ok_or(RelayChainError::GenericError(
+			"Overseer not running in full node.".to_string(),
+		))?,
 	};
 
 	task_manager.add_child(full_node.task_manager);
@@ -425,10 +427,12 @@ mod tests {
 		let block = block_builder.build().expect("Finalizes the block").block;
 		let dummy_network: Arc<dyn SyncOracle + Sync + Send> = Arc::new(DummyNetwork {});
 
+		let (tx, _rx) = metered::channel(30);
+		let mock_handle = Handle::new(tx);
 		(
 			client.clone(),
 			block,
-			RelayChainInProcessInterface::new(client, backend.clone(), dummy_network, None),
+			RelayChainInProcessInterface::new(client, backend.clone(), dummy_network, mock_handle),
 		)
 	}