rpc: add prometheus metrics for dropped subscriptions

bridges/relays/client-substrate/Cargo.toml

@@ -15,10 +15,11 @@ async-std = { version = "1.9.0", features = ["attributes"] }
 async-trait = "0.1.79"
 codec = { package = "parity-scale-codec", version = "3.6.12" }
 futures = "0.3.30"
-jsonrpsee = { version = "0.22", features = ["macros", "ws-client"] }
+jsonrpsee = { version = "0.23.1", features = ["macros", "ws-client"] }
 log = { workspace = true }
 num-traits = "0.2"
 rand = "0.8.5"
+serde_json = { workspace = true }
 scale-info = { version = "2.11.1", features = ["derive"] }
 tokio = { version = "1.37", features = ["rt-multi-thread"] }
 thiserror = { workspace = true }

bridges/relays/client-substrate/src/client/subscription.rs

@@ -21,7 +21,6 @@ use async_std::{
 	stream::StreamExt,
 };
 use futures::{FutureExt, Stream};
-use jsonrpsee::core::ClientError;
 use sp_runtime::DeserializeOwned;
 use std::{
 	fmt::Debug,
@@ -143,7 +142,7 @@ impl<T: 'static + Clone + DeserializeOwned + Send> Subscription<T> {
 	/// Create new forwarded subscription.
 	pub fn new_forwarded(
 		desc: StreamDescription,
-		subscription: impl Stream<Item = StdResult<T, ClientError>> + Unpin + Send + 'static,
+		subscription: impl Stream<Item = StdResult<T, serde_json::Error>> + Unpin + Send + 'static,
 	) -> Self {
 		Self {
 			desc: desc.clone(),

cumulus/client/relay-chain-interface/Cargo.toml

@@ -23,5 +23,5 @@ sp-version = { path = "../../../substrate/primitives/version", default-features
 futures = "0.3.28"
 async-trait = "0.1.79"
 thiserror = { workspace = true }
-jsonrpsee-core = "0.22"
+jsonrpsee-core = "0.23.1"
 codec = { package = "parity-scale-codec", version = "3.6.12" }

cumulus/client/relay-chain-rpc-interface/Cargo.toml

@@ -33,7 +33,7 @@ tokio-util = { version = "0.7.8", features = ["compat"] }
 futures = "0.3.28"
 futures-timer = "3.0.2"
 codec = { package = "parity-scale-codec", version = "3.6.12" }
-jsonrpsee = { version = "0.22", features = ["ws-client"] }
+jsonrpsee = { version = "0.23.1", features = ["ws-client"] }
 tracing = "0.1.37"
 async-trait = "0.1.79"
 url = "2.4.0"

cumulus/client/relay-chain-rpc-interface/src/light_client_worker.rs

@@ -20,7 +20,7 @@
 
 use futures::{channel::mpsc::Sender, prelude::*, stream::FuturesUnordered};
 use jsonrpsee::core::client::{
-	Client as JsonRpseeClient, ClientBuilder, ClientT, Error, ReceivedMessage, TransportReceiverT,
+	Client as JsonRpseeClient, ClientBuilder, ClientT, ReceivedMessage, TransportReceiverT,
 	TransportSenderT,
 };
 use smoldot_light::{ChainId, Client as SmoldotClient, JsonRpcResponses};
@@ -124,7 +124,7 @@ pub struct LightClientRpcWorker {
 }
 
 fn handle_notification(
-	maybe_header: Option<Result<RelayHeader, Error>>,
+	maybe_header: Option<Result<RelayHeader, serde_json::Error>>,
 	senders: &mut Vec<Sender<RelayHeader>>,
 ) -> Result<(), ()> {
 	match maybe_header {

cumulus/polkadot-parachain/Cargo.toml

@@ -39,7 +39,7 @@ coretime-rococo-runtime = { path = "../parachains/runtimes/coretime/coretime-roc
 coretime-westend-runtime = { path = "../parachains/runtimes/coretime/coretime-westend" }
 bridge-hub-westend-runtime = { path = "../parachains/runtimes/bridge-hubs/bridge-hub-westend" }
 penpal-runtime = { path = "../parachains/runtimes/testing/penpal" }
-jsonrpsee = { version = "0.22", features = ["server"] }
+jsonrpsee = { version = "0.23.1", features = ["server"] }
 people-rococo-runtime = { path = "../parachains/runtimes/people/people-rococo" }
 people-westend-runtime = { path = "../parachains/runtimes/people/people-westend" }
 parachains-common = { path = "../parachains/common" }

cumulus/test/service/Cargo.toml

@@ -17,7 +17,7 @@ async-trait = "0.1.79"
 clap = { version = "4.5.3", features = ["derive"] }
 codec = { package = "parity-scale-codec", version = "3.6.12" }
 criterion = { version = "0.5.1", features = ["async_tokio"] }
-jsonrpsee = { version = "0.22", features = ["server"] }
+jsonrpsee = { version = "0.23.1", features = ["server"] }
 rand = "0.8.5"
 serde = { features = ["derive"], workspace = true, default-features = true }
 serde_json = { workspace = true, default-features = true }

polkadot/rpc/Cargo.toml

@@ -10,7 +10,7 @@ description = "Polkadot specific RPC functionality."
 workspace = true
 
 [dependencies]
-jsonrpsee = { version = "0.22", features = ["server"] }
+jsonrpsee = { version = "0.23.1", features = ["server"] }
 polkadot-primitives = { path = "../primitives" }
 sc-client-api = { path = "../../substrate/client/api" }
 sp-blockchain = { path = "../../substrate/primitives/blockchain" }

polkadot/rpc/src/lib.rs

@@ -142,6 +142,8 @@ where
 	let chain_name = chain_spec.name().to_string();
 	let genesis_hash = client.hash(0).ok().flatten().expect("Genesis block exists; qed");
 	let properties = chain_spec.properties();
+	// TODO: This should be a shared metrics instance.
+	let metrics = sc_rpc::SubscriptionMetrics::disabled();
 
 	io.merge(ChainSpec::new(chain_name, genesis_hash, properties).into_rpc())?;
 	io.merge(StateMigration::new(client.clone(), backend.clone(), deny_unsafe).into_rpc())?;
@@ -167,6 +169,7 @@ where
 			shared_voter_state,
 			justification_stream,
 			finality_provider,
+			metrics.clone(),
 		)
 		.into_rpc(),
 	)?;
@@ -179,6 +182,7 @@ where
 			beefy.beefy_finality_proof_stream,
 			beefy.beefy_best_block_stream,
 			beefy.subscription_executor,
+			metrics,
 		)?
 		.into_rpc(),
 	)?;

substrate/bin/node/cli/Cargo.toml

@@ -41,7 +41,7 @@ array-bytes = "6.1"
 clap = { version = "4.5.3", features = ["derive"], optional = true }
 codec = { package = "parity-scale-codec", version = "3.6.12" }
 serde = { features = ["derive"], workspace = true, default-features = true }
-jsonrpsee = { version = "0.22", features = ["server"] }
+jsonrpsee = { version = "0.23.1", features = ["server"] }
 futures = "0.3.30"
 log = { workspace = true, default-features = true }
 rand = "0.8"

substrate/bin/node/rpc/Cargo.toml

@@ -16,7 +16,7 @@ workspace = true
 targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
-jsonrpsee = { version = "0.22", features = ["server"] }
+jsonrpsee = { version = "0.23.1", features = ["server"] }
 node-primitives = { path = "../primitives" }
 pallet-transaction-payment-rpc = { path = "../../../frame/transaction-payment/rpc" }
 mmr-rpc = { path = "../../../client/merkle-mountain-range/rpc" }

substrate/bin/node/rpc/src/lib.rs

@@ -179,6 +179,8 @@ where
 	let chain_name = chain_spec.name().to_string();
 	let genesis_hash = client.block_hash(0).ok().flatten().expect("Genesis block exists; qed");
 	let properties = chain_spec.properties();
+	let metrics = sc_rpc::SubscriptionMetrics::disabled();
+
 	io.merge(ChainSpec::new(chain_name, genesis_hash, properties).into_rpc())?;
 
 	io.merge(System::new(client.clone(), pool, deny_unsafe).into_rpc())?;
@@ -206,6 +208,7 @@ where
 			shared_voter_state,
 			justification_stream,
 			finality_provider,
+			metrics.clone(),
 		)
 		.into_rpc(),
 	)?;
@@ -231,6 +234,7 @@ where
 			beefy.beefy_finality_proof_stream,
 			beefy.beefy_best_block_stream,
 			beefy.subscription_executor,
+			metrics,
 		)?
 		.into_rpc(),
 	)?;

substrate/client/consensus/babe/rpc/Cargo.toml

@@ -16,7 +16,7 @@ workspace = true
 targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
-jsonrpsee = { version = "0.22.5", features = ["client-core", "macros", "server-core"] }
+jsonrpsee = { version = "0.23.1", features = ["client-core", "macros", "server-core"] }
 futures = "0.3.30"
 serde = { features = ["derive"], workspace = true, default-features = true }
 thiserror = { workspace = true }

substrate/client/consensus/beefy/rpc/Cargo.toml

@@ -14,7 +14,7 @@ workspace = true
 [dependencies]
 codec = { package = "parity-scale-codec", version = "3.6.12", features = ["derive"] }
 futures = "0.3.30"
-jsonrpsee = { version = "0.22.5", features = ["client-core", "macros", "server-core"] }
+jsonrpsee = { version = "0.23.1", features = ["client-core", "macros", "server-core"] }
 log = { workspace = true, default-features = true }
 parking_lot = "0.12.1"
 serde = { features = ["derive"], workspace = true, default-features = true }

substrate/client/consensus/beefy/rpc/src/lib.rs

@@ -24,7 +24,9 @@ use parking_lot::RwLock;
 use sp_consensus_beefy::AuthorityIdBound;
 use std::sync::Arc;
 
-use sc_rpc::{utils::pipe_from_stream, SubscriptionTaskExecutor};
+use sc_rpc::{
+	utils::pipe_from_stream, SubscriptionMetrics, SubscriptionParams, SubscriptionTaskExecutor,
+};
 use sp_application_crypto::RuntimeAppPublic;
 use sp_runtime::traits::Block as BlockT;
 
@@ -33,7 +35,7 @@ use jsonrpsee::{
 	core::async_trait,
 	proc_macros::rpc,
 	types::{ErrorObject, ErrorObjectOwned},
-	PendingSubscriptionSink,
+	ConnectionId, Extensions, PendingSubscriptionSink,
 };
 use log::warn;
 
@@ -87,6 +89,7 @@ pub trait BeefyApi<Notification, Hash> {
 		name = "beefy_subscribeJustifications" => "beefy_justifications",
 		unsubscribe = "beefy_unsubscribeJustifications",
 		item = Notification,
+		with_extensions
 	)]
 	fn subscribe_justifications(&self);
 
@@ -104,6 +107,7 @@ pub struct Beefy<Block: BlockT, AuthorityId: AuthorityIdBound> {
 	finality_proof_stream: BeefyVersionedFinalityProofStream<Block, AuthorityId>,
 	beefy_best_block: Arc<RwLock<Option<Block::Hash>>>,
 	executor: SubscriptionTaskExecutor,
+	metrics: SubscriptionMetrics,
 }
 
 impl<Block, AuthorityId> Beefy<Block, AuthorityId>
@@ -116,6 +120,7 @@ where
 		finality_proof_stream: BeefyVersionedFinalityProofStream<Block, AuthorityId>,
 		best_block_stream: BeefyBestBlockStream<Block>,
 		executor: SubscriptionTaskExecutor,
+		metrics: SubscriptionMetrics,
 	) -> Result<Self, Error> {
 		let beefy_best_block = Arc::new(RwLock::new(None));
 
@@ -127,7 +132,7 @@ where
 		});
 
 		executor.spawn("substrate-rpc-subscription", Some("rpc"), future.map(drop).boxed());
-		Ok(Self { finality_proof_stream, beefy_best_block, executor })
+		Ok(Self { finality_proof_stream, beefy_best_block, executor, metrics })
 	}
 }
 
@@ -139,13 +144,23 @@ where
 	AuthorityId: AuthorityIdBound,
 	<AuthorityId as RuntimeAppPublic>::Signature: Send + Sync,
 {
-	fn subscribe_justifications(&self, pending: PendingSubscriptionSink) {
+	fn subscribe_justifications(&self, pending: PendingSubscriptionSink, ext: &Extensions) {
+		let params = SubscriptionParams {
+			conn_id: *ext.get::<ConnectionId>().expect("ConnectionId is set"),
+			ip_addr: *ext.get::<std::net::IpAddr>().expect("IpAddr is set"),
+			method: "beefy_subscribeJustifications",
+			metrics: self.metrics.clone(),
+		};
+
 		let stream = self
 			.finality_proof_stream
 			.subscribe(100_000)
 			.map(|vfp| notification::EncodedVersionedFinalityProof::new::<Block, AuthorityId>(vfp));
 
-		sc_rpc::utils::spawn_subscription_task(&self.executor, pipe_from_stream(pending, stream));
+		sc_rpc::utils::spawn_subscription_task(
+			&self.executor,
+			pipe_from_stream(pending, stream, params),
+		);
 	}
 
 	async fn latest_finalized(&self) -> Result<Block::Hash, Error> {
@@ -163,6 +178,7 @@ mod tests {
 		communication::notification::BeefyVersionedFinalityProofSender,
 		justification::BeefyVersionedFinalityProof,
 	};
+	use sc_rpc::SubscriptionMetrics;
 	use sp_consensus_beefy::{ecdsa_crypto, known_payloads, Payload, SignedCommitment};
 	use sp_runtime::traits::{BlakeTwo256, Hash};
 	use substrate_test_runtime_client::runtime::Block;
@@ -184,9 +200,13 @@ mod tests {
 		let (finality_proof_sender, finality_proof_stream) =
 			BeefyVersionedFinalityProofStream::<Block, ecdsa_crypto::AuthorityId>::channel();
 
-		let handler =
-			Beefy::new(finality_proof_stream, best_block_stream, sc_rpc::testing::test_executor())
-				.expect("Setting up the BEEFY RPC handler works");
+		let handler = Beefy::new(
+			finality_proof_stream,
+			best_block_stream,
+			sc_rpc::testing::test_executor(),
+			SubscriptionMetrics::disabled()
+		)
+		.expect("Setting up the BEEFY RPC handler works");
 
 		(handler.into_rpc(), finality_proof_sender)
 	}

substrate/client/consensus/grandpa/rpc/Cargo.toml

@@ -15,7 +15,7 @@ workspace = true
 [dependencies]
 finality-grandpa = { version = "0.16.2", features = ["derive-codec"] }
 futures = "0.3.30"
-jsonrpsee = { version = "0.22.5", features = ["client-core", "macros", "server-core"] }
+jsonrpsee = { version = "0.23.1", features = ["client-core", "macros", "server-core"] }
 log = { workspace = true, default-features = true }
 codec = { package = "parity-scale-codec", version = "3.6.12", features = ["derive"] }
 serde = { features = ["derive"], workspace = true, default-features = true }

substrate/client/consensus/grandpa/rpc/src/lib.rs

@@ -26,6 +26,7 @@ use std::sync::Arc;
 use jsonrpsee::{
 	core::{async_trait, server::PendingSubscriptionSink},
 	proc_macros::rpc,
+	ConnectionId, Extensions,
 };
 
 mod error;
@@ -38,7 +39,9 @@ use finality::{EncodedFinalityProof, RpcFinalityProofProvider};
 use notification::JustificationNotification;
 use report::{ReportAuthoritySet, ReportVoterState, ReportedRoundStates};
 use sc_consensus_grandpa::GrandpaJustificationStream;
-use sc_rpc::{utils::pipe_from_stream, SubscriptionTaskExecutor};
+use sc_rpc::{
+	utils::pipe_from_stream, SubscriptionMetrics, SubscriptionParams, SubscriptionTaskExecutor,
+};
 use sp_runtime::traits::{Block as BlockT, NumberFor};
 
 /// Provides RPC methods for interacting with GRANDPA.
@@ -54,7 +57,8 @@ pub trait GrandpaApi<Notification, Hash, Number> {
 	#[subscription(
 		name = "grandpa_subscribeJustifications" => "grandpa_justifications",
 		unsubscribe = "grandpa_unsubscribeJustifications",
-		item = Notification
+		item = Notification,
+		with_extensions
 	)]
 	fn subscribe_justifications(&self);
 
@@ -71,6 +75,7 @@ pub struct Grandpa<AuthoritySet, VoterState, Block: BlockT, ProofProvider> {
 	voter_state: VoterState,
 	justification_stream: GrandpaJustificationStream<Block>,
 	finality_proof_provider: Arc<ProofProvider>,
+	metrics: SubscriptionMetrics,
 }
 impl<AuthoritySet, VoterState, Block: BlockT, ProofProvider>
 	Grandpa<AuthoritySet, VoterState, Block, ProofProvider>
@@ -82,8 +87,16 @@ impl<AuthoritySet, VoterState, Block: BlockT, ProofProvider>
 		voter_state: VoterState,
 		justification_stream: GrandpaJustificationStream<Block>,
 		finality_proof_provider: Arc<ProofProvider>,
+		metrics: SubscriptionMetrics,
 	) -> Self {
-		Self { executor, authority_set, voter_state, justification_stream, finality_proof_provider }
+		Self {
+			executor,
+			authority_set,
+			voter_state,
+			justification_stream,
+			finality_proof_provider,
+			metrics,
+		}
 	}
 }
 
@@ -101,14 +114,24 @@ where
 		ReportedRoundStates::from(&self.authority_set, &self.voter_state)
 	}
 
-	fn subscribe_justifications(&self, pending: PendingSubscriptionSink) {
+	fn subscribe_justifications(&self, pending: PendingSubscriptionSink, ext: &Extensions) {
+		let params = SubscriptionParams {
+			conn_id: *ext.get::<ConnectionId>().expect("ConnectionId is set"),
+			ip_addr: *ext.get::<std::net::IpAddr>().expect("IpAddr is set"),
+			method: "grandpa_subscribeJustifications",
+			metrics: self.metrics.clone(),
+		};
+
 		let stream = self.justification_stream.subscribe(100_000).map(
 			|x: sc_consensus_grandpa::GrandpaJustification<Block>| {
 				JustificationNotification::from(x)
 			},
 		);
 
-		sc_rpc::utils::spawn_subscription_task(&self.executor, pipe_from_stream(pending, stream));
+		sc_rpc::utils::spawn_subscription_task(
+			&self.executor,
+			pipe_from_stream(pending, stream, params),
+		);
 	}
 
 	async fn prove_finality(
@@ -260,6 +283,7 @@ mod tests {
 			voter_state,
 			justification_stream,
 			finality_proof_provider,
+			SubscriptionMetrics::disabled(),
 		)
 		.into_rpc();
 

substrate/client/consensus/manual-seal/Cargo.toml

@@ -16,7 +16,7 @@ workspace = true
 targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
-jsonrpsee = { version = "0.22.5", features = ["client-core", "macros", "server-core"] }
+jsonrpsee = { version = "0.23.1", features = ["client-core", "macros", "server-core"] }
 assert_matches = "1.3.0"
 async-trait = "0.1.79"
 codec = { package = "parity-scale-codec", version = "3.6.12" }

substrate/client/merkle-mountain-range/rpc/Cargo.toml

@@ -16,7 +16,7 @@ targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
 codec = { package = "parity-scale-codec", version = "3.6.12" }
-jsonrpsee = { version = "0.22.5", features = ["client-core", "macros", "server-core"] }
+jsonrpsee = { version = "0.23.1", features = ["client-core", "macros", "server-core"] }
 serde = { features = ["derive"], workspace = true, default-features = true }
 sp-api = { path = "../../../primitives/api" }
 sp-blockchain = { path = "../../../primitives/blockchain" }