chainHead: Allow methods to be called from within a single connection…

… context and limit connections (#3481)

This PR ensures that the chainHead RPC class can be called only from
within the same connection context.

The chainHead methods are now registered as raw methods. 
- paritytech/jsonrpsee#1297
The concept of raw methods is introduced in jsonrpsee, which is an async
method that exposes the connection ID:
The raw method doesn't have the concept of a blocking method. Previously
blocking methods are now spawning a blocking task to handle their
blocking (ie DB) access. We spawn the same number of tasks as before,
however we do that explicitly.

Another approach would be implementing a RPC middleware that captures
and decodes the method parameters:
- #3343
However, that approach is prone to errors since the methods are
hardcoded by name. Performace is affected by the double deserialization
that needs to happen to extract the subscription ID we'd like to limit.
Once from the middleware, and once from the methods itself.

This PR paves the way to implement the chainHead connection limiter:
- #1505
Registering tokens (subscription ID / operation ID) on the
`RpcConnections` could be extended to return an error when the maximum
number of operations is reached.

While at it, have added an integration-test to ensure that chainHead
methods can be called from within the same connection context.

Before this is merged, a new JsonRPC release should be made to expose
the `raw-methods`:
- [x] Use jsonrpsee from crates io (blocked by:
paritytech/jsonrpsee#1297)

Closes: #3207


cc @paritytech/subxt-team

---------

Signed-off-by: Alexandru Vasile <alexandru.vasile@parity.io>
Co-authored-by: Niklas Adolfsson <niklasadolfsson1@gmail.com>

substrate/client/rpc-spec-v2/Cargo.toml

@@ -44,6 +44,7 @@ futures-util = { version = "0.3.30", default-features = false }
 rand = "0.8.5"
 
 [dev-dependencies]
+jsonrpsee = { version = "0.22", features = ["server", "ws-client"] }
 serde_json = { workspace = true, default-features = true }
 tokio = { version = "1.22.0", features = ["macros"] }
 substrate-test-runtime-client = { path = "../../test-utils/runtime/client" }

substrate/client/rpc-spec-v2/src/chain_head/api.rs

@@ -27,7 +27,7 @@ use crate::{
 	common::events::StorageQuery,
 };
 use jsonrpsee::{proc_macros::rpc, server::ResponsePayload};
-use sp_rpc::list::ListOrValue;
+pub use sp_rpc::list::ListOrValue;
 
 #[rpc(client, server)]
 pub trait ChainHeadApi<Hash> {
@@ -54,8 +54,8 @@ pub trait ChainHeadApi<Hash> {
 	/// # Unstable
 	///
 	/// This method is unstable and subject to change in the future.
-	#[method(name = "chainHead_unstable_body", blocking)]
-	fn chain_head_unstable_body(
+	#[method(name = "chainHead_unstable_body", raw_method)]
+	async fn chain_head_unstable_body(
 		&self,
 		follow_subscription: String,
 		hash: Hash,
@@ -73,8 +73,8 @@ pub trait ChainHeadApi<Hash> {
 	/// # Unstable
 	///
 	/// This method is unstable and subject to change in the future.
-	#[method(name = "chainHead_unstable_header", blocking)]
-	fn chain_head_unstable_header(
+	#[method(name = "chainHead_unstable_header", raw_method)]
+	async fn chain_head_unstable_header(
 		&self,
 		follow_subscription: String,
 		hash: Hash,
@@ -85,8 +85,8 @@ pub trait ChainHeadApi<Hash> {
 	/// # Unstable
 	///
 	/// This method is unstable and subject to change in the future.
-	#[method(name = "chainHead_unstable_storage", blocking)]
-	fn chain_head_unstable_storage(
+	#[method(name = "chainHead_unstable_storage", raw_method)]
+	async fn chain_head_unstable_storage(
 		&self,
 		follow_subscription: String,
 		hash: Hash,
@@ -99,8 +99,8 @@ pub trait ChainHeadApi<Hash> {
 	/// # Unstable
 	///
 	/// This method is unstable and subject to change in the future.
-	#[method(name = "chainHead_unstable_call", blocking)]
-	fn chain_head_unstable_call(
+	#[method(name = "chainHead_unstable_call", raw_method)]
+	async fn chain_head_unstable_call(
 		&self,
 		follow_subscription: String,
 		hash: Hash,
@@ -118,8 +118,8 @@ pub trait ChainHeadApi<Hash> {
 	/// # Unstable
 	///
 	/// This method is unstable and subject to change in the future.
-	#[method(name = "chainHead_unstable_unpin", blocking)]
-	fn chain_head_unstable_unpin(
+	#[method(name = "chainHead_unstable_unpin", raw_method)]
+	async fn chain_head_unstable_unpin(
 		&self,
 		follow_subscription: String,
 		hash_or_hashes: ListOrValue<Hash>,
@@ -131,8 +131,8 @@ pub trait ChainHeadApi<Hash> {
 	/// # Unstable
 	///
 	/// This method is unstable and subject to change in the future.
-	#[method(name = "chainHead_unstable_continue", blocking)]
-	fn chain_head_unstable_continue(
+	#[method(name = "chainHead_unstable_continue", raw_method)]
+	async fn chain_head_unstable_continue(
 		&self,
 		follow_subscription: String,
 		operation_id: String,
@@ -145,8 +145,8 @@ pub trait ChainHeadApi<Hash> {
 	/// # Unstable
 	///
 	/// This method is unstable and subject to change in the future.
-	#[method(name = "chainHead_unstable_stopOperation", blocking)]
-	fn chain_head_unstable_stop_operation(
+	#[method(name = "chainHead_unstable_stopOperation", raw_method)]
+	async fn chain_head_unstable_stop_operation(
 		&self,
 		follow_subscription: String,
 		operation_id: String,