rpc: backpressured RPC server (bump jsonrpsee 0.20) (#1313)

This is a rather big change in jsonrpsee, the major things in this bump
are:
- Server backpressure (the subscription impls are modified to deal with
that)
- Allow custom error types / return types (remove jsonrpsee::core::Error
and jsonrpee::core::CallError)
- Bug fixes (graceful shutdown in particular not used by substrate
anyway)
   - Less dependencies for the clients in particular
   - Return type requires Clone in method call responses
   - Moved to tokio channels
   - Async subscription API (not used in this PR)

Major changes in this PR:
- The subscriptions are now bounded and if subscription can't keep up
with the server it is dropped
- CLI: add parameter to configure the jsonrpc server bounded message
buffer (default is 64)
- Add our own subscription helper to deal with the unbounded streams in
substrate

The most important things in this PR to review is the added helpers
functions in `substrate/client/rpc/src/utils.rs` and the rest is pretty
much chore.

Regarding the "bounded buffer limit" it may cause the server to handle
the JSON-RPC calls
slower than before.

The message size limit is bounded by "--rpc-response-size" thus "by
default 10MB * 64 = 640MB"
but the subscription message size is not covered by this limit and could
be capped as well.

Hopefully the last release prior to 1.0, sorry in advance for a big PR

Previous attempt: https://github.com/paritytech/substrate/pull/13992

Resolves https://github.com/paritytech/polkadot-sdk/issues/748, resolves
https://github.com/paritytech/polkadot-sdk/issues/627

substrate/client/rpc/src/author/mod.rs

@@ -23,15 +23,14 @@ mod tests;
 
 use std::sync::Arc;
 
-use crate::SubscriptionTaskExecutor;
+use crate::{
+	utils::{pipe_from_stream, spawn_subscription_task},
+	SubscriptionTaskExecutor,
+};
 
 use codec::{Decode, Encode};
-use futures::{FutureExt, TryFutureExt};
-use jsonrpsee::{
-	core::{async_trait, Error as JsonRpseeError, RpcResult},
-	types::SubscriptionResult,
-	SubscriptionSink,
-};
+use futures::TryFutureExt;
+use jsonrpsee::{core::async_trait, types::ErrorObject, PendingSubscriptionSink};
 use sc_rpc_api::DenyUnsafe;
 use sc_transaction_pool_api::{
 	error::IntoPoolError, BlockHash, InPoolTransaction, TransactionFor, TransactionPool,
@@ -91,7 +90,7 @@ where
 	P::Hash: Unpin,
 	<P::Block as BlockT>::Hash: Unpin,
 {
-	async fn submit_extrinsic(&self, ext: Bytes) -> RpcResult<TxHash<P>> {
+	async fn submit_extrinsic(&self, ext: Bytes) -> Result<TxHash<P>> {
 		let xt = match Decode::decode(&mut &ext[..]) {
 			Ok(xt) => xt,
 			Err(err) => return Err(Error::Client(Box::new(err)).into()),
@@ -105,7 +104,7 @@ where
 		})
 	}
 
-	fn insert_key(&self, key_type: String, suri: String, public: Bytes) -> RpcResult<()> {
+	fn insert_key(&self, key_type: String, suri: String, public: Bytes) -> Result<()> {
 		self.deny_unsafe.check_if_safe()?;
 
 		let key_type = key_type.as_str().try_into().map_err(|_| Error::BadKeyType)?;
@@ -115,7 +114,7 @@ where
 		Ok(())
 	}
 
-	fn rotate_keys(&self) -> RpcResult<Bytes> {
+	fn rotate_keys(&self) -> Result<Bytes> {
 		self.deny_unsafe.check_if_safe()?;
 
 		let best_block_hash = self.client.info().best_hash;
@@ -129,7 +128,7 @@ where
 			.map_err(|api_err| Error::Client(Box::new(api_err)).into())
 	}
 
-	fn has_session_keys(&self, session_keys: Bytes) -> RpcResult<bool> {
+	fn has_session_keys(&self, session_keys: Bytes) -> Result<bool> {
 		self.deny_unsafe.check_if_safe()?;
 
 		let best_block_hash = self.client.info().best_hash;
@@ -143,21 +142,21 @@ where
 		Ok(self.keystore.has_keys(&keys))
 	}
 
-	fn has_key(&self, public_key: Bytes, key_type: String) -> RpcResult<bool> {
+	fn has_key(&self, public_key: Bytes, key_type: String) -> Result<bool> {
 		self.deny_unsafe.check_if_safe()?;
 
 		let key_type = key_type.as_str().try_into().map_err(|_| Error::BadKeyType)?;
 		Ok(self.keystore.has_keys(&[(public_key.to_vec(), key_type)]))
 	}
 
-	fn pending_extrinsics(&self) -> RpcResult<Vec<Bytes>> {
+	fn pending_extrinsics(&self) -> Result<Vec<Bytes>> {
 		Ok(self.pool.ready().map(|tx| tx.data().encode().into()).collect())
 	}
 
 	fn remove_extrinsic(
 		&self,
 		bytes_or_hash: Vec<hash::ExtrinsicOrHash<TxHash<P>>>,
-	) -> RpcResult<Vec<TxHash<P>>> {
+	) -> Result<Vec<TxHash<P>>> {
 		self.deny_unsafe.check_if_safe()?;
 		let hashes = bytes_or_hash
 			.into_iter()
@@ -178,13 +177,13 @@ where
 			.collect())
 	}
 
-	fn watch_extrinsic(&self, mut sink: SubscriptionSink, xt: Bytes) -> SubscriptionResult {
+	fn watch_extrinsic(&self, pending: PendingSubscriptionSink, xt: Bytes) {
 		let best_block_hash = self.client.info().best_hash;
 		let dxt = match TransactionFor::<P>::decode(&mut &xt[..]).map_err(|e| Error::from(e)) {
 			Ok(dxt) => dxt,
 			Err(e) => {
-				let _ = sink.reject(JsonRpseeError::from(e));
-				return Ok(())
+				spawn_subscription_task(&self.executor, pending.reject(e));
+				return
 			},
 		};
 
@@ -198,15 +197,14 @@ where
 			let stream = match submit.await {
 				Ok(stream) => stream,
 				Err(err) => {
-					let _ = sink.reject(JsonRpseeError::from(err));
+					let _ = pending.reject(ErrorObject::from(err)).await;
 					return
 				},
 			};
 
-			sink.pipe_from_stream(stream).await;
+			pipe_from_stream(pending, stream).await;
 		};
 
-		self.executor.spawn("substrate-rpc-subscription", Some("rpc"), fut.boxed());
-		Ok(())
+		spawn_subscription_task(&self.executor, fut);
 	}
 }

substrate/client/rpc/src/author/tests.rs

@@ -22,8 +22,7 @@ use crate::testing::{test_executor, timeout_secs};
 use assert_matches::assert_matches;
 use codec::Encode;
 use jsonrpsee::{
-	core::Error as RpcError,
-	types::{error::CallError, EmptyServerParams as EmptyParams},
+	core::{EmptyServerParams as EmptyParams, Error as RpcError},
 	RpcModule,
 };
 use sc_transaction_pool::{BasicPool, FullChainApi};
@@ -104,7 +103,7 @@ async fn author_submit_transaction_should_not_cause_error() {
 
 	assert_matches!(
 		api.call::<_, H256>("author_submitExtrinsic", [xt]).await,
-		Err(RpcError::Call(CallError::Custom(err))) if err.message().contains("Already Imported") && err.code() == 1013
+		Err(RpcError::Call(err)) if err.message().contains("Already Imported") && err.code() == 1013
 	);
 }
 
@@ -119,7 +118,7 @@ async fn author_should_watch_extrinsic() {
 		true,
 	);
 
-	let mut sub = api.subscribe("author_submitAndWatchExtrinsic", [xt]).await.unwrap();
+	let mut sub = api.subscribe_unbounded("author_submitAndWatchExtrinsic", [xt]).await.unwrap();
 	let (tx, sub_id) = timeout_secs(10, sub.next::<TransactionStatus<H256, Block>>())
 		.await
 		.unwrap()
@@ -157,11 +156,11 @@ async fn author_should_return_watch_validation_error() {
 	let invalid_xt = ExtrinsicBuilder::new_fill_block(Perbill::from_percent(100)).build();
 
 	let api = TestSetup::into_rpc();
-	let failed_sub = api.subscribe(METHOD, [to_hex(&invalid_xt.encode(), true)]).await;
+	let failed_sub = api.subscribe_unbounded(METHOD, [to_hex(&invalid_xt.encode(), true)]).await;
 
 	assert_matches!(
 		failed_sub,
-		Err(RpcError::Call(CallError::Custom(err))) if err.message().contains("Invalid Transaction") && err.code() == 1010
+		Err(RpcError::Call(err)) if err.message().contains("Invalid Transaction") && err.code() == 1010
 	);
 }
 
@@ -277,7 +276,7 @@ async fn author_has_session_keys() {
 
 	assert_matches!(
 		api.call::<_, bool>("author_hasSessionKeys", vec![Bytes::from(vec![1, 2, 3])]).await,
-		Err(RpcError::Call(CallError::Custom(err))) if err.message().contains("Session keys are not encoded correctly")
+		Err(RpcError::Call(err)) if err.message().contains("Session keys are not encoded correctly")
 	);
 }
 

substrate/client/rpc/src/chain/chain_full.rs

@@ -19,14 +19,17 @@
 //! Blockchain API backend for full nodes.
 
 use super::{client_err, ChainBackend, Error};
-use crate::SubscriptionTaskExecutor;
+use crate::{
+	utils::{pipe_from_stream, spawn_subscription_task},
+	SubscriptionTaskExecutor,
+};
 use std::{marker::PhantomData, sync::Arc};
 
 use futures::{
-	future::{self, FutureExt},
+	future::{self},
 	stream::{self, Stream, StreamExt},
 };
-use jsonrpsee::SubscriptionSink;
+use jsonrpsee::{core::async_trait, PendingSubscriptionSink};
 use sc_client_api::{BlockBackend, BlockchainEvents};
 use sp_blockchain::HeaderBackend;
 use sp_runtime::{generic::SignedBlock, traits::Block as BlockT};
@@ -48,6 +51,7 @@ impl<Block: BlockT, Client> FullChain<Block, Client> {
 	}
 }
 
+#[async_trait]
 impl<Block, Client> ChainBackend<Client, Block> for FullChain<Block, Client>
 where
 	Block: BlockT + 'static,
@@ -66,11 +70,11 @@ where
 		self.client.block(self.unwrap_or_best(hash)).map_err(client_err)
 	}
 
-	fn subscribe_all_heads(&self, sink: SubscriptionSink) {
+	fn subscribe_all_heads(&self, pending: PendingSubscriptionSink) {
 		subscribe_headers(
 			&self.client,
 			&self.executor,
-			sink,
+			pending,
 			|| self.client().info().best_hash,
 			|| {
 				self.client()
@@ -80,11 +84,11 @@ where
 		)
 	}
 
-	fn subscribe_new_heads(&self, sink: SubscriptionSink) {
+	fn subscribe_new_heads(&self, pending: PendingSubscriptionSink) {
 		subscribe_headers(
 			&self.client,
 			&self.executor,
-			sink,
+			pending,
 			|| self.client().info().best_hash,
 			|| {
 				self.client()
@@ -95,11 +99,11 @@ where
 		)
 	}
 
-	fn subscribe_finalized_heads(&self, sink: SubscriptionSink) {
+	fn subscribe_finalized_heads(&self, pending: PendingSubscriptionSink) {
 		subscribe_headers(
 			&self.client,
 			&self.executor,
-			sink,
+			pending,
 			|| self.client().info().finalized_hash,
 			|| {
 				self.client()
@@ -114,7 +118,7 @@ where
 fn subscribe_headers<Block, Client, F, G, S>(
 	client: &Arc<Client>,
 	executor: &SubscriptionTaskExecutor,
-	mut sink: SubscriptionSink,
+	pending: PendingSubscriptionSink,
 	best_block_hash: G,
 	stream: F,
 ) where
@@ -139,9 +143,5 @@ fn subscribe_headers<Block, Client, F, G, S>(
 	// duplicates at the beginning of the stream though.
 	let stream = stream::iter(maybe_header).chain(stream());
 
-	let fut = async move {
-		sink.pipe_from_stream(stream).await;
-	};
-
-	executor.spawn("substrate-rpc-subscription", Some("rpc"), fut.boxed());
+	spawn_subscription_task(executor, pipe_from_stream(pending, stream));
 }

substrate/client/rpc/src/chain/mod.rs

@@ -27,7 +27,7 @@ use std::sync::Arc;
 
 use crate::SubscriptionTaskExecutor;
 
-use jsonrpsee::{core::RpcResult, types::SubscriptionResult, SubscriptionSink};
+use jsonrpsee::{core::async_trait, PendingSubscriptionSink};
 use sc_client_api::BlockchainEvents;
 use sp_rpc::{list::ListOrValue, number::NumberOrHex};
 use sp_runtime::{
@@ -42,6 +42,7 @@ pub use sc_rpc_api::chain::*;
 use sp_blockchain::HeaderBackend;
 
 /// Blockchain backend API
+#[async_trait]
 trait ChainBackend<Client, Block: BlockT>: Send + Sync + 'static
 where
 	Block: BlockT + 'static,
@@ -91,13 +92,13 @@ where
 	}
 
 	/// All new head subscription
-	fn subscribe_all_heads(&self, sink: SubscriptionSink);
+	fn subscribe_all_heads(&self, pending: PendingSubscriptionSink);
 
 	/// New best head subscription
-	fn subscribe_new_heads(&self, sink: SubscriptionSink);
+	fn subscribe_new_heads(&self, pending: PendingSubscriptionSink);
 
 	/// Finalized head subscription
-	fn subscribe_finalized_heads(&self, sink: SubscriptionSink);
+	fn subscribe_finalized_heads(&self, pending: PendingSubscriptionSink);
 }
 
 /// Create new state API that works on full node.
@@ -118,6 +119,7 @@ pub struct Chain<Block: BlockT, Client> {
 	backend: Box<dyn ChainBackend<Client, Block>>,
 }
 
+#[async_trait]
 impl<Block, Client> ChainApiServer<NumberFor<Block>, Block::Hash, Block::Header, SignedBlock<Block>>
 	for Chain<Block, Client>
 where
@@ -125,20 +127,20 @@ where
 	Block::Header: Unpin,
 	Client: HeaderBackend<Block> + BlockchainEvents<Block> + 'static,
 {
-	fn header(&self, hash: Option<Block::Hash>) -> RpcResult<Option<Block::Header>> {
-		self.backend.header(hash).map_err(Into::into)
+	fn header(&self, hash: Option<Block::Hash>) -> Result<Option<Block::Header>, Error> {
+		self.backend.header(hash)
 	}
 
-	fn block(&self, hash: Option<Block::Hash>) -> RpcResult<Option<SignedBlock<Block>>> {
-		self.backend.block(hash).map_err(Into::into)
+	fn block(&self, hash: Option<Block::Hash>) -> Result<Option<SignedBlock<Block>>, Error> {
+		self.backend.block(hash)
 	}
 
 	fn block_hash(
 		&self,
 		number: Option<ListOrValue<NumberOrHex>>,
-	) -> RpcResult<ListOrValue<Option<Block::Hash>>> {
+	) -> Result<ListOrValue<Option<Block::Hash>>, Error> {
 		match number {
-			None => self.backend.block_hash(None).map(ListOrValue::Value).map_err(Into::into),
+			None => self.backend.block_hash(None).map(ListOrValue::Value),
 			Some(ListOrValue::Value(number)) => self
 				.backend
 				.block_hash(Some(number))
@@ -152,23 +154,20 @@ where
 		}
 	}
 
-	fn finalized_head(&self) -> RpcResult<Block::Hash> {
-		self.backend.finalized_head().map_err(Into::into)
+	fn finalized_head(&self) -> Result<Block::Hash, Error> {
+		self.backend.finalized_head()
 	}
 
-	fn subscribe_all_heads(&self, sink: SubscriptionSink) -> SubscriptionResult {
-		self.backend.subscribe_all_heads(sink);
-		Ok(())
+	fn subscribe_all_heads(&self, pending: PendingSubscriptionSink) {
+		self.backend.subscribe_all_heads(pending);
 	}
 
-	fn subscribe_new_heads(&self, sink: SubscriptionSink) -> SubscriptionResult {
-		self.backend.subscribe_new_heads(sink);
-		Ok(())
+	fn subscribe_new_heads(&self, pending: PendingSubscriptionSink) {
+		self.backend.subscribe_new_heads(pending)
 	}
 
-	fn subscribe_finalized_heads(&self, sink: SubscriptionSink) -> SubscriptionResult {
-		self.backend.subscribe_finalized_heads(sink);
-		Ok(())
+	fn subscribe_finalized_heads(&self, pending: PendingSubscriptionSink) {
+		self.backend.subscribe_finalized_heads(pending)
 	}
 }
 

substrate/client/rpc/src/chain/tests.rs

@@ -19,7 +19,7 @@
 use super::*;
 use crate::testing::{test_executor, timeout_secs};
 use assert_matches::assert_matches;
-use jsonrpsee::types::EmptyServerParams as EmptyParams;
+use jsonrpsee::core::EmptyServerParams as EmptyParams;
 use sc_block_builder::BlockBuilderBuilder;
 use sp_consensus::BlockOrigin;
 use sp_rpc::list::ListOrValue;
@@ -252,7 +252,7 @@ async fn test_head_subscription(method: &str) {
 
 	let mut sub = {
 		let api = new_full(client.clone(), test_executor()).into_rpc();
-		let sub = api.subscribe(method, EmptyParams::new()).await.unwrap();
+		let sub = api.subscribe_unbounded(method, EmptyParams::new()).await.unwrap();
 		let block = BlockBuilderBuilder::new(&*client)
 			.on_parent_block(client.chain_info().best_hash)
 			.with_parent_block_number(client.chain_info().best_number)

substrate/client/rpc/src/dev/mod.rs

@@ -22,7 +22,6 @@
 #[cfg(test)]
 mod tests;
 
-use jsonrpsee::core::RpcResult;
 use sc_client_api::{BlockBackend, HeaderBackend};
 use sc_rpc_api::{dev::error::Error, DenyUnsafe};
 use sp_api::{ApiExt, Core, ProvideRuntimeApi};
@@ -65,7 +64,7 @@ where
 		+ 'static,
 	Client::Api: Core<Block>,
 {
-	fn block_stats(&self, hash: Block::Hash) -> RpcResult<Option<BlockStats>> {
+	fn block_stats(&self, hash: Block::Hash) -> Result<Option<BlockStats>, Error> {
 		self.deny_unsafe.check_if_safe()?;
 
 		let block = {

substrate/client/rpc/src/dev/tests.rs

@@ -97,7 +97,7 @@ async fn deny_unsafe_works() {
 		"{{\"jsonrpc\":\"2.0\",\"method\":\"dev_getBlockStats\",\"params\":[{}],\"id\":1}}",
 		best_hash_param
 	);
-	let (resp, _) = api.raw_json_request(&request).await.expect("Raw calls should succeed");
+	let (resp, _) = api.raw_json_request(&request, 1).await.expect("Raw calls should succeed");
 
 	assert_eq!(
 		resp.result,

substrate/client/rpc/src/lib.rs

@@ -39,6 +39,7 @@ pub mod offchain;
 pub mod state;
 pub mod statement;
 pub mod system;
+pub mod utils;
 
 #[cfg(any(test, feature = "test-helpers"))]
 pub mod testing;

substrate/client/rpc/src/mixnet/mod.rs

@@ -18,7 +18,7 @@
 
 //! Substrate mixnet API.
 
-use jsonrpsee::core::{async_trait, RpcResult};
+use jsonrpsee::core::async_trait;
 use sc_mixnet::Api;
 use sc_rpc_api::mixnet::error::Error;
 pub use sc_rpc_api::mixnet::MixnetApiServer;
@@ -36,7 +36,7 @@ impl Mixnet {
 
 #[async_trait]
 impl MixnetApiServer for Mixnet {
-	async fn submit_extrinsic(&self, extrinsic: Bytes) -> RpcResult<()> {
+	async fn submit_extrinsic(&self, extrinsic: Bytes) -> Result<(), Error> {
 		// We only hold the lock while pushing the request into the requests channel
 		let fut = {
 			let mut api = self.0.lock().await;

substrate/client/rpc/src/offchain/mod.rs

@@ -22,7 +22,7 @@
 mod tests;
 
 use self::error::Error;
-use jsonrpsee::core::{async_trait, Error as JsonRpseeError, RpcResult};
+use jsonrpsee::core::async_trait;
 use parking_lot::RwLock;
 /// Re-export the API for backward compatibility.
 pub use sc_rpc_api::offchain::*;
@@ -50,23 +50,23 @@ impl<T: OffchainStorage> Offchain<T> {
 
 #[async_trait]
 impl<T: OffchainStorage + 'static> OffchainApiServer for Offchain<T> {
-	fn set_local_storage(&self, kind: StorageKind, key: Bytes, value: Bytes) -> RpcResult<()> {
+	fn set_local_storage(&self, kind: StorageKind, key: Bytes, value: Bytes) -> Result<(), Error> {
 		self.deny_unsafe.check_if_safe()?;
 
 		let prefix = match kind {
 			StorageKind::PERSISTENT => sp_offchain::STORAGE_PREFIX,
-			StorageKind::LOCAL => return Err(JsonRpseeError::from(Error::UnavailableStorageKind)),
+			StorageKind::LOCAL => return Err(Error::UnavailableStorageKind),
 		};
 		self.storage.write().set(prefix, &key, &value);
 		Ok(())
 	}
 
-	fn get_local_storage(&self, kind: StorageKind, key: Bytes) -> RpcResult<Option<Bytes>> {
+	fn get_local_storage(&self, kind: StorageKind, key: Bytes) -> Result<Option<Bytes>, Error> {
 		self.deny_unsafe.check_if_safe()?;
 
 		let prefix = match kind {
 			StorageKind::PERSISTENT => sp_offchain::STORAGE_PREFIX,
-			StorageKind::LOCAL => return Err(JsonRpseeError::from(Error::UnavailableStorageKind)),
+			StorageKind::LOCAL => return Err(Error::UnavailableStorageKind),
 		};
 
 		Ok(self.storage.read().get(prefix, &key).map(Into::into))

substrate/client/rpc/src/offchain/tests.rs

@@ -39,7 +39,6 @@ fn local_storage_should_work() {
 
 #[test]
 fn offchain_calls_considered_unsafe() {
-	use jsonrpsee::types::error::CallError;
 	let storage = InMemOffchainStorage::default();
 	let offchain = Offchain::new(storage, DenyUnsafe::Yes);
 	let key = Bytes(b"offchain_storage".to_vec());
@@ -47,14 +46,14 @@ fn offchain_calls_considered_unsafe() {
 
 	assert_matches!(
 		offchain.set_local_storage(StorageKind::PERSISTENT, key.clone(), value.clone()),
-		Err(JsonRpseeError::Call(CallError::Custom(err))) => {
-			assert_eq!(err.message(), "RPC call is unsafe to be called externally")
+		Err(Error::UnsafeRpcCalled(e)) => {
+			assert_eq!(e.to_string(), "RPC call is unsafe to be called externally")
 		}
 	);
 	assert_matches!(
 		offchain.get_local_storage(StorageKind::PERSISTENT, key),
-		Err(JsonRpseeError::Call(CallError::Custom(err))) => {
-			assert_eq!(err.message(), "RPC call is unsafe to be called externally")
+		Err(Error::UnsafeRpcCalled(e)) => {
+			assert_eq!(e.to_string(), "RPC call is unsafe to be called externally")
 		}
 	);
 }

substrate/client/rpc/src/state/mod.rs

@@ -24,32 +24,23 @@ mod utils;
 #[cfg(test)]
 mod tests;
 
-use std::sync::Arc;
-
 use crate::SubscriptionTaskExecutor;
-
-use jsonrpsee::{
-	core::{async_trait, server::rpc_module::SubscriptionSink, Error as JsonRpseeError, RpcResult},
-	types::SubscriptionResult,
+use jsonrpsee::{core::async_trait, PendingSubscriptionSink};
+use sc_client_api::{
+	Backend, BlockBackend, BlockchainEvents, ExecutorProvider, ProofProvider, StorageProvider,
 };
-
 use sc_rpc_api::DenyUnsafe;
+use sp_api::{CallApiAt, Metadata, ProvideRuntimeApi};
+use sp_blockchain::{HeaderBackend, HeaderMetadata};
 use sp_core::{
 	storage::{PrefixedStorageKey, StorageChangeSet, StorageData, StorageKey},
 	Bytes,
 };
 use sp_runtime::traits::Block as BlockT;
 use sp_version::RuntimeVersion;
+use std::sync::Arc;
 
-use sp_api::{CallApiAt, Metadata, ProvideRuntimeApi};
-
-use self::error::Error;
-
-use sc_client_api::{
-	Backend, BlockBackend, BlockchainEvents, ExecutorProvider, ProofProvider, StorageProvider,
-};
 pub use sc_rpc_api::{child_state::*, state::*};
-use sp_blockchain::{HeaderBackend, HeaderMetadata};
 
 const STORAGE_KEYS_PAGED_MAX_COUNT: u32 = 1000;
 
@@ -158,10 +149,15 @@ where
 	) -> Result<sp_rpc::tracing::TraceBlockResponse, Error>;
 
 	/// New runtime version subscription
-	fn subscribe_runtime_version(&self, sink: SubscriptionSink);
+	fn subscribe_runtime_version(&self, pending: PendingSubscriptionSink);
 
 	/// New storage subscription
-	fn subscribe_storage(&self, sink: SubscriptionSink, keys: Option<Vec<StorageKey>>);
+	fn subscribe_storage(
+		&self,
+		pending: PendingSubscriptionSink,
+		keys: Option<Vec<StorageKey>>,
+		deny_unsafe: DenyUnsafe,
+	);
 }
 
 /// Create new state API that works on full node.
@@ -207,7 +203,12 @@ where
 	Block: BlockT + 'static,
 	Client: Send + Sync + 'static,
 {
-	fn call(&self, method: String, data: Bytes, block: Option<Block::Hash>) -> RpcResult<Bytes> {
+	fn call(
+		&self,
+		method: String,
+		data: Bytes,
+		block: Option<Block::Hash>,
+	) -> Result<Bytes, Error> {
 		self.backend.call(block, method, data).map_err(Into::into)
 	}
 
@@ -215,7 +216,7 @@ where
 		&self,
 		key_prefix: StorageKey,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Vec<StorageKey>> {
+	) -> Result<Vec<StorageKey>, Error> {
 		self.backend.storage_keys(block, key_prefix).map_err(Into::into)
 	}
 
@@ -223,7 +224,7 @@ where
 		&self,
 		key_prefix: StorageKey,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Vec<(StorageKey, StorageData)>> {
+	) -> Result<Vec<(StorageKey, StorageData)>, Error> {
 		self.deny_unsafe.check_if_safe()?;
 		self.backend.storage_pairs(block, key_prefix).map_err(Into::into)
 	}
@@ -234,12 +235,9 @@ where
 		count: u32,
 		start_key: Option<StorageKey>,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Vec<StorageKey>> {
+	) -> Result<Vec<StorageKey>, Error> {
 		if count > STORAGE_KEYS_PAGED_MAX_COUNT {
-			return Err(JsonRpseeError::from(Error::InvalidCount {
-				value: count,
-				max: STORAGE_KEYS_PAGED_MAX_COUNT,
-			}))
+			return Err(Error::InvalidCount { value: count, max: STORAGE_KEYS_PAGED_MAX_COUNT })
 		}
 		self.backend
 			.storage_keys_paged(block, prefix, count, start_key)
@@ -250,7 +248,7 @@ where
 		&self,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Option<StorageData>> {
+	) -> Result<Option<StorageData>, Error> {
 		self.backend.storage(block, key).map_err(Into::into)
 	}
 
@@ -258,7 +256,7 @@ where
 		&self,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Option<Block::Hash>> {
+	) -> Result<Option<Block::Hash>, Error> {
 		self.backend.storage_hash(block, key).map_err(Into::into)
 	}
 
@@ -266,18 +264,18 @@ where
 		&self,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Option<u64>> {
+	) -> Result<Option<u64>, Error> {
 		self.backend
 			.storage_size(block, key, self.deny_unsafe)
 			.await
 			.map_err(Into::into)
 	}
 
-	fn metadata(&self, block: Option<Block::Hash>) -> RpcResult<Bytes> {
+	fn metadata(&self, block: Option<Block::Hash>) -> Result<Bytes, Error> {
 		self.backend.metadata(block).map_err(Into::into)
 	}
 
-	fn runtime_version(&self, at: Option<Block::Hash>) -> RpcResult<RuntimeVersion> {
+	fn runtime_version(&self, at: Option<Block::Hash>) -> Result<RuntimeVersion, Error> {
 		self.backend.runtime_version(at).map_err(Into::into)
 	}
 
@@ -286,7 +284,7 @@ where
 		keys: Vec<StorageKey>,
 		from: Block::Hash,
 		to: Option<Block::Hash>,
-	) -> RpcResult<Vec<StorageChangeSet<Block::Hash>>> {
+	) -> Result<Vec<StorageChangeSet<Block::Hash>>, Error> {
 		self.deny_unsafe.check_if_safe()?;
 		self.backend.query_storage(from, to, keys).map_err(Into::into)
 	}
@@ -295,7 +293,7 @@ where
 		&self,
 		keys: Vec<StorageKey>,
 		at: Option<Block::Hash>,
-	) -> RpcResult<Vec<StorageChangeSet<Block::Hash>>> {
+	) -> Result<Vec<StorageChangeSet<Block::Hash>>, Error> {
 		self.backend.query_storage_at(keys, at).map_err(Into::into)
 	}
 
@@ -303,7 +301,7 @@ where
 		&self,
 		keys: Vec<StorageKey>,
 		block: Option<Block::Hash>,
-	) -> RpcResult<ReadProof<Block::Hash>> {
+	) -> Result<ReadProof<Block::Hash>, Error> {
 		self.backend.read_proof(block, keys).map_err(Into::into)
 	}
 
@@ -318,32 +316,19 @@ where
 		targets: Option<String>,
 		storage_keys: Option<String>,
 		methods: Option<String>,
-	) -> RpcResult<sp_rpc::tracing::TraceBlockResponse> {
+	) -> Result<sp_rpc::tracing::TraceBlockResponse, Error> {
 		self.deny_unsafe.check_if_safe()?;
 		self.backend
 			.trace_block(block, targets, storage_keys, methods)
 			.map_err(Into::into)
 	}
 
-	fn subscribe_runtime_version(&self, sink: SubscriptionSink) -> SubscriptionResult {
-		self.backend.subscribe_runtime_version(sink);
-		Ok(())
+	fn subscribe_runtime_version(&self, pending: PendingSubscriptionSink) {
+		self.backend.subscribe_runtime_version(pending)
 	}
 
-	fn subscribe_storage(
-		&self,
-		mut sink: SubscriptionSink,
-		keys: Option<Vec<StorageKey>>,
-	) -> SubscriptionResult {
-		if keys.is_none() {
-			if let Err(err) = self.deny_unsafe.check_if_safe() {
-				let _ = sink.reject(JsonRpseeError::from(err));
-				return Ok(())
-			}
-		}
-
-		self.backend.subscribe_storage(sink, keys);
-		Ok(())
+	fn subscribe_storage(&self, pending: PendingSubscriptionSink, keys: Option<Vec<StorageKey>>) {
+		self.backend.subscribe_storage(pending, keys, self.deny_unsafe)
 	}
 }
 
@@ -430,7 +415,7 @@ where
 		storage_key: PrefixedStorageKey,
 		key_prefix: StorageKey,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Vec<StorageKey>> {
+	) -> Result<Vec<StorageKey>, Error> {
 		self.backend.storage_keys(block, storage_key, key_prefix).map_err(Into::into)
 	}
 
@@ -441,7 +426,7 @@ where
 		count: u32,
 		start_key: Option<StorageKey>,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Vec<StorageKey>> {
+	) -> Result<Vec<StorageKey>, Error> {
 		self.backend
 			.storage_keys_paged(block, storage_key, prefix, count, start_key)
 			.map_err(Into::into)
@@ -452,7 +437,7 @@ where
 		storage_key: PrefixedStorageKey,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Option<StorageData>> {
+	) -> Result<Option<StorageData>, Error> {
 		self.backend.storage(block, storage_key, key).map_err(Into::into)
 	}
 
@@ -461,7 +446,7 @@ where
 		storage_key: PrefixedStorageKey,
 		keys: Vec<StorageKey>,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Vec<Option<StorageData>>> {
+	) -> Result<Vec<Option<StorageData>>, Error> {
 		self.backend.storage_entries(block, storage_key, keys).map_err(Into::into)
 	}
 
@@ -470,7 +455,7 @@ where
 		storage_key: PrefixedStorageKey,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Option<Block::Hash>> {
+	) -> Result<Option<Block::Hash>, Error> {
 		self.backend.storage_hash(block, storage_key, key).map_err(Into::into)
 	}
 
@@ -479,7 +464,7 @@ where
 		storage_key: PrefixedStorageKey,
 		key: StorageKey,
 		block: Option<Block::Hash>,
-	) -> RpcResult<Option<u64>> {
+	) -> Result<Option<u64>, Error> {
 		self.backend.storage_size(block, storage_key, key).map_err(Into::into)
 	}
 
@@ -488,7 +473,7 @@ where
 		child_storage_key: PrefixedStorageKey,
 		keys: Vec<StorageKey>,
 		block: Option<Block::Hash>,
-	) -> RpcResult<ReadProof<Block::Hash>> {
+	) -> Result<ReadProof<Block::Hash>, Error> {
 		self.backend
 			.read_child_proof(block, child_storage_key, keys)
 			.map_err(Into::into)

substrate/client/rpc/src/state/state_full.rs

@@ -25,13 +25,13 @@ use super::{
 	error::{Error, Result},
 	ChildStateBackend, StateBackend,
 };
-use crate::{DenyUnsafe, SubscriptionTaskExecutor};
-
-use futures::{future, stream, FutureExt, StreamExt};
-use jsonrpsee::{
-	core::{async_trait, Error as JsonRpseeError},
-	SubscriptionSink,
+use crate::{
+	utils::{pipe_from_stream, spawn_subscription_task},
+	DenyUnsafe, SubscriptionTaskExecutor,
 };
+
+use futures::{future, stream, StreamExt};
+use jsonrpsee::{core::async_trait, types::ErrorObject, PendingSubscriptionSink};
 use sc_client_api::{
 	Backend, BlockBackend, BlockchainEvents, CallExecutor, ExecutorProvider, ProofProvider,
 	StorageProvider,
@@ -371,9 +371,7 @@ where
 			.map_err(client_err)
 	}
 
-	fn subscribe_runtime_version(&self, mut sink: SubscriptionSink) {
-		let client = self.client.clone();
-
+	fn subscribe_runtime_version(&self, pending: PendingSubscriptionSink) {
 		let initial = match self
 			.block_or_best(None)
 			.and_then(|block| self.client.runtime_version_at(block).map_err(Into::into))
@@ -381,12 +379,13 @@ where
 		{
 			Ok(initial) => initial,
 			Err(e) => {
-				let _ = sink.reject(JsonRpseeError::from(e));
+				spawn_subscription_task(&self.executor, pending.reject(e));
 				return
 			},
 		};
 
 		let mut previous_version = initial.clone();
+		let client = self.client.clone();
 
 		// A stream of new versions
 		let version_stream = client
@@ -406,24 +405,33 @@ where
 			});
 
 		let stream = futures::stream::once(future::ready(initial)).chain(version_stream);
-
-		let fut = async move {
-			sink.pipe_from_stream(stream).await;
-		};
-
-		self.executor.spawn("substrate-rpc-subscription", Some("rpc"), fut.boxed());
+		spawn_subscription_task(&self.executor, pipe_from_stream(pending, stream));
 	}
 
-	fn subscribe_storage(&self, mut sink: SubscriptionSink, keys: Option<Vec<StorageKey>>) {
+	fn subscribe_storage(
+		&self,
+		pending: PendingSubscriptionSink,
+		keys: Option<Vec<StorageKey>>,
+		deny_unsafe: DenyUnsafe,
+	) {
+		if keys.is_none() {
+			if let Err(err) = deny_unsafe.check_if_safe() {
+				spawn_subscription_task(&self.executor, pending.reject(ErrorObject::from(err)));
+				return
+			}
+		}
+
 		let stream = match self.client.storage_changes_notification_stream(keys.as_deref(), None) {
 			Ok(stream) => stream,
 			Err(blockchain_err) => {
-				let _ = sink.reject(JsonRpseeError::from(Error::Client(Box::new(blockchain_err))));
+				spawn_subscription_task(
+					&self.executor,
+					pending.reject(Error::Client(Box::new(blockchain_err))),
+				);
 				return
 			},
 		};
 
-		// initial values
 		let initial = stream::iter(keys.map(|keys| {
 			let block = self.client.info().best_hash;
 			let changes = keys
@@ -436,7 +444,6 @@ where
 			StorageChangeSet { block, changes }
 		}));
 
-		// let storage_stream = stream.map(|(block, changes)| StorageChangeSet {
 		let storage_stream = stream.map(|storage_notif| StorageChangeSet {
 			block: storage_notif.block,
 			changes: storage_notif
@@ -450,11 +457,7 @@ where
 			.chain(storage_stream)
 			.filter(|storage| future::ready(!storage.changes.is_empty()));
 
-		let fut = async move {
-			sink.pipe_from_stream(stream).await;
-		};
-
-		self.executor.spawn("substrate-rpc-subscription", Some("rpc"), fut.boxed());
+		spawn_subscription_task(&self.executor, pipe_from_stream(pending, stream));
 	}
 
 	fn trace_block(

substrate/client/rpc/src/state/tests.rs

@@ -21,10 +21,7 @@ use super::*;
 use crate::testing::{test_executor, timeout_secs};
 use assert_matches::assert_matches;
 use futures::executor;
-use jsonrpsee::{
-	core::Error as RpcError,
-	types::{error::CallError as RpcCallError, EmptyServerParams as EmptyParams, ErrorObject},
-};
+use jsonrpsee::core::{EmptyServerParams as EmptyParams, Error as RpcError};
 use sc_block_builder::BlockBuilderBuilder;
 use sc_rpc_api::DenyUnsafe;
 use sp_consensus::BlockOrigin;
@@ -42,6 +39,14 @@ fn prefixed_storage_key() -> PrefixedStorageKey {
 	child_info.prefixed_storage_key()
 }
 
+fn init_logger() {
+	use tracing_subscriber::{EnvFilter, FmtSubscriber};
+
+	let _ = FmtSubscriber::builder()
+		.with_env_filter(EnvFilter::from_default_env())
+		.try_init();
+}
+
 #[tokio::test]
 async fn should_return_storage() {
 	const KEY: &[u8] = b":mock";
@@ -200,22 +205,25 @@ async fn should_call_contract() {
 	let genesis_hash = client.genesis_hash();
 	let (client, _child) = new_full(client, test_executor(), DenyUnsafe::No);
 
-	use jsonrpsee::{core::Error, types::error::CallError};
-
 	assert_matches!(
 		client.call("balanceOf".into(), Bytes(vec![1, 2, 3]), Some(genesis_hash).into()),
-		Err(Error::Call(CallError::Failed(_)))
+		Err(Error::Client(_))
 	)
 }
 
 #[tokio::test]
 async fn should_notify_about_storage_changes() {
+	init_logger();
+
 	let mut sub = {
 		let mut client = Arc::new(substrate_test_runtime_client::new());
 		let (api, _child) = new_full(client.clone(), test_executor(), DenyUnsafe::No);
 
 		let api_rpc = api.into_rpc();
-		let sub = api_rpc.subscribe("state_subscribeStorage", EmptyParams::new()).await.unwrap();
+		let sub = api_rpc
+			.subscribe_unbounded("state_subscribeStorage", EmptyParams::new())
+			.await
+			.unwrap();
 
 		// Cause a change:
 		let mut builder = BlockBuilderBuilder::new(&*client)
@@ -241,11 +249,12 @@ async fn should_notify_about_storage_changes() {
 	// NOTE: previous versions of the subscription code used to return an empty value for the
 	// "initial" storage change here
 	assert_matches!(timeout_secs(1, sub.next::<StorageChangeSet<H256>>()).await, Ok(Some(_)));
-	assert_matches!(timeout_secs(1, sub.next::<StorageChangeSet<H256>>()).await, Ok(None));
 }
 
 #[tokio::test]
 async fn should_send_initial_storage_changes_and_notifications() {
+	init_logger();
+
 	let mut sub = {
 		let mut client = Arc::new(substrate_test_runtime_client::new());
 		let (api, _child) = new_full(client.clone(), test_executor(), DenyUnsafe::No);
@@ -263,7 +272,10 @@ async fn should_send_initial_storage_changes_and_notifications() {
 
 		let api_rpc = api.into_rpc();
 		let sub = api_rpc
-			.subscribe("state_subscribeStorage", [[StorageKey(alice_balance_key)]])
+			.subscribe_unbounded(
+				"state_subscribeStorage",
+				[[StorageKey(alice_balance_key.to_vec())]],
+			)
 			.await
 			.unwrap();
 
@@ -288,9 +300,6 @@ async fn should_send_initial_storage_changes_and_notifications() {
 
 	assert_matches!(timeout_secs(1, sub.next::<StorageChangeSet<H256>>()).await, Ok(Some(_)));
 	assert_matches!(timeout_secs(1, sub.next::<StorageChangeSet<H256>>()).await, Ok(Some(_)));
-
-	// No more messages to follow
-	assert_matches!(timeout_secs(1, sub.next::<StorageChangeSet<H256>>()).await, Ok(None));
 }
 
 #[tokio::test]
@@ -393,108 +402,48 @@ async fn should_query_storage() {
 		assert_eq!(result.unwrap(), expected);
 
 		// Inverted range.
-		let result = api.query_storage(keys.clone(), block1_hash, Some(genesis_hash));
-
-		assert_eq!(
-			result.map_err(|e| e.to_string()),
-			Err(RpcError::Call(RpcCallError::Custom(ErrorObject::owned(
-				4001,
-				Error::InvalidBlockRange {
-					from: format!("1 ({:?})", block1_hash),
-					to: format!("0 ({:?})", genesis_hash),
-					details: "from number > to number".to_owned(),
-				}
-				.to_string(),
-				None::<()>,
-			))))
-			.map_err(|e| e.to_string())
+		assert_matches!(
+			api.query_storage(keys.clone(), block1_hash, Some(genesis_hash)),
+			Err(Error::InvalidBlockRange { from, to, details }) if from == format!("1 ({:?})", block1_hash) && to == format!("0 ({:?})", genesis_hash) && details == "from number > to number".to_owned()
 		);
 
 		let random_hash1 = H256::random();
 		let random_hash2 = H256::random();
 
 		// Invalid second hash.
-		let result = api.query_storage(keys.clone(), genesis_hash, Some(random_hash1));
-
-		assert_eq!(
-			result.map_err(|e| e.to_string()),
-			Err(RpcError::Call(RpcCallError::Custom(ErrorObject::owned(
-				4001,
-				Error::InvalidBlockRange {
-					from: format!("{:?}", genesis_hash),
-					to: format!("{:?}", Some(random_hash1)),
-					details: format!(
-						"UnknownBlock: Header was not found in the database: {:?}",
-						random_hash1
-					),
-				}
-				.to_string(),
-				None::<()>,
-			))))
-			.map_err(|e| e.to_string())
+		assert_matches!(
+			api.query_storage(keys.clone(), genesis_hash, Some(random_hash1)),
+			Err(Error::InvalidBlockRange { from, to, details }) if from == format!("{:?}", genesis_hash) && to == format!("{:?}", Some(random_hash1)) && details == format!(
+				"UnknownBlock: Header was not found in the database: {:?}",
+				random_hash1
+			)
 		);
 
 		// Invalid first hash with Some other hash.
-		let result = api.query_storage(keys.clone(), random_hash1, Some(genesis_hash));
-
-		assert_eq!(
-			result.map_err(|e| e.to_string()),
-			Err(RpcError::Call(RpcCallError::Custom(ErrorObject::owned(
-				4001,
-				Error::InvalidBlockRange {
-					from: format!("{:?}", random_hash1),
-					to: format!("{:?}", Some(genesis_hash)),
-					details: format!(
-						"UnknownBlock: Header was not found in the database: {:?}",
-						random_hash1
-					),
-				}
-				.to_string(),
-				None::<()>,
-			))))
-			.map_err(|e| e.to_string()),
+		assert_matches!(
+			api.query_storage(keys.clone(), random_hash1, Some(genesis_hash)),
+			Err(Error::InvalidBlockRange { from, to, details }) if from == format!("{:?}", random_hash1) && to == format!("{:?}", Some(genesis_hash)) && details == format!(
+				"UnknownBlock: Header was not found in the database: {:?}",
+				random_hash1
+			)
 		);
 
 		// Invalid first hash with None.
-		let result = api.query_storage(keys.clone(), random_hash1, None);
-
-		assert_eq!(
-			result.map_err(|e| e.to_string()),
-			Err(RpcError::Call(RpcCallError::Custom(ErrorObject::owned(
-				4001,
-				Error::InvalidBlockRange {
-					from: format!("{:?}", random_hash1),
-					to: format!("{:?}", Some(block2_hash)), // Best block hash.
-					details: format!(
-						"UnknownBlock: Header was not found in the database: {:?}",
-						random_hash1
-					),
-				}
-				.to_string(),
-				None::<()>,
-			))))
-			.map_err(|e| e.to_string()),
+		assert_matches!(
+			api.query_storage(keys.clone(), random_hash1, None),
+			Err(Error::InvalidBlockRange { from, to, details }) if from == format!("{:?}", random_hash1) && to == format!("{:?}", Some(block2_hash)) && details == format!(
+				"UnknownBlock: Header was not found in the database: {:?}",
+				random_hash1
+			)
 		);
 
 		// Both hashes invalid.
-		let result = api.query_storage(keys.clone(), random_hash1, Some(random_hash2));
-
-		assert_eq!(
-			result.map_err(|e| e.to_string()),
-			Err(RpcError::Call(RpcCallError::Custom(ErrorObject::owned(
-				4001,
-				Error::InvalidBlockRange {
-					from: format!("{:?}", random_hash1), // First hash not found.
-					to: format!("{:?}", Some(random_hash2)),
-					details: format!(
-						"UnknownBlock: Header was not found in the database: {:?}",
-						random_hash1
-					),
-				}
-				.to_string(),
-				None::<()>
-			))))
-			.map_err(|e| e.to_string()),
+		assert_matches!(
+			api.query_storage(keys.clone(), random_hash1, Some(random_hash2)),
+			Err(Error::InvalidBlockRange { from, to, details }) if from == format!("{:?}", random_hash1) && to == format!("{:?}", Some(random_hash2)) && details == format!(
+				"UnknownBlock: Header was not found in the database: {:?}",
+				random_hash1
+			)
 		);
 
 		// single block range
@@ -548,7 +497,7 @@ async fn should_notify_on_runtime_version_initially() {
 
 		let api_rpc = api.into_rpc();
 		let sub = api_rpc
-			.subscribe("state_subscribeRuntimeVersion", EmptyParams::new())
+			.subscribe_unbounded("state_subscribeRuntimeVersion", EmptyParams::new())
 			.await
 			.unwrap();
 
@@ -557,9 +506,6 @@ async fn should_notify_on_runtime_version_initially() {
 
 	// assert initial version sent.
 	assert_matches!(timeout_secs(10, sub.next::<RuntimeVersion>()).await, Ok(Some(_)));
-
-	sub.close();
-	assert_matches!(timeout_secs(10, sub.next::<RuntimeVersion>()).await, Ok(None));
 }
 
 #[test]
@@ -572,12 +518,14 @@ fn should_deserialize_storage_key() {
 
 #[tokio::test]
 async fn wildcard_storage_subscriptions_are_rpc_unsafe() {
+	init_logger();
+
 	let client = Arc::new(substrate_test_runtime_client::new());
 	let (api, _child) = new_full(client, test_executor(), DenyUnsafe::Yes);
 
 	let api_rpc = api.into_rpc();
-	let err = api_rpc.subscribe("state_subscribeStorage", EmptyParams::new()).await;
-	assert_matches!(err, Err(RpcError::Call(RpcCallError::Custom(e))) if e.message() == "RPC call is unsafe to be called externally");
+	let err = api_rpc.subscribe_unbounded("state_subscribeStorage", EmptyParams::new()).await;
+	assert_matches!(err, Err(RpcError::Call(e)) if e.message() == "RPC call is unsafe to be called externally");
 }
 
 #[tokio::test]
@@ -587,7 +535,7 @@ async fn concrete_storage_subscriptions_are_rpc_safe() {
 	let api_rpc = api.into_rpc();
 
 	let key = StorageKey(STORAGE_KEY.to_vec());
-	let sub = api_rpc.subscribe("state_subscribeStorage", [[key]]).await;
+	let sub = api_rpc.subscribe_unbounded("state_subscribeStorage", [[key]]).await;
 
 	assert!(sub.is_ok());
 }

substrate/client/rpc/src/system/mod.rs

@@ -22,17 +22,12 @@
 mod tests;
 
 use futures::channel::oneshot;
-use jsonrpsee::{
-	core::{async_trait, error::Error as JsonRpseeError, JsonValue, RpcResult},
-	types::error::{CallError, ErrorCode, ErrorObject},
-};
+use jsonrpsee::core::{async_trait, JsonValue};
 use sc_rpc_api::DenyUnsafe;
 use sc_tracing::logging;
 use sc_utils::mpsc::TracingUnboundedSender;
 use sp_runtime::traits::{self, Header as HeaderT};
 
-use self::error::Result;
-
 pub use self::helpers::{Health, NodeRole, PeerInfo, SyncState, SystemInfo};
 pub use sc_rpc_api::system::*;
 
@@ -57,9 +52,9 @@ pub enum Request<B: traits::Block> {
 	/// Must return the state of the network.
 	NetworkState(oneshot::Sender<serde_json::Value>),
 	/// Must return any potential parse error.
-	NetworkAddReservedPeer(String, oneshot::Sender<Result<()>>),
+	NetworkAddReservedPeer(String, oneshot::Sender<error::Result<()>>),
 	/// Must return any potential parse error.
-	NetworkRemoveReservedPeer(String, oneshot::Sender<Result<()>>),
+	NetworkRemoveReservedPeer(String, oneshot::Sender<error::Result<()>>),
 	/// Must return the list of reserved peers
 	NetworkReservedPeers(oneshot::Sender<Vec<String>>),
 	/// Must return the node role.
@@ -84,121 +79,109 @@ impl<B: traits::Block> System<B> {
 
 #[async_trait]
 impl<B: traits::Block> SystemApiServer<B::Hash, <B::Header as HeaderT>::Number> for System<B> {
-	fn system_name(&self) -> RpcResult<String> {
+	fn system_name(&self) -> Result<String, Error> {
 		Ok(self.info.impl_name.clone())
 	}
 
-	fn system_version(&self) -> RpcResult<String> {
+	fn system_version(&self) -> Result<String, Error> {
 		Ok(self.info.impl_version.clone())
 	}
 
-	fn system_chain(&self) -> RpcResult<String> {
+	fn system_chain(&self) -> Result<String, Error> {
 		Ok(self.info.chain_name.clone())
 	}
 
-	fn system_type(&self) -> RpcResult<sc_chain_spec::ChainType> {
+	fn system_type(&self) -> Result<sc_chain_spec::ChainType, Error> {
 		Ok(self.info.chain_type.clone())
 	}
 
-	fn system_properties(&self) -> RpcResult<sc_chain_spec::Properties> {
+	fn system_properties(&self) -> Result<sc_chain_spec::Properties, Error> {
 		Ok(self.info.properties.clone())
 	}
 
-	async fn system_health(&self) -> RpcResult<Health> {
+	async fn system_health(&self) -> Result<Health, Error> {
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::Health(tx));
-		rx.await.map_err(|e| JsonRpseeError::to_call_error(e))
+		rx.await.map_err(|e| Error::Internal(e.to_string()))
 	}
 
-	async fn system_local_peer_id(&self) -> RpcResult<String> {
+	async fn system_local_peer_id(&self) -> Result<String, Error> {
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::LocalPeerId(tx));
-		rx.await.map_err(|e| JsonRpseeError::to_call_error(e))
+		rx.await.map_err(|e| Error::Internal(e.to_string()))
 	}
 
-	async fn system_local_listen_addresses(&self) -> RpcResult<Vec<String>> {
+	async fn system_local_listen_addresses(&self) -> Result<Vec<String>, Error> {
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::LocalListenAddresses(tx));
-		rx.await.map_err(|e| JsonRpseeError::to_call_error(e))
+		rx.await.map_err(|e| Error::Internal(e.to_string()))
 	}
 
 	async fn system_peers(
 		&self,
-	) -> RpcResult<Vec<PeerInfo<B::Hash, <B::Header as HeaderT>::Number>>> {
+	) -> Result<Vec<PeerInfo<B::Hash, <B::Header as HeaderT>::Number>>, Error> {
 		self.deny_unsafe.check_if_safe()?;
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::Peers(tx));
-		rx.await.map_err(|e| JsonRpseeError::to_call_error(e))
+		rx.await.map_err(|e| Error::Internal(e.to_string()))
 	}
 
-	async fn system_network_state(&self) -> RpcResult<JsonValue> {
+	async fn system_network_state(&self) -> Result<JsonValue, Error> {
 		self.deny_unsafe.check_if_safe()?;
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::NetworkState(tx));
-		rx.await.map_err(|e| JsonRpseeError::to_call_error(e))
+		rx.await.map_err(|e| Error::Internal(e.to_string()))
 	}
 
-	async fn system_add_reserved_peer(&self, peer: String) -> RpcResult<()> {
+	async fn system_add_reserved_peer(&self, peer: String) -> Result<(), Error> {
 		self.deny_unsafe.check_if_safe()?;
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::NetworkAddReservedPeer(peer, tx));
 		match rx.await {
 			Ok(Ok(())) => Ok(()),
-			Ok(Err(e)) => Err(JsonRpseeError::from(e)),
-			Err(e) => Err(JsonRpseeError::to_call_error(e)),
+			Ok(Err(e)) => Err(e),
+			Err(e) => Err(Error::Internal(e.to_string())),
 		}
 	}
 
-	async fn system_remove_reserved_peer(&self, peer: String) -> RpcResult<()> {
+	async fn system_remove_reserved_peer(&self, peer: String) -> Result<(), Error> {
 		self.deny_unsafe.check_if_safe()?;
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::NetworkRemoveReservedPeer(peer, tx));
 		match rx.await {
 			Ok(Ok(())) => Ok(()),
-			Ok(Err(e)) => Err(JsonRpseeError::from(e)),
-			Err(e) => Err(JsonRpseeError::to_call_error(e)),
+			Ok(Err(e)) => Err(e),
+			Err(e) => Err(Error::Internal(e.to_string())),
 		}
 	}
 
-	async fn system_reserved_peers(&self) -> RpcResult<Vec<String>> {
+	async fn system_reserved_peers(&self) -> Result<Vec<String>, Error> {
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::NetworkReservedPeers(tx));
-		rx.await.map_err(|e| JsonRpseeError::to_call_error(e))
+		rx.await.map_err(|e| Error::Internal(e.to_string()))
 	}
 
-	async fn system_node_roles(&self) -> RpcResult<Vec<NodeRole>> {
+	async fn system_node_roles(&self) -> Result<Vec<NodeRole>, Error> {
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::NodeRoles(tx));
-		rx.await.map_err(|e| JsonRpseeError::to_call_error(e))
+		rx.await.map_err(|e| Error::Internal(e.to_string()))
 	}
 
-	async fn system_sync_state(&self) -> RpcResult<SyncState<<B::Header as HeaderT>::Number>> {
+	async fn system_sync_state(&self) -> Result<SyncState<<B::Header as HeaderT>::Number>, Error> {
 		let (tx, rx) = oneshot::channel();
 		let _ = self.send_back.unbounded_send(Request::SyncState(tx));
-		rx.await.map_err(|e| JsonRpseeError::to_call_error(e))
+		rx.await.map_err(|e| Error::Internal(e.to_string()))
 	}
 
-	fn system_add_log_filter(&self, directives: String) -> RpcResult<()> {
+	fn system_add_log_filter(&self, directives: String) -> Result<(), Error> {
 		self.deny_unsafe.check_if_safe()?;
 
 		logging::add_directives(&directives);
-		logging::reload_filter().map_err(|e| {
-			JsonRpseeError::Call(CallError::Custom(ErrorObject::owned(
-				ErrorCode::InternalError.code(),
-				e,
-				None::<()>,
-			)))
-		})
+		logging::reload_filter().map_err(|e| Error::Internal(e))
 	}
 
-	fn system_reset_log_filter(&self) -> RpcResult<()> {
+	fn system_reset_log_filter(&self) -> Result<(), Error> {
 		self.deny_unsafe.check_if_safe()?;
-		logging::reset_log_filter().map_err(|e| {
-			JsonRpseeError::Call(CallError::Custom(ErrorObject::owned(
-				ErrorCode::InternalError.code(),
-				e,
-				None::<()>,
-			)))
-		})
+		logging::reset_log_filter().map_err(|e| Error::Internal(e))
 	}
 }

substrate/client/rpc/src/system/tests.rs

@@ -20,8 +20,7 @@ use super::{helpers::SyncState, *};
 use assert_matches::assert_matches;
 use futures::prelude::*;
 use jsonrpsee::{
-	core::Error as RpcError,
-	types::{error::CallError, EmptyServerParams as EmptyParams},
+	core::{EmptyServerParams as EmptyParams, Error as RpcError},
 	RpcModule,
 };
 use sc_network::{self, config::Role, PeerId};
@@ -312,7 +311,7 @@ async fn system_network_add_reserved() {
 	let bad_peer_id = ["/ip4/198.51.100.19/tcp/30333"];
 	assert_matches!(
 		api(None).call::<_, ()>("system_addReservedPeer", bad_peer_id).await,
-		Err(RpcError::Call(CallError::Custom(err))) if err.message().contains("Peer id is missing from the address")
+		Err(RpcError::Call(err)) if err.message().contains("Peer id is missing from the address")
 	);
 }
 
@@ -328,7 +327,7 @@ async fn system_network_remove_reserved() {
 
 	assert_matches!(
 		api(None).call::<_, String>("system_removeReservedPeer", bad_peer_id).await,
-		Err(RpcError::Call(CallError::Custom(err))) if err.message().contains("base-58 decode error: provided string contained invalid character '/' at byte 0")
+		Err(RpcError::Call(err)) if err.message().contains("base-58 decode error: provided string contained invalid character '/' at byte 0")
 	);
 }
 #[tokio::test]

substrate/client/rpc/src/testing.rs

@@ -20,11 +20,50 @@
 
 use std::{future::Future, sync::Arc};
 
-use sp_core::testing::TaskExecutor;
+/// A task executor that can be used for running RPC tests.
+///
+/// Warning: the tokio runtime must be initialized before calling this.
+#[derive(Clone)]
+pub struct TokioTestExecutor(tokio::runtime::Handle);
+
+impl TokioTestExecutor {
+	/// Create a new instance of `Self`.
+	pub fn new() -> Self {
+		Self(tokio::runtime::Handle::current())
+	}
+}
+
+impl Default for TokioTestExecutor {
+	fn default() -> Self {
+		Self::new()
+	}
+}
+
+impl sp_core::traits::SpawnNamed for TokioTestExecutor {
+	fn spawn_blocking(
+		&self,
+		_name: &'static str,
+		_group: Option<&'static str>,
+		future: futures::future::BoxFuture<'static, ()>,
+	) {
+		let handle = self.0.clone();
+		self.0.spawn_blocking(move || {
+			handle.block_on(future);
+		});
+	}
+	fn spawn(
+		&self,
+		_name: &'static str,
+		_group: Option<&'static str>,
+		future: futures::future::BoxFuture<'static, ()>,
+	) {
+		self.0.spawn(future);
+	}
+}
 
 /// Executor for testing.
-pub fn test_executor() -> Arc<sp_core::testing::TaskExecutor> {
-	Arc::new(TaskExecutor::default())
+pub fn test_executor() -> Arc<TokioTestExecutor> {
+	Arc::new(TokioTestExecutor::default())
 }
 
 /// Wrap a future in a timeout a little more concisely

substrate/client/rpc/src/utils.rs

@@ -0,0 +1,228 @@
+// This file is part of Substrate.
+
+// Copyright (C) Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! JSON-RPC helpers.
+
+use crate::SubscriptionTaskExecutor;
+use futures::{
+	future::{self, Either, Fuse, FusedFuture},
+	Future, FutureExt, Stream, StreamExt,
+};
+use jsonrpsee::{PendingSubscriptionSink, SubscriptionMessage, SubscriptionSink};
+use sp_runtime::Serialize;
+use std::collections::VecDeque;
+
+const DEFAULT_BUF_SIZE: usize = 16;
+
+/// A simple bounded VecDeque.
+struct BoundedVecDeque<T> {
+	inner: VecDeque<T>,
+	max_cap: usize,
+}
+
+impl<T> BoundedVecDeque<T> {
+	/// Create a new bounded VecDeque.
+	fn new() -> Self {
+		Self { inner: VecDeque::with_capacity(DEFAULT_BUF_SIZE), max_cap: DEFAULT_BUF_SIZE }
+	}
+
+	fn push_back(&mut self, item: T) -> Result<(), ()> {
+		if self.inner.len() >= self.max_cap {
+			Err(())
+		} else {
+			self.inner.push_back(item);
+			Ok(())
+		}
+	}
+
+	fn pop_front(&mut self) -> Option<T> {
+		self.inner.pop_front()
+	}
+}
+
+/// Feed items to the subscription from the underlying stream.
+///
+/// This is bounded because the underlying streams in substrate are
+/// unbounded and if the subscription can't keep with stream it can
+/// cause the buffer to become very large and consume lots of memory.
+///
+/// In such cases the subscription is dropped.
+pub async fn pipe_from_stream<S, T>(pending: PendingSubscriptionSink, mut stream: S)
+where
+	S: Stream<Item = T> + Unpin + Send + 'static,
+	T: Serialize + Send + 'static,
+{
+	let mut buf = BoundedVecDeque::new();
+	let accept_fut = pending.accept();
+
+	futures::pin_mut!(accept_fut);
+
+	// Poll the stream while waiting for the subscription to be accepted
+	//
+	// If the `max_cap` is exceeded then the subscription is dropped.
+	let sink = loop {
+		match future::select(accept_fut, stream.next()).await {
+			Either::Left((Ok(sink), _)) => break sink,
+			Either::Right((Some(msg), f)) => {
+				if buf.push_back(msg).is_err() {
+					log::warn!(target: "rpc", "Subscription::accept failed buffer limit={} exceed; dropping subscription", buf.max_cap);
+					return
+				}
+				accept_fut = f;
+			},
+			// The connection was closed or the stream was closed.
+			_ => return,
+		}
+	};
+
+	inner_pipe_from_stream(sink, stream, buf).await
+}
+
+async fn inner_pipe_from_stream<S, T>(
+	sink: SubscriptionSink,
+	mut stream: S,
+	mut buf: BoundedVecDeque<T>,
+) where
+	S: Stream<Item = T> + Unpin + Send + 'static,
+	T: Serialize + Send + 'static,
+{
+	let mut next_fut = Box::pin(Fuse::terminated());
+	let mut next_item = stream.next();
+	let closed = sink.closed();
+
+	futures::pin_mut!(closed);
+
+	loop {
+		if next_fut.is_terminated() {
+			if let Some(v) = buf.pop_front() {
+				let val = to_sub_message(&sink, &v);
+				next_fut.set(async { sink.send(val).await }.fuse());
+			}
+		}
+
+		match future::select(closed, future::select(next_fut, next_item)).await {
+			// Send operation finished.
+			Either::Right((Either::Left((_, n)), c)) => {
+				next_item = n;
+				closed = c;
+				next_fut = Box::pin(Fuse::terminated());
+			},
+			// New item from the stream
+			Either::Right((Either::Right((Some(v), n)), c)) => {
+				if buf.push_back(v).is_err() {
+					log::warn!(target: "rpc", "Subscription buffer limit={} exceed; dropping subscription", buf.max_cap);
+					return
+				}
+
+				next_fut = n;
+				closed = c;
+				next_item = stream.next();
+			},
+			// Stream "finished".
+			//
+			// Process remaining items and terminate.
+			Either::Right((Either::Right((None, pending_fut)), _)) => {
+				if pending_fut.await.is_err() {
+					return;
+				}
+
+				while let Some(v) = buf.pop_front() {
+					let val = to_sub_message(&sink, &v);
+					if sink.send(val).await.is_err() {
+						return;
+					}
+				}
+
+				return;
+			},
+			// Subscription was closed.
+			Either::Left(_) => return,
+		}
+	}
+}
+
+/// Builds a subscription message.
+///
+/// # Panics
+///
+/// This function panics `Serialize` fails and it is treated a bug.
+pub fn to_sub_message(sink: &SubscriptionSink, result: &impl Serialize) -> SubscriptionMessage {
+	SubscriptionMessage::new(sink.method_name(), sink.subscription_id(), result)
+		.expect("Serialize infallible; qed")
+}
+
+/// Helper for spawning non-blocking rpc subscription task.
+pub fn spawn_subscription_task(
+	executor: &SubscriptionTaskExecutor,
+	fut: impl Future<Output = ()> + Send + 'static,
+) {
+	executor.spawn("substrate-rpc-subscription", Some("rpc"), fut.boxed());
+}
+
+#[cfg(test)]
+mod tests {
+	use super::pipe_from_stream;
+	use futures::StreamExt;
+	use jsonrpsee::{core::EmptyServerParams, RpcModule, Subscription};
+
+	async fn subscribe() -> Subscription {
+		let mut module = RpcModule::new(());
+		module
+			.register_subscription("sub", "my_sub", "unsub", |_, pending, _| async move {
+				let stream = futures::stream::iter([0; 16]);
+				pipe_from_stream(pending, stream).await;
+				Ok(())
+			})
+			.unwrap();
+
+		module.subscribe("sub", EmptyServerParams::new(), 1).await.unwrap()
+	}
+
+	#[tokio::test]
+	async fn pipe_from_stream_works() {
+		let mut sub = subscribe().await;
+		let mut rx = 0;
+
+		while let Some(Ok(_)) = sub.next::<usize>().await {
+			rx += 1;
+		}
+
+		assert_eq!(rx, 16);
+	}
+
+	#[tokio::test]
+	async fn pipe_from_stream_is_bounded() {
+		let (tx, mut rx) = futures::channel::mpsc::unbounded::<()>();
+
+		let mut module = RpcModule::new(tx);
+		module
+			.register_subscription("sub", "my_sub", "unsub", |_, pending, ctx| async move {
+				let stream = futures::stream::iter([0; 32]);
+				pipe_from_stream(pending, stream).await;
+				_ = ctx.unbounded_send(());
+				Ok(())
+			})
+			.unwrap();
+
+		let mut sub = module.subscribe("sub", EmptyServerParams::new(), 1).await.unwrap();
+
+		// When the 17th item arrives the subscription is dropped
+		_ = rx.next().await.unwrap();
+		assert!(sub.next::<usize>().await.is_none());
+	}
+}