frame/utils: introduce substrate-rpc-client crate for RPC utils (#12212)

* hack together a PoC

* Update utils/frame/rpc-utils/Cargo.toml

Co-authored-by: Kian Paimani <5588131+kianenigma@users.noreply.github.com>

* Update utils/frame/rpc-utils/src/lib.rs

Co-authored-by: Kian Paimani <5588131+kianenigma@users.noreply.github.com>

* rpc_utils -> substrate_rpc_client

* try runtime: remove keep connection

* make CI happy

* cargo fmt

* fix ci

* update lock file

* fix

* fix

Co-authored-by: Kian Paimani <5588131+kianenigma@users.noreply.github.com>
Co-authored-by: kianenigma <kian@parity.io>

substrate/utils/frame/try-runtime/cli/src/commands/execute_block.rs

@@ -20,11 +20,11 @@ use crate::{
 	state_machine_call_with_proof, SharedParams, State, LOG_TARGET,
 };
 use parity_scale_codec::Encode;
-use remote_externalities::rpc_api;
 use sc_service::{Configuration, NativeExecutionDispatch};
 use sp_core::storage::well_known_keys;
 use sp_runtime::traits::{Block as BlockT, Header as HeaderT, NumberFor};
 use std::{fmt::Debug, str::FromStr};
+use substrate_rpc_client::{ws_client, ChainApi};
 
 /// Configurations of the [`Command::ExecuteBlock`].
 ///
@@ -84,10 +84,11 @@ pub struct ExecuteBlockCmd {
 impl ExecuteBlockCmd {
 	async fn block_at<Block: BlockT>(&self, ws_uri: String) -> sc_cli::Result<Block::Hash>
 	where
-		Block::Hash: FromStr,
+		Block::Hash: FromStr + serde::de::DeserializeOwned,
 		<Block::Hash as FromStr>::Err: Debug,
+		Block::Header: serde::de::DeserializeOwned,
 	{
-		let rpc_service = rpc_api::RpcService::new(ws_uri, false).await?;
+		let rpc = ws_client(&ws_uri).await?;
 
 		match (&self.block_at, &self.state) {
 			(Some(block_at), State::Snap { .. }) => hash_of::<Block>(block_at),
@@ -100,7 +101,9 @@ impl ExecuteBlockCmd {
 					target: LOG_TARGET,
 					"No --block-at or --at provided, using the latest finalized block instead"
 				);
-				rpc_service.get_finalized_head::<Block>().await.map_err(Into::into)
+				ChainApi::<(), Block::Hash, Block::Header, ()>::finalized_head(&rpc)
+					.await
+					.map_err(|e| e.to_string().into())
 			},
 			(None, State::Live { at: Some(at), .. }) => hash_of::<Block>(at),
 			_ => {
@@ -137,6 +140,8 @@ where
 	Block: BlockT + serde::de::DeserializeOwned,
 	Block::Hash: FromStr,
 	<Block::Hash as FromStr>::Err: Debug,
+	Block::Hash: serde::de::DeserializeOwned,
+	Block::Header: serde::de::DeserializeOwned,
 	NumberFor<Block>: FromStr,
 	<NumberFor<Block> as FromStr>::Err: Debug,
 	ExecDispatch: NativeExecutionDispatch + 'static,
@@ -146,8 +151,11 @@ where
 
 	let block_ws_uri = command.block_ws_uri::<Block>();
 	let block_at = command.block_at::<Block>(block_ws_uri.clone()).await?;
-	let rpc_service = rpc_api::RpcService::new(block_ws_uri.clone(), false).await?;
-	let block: Block = rpc_service.get_block::<Block>(block_at).await?;
+	let rpc = ws_client(&block_ws_uri).await?;
+	let block: Block = ChainApi::<(), Block::Hash, Block::Header, _>::block(&rpc, Some(block_at))
+		.await
+		.unwrap()
+		.unwrap();
 	let parent_hash = block.header().parent_hash();
 	log::info!(
 		target: LOG_TARGET,

substrate/utils/frame/try-runtime/cli/src/commands/follow_chain.rs

@@ -19,22 +19,15 @@ use crate::{
 	build_executor, ensure_matching_spec, extract_code, full_extensions, local_spec, parse,
 	state_machine_call_with_proof, SharedParams, LOG_TARGET,
 };
-use jsonrpsee::{
-	core::{
-		async_trait,
-		client::{Client, Subscription, SubscriptionClientT},
-	},
-	ws_client::WsClientBuilder,
-};
 use parity_scale_codec::{Decode, Encode};
-use remote_externalities::{rpc_api::RpcService, Builder, Mode, OnlineConfig};
+use remote_externalities::{Builder, Mode, OnlineConfig};
 use sc_executor::NativeExecutionDispatch;
 use sc_service::Configuration;
-use serde::de::DeserializeOwned;
+use serde::{de::DeserializeOwned, Serialize};
 use sp_core::H256;
 use sp_runtime::traits::{Block as BlockT, Header as HeaderT, NumberFor};
-use sp_weights::Weight;
-use std::{collections::VecDeque, fmt::Debug, str::FromStr};
+use std::{fmt::Debug, str::FromStr};
+use substrate_rpc_client::{ws_client, ChainApi, FinalizedHeaders, Subscription, WsClient};
 
 const SUB: &str = "chain_subscribeFinalizedHeads";
 const UN_SUB: &str = "chain_unsubscribeFinalizedHeads";
@@ -71,144 +64,19 @@ pub struct FollowChainCmd {
 ///
 /// Returns a pair `(client, subscription)` - `subscription` alone will be useless, because it
 /// relies on the related alive `client`.
-async fn start_subscribing<Header: DeserializeOwned>(
+async fn start_subscribing<Header: DeserializeOwned + Serialize + Send + Sync + 'static>(
 	url: &str,
-) -> sc_cli::Result<(Client, Subscription<Header>)> {
-	let client = WsClientBuilder::default()
-		.connection_timeout(std::time::Duration::new(20, 0))
-		.max_notifs_per_subscription(1024)
-		.max_request_body_size(u32::MAX)
-		.build(url)
-		.await
-		.map_err(|e| sc_cli::Error::Application(e.into()))?;
+) -> sc_cli::Result<(WsClient, Subscription<Header>)> {
+	let client = ws_client(url).await.map_err(|e| sc_cli::Error::Application(e.into()))?;
 
 	log::info!(target: LOG_TARGET, "subscribing to {:?} / {:?}", SUB, UN_SUB);
 
-	let sub = client
-		.subscribe(SUB, None, UN_SUB)
+	let sub = ChainApi::<(), (), Header, ()>::subscribe_finalized_heads(&client)
 		.await
 		.map_err(|e| sc_cli::Error::Application(e.into()))?;
 	Ok((client, sub))
 }
 
-/// Abstraction over RPC calling for headers.
-#[async_trait]
-trait HeaderProvider<Block: BlockT>
-where
-	Block::Header: HeaderT,
-{
-	/// Awaits for the header of the block with hash `hash`.
-	async fn get_header(&self, hash: Block::Hash) -> Block::Header;
-}
-
-#[async_trait]
-impl<Block: BlockT> HeaderProvider<Block> for RpcService
-where
-	Block::Header: DeserializeOwned,
-{
-	async fn get_header(&self, hash: Block::Hash) -> Block::Header {
-		self.get_header::<Block>(hash).await.unwrap()
-	}
-}
-
-/// Abstraction over RPC subscription for finalized headers.
-#[async_trait]
-trait HeaderSubscription<Block: BlockT>
-where
-	Block::Header: HeaderT,
-{
-	/// Await for the next finalized header from the subscription.
-	///
-	/// Returns `None` if either the subscription has been closed or there was an error when reading
-	/// an object from the client.
-	async fn next_header(&mut self) -> Option<Block::Header>;
-}
-
-#[async_trait]
-impl<Block: BlockT> HeaderSubscription<Block> for Subscription<Block::Header>
-where
-	Block::Header: DeserializeOwned,
-{
-	async fn next_header(&mut self) -> Option<Block::Header> {
-		match self.next().await {
-			Some(Ok(header)) => Some(header),
-			None => {
-				log::warn!("subscription closed");
-				None
-			},
-			Some(Err(why)) => {
-				log::warn!("subscription returned error: {:?}. Probably decoding has failed.", why);
-				None
-			},
-		}
-	}
-}
-
-/// Stream of all finalized headers.
-///
-/// Returned headers are guaranteed to be ordered. There are no missing headers (even if some of
-/// them lack justification).
-struct FinalizedHeaders<'a, Block: BlockT, HP: HeaderProvider<Block>, HS: HeaderSubscription<Block>>
-{
-	header_provider: &'a HP,
-	subscription: HS,
-	fetched_headers: VecDeque<Block::Header>,
-	last_returned: Option<<Block::Header as HeaderT>::Hash>,
-}
-
-impl<'a, Block: BlockT, HP: HeaderProvider<Block>, HS: HeaderSubscription<Block>>
-	FinalizedHeaders<'a, Block, HP, HS>
-where
-	<Block as BlockT>::Header: DeserializeOwned,
-{
-	pub fn new(header_provider: &'a HP, subscription: HS) -> Self {
-		Self {
-			header_provider,
-			subscription,
-			fetched_headers: VecDeque::new(),
-			last_returned: None,
-		}
-	}
-
-	/// Reads next finalized header from the subscription. If some headers (without justification)
-	/// have been skipped, fetches them as well. Returns number of headers that have been fetched.
-	///
-	/// All fetched headers are stored in `self.fetched_headers`.
-	async fn fetch(&mut self) -> usize {
-		let last_finalized = match self.subscription.next_header().await {
-			Some(header) => header,
-			None => return 0,
-		};
-
-		self.fetched_headers.push_front(last_finalized.clone());
-
-		let mut last_finalized_parent = *last_finalized.parent_hash();
-		let last_returned = self.last_returned.unwrap_or(last_finalized_parent);
-
-		while last_finalized_parent != last_returned {
-			let parent_header = self.header_provider.get_header(last_finalized_parent).await;
-			self.fetched_headers.push_front(parent_header.clone());
-			last_finalized_parent = *parent_header.parent_hash();
-		}
-
-		self.fetched_headers.len()
-	}
-
-	/// Get the next finalized header.
-	pub async fn next(&mut self) -> Option<Block::Header> {
-		if self.fetched_headers.is_empty() {
-			self.fetch().await;
-		}
-
-		if let Some(header) = self.fetched_headers.pop_front() {
-			self.last_returned = Some(header.hash());
-			Some(header)
-		} else {
-			None
-		}
-	}
-}
-
 pub(crate) async fn follow_chain<Block, ExecDispatch>(
 	shared: SharedParams,
 	command: FollowChainCmd,
@@ -224,22 +92,23 @@ where
 	ExecDispatch: NativeExecutionDispatch + 'static,
 {
 	let mut maybe_state_ext = None;
-	let (_client, subscription) = start_subscribing::<Block::Header>(&command.uri).await?;
+	let (rpc, subscription) = start_subscribing::<Block::Header>(&command.uri).await?;
 
 	let (code_key, code) = extract_code(&config.chain_spec)?;
 	let executor = build_executor::<ExecDispatch>(&shared, &config);
 	let execution = shared.execution;
 
-	let rpc_service = RpcService::new(&command.uri, command.keep_connection).await?;
-
-	let mut finalized_headers: FinalizedHeaders<Block, RpcService, Subscription<Block::Header>> =
-		FinalizedHeaders::new(&rpc_service, subscription);
+	let mut finalized_headers: FinalizedHeaders<Block, _, _> =
+		FinalizedHeaders::new(&rpc, subscription);
 
 	while let Some(header) = finalized_headers.next().await {
 		let hash = header.hash();
 		let number = header.number();
 
-		let block = rpc_service.get_block::<Block>(hash).await.unwrap();
+		let block: Block = ChainApi::<(), Block::Hash, Block::Header, _>::block(&rpc, Some(hash))
+			.await
+			.unwrap()
+			.unwrap();
 
 		log::debug!(
 			target: LOG_TARGET,
@@ -295,7 +164,7 @@ where
 			full_extensions(),
 		)?;
 
-		let consumed_weight = <Weight as Decode>::decode(&mut &*encoded_result)
+		let consumed_weight = <sp_weights::Weight as Decode>::decode(&mut &*encoded_result)
 			.map_err(|e| format!("failed to decode weight: {:?}", e))?;
 
 		let storage_changes = changes
@@ -326,76 +195,3 @@ where
 	log::error!(target: LOG_TARGET, "ws subscription must have terminated.");
 	Ok(())
 }
-
-#[cfg(test)]
-mod tests {
-	use super::*;
-	use sp_runtime::testing::{Block as TBlock, ExtrinsicWrapper, Header};
-	use std::sync::Arc;
-	use tokio::sync::Mutex;
-
-	type Block = TBlock<ExtrinsicWrapper<()>>;
-	type BlockNumber = u64;
-	type Hash = H256;
-
-	struct MockHeaderProvider(pub Arc<Mutex<VecDeque<BlockNumber>>>);
-
-	fn headers() -> Vec<Header> {
-		let mut headers = vec![Header::new_from_number(0)];
-		for n in 1..11 {
-			headers.push(Header {
-				parent_hash: headers.last().unwrap().hash(),
-				..Header::new_from_number(n)
-			})
-		}
-		headers
-	}
-
-	#[async_trait]
-	impl HeaderProvider<Block> for MockHeaderProvider {
-		async fn get_header(&self, _hash: Hash) -> Header {
-			let height = self.0.lock().await.pop_front().unwrap();
-			headers()[height as usize].clone()
-		}
-	}
-
-	struct MockHeaderSubscription(pub VecDeque<BlockNumber>);
-
-	#[async_trait]
-	impl HeaderSubscription<Block> for MockHeaderSubscription {
-		async fn next_header(&mut self) -> Option<Header> {
-			self.0.pop_front().map(|h| headers()[h as usize].clone())
-		}
-	}
-
-	#[tokio::test]
-	async fn finalized_headers_works_when_every_block_comes_from_subscription() {
-		let heights = vec![4, 5, 6, 7];
-
-		let provider = MockHeaderProvider(Default::default());
-		let subscription = MockHeaderSubscription(heights.clone().into());
-		let mut headers = FinalizedHeaders::new(&provider, subscription);
-
-		for h in heights {
-			assert_eq!(h, headers.next().await.unwrap().number);
-		}
-		assert_eq!(None, headers.next().await);
-	}
-
-	#[tokio::test]
-	async fn finalized_headers_come_from_subscription_and_provider_if_in_need() {
-		let all_heights = 3..11;
-		let heights_in_subscription = vec![3, 4, 6, 10];
-		// Consecutive headers will be requested in the reversed order.
-		let heights_not_in_subscription = vec![5, 9, 8, 7];
-
-		let provider = MockHeaderProvider(Arc::new(Mutex::new(heights_not_in_subscription.into())));
-		let subscription = MockHeaderSubscription(heights_in_subscription.into());
-		let mut headers = FinalizedHeaders::new(&provider, subscription);
-
-		for h in all_heights {
-			assert_eq!(h, headers.next().await.unwrap().number);
-		}
-		assert_eq!(None, headers.next().await);
-	}
-}

substrate/utils/frame/try-runtime/cli/src/commands/offchain_worker.rs

@@ -20,12 +20,12 @@ use crate::{
 	parse, state_machine_call, SharedParams, State, LOG_TARGET,
 };
 use parity_scale_codec::Encode;
-use remote_externalities::rpc_api;
 use sc_executor::NativeExecutionDispatch;
 use sc_service::Configuration;
 use sp_core::storage::well_known_keys;
 use sp_runtime::traits::{Block as BlockT, Header, NumberFor};
 use std::{fmt::Debug, str::FromStr};
+use substrate_rpc_client::{ws_client, ChainApi};
 
 /// Configurations of the [`Command::OffchainWorker`].
 #[derive(Debug, Clone, clap::Parser)]
@@ -117,8 +117,11 @@ where
 	let header_at = command.header_at::<Block>()?;
 	let header_ws_uri = command.header_ws_uri::<Block>();
 
-	let rpc_service = rpc_api::RpcService::new(header_ws_uri.clone(), false).await?;
-	let header = rpc_service.get_header::<Block>(header_at).await?;
+	let rpc = ws_client(&header_ws_uri).await?;
+	let header = ChainApi::<(), Block::Hash, Block::Header, ()>::header(&rpc, Some(header_at))
+		.await
+		.unwrap()
+		.unwrap();
 	log::info!(
 		target: LOG_TARGET,
 		"fetched header from {:?}, block number: {:?}",

substrate/utils/frame/try-runtime/cli/src/commands/on_runtime_upgrade.rs

@@ -45,6 +45,7 @@ where
 	Block: BlockT + serde::de::DeserializeOwned,
 	Block::Hash: FromStr,
 	<Block::Hash as FromStr>::Err: Debug,
+	Block::Header: serde::de::DeserializeOwned,
 	NumberFor<Block>: FromStr,
 	<NumberFor<Block> as FromStr>::Err: Debug,
 	ExecDispatch: NativeExecutionDispatch + 'static,

substrate/utils/frame/try-runtime/cli/src/lib.rs

@@ -267,8 +267,7 @@
 
 use parity_scale_codec::Decode;
 use remote_externalities::{
-	rpc_api::RpcService, Builder, Mode, OfflineConfig, OnlineConfig, SnapshotConfig,
-	TestExternalities,
+	Builder, Mode, OfflineConfig, OnlineConfig, SnapshotConfig, TestExternalities,
 };
 use sc_chain_spec::ChainSpec;
 use sc_cli::{
@@ -297,6 +296,7 @@ use sp_runtime::{
 use sp_state_machine::{OverlayedChanges, StateMachine, TrieBackendBuilder};
 use sp_version::StateVersion;
 use std::{fmt::Debug, path::PathBuf, str::FromStr};
+use substrate_rpc_client::{ws_client, StateApi};
 
 mod commands;
 pub(crate) mod parse;
@@ -633,9 +633,8 @@ pub(crate) async fn ensure_matching_spec<Block: BlockT + DeserializeOwned>(
 	expected_spec_version: u32,
 	relaxed: bool,
 ) {
-	let rpc_service = RpcService::new(uri.clone(), false).await.unwrap();
-	match rpc_service
-		.get_runtime_version::<Block>(None)
+	let rpc = ws_client(&uri).await.unwrap();
+	match StateApi::<Block::Hash>::runtime_version(&rpc, None)
 		.await
 		.map(|version| (String::from(version.spec_name.clone()), version.spec_version))
 		.map(|(spec_name, spec_version)| (spec_name.to_lowercase(), spec_version))