Enable collation via RPC relay chain node (#1585)

* Add minimal overseer gen with dummy subsystems

* Fix dependencies

* no-compile: only client transaction pool missing

* Remove unused imports

* Continue to hack towards PoC

* Continue

* Make mini node compile

* Compiling version with blockchainevents trait

* Continue

* Check in lockfile

* Block with tokio

* update patches

* Update polkadot patches

* Use polkadot-primitives v2

* Fix build problems

* First working version

* Adjust cargo.lock

* Add integration test

* Make integration test work

* Allow startinc collator without relay-chain args

* Make OverseerRuntimeClient async

* Create separate integration test

* Remove unused ChainSelection code

* Remove unused parameters on new-mini

* Connect collator node in test to relay chain nodes

* Make BlockChainRPCClient obsolete

* Clean up

* Clean up

* Reimplement blockchain-rpc-events

* Revert "Allow startinc collator without relay-chain args"

This reverts commit f22c70e16521f375fe125df5616d48ceea926b1a.

* Add `strict_record_validation` to AuthorityDiscovery

* Move network to cumulus

* Remove BlockchainRPCEvents

* Remove `BlockIdTo` and `BlockchainEvents`

* Make AuthorityDiscovery async

* Use hash in OverseerRuntime

* Adjust naming of runtime client trait

* Implement more rpc-client methods

* Improve error handling for `ApiError`

* Extract authority-discovery creationand cleanup

* RPC -> Rpc

* Extract bitswap

* Adjust to changes on master

* Implement `hash` method

* Introduce DummyChainSync, remove ProofProvider and BlockBackend

* Remove `HeaderMetadata` from blockchain-rpc-client

* Make ChainSync work

* Implement NetworkHeaderBackend

* Cleanup

* Adjustments after master merge

* Remove ImportQueue from network parameters

* Remove cargo patches

* Eliminate warnings

* Revert to HeaderBackend

* Add zombienet test

* Implement `status()` method

* Add more comments, improve readability

* Remove patches from Cargo.toml

* Remove integration test in favor of zombienet

* Remove unused dependencies, rename minimal node crate

* Adjust to latest master changes

* fmt

* Execute zombienet test on gitlab ci

* Reuse network metrics

* Chainsync metrics

* fmt

* Feed RPC node as boot node to the relay chain minimal node

* fmt

* Add bootnodes to zombienet collators

* Allow specification of relay chain args

* Apply review suggestions

* Remove unnecessary casts

* Enable PoV recovery for rpc full nodes

* Revert unwanted changes

* Make overseerHandle non-optional

* Add availability-store subsystem

* Add AuxStore and ChainApiSubsystem

* Add availability distribution subsystem

* Improve pov-recovery logging and add RPC nodes to tests

* fmt

* Make availability config const

* lock

* Enable debug logs for pov-recovery in zombienet

* Add log filters to test binary

* Allow wss

* Address review comments

* Apply reviewer comments

* Adjust to master changes

* Apply reviewer suggestions

* Bump polkadot

* Add builder method for minimal node

* Bump substrate and polkadot

* Clean up overseer building

* Add bootnode to two in pov_recovery test

* Fix missing quote in pov recovery zombienet test

* Improve zombienet pov test

* More debug logs for pov-recovery

* Remove reserved nodes like on original test

* Revert zombienet test to master

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

@@ -44,11 +44,12 @@ const TIMEOUT_IN_SECONDS: u64 = 6;
 #[derive(Clone)]
 pub struct RelayChainRpcInterface {
 	rpc_client: RelayChainRpcClient,
+	overseer_handle: Handle,
 }
 
 impl RelayChainRpcInterface {
-	pub fn new(rpc_client: RelayChainRpcClient) -> Self {
-		Self { rpc_client }
+	pub fn new(rpc_client: RelayChainRpcClient, overseer_handle: Handle) -> Self {
+		Self { rpc_client, overseer_handle }
 	}
 }
 
@@ -118,15 +119,15 @@ impl RelayChainInterface for RelayChainRpcInterface {
 	}
 
 	async fn best_block_hash(&self) -> RelayChainResult<PHash> {
-		self.rpc_client.chain_get_head().await
+		self.rpc_client.chain_get_head(None).await
 	}
 
 	async fn is_major_syncing(&self) -> RelayChainResult<bool> {
 		self.rpc_client.system_health().await.map(|h| h.is_syncing)
 	}
 
-	fn overseer_handle(&self) -> RelayChainResult<Option<Handle>> {
-		unimplemented!("Overseer handle is not available on relay-chain-rpc-interface");
+	fn overseer_handle(&self) -> RelayChainResult<Handle> {
+		Ok(self.overseer_handle.clone())
 	}
 
 	async fn get_storage_by_key(

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

@@ -17,8 +17,13 @@
 use backoff::{future::retry_notify, ExponentialBackoff};
 use cumulus_primitives_core::{
 	relay_chain::{
-		v2::{CommittedCandidateReceipt, OccupiedCoreAssumption, SessionIndex, ValidatorId},
-		Hash as PHash, Header as PHeader, InboundHrmpMessage,
+		v2::{
+			CandidateCommitments, CandidateEvent, CommittedCandidateReceipt, CoreState,
+			DisputeState, GroupRotationInfo, OccupiedCoreAssumption, OldV1SessionInfo,
+			PvfCheckStatement, ScrapedOnChainVotes, SessionIndex, SessionInfo, ValidationCode,
+			ValidationCodeHash, ValidatorId, ValidatorIndex, ValidatorSignature,
+		},
+		CandidateHash, Hash as PHash, Header as PHeader, InboundHrmpMessage,
 	},
 	InboundDownwardMessage, ParaId, PersistedValidationData,
 };
@@ -37,9 +42,11 @@ use jsonrpsee::{
 	ws_client::WsClientBuilder,
 };
 use parity_scale_codec::{Decode, Encode};
-use polkadot_service::TaskManager;
+use polkadot_service::{BlockNumber, TaskManager};
 use sc_client_api::StorageData;
 use sc_rpc_api::{state::ReadProof, system::Health};
+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;
@@ -253,8 +260,6 @@ impl RelayChainRpcClient {
 		Decode::decode(&mut &*res.0).map_err(Into::into)
 	}
 
-	/// Subscribe to a notification stream via RPC
-
 	/// Perform RPC request
 	async fn request<'a, R>(
 		&self,
@@ -300,10 +305,69 @@ impl RelayChainRpcClient {
 			RelayChainError::RpcCallError(method.to_string(), err)})
 	}
 
+	/// Returns information regarding the current epoch.
+	pub async fn babe_api_current_epoch(&self, at: PHash) -> Result<Epoch, RelayChainError> {
+		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,
+		index: SessionIndex,
+	) -> Result<Option<OldV1SessionInfo>, RelayChainError> {
+		self.call_remote_runtime_function(
+			"ParachainHost_session_info_before_version_2",
+			at,
+			Some(index),
+		)
+		.await
+	}
+
+	/// Scrape dispute relevant from on-chain, backing votes and resolved disputes.
+	pub async fn parachain_host_on_chain_votes(
+		&self,
+		at: PHash,
+	) -> Result<Option<ScrapedOnChainVotes<PHash>>, RelayChainError> {
+		self.call_remote_runtime_function("ParachainHost_on_chain_votes", at, None::<()>)
+			.await
+	}
+
+	/// 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,
+	) -> Result<Vec<ValidationCodeHash>, RelayChainError> {
+		self.call_remote_runtime_function("ParachainHost_pvfs_require_precheck", at, None::<()>)
+			.await
+	}
+
+	/// Submits a PVF pre-checking statement into the transaction pool.
+	pub async fn parachain_host_submit_pvf_check_statement(
+		&self,
+		at: PHash,
+		stmt: PvfCheckStatement,
+		signature: ValidatorSignature,
+	) -> Result<(), RelayChainError> {
+		self.call_remote_runtime_function(
+			"ParachainHost_submit_pvf_check_statement",
+			at,
+			Some((stmt, signature)),
+		)
+		.await
+	}
+
+	/// Get local listen address of the node
+	pub async fn system_local_listen_addresses(&self) -> Result<Vec<String>, RelayChainError> {
+		self.request("system_localListenAddresses", None).await
+	}
+
+	/// Get system health information
 	pub async fn system_health(&self) -> Result<Health, RelayChainError> {
 		self.request("system_health", None).await
 	}
 
+	/// Get read proof for `storage_keys`
 	pub async fn state_get_read_proof(
 		&self,
 		storage_keys: Vec<StorageKey>,
@@ -313,6 +377,7 @@ impl RelayChainRpcClient {
 		self.request("state_getReadProof", params).await
 	}
 
+	/// Retrieve storage item at `storage_key`
 	pub async fn state_get_storage(
 		&self,
 		storage_key: StorageKey,
@@ -322,47 +387,85 @@ impl RelayChainRpcClient {
 		self.request("state_getStorage", params).await
 	}
 
-	pub async fn chain_get_head(&self) -> Result<PHash, RelayChainError> {
-		self.request("chain_getHead", None).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<u64>) -> Result<PHash, RelayChainError> {
+		let params = rpc_params!(at);
+		self.request("chain_getHead", params).await
 	}
 
-	pub async fn chain_get_header(
+	/// Returns the validator groups and rotation info localized based on the hypothetical child
+	///  of a block whose state  this is invoked on. Note that `now` in the `GroupRotationInfo`
+	/// should be the successor of the number of the block.
+	pub async fn parachain_host_validator_groups(
 		&self,
-		hash: Option<PHash>,
-	) -> Result<Option<PHeader>, RelayChainError> {
-		let params = rpc_params!(hash);
-		self.request("chain_getHeader", params).await
+		at: PHash,
+	) -> Result<(Vec<Vec<ValidatorIndex>>, GroupRotationInfo), RelayChainError> {
+		self.call_remote_runtime_function("ParachainHost_validator_groups", at, None::<()>)
+			.await
 	}
 
-	pub async fn parachain_host_candidate_pending_availability(
+	/// Get a vector of events concerning candidates that occurred within a block.
+	pub async fn parachain_host_candidate_events(
+		&self,
+		at: PHash,
+	) -> Result<Vec<CandidateEvent>, RelayChainError> {
+		self.call_remote_runtime_function("ParachainHost_candidate_events", at, None::<()>)
+			.await
+	}
+
+	/// Checks if the given validation outputs pass the acceptance criteria.
+	pub async fn parachain_host_check_validation_outputs(
 		&self,
 		at: PHash,
 		para_id: ParaId,
-	) -> Result<Option<CommittedCandidateReceipt>, RelayChainError> {
+		outputs: CandidateCommitments,
+	) -> Result<bool, RelayChainError> {
 		self.call_remote_runtime_function(
-			"ParachainHost_candidate_pending_availability",
+			"ParachainHost_check_validation_outputs",
 			at,
-			Some(para_id),
+			Some((para_id, outputs)),
 		)
 		.await
 	}
 
-	pub async fn parachain_host_session_index_for_child(
+	/// Returns the persisted validation data for the given `ParaId` along with the corresponding
+	/// validation code hash. Instead of accepting assumption about the para, matches the validation
+	/// 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,
-	) -> Result<SessionIndex, RelayChainError> {
-		self.call_remote_runtime_function("ParachainHost_session_index_for_child", at, None::<()>)
-			.await
+		para_id: ParaId,
+		expected_hash: PHash,
+	) -> Result<Option<(PersistedValidationData, ValidationCodeHash)>, RelayChainError> {
+		self.call_remote_runtime_function(
+			"ParachainHost_persisted_assumed_validation_data",
+			at,
+			Some((para_id, expected_hash)),
+		)
+		.await
 	}
 
-	pub async fn parachain_host_validators(
+	/// Get hash of last finalized block.
+	pub async fn chain_get_finalized_head(&self) -> Result<PHash, RelayChainError> {
+		self.request("chain_getFinalizedHead", None).await
+	}
+
+	/// Get hash of n-th block.
+	pub async fn chain_get_block_hash(
 		&self,
-		at: PHash,
-	) -> Result<Vec<ValidatorId>, RelayChainError> {
-		self.call_remote_runtime_function("ParachainHost_validators", at, None::<()>)
-			.await
+		block_number: Option<polkadot_service::BlockNumber>,
+	) -> Result<Option<PHash>, RelayChainError> {
+		let params = rpc_params!(block_number);
+		self.request("chain_getBlockHash", params).await
 	}
 
+	/// Yields the persisted validation data for the given `ParaId` along with an assumption that
+	/// should be used if the para currently occupies a core.
+	///
+	/// Returns `None` if either the para is not registered or the assumption is `Freed`
+	/// and the para already occupies a core.
 	pub async fn parachain_host_persisted_validation_data(
 		&self,
 		at: PHash,
@@ -377,6 +480,143 @@ impl RelayChainRpcClient {
 		.await
 	}
 
+	/// Get the validation code from its hash.
+	pub async fn parachain_host_validation_code_by_hash(
+		&self,
+		at: PHash,
+		validation_code_hash: ValidationCodeHash,
+	) -> Result<Option<ValidationCode>, RelayChainError> {
+		self.call_remote_runtime_function(
+			"ParachainHost_validation_code_by_hash",
+			at,
+			Some(validation_code_hash),
+		)
+		.await
+	}
+
+	/// Yields information on all availability cores as relevant to the child block.
+	/// 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<Vec<CoreState<PHash, BlockNumber>>, RelayChainError> {
+		self.call_remote_runtime_function("ParachainHost_availability_cores", at, None::<()>)
+			.await
+	}
+
+	/// Get runtime version
+	pub async fn runtime_version(&self, at: PHash) -> Result<RuntimeVersion, RelayChainError> {
+		let params = rpc_params!(at);
+		self.request("state_getRuntimeVersion", params).await
+	}
+
+	/// Returns all onchain disputes.
+	/// This is a staging method! Do not use on production runtimes!
+	pub async fn parachain_host_staging_get_disputes(
+		&self,
+		at: PHash,
+	) -> Result<Vec<(SessionIndex, CandidateHash, DisputeState<BlockNumber>)>, RelayChainError> {
+		self.call_remote_runtime_function("ParachainHost_staging_get_disputes", at, None::<()>)
+			.await
+	}
+
+	pub async fn authority_discovery_authorities(
+		&self,
+		at: PHash,
+	) -> Result<Vec<sp_authority_discovery::AuthorityId>, RelayChainError> {
+		self.call_remote_runtime_function("AuthorityDiscoveryApi_authorities", at, None::<()>)
+			.await
+	}
+
+	/// Fetch the validation code used by a para, making the given `OccupiedCoreAssumption`.
+	///
+	/// Returns `None` if either the para is not registered or the assumption is `Freed`
+	/// and the para already occupies a core.
+	pub async fn parachain_host_validation_code(
+		&self,
+		at: PHash,
+		para_id: ParaId,
+		occupied_core_assumption: OccupiedCoreAssumption,
+	) -> Result<Option<ValidationCode>, RelayChainError> {
+		self.call_remote_runtime_function(
+			"ParachainHost_validation_code",
+			at,
+			Some((para_id, occupied_core_assumption)),
+		)
+		.await
+	}
+
+	/// 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,
+		para_id: ParaId,
+		occupied_core_assumption: OccupiedCoreAssumption,
+	) -> Result<Option<ValidationCodeHash>, RelayChainError> {
+		self.call_remote_runtime_function(
+			"ParachainHost_validation_code_hash",
+			at,
+			Some((para_id, occupied_core_assumption)),
+		)
+		.await
+	}
+
+	/// Get the session info for the given session, if stored.
+	pub async fn parachain_host_session_info(
+		&self,
+		at: PHash,
+		index: SessionIndex,
+	) -> Result<Option<SessionInfo>, RelayChainError> {
+		self.call_remote_runtime_function("ParachainHost_session_info", at, Some(index))
+			.await
+	}
+
+	/// Get header at specified hash
+	pub async fn chain_get_header(
+		&self,
+		hash: Option<PHash>,
+	) -> Result<Option<PHeader>, RelayChainError> {
+		let params = rpc_params!(hash);
+		self.request("chain_getHeader", params).await
+	}
+
+	/// Get the receipt of a candidate pending availability. This returns `Some` for any paras
+	/// assigned to occupied cores in `availability_cores` and `None` otherwise.
+	pub async fn parachain_host_candidate_pending_availability(
+		&self,
+		at: PHash,
+		para_id: ParaId,
+	) -> Result<Option<CommittedCandidateReceipt>, RelayChainError> {
+		self.call_remote_runtime_function(
+			"ParachainHost_candidate_pending_availability",
+			at,
+			Some(para_id),
+		)
+		.await
+	}
+
+	/// Returns the session index expected at a child of the block.
+	///
+	/// This can be used to instantiate a `SigningContext`.
+	pub async fn parachain_host_session_index_for_child(
+		&self,
+		at: PHash,
+	) -> Result<SessionIndex, RelayChainError> {
+		self.call_remote_runtime_function("ParachainHost_session_index_for_child", at, None::<()>)
+			.await
+	}
+
+	/// Get the current validators.
+	pub async fn parachain_host_validators(
+		&self,
+		at: PHash,
+	) -> Result<Vec<ValidatorId>, RelayChainError> {
+		self.call_remote_runtime_function("ParachainHost_validators", at, None::<()>)
+			.await
+	}
+
+	/// Get the contents of all channels addressed to the given recipient. Channels that have no
+	/// messages in them are also included.
 	pub async fn parachain_host_inbound_hrmp_channels_contents(
 		&self,
 		para_id: ParaId,
@@ -390,6 +630,7 @@ impl RelayChainRpcClient {
 		.await
 	}
 
+	/// Get all the pending inbound messages in the downward message queue for a para.
 	pub async fn parachain_host_dmq_contents(
 		&self,
 		para_id: ParaId,
@@ -399,15 +640,7 @@ impl RelayChainRpcClient {
 			.await
 	}
 
-	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()))
-	}
-
+	/// Get a stream of all imported relay chain headers
 	pub async fn get_imported_heads_stream(&self) -> Result<Receiver<PHeader>, RelayChainError> {
 		let (tx, rx) = futures::channel::mpsc::channel::<PHeader>(NOTIFICATION_CHANNEL_SIZE_LIMIT);
 		self.send_register_message_to_worker(NotificationRegisterMessage::RegisterImportListener(
@@ -416,6 +649,7 @@ impl RelayChainRpcClient {
 		Ok(rx)
 	}
 
+	/// Get a stream of new best relay chain headers
 	pub async fn get_best_heads_stream(&self) -> Result<Receiver<PHeader>, RelayChainError> {
 		let (tx, rx) = futures::channel::mpsc::channel::<PHeader>(NOTIFICATION_CHANNEL_SIZE_LIMIT);
 		self.send_register_message_to_worker(
@@ -424,6 +658,7 @@ impl RelayChainRpcClient {
 		Ok(rx)
 	}
 
+	/// Get a stream of finalized relay chain headers
 	pub async fn get_finalized_heads_stream(&self) -> Result<Receiver<PHeader>, RelayChainError> {
 		let (tx, rx) = futures::channel::mpsc::channel::<PHeader>(NOTIFICATION_CHANNEL_SIZE_LIMIT);
 		self.send_register_message_to_worker(
@@ -432,6 +667,15 @@ impl RelayChainRpcClient {
 		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<Subscription<PHeader>, RelayChainError> {