update jsonrpsee again

Cargo.toml

@@ -553,6 +553,6 @@ yamux = { opt-level = 3 }
 zeroize = { opt-level = 3 }
 
 [patch.crates-io]
-jsonrpsee = { git = "https://github.com/paritytech/jsonrpsee", branch = "na-jsonrpsee-service" }
-jsonrpsee-core = { git = "https://github.com/paritytech/jsonrpsee", branch = "na-jsonrpsee-service" }
+jsonrpsee = { git = "https://github.com/paritytech/jsonrpsee", branch = "master" }
+jsonrpsee-core = { git = "https://github.com/paritytech/jsonrpsee", branch = "master" }
 

substrate/client/rpc-servers/Cargo.toml

@@ -23,3 +23,4 @@ tower = { version = "0.4.13", features = ["util"] }
 http = "0.2.8"
 hyper = "0.14.27"
 futures = "0.3.29"
+pin-project = "1.1.3"

substrate/client/rpc-servers/src/lib.rs

@@ -118,8 +118,11 @@ pub async fn start_server<M: Send + Sync + 'static>(
 		.max_response_body_size(max_payload_out_mb.saturating_mul(MEGABYTE))
 		.max_connections(max_connections)
 		.max_subscriptions_per_connection(max_subs_per_conn)
-		.ping_interval(PingConfig::WithoutInactivityCheck(Duration::from_secs(30)))
-		.unwrap()
+		.enable_ws_ping(
+			PingConfig::new()
+				.ping_interval(Duration::from_secs(30))
+				.inactive_limit(Duration::from_secs(40)),
+		)
 		.set_http_middleware(http_middleware)
 		.set_message_buffer_capacity(message_buffer_capacity)
 		.custom_tokio_runtime(tokio_handle);

substrate/client/rpc-servers/src/middleware/metrics.rs

@@ -1,12 +1,18 @@
-use prometheus_endpoint::{
-	register, Counter, CounterVec, HistogramOpts, HistogramVec, Opts, PrometheusError, Registry,
-	U64,
+use std::{
+	future::Future,
+	pin::Pin,
+	task::{Context, Poll},
+	time::Instant,
 };
 
 use jsonrpsee::{
 	core::async_trait, server::middleware::rpc::RpcServiceT, types::Request, MethodResponse,
 };
-use std::time::Instant;
+use pin_project::pin_project;
+use prometheus_endpoint::{
+	register, Counter, CounterVec, HistogramOpts, HistogramVec, Opts, PrometheusError, Registry,
+	U64,
+};
 
 /// Histogram time buckets in microseconds.
 const HISTOGRAM_BUCKETS: [f64; 11] = [
@@ -172,7 +178,9 @@ impl<'a, S> RpcServiceT<'a> for Metrics<S>
 where
 	S: Send + Sync + RpcServiceT<'a>,
 {
-	async fn call(&self, req: Request<'a>) -> MethodResponse {
+	type Future = ResponseFuture<'a, S::Future>;
+
+	fn call(&self, req: Request<'a>) -> Self::Future {
 		let now = Instant::now();
 
 		log::trace!(
@@ -187,33 +195,70 @@ where
 			.with_label_values(&[self.transport_label, req.method_name()])
 			.inc();
 
-		let rp = self.service.call(req.clone()).await;
+		ResponseFuture {
+			fut: self.service.call(req.clone()),
+			metrics: self.metrics.clone(),
+			req,
+			now,
+			transport_label: self.transport_label,
+		}
+	}
+}
 
-		log::trace!(target: "rpc_metrics", "[{}] on_response started_at={:?}", self.transport_label, now);
-		log::trace!(target: "rpc_metrics::extra", "[{}] result={:?}", self.transport_label, rp);
+/// Response future for metrics.
+#[pin_project]
+pub struct ResponseFuture<'a, F> {
+	#[pin]
+	fut: F,
+	metrics: RpcMetrics,
+	req: Request<'a>,
+	now: Instant,
+	transport_label: &'static str,
+}
 
-		let micros = now.elapsed().as_micros();
-		log::debug!(
-			target: "rpc_metrics",
-			"[{}] {} call took {} μs",
-			self.transport_label,
-			req.method_name(),
-			micros,
-		);
-		self.metrics
-			.calls_time
-			.with_label_values(&[self.transport_label, req.method_name()])
-			.observe(micros as _);
-		self.metrics
-			.calls_finished
-			.with_label_values(&[
-				self.transport_label,
-				req.method_name(),
-				// the label "is_error", so `success` should be regarded as false
-				// and vice-versa to be registrered correctly.
-				if rp.success_or_error.is_success() { "false" } else { "true" },
-			])
-			.inc();
-		rp
+impl<'a, F> std::fmt::Debug for ResponseFuture<'a, F> {
+	fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+		f.write_str("ResponseFuture")
+	}
+}
+
+impl<'a, F: Future<Output = MethodResponse>> Future for ResponseFuture<'a, F> {
+	type Output = F::Output;
+
+	fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+		let this = self.project();
+
+		let res = this.fut.poll(cx);
+		if let Poll::Ready(rp) = &res {
+			let method_name = this.req.method_name();
+			let transport_label = &this.transport_label;
+			let now = this.now;
+			let metrics = &this.metrics;
+
+			log::trace!(target: "rpc_metrics", "[{transport_label}] on_response started_at={:?}", now);
+			log::trace!(target: "rpc_metrics::extra", "[{transport_label}] result={:?}", rp);
+
+			let micros = now.elapsed().as_micros();
+			log::debug!(
+				target: "rpc_metrics",
+				"[{transport_label}] {method_name} call took {} μs",
+				micros,
+			);
+			metrics
+				.calls_time
+				.with_label_values(&[transport_label, method_name])
+				.observe(micros as _);
+			metrics
+				.calls_finished
+				.with_label_values(&[
+					transport_label,
+					method_name,
+					// the label "is_error", so `success` should be regarded as false
+					// and vice-versa to be registrered correctly.
+					if rp.success_or_error.is_success() { "false" } else { "true" },
+				])
+				.inc();
+		}
+		res
 	}
 }

substrate/client/rpc-servers/src/middleware/rate_limit.rs

@@ -1,6 +1,6 @@
 use jsonrpsee::{
 	core::async_trait,
-	server::middleware::rpc::RpcServiceT,
+	server::middleware::rpc::{ResponseFuture, RpcServiceT},
 	types::{ErrorObject, Request},
 	MethodResponse,
 };
@@ -51,13 +51,7 @@ enum State {
 	Allow { until: Instant, rem: u64 },
 }
 
-/// Depending on how the rate limit is instantiated
-/// it's possible to select whether the rate limit
-/// is be applied per connection or shared by
-/// all connections.
-///
-/// Have a look at `async fn run_server` below which
-/// shows how do it.
+/// Rate limit middleware.
 #[derive(Clone)]
 pub struct RateLimit<S> {
 	service: S,
@@ -87,7 +81,9 @@ impl<'a, S> RpcServiceT<'a> for RateLimit<S>
 where
 	S: Send + Sync + RpcServiceT<'a>,
 {
-	async fn call(&self, req: Request<'a>) -> MethodResponse {
+	type Future = ResponseFuture<S::Future>;
+
+	fn call(&self, req: Request<'a>) -> Self::Future {
 		let now = Instant::now();
 
 		let is_denied = {
@@ -117,16 +113,18 @@ where
 		};
 
 		if is_denied {
-			MethodResponse::error(
+			let rp = MethodResponse::error(
 				req.id,
 				ErrorObject::owned(
 					-32000,
 					"RPC rate limit",
 					Some(format!("{} calls/min is allowed", self.rate.num)),
 				),
-			)
+			);
+
+			ResponseFuture::ready(rp)
 		} else {
-			self.service.call(req).await
+			ResponseFuture::future(self.service.call(req))
 		}
 	}
 }