rpc: bump jsonrpsee v0.22 and fix race in rpc v2 chain_head (#3230)

Close #2992 Breaking changes: - rpc server grafana metric `substrate_rpc_requests_started` is removed (not possible to implement anymore) - rpc server grafana metric `substrate_rpc_requests_finished` is removed (not possible to implement anymore) - rpc server ws ping/pong not ACK:ed within 30 seconds more than three times then the connection will be closed Added - rpc server grafana metric `substrate_rpc_sessions_time` is added to get the duration for each websocket session
2026-07-21 21:25:41 +00:00 · 2024-02-14 23:18:22 +01:00
parent 7e7c488ba8
commit c7c4fe0184
53 changed files with 777 additions and 468 deletions
@@ -36,7 +36,8 @@ use crate::{
 use codec::Encode;
 use futures::future::FutureExt;
 use jsonrpsee::{
-	core::async_trait, types::SubscriptionId, PendingSubscriptionSink, SubscriptionSink,
+	core::async_trait, server::ResponsePayload, types::SubscriptionId, MethodResponseFuture,
+	PendingSubscriptionSink, SubscriptionSink,
 };
 use log::debug;
 use sc_client_api::{
@@ -218,16 +219,17 @@ where
 		&self,
 		follow_subscription: String,
 		hash: Block::Hash,
-	) -> Result<MethodResponse, ChainHeadRpcError> {
+	) -> ResponsePayload<'static, MethodResponse> {
 		let mut block_guard = match self.subscriptions.lock_block(&follow_subscription, hash, 1) {
 			Ok(block) => block,
 			Err(SubscriptionManagementError::SubscriptionAbsent) |
-			Err(SubscriptionManagementError::ExceededLimits) => return Ok(MethodResponse::LimitReached),
+			Err(SubscriptionManagementError::ExceededLimits) =>
+				return ResponsePayload::success(MethodResponse::LimitReached),
 			Err(SubscriptionManagementError::BlockHashAbsent) => {
 				// Block is not part of the subscription.
-				return Err(ChainHeadRpcError::InvalidBlock.into())
+				return ResponsePayload::error(ChainHeadRpcError::InvalidBlock);
 			},
-			Err(_) => return Err(ChainHeadRpcError::InvalidBlock.into()),
+			Err(_) => return ResponsePayload::error(ChainHeadRpcError::InvalidBlock),
 		};

 		let operation_id = block_guard.operation().operation_id();
@@ -254,7 +256,7 @@ where
 					hash
 				);
 				self.subscriptions.remove_subscription(&follow_subscription);
-				return Err(ChainHeadRpcError::InvalidBlock.into())
+				return ResponsePayload::error(ChainHeadRpcError::InvalidBlock)
 			},
 			Err(error) => FollowEvent::<Block::Hash>::OperationError(OperationError {
 				operation_id: operation_id.clone(),
@@ -262,8 +264,20 @@ where
 			}),
 		};

-		let _ = block_guard.response_sender().unbounded_send(event);
-		Ok(MethodResponse::Started(MethodResponseStarted { operation_id, discarded_items: None }))
+		let (rp, rp_fut) = method_started_response(operation_id, None);
+
+		let fut = async move {
+			// Events should only by generated
+			// if the response was successfully propagated.
+			if rp_fut.await.is_err() {
+				return;
+			}
+			let _ = block_guard.response_sender().unbounded_send(event);
+		};
+
+		self.executor.spawn("substrate-rpc-subscription", Some("rpc"), fut.boxed());
+
+		rp
 	}

 	fn chain_head_unstable_header(
@@ -294,31 +308,40 @@ where
 		hash: Block::Hash,
 		items: Vec<StorageQuery<String>>,
 		child_trie: Option<String>,
-	) -> Result<MethodResponse, ChainHeadRpcError> {
+	) -> ResponsePayload<'static, MethodResponse> {
 		// Gain control over parameter parsing and returned error.
-		let items = items
+		let items = match items
 			.into_iter()
 			.map(|query| {
 				let key = StorageKey(parse_hex_param(query.key)?);
 				Ok(StorageQuery { key, query_type: query.query_type })
 			})
-			.collect::<Result<Vec<_>, ChainHeadRpcError>>()?;
+			.collect::<Result<Vec<_>, ChainHeadRpcError>>()
+		{
+			Ok(items) => items,
+			Err(err) => {
+				return ResponsePayload::error(err);
+			},
+		};

-		let child_trie = child_trie
-			.map(|child_trie| parse_hex_param(child_trie))
-			.transpose()?
-			.map(ChildInfo::new_default_from_vec);
+		let child_trie = match child_trie.map(|child_trie| parse_hex_param(child_trie)).transpose()
+		{
+			Ok(c) => c.map(ChildInfo::new_default_from_vec),
+			Err(e) => return ResponsePayload::error(e),
+		};

 		let mut block_guard =
 			match self.subscriptions.lock_block(&follow_subscription, hash, items.len()) {
 				Ok(block) => block,
 				Err(SubscriptionManagementError::SubscriptionAbsent) |
-				Err(SubscriptionManagementError::ExceededLimits) => return Ok(MethodResponse::LimitReached),
+				Err(SubscriptionManagementError::ExceededLimits) => {
+					return ResponsePayload::success(MethodResponse::LimitReached);
+				},
 				Err(SubscriptionManagementError::BlockHashAbsent) => {
 					// Block is not part of the subscription.
-					return Err(ChainHeadRpcError::InvalidBlock.into())
+					return ResponsePayload::error(ChainHeadRpcError::InvalidBlock)
 				},
-				Err(_) => return Err(ChainHeadRpcError::InvalidBlock.into()),
+				Err(_) => return ResponsePayload::error(ChainHeadRpcError::InvalidBlock),
 			};

 		let mut storage_client = ChainHeadStorage::<Client, Block, BE>::new(
@@ -334,16 +357,21 @@ where
 		let mut items = items;
 		items.truncate(num_operations);

+		let (rp, rp_is_success) = method_started_response(operation_id, Some(discarded));
+
 		let fut = async move {
+			// Events should only by generated
+			// if the response was successfully propagated.
+			if rp_is_success.await.is_err() {
+				return;
+			}
 			storage_client.generate_events(block_guard, hash, items, child_trie).await;
 		};

 		self.executor
 			.spawn_blocking("substrate-rpc-subscription", Some("rpc"), fut.boxed());
-		Ok(MethodResponse::Started(MethodResponseStarted {
-			operation_id,
-			discarded_items: Some(discarded),
-		}))
+
+		rp
 	}

 	fn chain_head_unstable_call(
@@ -352,29 +380,31 @@ where
 		hash: Block::Hash,
 		function: String,
 		call_parameters: String,
-	) -> Result<MethodResponse, ChainHeadRpcError> {
-		let call_parameters = Bytes::from(parse_hex_param(call_parameters)?);
+	) -> ResponsePayload<'static, MethodResponse> {
+		let call_parameters = match parse_hex_param(call_parameters) {
+			Ok(hex) => Bytes::from(hex),
+			Err(err) => return ResponsePayload::error(err),
+		};

 		let mut block_guard = match self.subscriptions.lock_block(&follow_subscription, hash, 1) {
 			Ok(block) => block,
 			Err(SubscriptionManagementError::SubscriptionAbsent) |
 			Err(SubscriptionManagementError::ExceededLimits) => {
 				// Invalid invalid subscription ID.
-				return Ok(MethodResponse::LimitReached)
+				return ResponsePayload::success(MethodResponse::LimitReached)
 			},
 			Err(SubscriptionManagementError::BlockHashAbsent) => {
 				// Block is not part of the subscription.
-				return Err(ChainHeadRpcError::InvalidBlock.into())
+				return ResponsePayload::error(ChainHeadRpcError::InvalidBlock)
 			},
-			Err(_) => return Err(ChainHeadRpcError::InvalidBlock.into()),
+			Err(_) => return ResponsePayload::error(ChainHeadRpcError::InvalidBlock),
 		};

 		// Reject subscription if with_runtime is false.
 		if !block_guard.has_runtime() {
-			return Err(ChainHeadRpcError::InvalidRuntimeCall(
+			return ResponsePayload::error(ChainHeadRpcError::InvalidRuntimeCall(
 				"The runtime updates flag must be set".to_string(),
-			)
-			.into())
+			));
 		}

 		let operation_id = block_guard.operation().operation_id();
@@ -395,11 +425,20 @@ where
 				})
 			});

-		let _ = block_guard.response_sender().unbounded_send(event);
-		Ok(MethodResponse::Started(MethodResponseStarted {
-			operation_id: operation_id.clone(),
-			discarded_items: None,
-		}))
+		let (rp, rp_fut) = method_started_response(operation_id, None);
+
+		let fut = async move {
+			// Events should only by generated
+			// if the response was successfully propagated.
+			if rp_fut.await.is_err() {
+				return;
+			}
+			let _ = block_guard.response_sender().unbounded_send(event);
+		};
+
+		self.executor.spawn("substrate-rpc-subscription", Some("rpc"), fut.boxed());
+
+		rp
 	}

 	fn chain_head_unstable_unpin(
@@ -463,3 +502,11 @@ where
 		Ok(())
 	}
 }
+
+fn method_started_response(
+	operation_id: String,
+	discarded_items: Option<usize>,
+) -> (ResponsePayload<'static, MethodResponse>, MethodResponseFuture) {
+	let rp = MethodResponse::Started(MethodResponseStarted { operation_id, discarded_items });
+	ResponsePayload::success(rp).notify_on_completion()
+}