staking miner: reuse ws conn for remote-ext

utils/staking-miner/Cargo.toml

@@ -7,8 +7,8 @@ edition = "2018"
 [dependencies]
 clap = { version = "3.0", features = ["derive", "env"] }
 codec = { package = "parity-scale-codec", version = "2.0.0" }
-env_logger = "0.9.0"
-jsonrpsee = { version = "0.8", features = ["ws-client"] }
+tracing-subscriber = { version = "0.3.8", features = ["env-filter"]   }
+jsonrpsee = { version = "0.8", features = ["ws-client", "macros"] }
 log = "0.4.11"
 paste = "1.0.6"
 serde = "1.0.136"

utils/staking-miner/src/dry_run.rs

@@ -16,12 +16,10 @@
 
 //! The dry-run command.
 
-use crate::{
-	prelude::*, rpc_helpers::*, signer::Signer, DryRunConfig, Error, SharedConfig, WsClient,
-};
+use crate::{prelude::*, rpc::*, signer::Signer, DryRunConfig, Error, SharedRpcClient};
 use codec::Encode;
 use frame_support::traits::Currency;
-use jsonrpsee::rpc_params;
+use sp_core::Bytes;
 use sp_npos_elections::ElectionScore;
 
 /// Forcefully create the snapshot. This can be used to compute the election at anytime.
@@ -39,10 +37,10 @@ fn force_create_snapshot<T: EPM::Config>(ext: &mut Ext) -> Result<(), Error<T>>
 
 /// Helper method to print the encoded size of the snapshot.
 async fn print_info<T: EPM::Config>(
-	client: &WsClient,
+	rpc: &SharedRpcClient,
 	ext: &mut Ext,
 	raw_solution: &EPM::RawSolution<EPM::SolutionOf<T>>,
-	extrinsic: sp_core::Bytes,
+	extrinsic: &Bytes,
 ) where
 	<T as EPM::Config>::Currency: Currency<T::AccountId, Balance = Balance>,
 {
@@ -79,12 +77,8 @@ async fn print_info<T: EPM::Config>(
 		);
 	});
 
-	let info = rpc::<pallet_transaction_payment::RuntimeDispatchInfo<Balance>>(
-		client,
-		"payment_queryInfo",
-		rpc_params! { extrinsic },
-	)
-	.await;
+	let info = rpc.payment_query_info(&extrinsic, None).await;
+
 	log::info!(
 		target: LOG_TARGET,
 		"payment_queryInfo: (fee = {}) {:?}",
@@ -116,22 +110,21 @@ fn find_threshold<T: EPM::Config>(ext: &mut Ext, count: usize) {
 macro_rules! dry_run_cmd_for { ($runtime:ident) => { paste::paste! {
 	/// Execute the dry-run command.
 	pub(crate) async fn [<dry_run_cmd_ $runtime>](
-		client: &WsClient,
-		shared: SharedConfig,
+		rpc: SharedRpcClient,
 		config: DryRunConfig,
 		signer: Signer,
 	) -> Result<(), Error<$crate::[<$runtime _runtime_exports>]::Runtime>> {
 		use $crate::[<$runtime _runtime_exports>]::*;
 		let mut ext = crate::create_election_ext::<Runtime, Block>(
-			shared.uri.clone(),
+			rpc.clone(),
 			config.at,
 			vec!["Staking".to_string(), "System".to_string()],
 		).await?;
 		force_create_snapshot::<Runtime>(&mut ext)?;
 
 		let (raw_solution, witness) = crate::mine_with::<Runtime>(&config.solver, &mut ext, false)?;
 
-		let nonce = crate::get_account_info::<Runtime>(client, &signer.account, config.at)
+		let nonce = crate::get_account_info::<Runtime>(&rpc, &signer.account, config.at)
 			.await?
 			.map(|i| i.nonce)
 			.expect("signer account is checked to exist upon startup; it can only die if it \
@@ -143,7 +136,7 @@ macro_rules! dry_run_cmd_for { ($runtime:ident) => { paste::paste! {
 		let extrinsic = ext.execute_with(|| create_uxt(raw_solution.clone(), witness, signer.clone(), nonce, tip, era));
 
 		let bytes = sp_core::Bytes(extrinsic.encode().to_vec());
-		print_info::<Runtime>(client, &mut ext, &raw_solution, bytes.clone()).await;
+		print_info::<Runtime>(&rpc, &mut ext, &raw_solution, &bytes).await;
 
 		let feasibility_result = ext.execute_with(|| {
 			EPM::Pallet::<Runtime>::feasibility_check(raw_solution.clone(), EPM::ElectionCompute::Signed)
@@ -157,9 +150,8 @@ macro_rules! dry_run_cmd_for { ($runtime:ident) => { paste::paste! {
 		});
 		log::info!(target: LOG_TARGET, "dispatch result is {:?}", dispatch_result);
 
-		let outcome = rpc_decode::<sp_runtime::ApplyExtrinsicResult>(client, "system_dryRun", rpc_params!{ bytes })
-			.await
-			.map_err::<Error<Runtime>, _>(Into::into)?;
+		let dry_run_fut = rpc.dry_run(&bytes, None);
+		let outcome: sp_runtime::ApplyExtrinsicResult = await_request_and_decode(dry_run_fut).await.map_err::<Error<Runtime>, _>(Into::into)?;
 		log::info!(target: LOG_TARGET, "dry-run outcome is {:?}", outcome);
 		Ok(())
 	}

utils/staking-miner/src/emergency_solution.rs

@@ -16,19 +16,19 @@
 
 //! The emergency-solution command.
 
-use crate::{prelude::*, EmergencySolutionConfig, Error, SharedConfig};
+use crate::{prelude::*, EmergencySolutionConfig, Error, SharedRpcClient};
 use codec::Encode;
 use std::io::Write;
 
 macro_rules! emergency_solution_cmd_for { ($runtime:ident) => { paste::paste! {
 	/// Execute the emergency-solution command.
 	pub(crate) async fn [<emergency_solution_cmd_ $runtime>](
-		shared: SharedConfig,
+		client: SharedRpcClient,
 		config: EmergencySolutionConfig,
 	) -> Result<(), Error<$crate::[<$runtime _runtime_exports>]::Runtime>> {
 		use $crate::[<$runtime _runtime_exports>]::*;
 
-		let mut ext = crate::create_election_ext::<Runtime, Block>(shared.uri.clone(), config.at, vec![]).await?;
+		let mut ext = crate::create_election_ext::<Runtime, Block>(client, config.at, vec![]).await?;
 		let (raw_solution, _witness) = crate::mine_with::<Runtime>(&config.solver, &mut ext, false)?;
 
 		ext.execute_with(|| {

utils/staking-miner/src/main.rs

@@ -32,7 +32,7 @@ mod dry_run;
 mod emergency_solution;
 mod monitor;
 mod prelude;
-mod rpc_helpers;
+mod rpc;
 mod signer;
 
 pub(crate) use prelude::*;
@@ -43,8 +43,12 @@ use frame_election_provider_support::NposSolver;
 use frame_support::traits::Get;
 use jsonrpsee::ws_client::{WsClient, WsClientBuilder};
 use remote_externalities::{Builder, Mode, OnlineConfig};
+use rpc::{RpcApiClient, SharedRpcClient};
 use sp_npos_elections::ExtendedBalance;
 use sp_runtime::{traits::Block as BlockT, DeserializeOwned};
+use tracing_subscriber::{fmt, EnvFilter};
+
+use std::{ops::Deref, sync::Arc};
 
 pub(crate) enum AnyRuntime {
 	Polkadot,
@@ -90,10 +94,10 @@ macro_rules! construct_runtime_prelude {
 					let extra: SignedExtra = crate::[<signed_ext_builder_ $runtime>](nonce, tip, era);
 					let raw_payload = SignedPayload::new(call, extra).expect("creating signed payload infallible; qed.");
 					let signature = raw_payload.using_encoded(|payload| {
-						pair.clone().sign(payload)
+						pair.sign(payload)
 					});
 					let (call, extra, _) = raw_payload.deconstruct();
-					let address = <Runtime as frame_system::Config>::Lookup::unlookup(account.clone());
+					let address = <Runtime as frame_system::Config>::Lookup::unlookup(account);
 					let extrinsic = UncheckedExtrinsic::new_signed(call, address, signature.into(), extra);
 					log::debug!(
 						target: crate::LOG_TARGET, "constructed extrinsic {} with length {}",
@@ -229,7 +233,7 @@ macro_rules! any_runtime_unit {
 enum Error<T: EPM::Config> {
 	Io(#[from] std::io::Error),
 	JsonRpsee(#[from] jsonrpsee::core::Error),
-	RpcHelperError(#[from] rpc_helpers::RpcHelperError),
+	RpcHelperError(#[from] rpc::RpcHelperError),
 	Codec(#[from] codec::Error),
 	Crypto(sp_core::crypto::SecretStringError),
 	RemoteExternalities(&'static str),
@@ -368,7 +372,7 @@ struct Opt {
 /// Build the Ext at hash with all the data of `ElectionProviderMultiPhase` and any additional
 /// pallets.
 async fn create_election_ext<T: EPM::Config, B: BlockT + DeserializeOwned>(
-	uri: String,
+	client: SharedRpcClient,
 	at: Option<B::Hash>,
 	additional: Vec<String>,
 ) -> Result<Ext, Error<T>> {
@@ -381,7 +385,7 @@ async fn create_election_ext<T: EPM::Config, B: BlockT + DeserializeOwned>(
 	pallets.extend(additional);
 	Builder::<B>::new()
 		.mode(Mode::Online(OnlineConfig {
-			transport: uri.into(),
+			transport: client.into_inner().into(),
 			at,
 			pallets,
 			..Default::default()
@@ -493,13 +497,13 @@ fn mine_dpos<T: EPM::Config>(ext: &mut Ext) -> Result<(), Error<T>> {
 }
 
 pub(crate) async fn check_versions<T: frame_system::Config + EPM::Config>(
-	client: &WsClient,
+	rpc: &SharedRpcClient,
 ) -> Result<(), Error<T>> {
 	let linked_version = T::Version::get();
-	let on_chain_version =
-		rpc_helpers::rpc::<sp_version::RuntimeVersion>(client, "state_getRuntimeVersion", None)
-			.await
-			.expect("runtime version RPC should always work; qed");
+	let on_chain_version = rpc
+		.runtime_version(None)
+		.await
+		.expect("runtime version RPC should always work; qed");
 
 	log::debug!(target: LOG_TARGET, "linked version {:?}", linked_version);
 	log::debug!(target: LOG_TARGET, "on-chain version {:?}", on_chain_version);
@@ -517,35 +521,26 @@ pub(crate) async fn check_versions<T: frame_system::Config + EPM::Config>(
 
 #[tokio::main]
 async fn main() {
-	env_logger::Builder::from_default_env()
-		.format_module_path(true)
-		.format_level(true)
-		.init();
+	fmt().with_env_filter(EnvFilter::from_default_env()).init();
+
 	let Opt { shared, command } = Opt::parse();
 	log::debug!(target: LOG_TARGET, "attempting to connect to {:?}", shared.uri);
 
-	let client = loop {
-		let maybe_client = WsClientBuilder::default()
-			.connection_timeout(std::time::Duration::new(20, 0))
-			.max_request_body_size(u32::MAX)
-			.build(&shared.uri)
-			.await;
-		match maybe_client {
+	let rpc = loop {
+		match SharedRpcClient::new(&shared.uri).await {
 			Ok(client) => break client,
 			Err(why) => {
 				log::warn!(
 					target: LOG_TARGET,
 					"failed to connect to client due to {:?}, retrying soon..",
 					why
 				);
-				std::thread::sleep(std::time::Duration::from_millis(2500));
+				tokio::time::sleep(std::time::Duration::from_millis(2500)).await;
 			},
 		}
 	};
 
-	let chain = rpc_helpers::rpc::<String>(&client, "system_chain", None)
-		.await
-		.expect("system_chain infallible; qed.");
+	let chain: String = rpc.system_chain().await.expect("system_chain infallible; qed.");
 	match chain.to_lowercase().as_str() {
 		"polkadot" | "development" => {
 			sp_core::crypto::set_default_ss58_version(
@@ -591,26 +586,26 @@ async fn main() {
 	log::info!(target: LOG_TARGET, "connected to chain {:?}", chain);
 
 	any_runtime_unit! {
-		check_versions::<Runtime>(&client).await
+		check_versions::<Runtime>(&rpc).await
 	};
 
 	let signer_account = any_runtime! {
-		signer::signer_uri_from_string::<Runtime>(&shared.seed, &client)
+		signer::signer_uri_from_string::<Runtime>(&shared.seed, &rpc)
 			.await
 			.expect("Provided account is invalid, terminating.")
 	};
 
 	let outcome = any_runtime! {
-		match command.clone() {
-			Command::Monitor(c) => monitor_cmd(client, shared, c, signer_account).await
+		match command {
+			Command::Monitor(cmd) => monitor_cmd(rpc, cmd, signer_account).await
 				.map_err(|e| {
 					log::error!(target: LOG_TARGET, "Monitor error: {:?}", e);
 				}),
-			Command::DryRun(c) => dry_run_cmd(&client, shared, c, signer_account).await
+			Command::DryRun(cmd) => dry_run_cmd(rpc, cmd, signer_account).await
 				.map_err(|e| {
 					log::error!(target: LOG_TARGET, "DryRun error: {:?}", e);
 				}),
-			Command::EmergencySolution(c) => emergency_solution_cmd(shared.clone(), c).await
+			Command::EmergencySolution(cmd) => emergency_solution_cmd(rpc, cmd).await
 				.map_err(|e| {
 					log::error!(target: LOG_TARGET, "EmergencySolution error: {:?}", e);
 				}),