companion for #8783 - jsonrpsee (#4344)

* add jsonrpsee glue code

* diener --substrate dp-jsonrpsee-integration-2

* cargo fmt

* update substrate

* fix build

* update substrate

* fix tests

* update substrate

* update substrate

* revert Cargo.toml

* revert changes in Cargo.toml

* jsonrpsee v0.11

* fix staking miner

* chore: update jsonrpsee v0.12

* update companion

* update companion

* fix changes in substrate

* revert requires_full_sync removal

* fix: read WS address from polkadot output

* fit nits

* fix more nits

* update lockfile for {"substrate"}

* cargo fmt

Co-authored-by: parity-processbot <>

polkadot/node/metrics/src/tests.rs

@@ -25,7 +25,7 @@ use std::collections::HashMap;
 
 const DEFAULT_PROMETHEUS_PORT: u16 = 9616;
 
-#[substrate_test_utils::test]
+#[substrate_test_utils::test(flavor = "multi_thread")]
 async fn runtime_can_publish_metrics() {
 	let mut alice_config =
 		node_config(|| {}, tokio::runtime::Handle::current(), Alice, Vec::new(), true);

polkadot/node/network/protocol/Cargo.toml

@@ -21,4 +21,4 @@ rand = "0.8"
 derive_more = "0.99"
 
 [dev-dependencies]
-rand_chacha = "0.3.1"
+rand_chacha = "0.3.1"

polkadot/node/service/Cargo.toml

@@ -199,4 +199,4 @@ runtime-metrics = [
 	"polkadot-runtime-parachains/runtime-metrics"
 ]
 
-staging-client = ["polkadot-node-core-provisioner/staging-client"]
+staging-client = ["polkadot-node-core-provisioner/staging-client"]

polkadot/node/service/src/lib.rs

@@ -444,7 +444,10 @@ fn new_partial<RuntimeApi, ExecutorDispatch, ChainSelection>(
 		sc_consensus::DefaultImportQueue<Block, FullClient<RuntimeApi, ExecutorDispatch>>,
 		sc_transaction_pool::FullPool<Block, FullClient<RuntimeApi, ExecutorDispatch>>,
 		(
-			impl service::RpcExtensionBuilder,
+			impl Fn(
+				polkadot_rpc::DenyUnsafe,
+				polkadot_rpc::SubscriptionTaskExecutor,
+			) -> Result<polkadot_rpc::RpcExtension, SubstrateServiceError>,
 			(
 				babe::BabeBlockImport<
 					Block,
@@ -938,7 +941,7 @@ where
 		client: client.clone(),
 		keystore: keystore_container.sync_keystore(),
 		network: network.clone(),
-		rpc_extensions_builder: Box::new(rpc_extensions_builder),
+		rpc_builder: Box::new(rpc_extensions_builder),
 		transaction_pool: transaction_pool.clone(),
 		task_manager: &mut task_manager,
 		system_rpc_tx,

polkadot/node/test/service/src/lib.rs

@@ -190,9 +190,13 @@ pub fn node_config(
 		rpc_ws: None,
 		rpc_ipc: None,
 		rpc_max_payload: None,
+		rpc_max_request_size: None,
+		rpc_max_response_size: None,
 		rpc_ws_max_connections: None,
 		rpc_cors: None,
 		rpc_methods: Default::default(),
+		rpc_id_provider: None,
+		rpc_max_subs_per_conn: None,
 		ws_max_out_buffer_capacity: None,
 		prometheus_config: None,
 		telemetry_endpoints: None,

polkadot/node/test/service/tests/build-blocks.rs

@@ -18,7 +18,7 @@ use futures::{future, pin_mut, select, FutureExt};
 use polkadot_test_service::*;
 use sp_keyring::Sr25519Keyring;
 
-#[substrate_test_utils::test]
+#[substrate_test_utils::test(flavor = "multi_thread")]
 async fn ensure_test_service_build_blocks() {
 	let mut builder = sc_cli::LoggerBuilder::new("");
 	builder.with_colors(false);

polkadot/node/test/service/tests/call-function.rs

@@ -17,7 +17,7 @@
 use polkadot_test_service::*;
 use sp_keyring::Sr25519Keyring::{Alice, Bob, Charlie};
 
-#[substrate_test_utils::test]
+#[substrate_test_utils::test(flavor = "multi_thread")]
 async fn call_function_actually_work() {
 	let alice_config =
 		node_config(|| {}, tokio::runtime::Handle::current(), Alice, Vec::new(), true);
@@ -30,7 +30,7 @@ async fn call_function_actually_work() {
 	});
 	let output = alice.send_extrinsic(function, Bob).await.unwrap();
 
-	let res = output.result.expect("return value expected");
+	let res = output.result;
 	let json = serde_json::from_str::<serde_json::Value>(res.as_str()).expect("valid JSON");
 	let object = json.as_object().expect("JSON is an object");
 	assert!(object.contains_key("jsonrpc"), "key jsonrpc exists");

polkadot/parachain/test-parachains/adder/collator/tests/integration.rs

@@ -20,7 +20,8 @@
 const PUPPET_EXE: &str = env!("CARGO_BIN_EXE_adder_collator_puppet_worker");
 
 // If this test is failing, make sure to run all tests with the `real-overseer` feature being enabled.
-#[substrate_test_utils::test]
+
+#[substrate_test_utils::test(flavor = "multi_thread")]
 async fn collating_using_adder_collator() {
 	use polkadot_primitives::v2::Id as ParaId;
 	use sp_keyring::AccountKeyring::*;

polkadot/parachain/test-parachains/undying/collator/tests/integration.rs

@@ -20,7 +20,7 @@
 const PUPPET_EXE: &str = env!("CARGO_BIN_EXE_undying_collator_puppet_worker");
 
 // If this test is failing, make sure to run all tests with the `real-overseer` feature being enabled.
-#[substrate_test_utils::test]
+#[substrate_test_utils::test(flavor = "multi_thread")]
 async fn collating_using_undying_collator() {
 	use polkadot_primitives::v2::Id as ParaId;
 	use sp_keyring::AccountKeyring::*;

polkadot/rpc/Cargo.toml

@@ -5,7 +5,7 @@ authors = ["Parity Technologies <admin@parity.io>"]
 edition = "2021"
 
 [dependencies]
-jsonrpc-core = "18.0.0"
+jsonrpsee = { version = "0.12.0", features = ["server"] }
 polkadot-primitives = { path = "../primitives" }
 sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "master" }

polkadot/rpc/src/lib.rs

@@ -20,12 +20,12 @@
 
 use std::sync::Arc;
 
+use jsonrpsee::RpcModule;
 use polkadot_primitives::v2::{AccountId, Balance, Block, BlockNumber, Hash, Nonce};
 use sc_client_api::AuxStore;
 use sc_consensus_babe::Epoch;
 use sc_finality_grandpa::FinalityProofProvider;
 pub use sc_rpc::{DenyUnsafe, SubscriptionTaskExecutor};
-use sc_sync_state_rpc::{SyncStateRpcApi, SyncStateRpcHandler};
 use sp_api::ProvideRuntimeApi;
 use sp_block_builder::BlockBuilder;
 use sp_blockchain::{Error as BlockChainError, HeaderBackend, HeaderMetadata};
@@ -35,7 +35,7 @@ use sp_keystore::SyncCryptoStorePtr;
 use txpool_api::TransactionPool;
 
 /// A type representing all RPC extensions.
-pub type RpcExtension = jsonrpc_core::IoHandler<sc_rpc::Metadata>;
+pub type RpcExtension = RpcModule<()>;
 
 /// Extra dependencies for BABE.
 pub struct BabeDeps {
@@ -115,13 +115,16 @@ where
 	B: sc_client_api::Backend<Block> + Send + Sync + 'static,
 	B::State: sc_client_api::StateBackend<sp_runtime::traits::HashFor<Block>>,
 {
-	use frame_rpc_system::{FullSystem, SystemApi};
-	use pallet_mmr_rpc::{Mmr, MmrApi};
-	use pallet_transaction_payment_rpc::{TransactionPayment, TransactionPaymentApi};
-	use sc_consensus_babe_rpc::BabeRpcHandler;
-	use sc_finality_grandpa_rpc::{GrandpaApi, GrandpaRpcHandler};
+	use beefy_gadget_rpc::{BeefyApiServer, BeefyRpcHandler};
+	use frame_rpc_system::{SystemApiServer, SystemRpc};
+	use pallet_mmr_rpc::{MmrApiServer, MmrRpc};
+	use pallet_transaction_payment_rpc::{TransactionPaymentApiServer, TransactionPaymentRpc};
+	use sc_consensus_babe_rpc::{BabeApiServer, BabeRpc};
+	use sc_finality_grandpa_rpc::{GrandpaApiServer, GrandpaRpc};
+	use sc_sync_state_rpc::{SyncStateRpc, SyncStateRpcApiServer};
+	use substrate_state_trie_migration_rpc::StateMigrationApiServer;
 
-	let mut io = jsonrpc_core::IoHandler::default();
+	let mut io = RpcModule::new(());
 	let FullDeps { client, pool, select_chain, chain_spec, deny_unsafe, babe, grandpa, beefy } =
 		deps;
 	let BabeDeps { keystore, babe_config, shared_epoch_changes } = babe;
@@ -133,41 +136,47 @@ where
 		finality_provider,
 	} = grandpa;
 
-	io.extend_with(substrate_state_trie_migration_rpc::StateMigrationApi::to_delegate(
-		substrate_state_trie_migration_rpc::MigrationRpc::new(client.clone(), backend, deny_unsafe),
-	));
-
-	io.extend_with(SystemApi::to_delegate(FullSystem::new(client.clone(), pool, deny_unsafe)));
-	io.extend_with(TransactionPaymentApi::to_delegate(TransactionPayment::new(client.clone())));
-	io.extend_with(MmrApi::to_delegate(Mmr::new(client.clone())));
-	io.extend_with(sc_consensus_babe_rpc::BabeApi::to_delegate(BabeRpcHandler::new(
-		client.clone(),
-		shared_epoch_changes.clone(),
-		keystore,
-		babe_config,
-		select_chain,
-		deny_unsafe,
-	)));
-	io.extend_with(GrandpaApi::to_delegate(GrandpaRpcHandler::new(
-		shared_authority_set.clone(),
-		shared_voter_state,
-		justification_stream,
-		subscription_executor,
-		finality_provider,
-	)));
-	io.extend_with(SyncStateRpcApi::to_delegate(SyncStateRpcHandler::new(
-		chain_spec,
-		client,
-		shared_authority_set,
-		shared_epoch_changes,
-	)?));
-
-	let handler: beefy_gadget_rpc::BeefyRpcHandler<Block> = beefy_gadget_rpc::BeefyRpcHandler::new(
-		beefy.beefy_commitment_stream,
-		beefy.beefy_best_block_stream,
-		beefy.subscription_executor,
+	io.merge(
+		substrate_state_trie_migration_rpc::MigrationRpc::new(client.clone(), backend, deny_unsafe)
+			.into_rpc(),
+	)?;
+	io.merge(SystemRpc::new(client.clone(), pool.clone(), deny_unsafe).into_rpc())?;
+	io.merge(TransactionPaymentRpc::new(client.clone()).into_rpc())?;
+	io.merge(MmrRpc::new(client.clone()).into_rpc())?;
+	io.merge(
+		BabeRpc::new(
+			client.clone(),
+			shared_epoch_changes.clone(),
+			keystore,
+			babe_config,
+			select_chain,
+			deny_unsafe,
+		)
+		.into_rpc(),
+	)?;
+	io.merge(
+		GrandpaRpc::new(
+			subscription_executor,
+			shared_authority_set.clone(),
+			shared_voter_state,
+			justification_stream,
+			finality_provider,
+		)
+		.into_rpc(),
+	)?;
+	io.merge(
+		SyncStateRpc::new(chain_spec, client, shared_authority_set, shared_epoch_changes)?
+			.into_rpc(),
+	)?;
+
+	io.merge(
+		BeefyRpcHandler::<Block>::new(
+			beefy.beefy_commitment_stream,
+			beefy.beefy_best_block_stream,
+			beefy.subscription_executor,
+		)?
+		.into_rpc(),
 	)?;
-	io.extend_with(beefy_gadget_rpc::BeefyApi::to_delegate(handler));
 
 	Ok(io)
 }

polkadot/runtime/parachains/Cargo.toml

@@ -109,4 +109,4 @@ try-runtime = [
 	"pallet-vesting/try-runtime",
 ]
 runtime-metrics = ["sp-tracing/with-tracing", "polkadot-runtime-metrics/runtime-metrics"]
-vstaging = []
+vstaging = []

polkadot/tests/benchmark_block_works.rs

@@ -22,7 +22,12 @@ use nix::{
 	sys::signal::{kill, Signal::SIGINT},
 	unistd::Pid,
 };
-use std::{path::Path, process::Command, result::Result, time::Duration};
+use std::{
+	path::Path,
+	process::{self, Command},
+	result::Result,
+	time::Duration,
+};
 use tempfile::tempdir;
 
 pub mod common;
@@ -47,6 +52,8 @@ async fn benchmark_block_works() {
 /// Builds a chain with one block for the given runtime and base path.
 async fn build_chain(runtime: &str, base_path: &Path) -> Result<(), String> {
 	let mut cmd = Command::new(cargo_bin("polkadot"))
+		.stdout(process::Stdio::piped())
+		.stderr(process::Stdio::piped())
 		.args(["--chain", &runtime, "--force-authoring", "--alice"])
 		.arg("-d")
 		.arg(base_path)
@@ -55,8 +62,10 @@ async fn build_chain(runtime: &str, base_path: &Path) -> Result<(), String> {
 		.spawn()
 		.unwrap();
 
+	let (ws_url, _) = common::find_ws_url_from_output(cmd.stderr.take().unwrap());
+
 	// Wait for the chain to produce one block.
-	let ok = common::wait_n_finalized_blocks(1, Duration::from_secs(60)).await;
+	let ok = common::wait_n_finalized_blocks(1, Duration::from_secs(60), &ws_url).await;
 	// Send SIGINT to node.
 	kill(Pid::from_raw(cmd.id().try_into().unwrap()), SIGINT).unwrap();
 	// Wait for the node to handle it and exit.

polkadot/tests/common.rs

@@ -17,14 +17,13 @@
 use polkadot_core_primitives::Block;
 use remote_externalities::rpc_api::get_finalized_head;
 use std::{
+	io::{BufRead, BufReader, Read},
 	process::{Child, ExitStatus},
 	thread,
 	time::Duration,
 };
 use tokio::time::timeout;
 
-static LOCALHOST_WS: &str = "ws://127.0.0.1:9944/";
-
 /// Wait for the given `child` the given amount of `secs`.
 ///
 /// Returns the `Some(exit status)` or `None` if the process did not finish in the given time.
@@ -46,8 +45,9 @@ pub fn wait_for(child: &mut Child, secs: usize) -> Option<ExitStatus> {
 pub async fn wait_n_finalized_blocks(
 	n: usize,
 	timeout_duration: Duration,
+	url: &str,
 ) -> Result<(), tokio::time::error::Elapsed> {
-	timeout(timeout_duration, wait_n_finalized_blocks_from(n, LOCALHOST_WS)).await
+	timeout(timeout_duration, wait_n_finalized_blocks_from(n, url)).await
 }
 
 /// Wait for at least `n` blocks to be finalized from a specified node.
@@ -65,3 +65,32 @@ async fn wait_n_finalized_blocks_from(n: usize, url: &str) {
 		interval.tick().await;
 	}
 }
+
+/// Read the WS address from the output.
+///
+/// This is hack to get the actual binded sockaddr because
+/// polkadot assigns a random port if the specified port was already binded.
+///
+/// You must call `Command::new("cmd").stdout(process::Stdio::piped()).stderr(process::Stdio::piped())`
+/// for this to work.
+pub fn find_ws_url_from_output(read: impl Read + Send) -> (String, String) {
+	let mut data = String::new();
+
+	let ws_url = BufReader::new(read)
+		.lines()
+		.find_map(|line| {
+			let line = line.expect("failed to obtain next line from stdout for port discovery");
+
+			data.push_str(&line);
+
+			// does the line contain our port (we expect this specific output from substrate).
+			let sock_addr = match line.split_once("Running JSON-RPC WS server: addr=") {
+				None => return None,
+				Some((_, after)) => after.split_once(",").unwrap().0,
+			};
+
+			Some(format!("ws://{}", sock_addr))
+		})
+		.expect("We should get a WebSocket address");
+	(ws_url, data)
+}

polkadot/tests/purge_chain_works.rs

@@ -15,7 +15,10 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use assert_cmd::cargo::cargo_bin;
-use std::{process::Command, time::Duration};
+use std::{
+	process::{self, Command},
+	time::Duration,
+};
 use tempfile::tempdir;
 
 pub mod common;
@@ -31,6 +34,8 @@ async fn purge_chain_rocksdb_works() {
 	let tmpdir = tempdir().expect("could not create temp dir");
 
 	let mut cmd = Command::new(cargo_bin("polkadot"))
+		.stdout(process::Stdio::piped())
+		.stderr(process::Stdio::piped())
 		.args(&["--dev", "-d"])
 		.arg(tmpdir.path())
 		.arg("--port")
@@ -39,8 +44,12 @@ async fn purge_chain_rocksdb_works() {
 		.spawn()
 		.unwrap();
 
+	let (ws_url, _) = common::find_ws_url_from_output(cmd.stderr.take().unwrap());
+
 	// Let it produce 1 block.
-	common::wait_n_finalized_blocks(1, Duration::from_secs(60)).await.unwrap();
+	common::wait_n_finalized_blocks(1, Duration::from_secs(60), &ws_url)
+		.await
+		.unwrap();
 
 	// Send SIGINT to node.
 	kill(Pid::from_raw(cmd.id().try_into().unwrap()), SIGINT).unwrap();
@@ -75,6 +84,8 @@ async fn purge_chain_paritydb_works() {
 	let tmpdir = tempdir().expect("could not create temp dir");
 
 	let mut cmd = Command::new(cargo_bin("polkadot"))
+		.stdout(process::Stdio::piped())
+		.stderr(process::Stdio::piped())
 		.args(&["--dev", "-d"])
 		.arg(tmpdir.path())
 		.arg("--database")
@@ -83,8 +94,12 @@ async fn purge_chain_paritydb_works() {
 		.spawn()
 		.unwrap();
 
+	let (ws_url, _) = common::find_ws_url_from_output(cmd.stderr.take().unwrap());
+
 	// Let it produce 1 block.
-	common::wait_n_finalized_blocks(1, Duration::from_secs(60)).await.unwrap();
+	common::wait_n_finalized_blocks(1, Duration::from_secs(60), &ws_url)
+		.await
+		.unwrap();
 
 	// Send SIGINT to node.
 	kill(Pid::from_raw(cmd.id().try_into().unwrap()), SIGINT).unwrap();

polkadot/tests/running_the_node_and_interrupt.rs

@@ -15,7 +15,10 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use assert_cmd::cargo::cargo_bin;
-use std::{process::Command, time::Duration};
+use std::{
+	process::{self, Command},
+	time::Duration,
+};
 use tempfile::tempdir;
 
 pub mod common;
@@ -35,21 +38,27 @@ async fn running_the_node_works_and_can_be_interrupted() {
 		let tmpdir = tempdir().expect("coult not create temp dir");
 
 		let mut cmd = Command::new(cargo_bin("polkadot"))
+			.stdout(process::Stdio::piped())
+			.stderr(process::Stdio::piped())
 			.args(&["--dev", "-d"])
 			.arg(tmpdir.path())
 			.arg("--no-hardware-benchmarks")
 			.spawn()
 			.unwrap();
 
+		let (ws_url, _) = common::find_ws_url_from_output(cmd.stderr.take().unwrap());
+
 		// Let it produce three blocks.
-		common::wait_n_finalized_blocks(3, Duration::from_secs(60)).await.unwrap();
+		common::wait_n_finalized_blocks(3, Duration::from_secs(60), &ws_url)
+			.await
+			.unwrap();
 
 		assert!(cmd.try_wait().unwrap().is_none(), "the process should still be running");
 		kill(Pid::from_raw(cmd.id().try_into().unwrap()), signal).unwrap();
 		assert_eq!(
 			common::wait_for(&mut cmd, 30).map(|x| x.success()),
 			Some(true),
-			"the pocess must exit gracefully after signal {}",
+			"the process must exit gracefully after signal {}",
 			signal,
 		);
 	}

polkadot/utils/staking-miner/Cargo.toml

@@ -8,7 +8,7 @@ edition = "2021"
 codec = { package = "parity-scale-codec", version = "3.0.0" }
 clap = { version = "3.1", features = ["derive", "env"] }
 tracing-subscriber = { version = "0.3.10", features = ["env-filter"]   }
-jsonrpsee = { version = "0.10.1", features = ["ws-client", "macros"] }
+jsonrpsee = { version = "0.12.0", features = ["ws-client", "macros"] }
 log = "0.4.17"
 paste = "1.0.7"
 serde = "1.0.137"

polkadot/utils/staking-miner/src/monitor.rs

@@ -139,12 +139,6 @@ macro_rules! monitor_cmd_for { ($runtime:tt) => { paste::paste! {
 				maybe_rp = subscription.next() => {
 					match maybe_rp {
 						Some(Ok(r)) => r,
-						// Custom `jsonrpsee` message sent by the server if the subscription was closed on the server side.
-						Some(Err(RpcError::SubscriptionClosed(reason))) => {
-							log::warn!(target: LOG_TARGET, "subscription to `subscribeNewHeads/subscribeFinalizedHeads` terminated: {:?}. Retrying..", reason);
-							subscription = heads_subscription().await?;
-							continue;
-						}
 						Some(Err(e)) => {
 							log::error!(target: LOG_TARGET, "subscription failed to decode Header {:?}, this is bug please file an issue", e);
 							return Err(e.into());
@@ -312,15 +306,6 @@ macro_rules! monitor_cmd_for { ($runtime:tt) => { paste::paste! {
 			while let Some(rp) = tx_subscription.next().await {
 				let status_update = match rp {
 					Ok(r) => r,
-					// Custom `jsonrpsee` message sent by the server if the subscription was closed on the server side.
-					Err(RpcError::SubscriptionClosed(reason)) => {
-						log::warn!(
-							target: LOG_TARGET,
-							"tx subscription closed by the server: {:?}; skip block: {}",
-							reason, at.number
-						);
-						return;
-					},
 					Err(e) => {
 						log::error!(target: LOG_TARGET, "subscription failed to decode TransactionStatus {:?}, this is a bug please file an issue", e);
 						let _ = tx.send(e.into());

polkadot/utils/staking-miner/src/rpc.rs

@@ -77,7 +77,7 @@ pub trait RpcApi {
 		unsubscribe = "author_unwatchExtrinsic",
 		item = TransactionStatus<Hash, Hash>
 	)]
-	fn watch_extrinsic(&self, bytes: &Bytes) -> RpcResult<()>;
+	fn watch_extrinsic(&self, bytes: &Bytes);
 
 	/// New head subscription.
 	#[subscription(
@@ -85,7 +85,7 @@ pub trait RpcApi {
 		unsubscribe = "chain_unsubscribeNewHeads",
 		item = Header
 	)]
-	fn subscribe_new_heads(&self) -> RpcResult<()>;
+	fn subscribe_new_heads(&self);
 
 	/// Finalized head subscription.
 	#[subscription(
@@ -93,7 +93,7 @@ pub trait RpcApi {
 		unsubscribe = "chain_unsubscribeFinalizedHeads",
 		item = Header
 	)]
-	fn subscribe_finalized_heads(&self) -> RpcResult<()>;
+	fn subscribe_finalized_heads(&self);
 }
 
 /// Wraps a shared web-socket JSON-RPC client that can be cloned.