tx/metrics: Add metrics for the RPC v2 transactionWatch_v1_submitAndWatch (#8345)

This PR adds metrics for the following RPC subscription:
[transactionWatch_v1_submitAndWatch](https://paritytech.github.io/json-rpc-interface-spec/api/transactionWatch_v1_submitAndWatch.html)

Metrics are exposed in two ways:
- simple counters of how many events we've seen globally
- a histogram vector of execution times, which is labeled by `initial
event` -> `final event`
- This helps us identify how long it takes the transaction pool to
advance the state of the events, and further debug issues
  
Part of: #8336

### (outdated) PoC Dashboards

![Screenshot 2025-04-28 at 17 50
48](https://github.com/user-attachments/assets/9fd0bf30-a321-4362-a10b-dfc3de1eb474)


### Next steps
- [x] initial dashboards with a live node
- [x] adjust testing

---------

Signed-off-by: Alexandru Vasile <[email protected]>
Co-authored-by: cmd[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: 2 people

Cargo.lock

@@ -0,0 +1,26 @@
+title: 'tx/metrics: Add metrics for the RPC v2 `transactionWatch_v1_submitAndWatch`'
+doc:
+- audience: Node Operator
+  description: |-
+    This PR adds metrics for the following RPC subscription: [transactionWatch_v1_submitAndWatch](https://paritytech.github.io/json-rpc-interface-spec/api/transactionWatch_v1_submitAndWatch.html)
+
+    Metrics are exposed in two ways:
+    - simple counters of how many events we've seen globally
+    - a histogram vector of execution times, which is labeled by `initial event` -> `final event`
+      - This helps us identify how long it takes the transaction pool to advance the state of the events, and further debug issues
+
+    Part of: https://github.com/paritytech/polkadot-sdk/issues/8336
+
+    ### (outdated) PoC Dashboards
+
+    ![Screenshot 2025-04-28 at 17 50 48](https://github.com/user-attachments/assets/9fd0bf30-a321-4362-a10b-dfc3de1eb474)
+
+
+    ### Next steps
+    - [x] initial dashboards with a live node
+    - [x] adjust testing
+crates:
+- name: sc-service
+  bump: major
+- name: sc-rpc-spec-v2
+  bump: major

prdoc/pr_8345.prdoc

@@ -28,6 +28,7 @@ hex = { workspace = true, default-features = true }
 itertools = { workspace = true }
 log = { workspace = true, default-features = true }
 parking_lot = { workspace = true, default-features = true }
+prometheus-endpoint = { workspace = true, default-features = true }
 rand = { workspace = true, default-features = true }
 sc-client-api = { workspace = true, default-features = true }
 sc-rpc = { workspace = true, default-features = true }

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

@@ -95,6 +95,19 @@ pub enum TransactionEvent<Hash> {
 	Dropped(TransactionDropped),
 }
 
+impl<Hash> TransactionEvent<Hash> {
+	/// Returns true if this is the last event emitted by the RPC subscription.
+	pub fn is_final(&self) -> bool {
+		matches!(
+			&self,
+			TransactionEvent::Finalized(_) |
+				TransactionEvent::Error(_) |
+				TransactionEvent::Invalid(_) |
+				TransactionEvent::Dropped(_)
+		)
+	}
+}
+
 /// Intermediate representation (IR) for the transaction events
 /// that handles block events only.
 ///

substrate/client/rpc-spec-v2/src/transaction/event.rs

@@ -0,0 +1,155 @@
+// 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/>.
+
+//! Metrics for recording transaction events.
+
+use std::{collections::HashSet, time::Instant};
+
+use prometheus_endpoint::{
+	exponential_buckets, linear_buckets, register, Histogram, HistogramOpts, PrometheusError,
+	Registry,
+};
+
+use super::TransactionEvent;
+
+/// RPC layer metrics for transaction pool.
+#[derive(Debug, Clone)]
+pub struct Metrics {
+	validated: Histogram,
+	in_block: Histogram,
+	finalized: Histogram,
+	dropped: Histogram,
+	invalid: Histogram,
+	error: Histogram,
+}
+
+impl Metrics {
+	/// Creates a new [`Metrics`] instance.
+	pub fn new(registry: &Registry) -> Result<Self, PrometheusError> {
+		let validated = register(
+			Histogram::with_opts(
+				HistogramOpts::new(
+					"rpc_transaction_validation_time",
+					"RPC Transaction validation time in seconds",
+				)
+				.buckets(exponential_buckets(0.01, 2.0, 16).expect("Valid buckets; qed")),
+			)?,
+			registry,
+		)?;
+
+		let in_block = register(
+			Histogram::with_opts(
+				HistogramOpts::new(
+					"rpc_transaction_in_block_time",
+					"RPC Transaction in block time in seconds",
+				)
+				.buckets(linear_buckets(0.0, 3.0, 20).expect("Valid buckets; qed")),
+			)?,
+			registry,
+		)?;
+
+		let finalized = register(
+			Histogram::with_opts(
+				HistogramOpts::new(
+					"rpc_transaction_finalized_time",
+					"RPC Transaction finalized time in seconds",
+				)
+				.buckets(linear_buckets(0.01, 40.0, 20).expect("Valid buckets; qed")),
+			)?,
+			registry,
+		)?;
+
+		let dropped = register(
+			Histogram::with_opts(
+				HistogramOpts::new(
+					"rpc_transaction_dropped_time",
+					"RPC Transaction dropped time in seconds",
+				)
+				.buckets(linear_buckets(0.01, 3.0, 20).expect("Valid buckets; qed")),
+			)?,
+			registry,
+		)?;
+
+		let invalid = register(
+			Histogram::with_opts(
+				HistogramOpts::new(
+					"rpc_transaction_invalid_time",
+					"RPC Transaction invalid time in seconds",
+				)
+				.buckets(linear_buckets(0.01, 3.0, 20).expect("Valid buckets; qed")),
+			)?,
+			registry,
+		)?;
+
+		let error = register(
+			Histogram::with_opts(
+				HistogramOpts::new(
+					"rpc_transaction_error_time",
+					"RPC Transaction error time in seconds",
+				)
+				.buckets(linear_buckets(0.01, 3.0, 20).expect("Valid buckets; qed")),
+			)?,
+			registry,
+		)?;
+
+		Ok(Metrics { validated, in_block, finalized, dropped, invalid, error })
+	}
+}
+
+/// Transaction metrics for a single transaction instance.
+pub struct InstanceMetrics {
+	/// The metrics instance.
+	metrics: Option<Metrics>,
+	/// The time when the transaction was submitted.
+	submitted_at: Instant,
+	/// Ensure the states are reported once.
+	reported_states: HashSet<&'static str>,
+}
+
+impl InstanceMetrics {
+	/// Creates a new [`InstanceMetrics`] instance.
+	pub fn new(metrics: Option<Metrics>) -> Self {
+		Self { metrics, submitted_at: Instant::now(), reported_states: HashSet::new() }
+	}
+
+	/// Record the execution time of a transaction state.
+	///
+	/// This represents how long it took for the transaction to move to the next state.
+	///
+	/// The method must be called before the transaction event is provided to the user.
+	pub fn register_event<Hash>(&mut self, event: &TransactionEvent<Hash>) {
+		let Some(ref metrics) = self.metrics else {
+			return;
+		};
+
+		let (histogram, target_state) = match event {
+			TransactionEvent::Validated => (&metrics.validated, "validated"),
+			TransactionEvent::BestChainBlockIncluded(Some(_)) => (&metrics.in_block, "in_block"),
+			TransactionEvent::BestChainBlockIncluded(None) => (&metrics.in_block, "retracted"),
+			TransactionEvent::Finalized(..) => (&metrics.finalized, "finalized"),
+			TransactionEvent::Error(..) => (&metrics.error, "error"),
+			TransactionEvent::Dropped(..) => (&metrics.dropped, "dropped"),
+			TransactionEvent::Invalid(..) => (&metrics.invalid, "invalid"),
+		};
+
+		// Only record the state if it hasn't been reported before.
+		if self.reported_states.insert(target_state) {
+			histogram.observe(self.submitted_at.elapsed().as_secs_f64());
+		}
+	}
+}

substrate/client/rpc-spec-v2/src/transaction/metrics.rs

@@ -28,6 +28,8 @@
 #[cfg(test)]
 mod tests;
 
+mod metrics;
+
 pub mod api;
 pub mod error;
 pub mod event;
@@ -36,5 +38,6 @@ pub mod transaction_broadcast;
 
 pub use api::{TransactionApiServer, TransactionBroadcastApiServer};
 pub use event::{TransactionBlock, TransactionDropped, TransactionError, TransactionEvent};
+pub use metrics::Metrics as TransactionMetrics;
 pub use transaction::Transaction;
 pub use transaction_broadcast::TransactionBroadcast;

substrate/client/rpc-spec-v2/src/transaction/mod.rs

@@ -115,7 +115,8 @@ pub fn setup_api_tx() -> (
 	let (task_executor, executor_recv) = TaskExecutorBroadcast::new();
 
 	let tx_api =
-		RpcTransaction::new(client_mock.clone(), pool.clone(), Arc::new(task_executor)).into_rpc();
+		RpcTransaction::new(client_mock.clone(), pool.clone(), Arc::new(task_executor), None)
+			.into_rpc();
 
 	(api, pool, client_mock, tx_api, executor_recv, pool_state)
 }

substrate/client/rpc-spec-v2/src/transaction/tests/setup.rs

@@ -30,6 +30,9 @@ use crate::{
 use codec::Decode;
 use futures::{StreamExt, TryFutureExt};
 use jsonrpsee::{core::async_trait, PendingSubscriptionSink};
+
+use super::metrics::{InstanceMetrics, Metrics};
+
 use sc_rpc::utils::{RingBuffer, Subscription};
 use sc_transaction_pool_api::{
 	error::IntoPoolError, BlockHash, TransactionFor, TransactionPool, TransactionSource,
@@ -50,12 +53,19 @@ pub struct Transaction<Pool, Client> {
 	pool: Arc<Pool>,
 	/// Executor to spawn subscriptions.
 	executor: SubscriptionTaskExecutor,
+	/// Metrics for transactions.
+	metrics: Option<Metrics>,
 }
 
 impl<Pool, Client> Transaction<Pool, Client> {
 	/// Creates a new [`Transaction`].
-	pub fn new(client: Arc<Client>, pool: Arc<Pool>, executor: SubscriptionTaskExecutor) -> Self {
-		Transaction { client, pool, executor }
+	pub fn new(
+		client: Arc<Client>,
+		pool: Arc<Pool>,
+		executor: SubscriptionTaskExecutor,
+		metrics: Option<Metrics>,
+	) -> Self {
+		Transaction { client, pool, executor, metrics }
 	}
 }
 
@@ -78,6 +88,9 @@ where
 		let client = self.client.clone();
 		let pool = self.pool.clone();
 
+		// Get a new transaction metrics instance and increment the counter.
+		let mut metrics = InstanceMetrics::new(self.metrics.clone());
+
 		let fut = async move {
 			let decoded_extrinsic = match TransactionFor::<Pool>::decode(&mut &xt[..]) {
 				Ok(decoded_extrinsic) => decoded_extrinsic,
@@ -86,12 +99,14 @@ where
 
 					let Ok(sink) = pending.accept().await.map(Subscription::from) else { return };
 
+					let event = TransactionEvent::Invalid::<BlockHash<Pool>>(TransactionError {
+						error: "Extrinsic bytes cannot be decoded".into(),
+					});
+
+					metrics.register_event(&event);
+
 					// The transaction is invalid.
-					let _ = sink
-						.send(&TransactionEvent::Invalid::<BlockHash<Pool>>(TransactionError {
-							error: "Extrinsic bytes cannot be decoded".into(),
-						}))
-						.await;
+					let _ = sink.send(&event).await;
 					return
 				},
 			};
@@ -112,8 +127,17 @@ where
 
 			match submit.await {
 				Ok(stream) => {
-					let stream =
-						stream.filter_map(move |event| async move { handle_event(event) }).boxed();
+					let stream = stream
+						.filter_map(|event| {
+							let event = handle_event(event);
+
+							event.as_ref().inspect(|event| {
+								metrics.register_event(event);
+							});
+
+							async move { event }
+						})
+						.boxed();
 
 					// If the subscription is too slow older events will be overwritten.
 					sink.pipe_from_stream(stream, RingBuffer::new(3)).await;
@@ -122,6 +146,9 @@ where
 					// We have not created an `Watcher` for the tx. Make sure the
 					// error is still propagated as an event.
 					let event: TransactionEvent<<Pool::Block as BlockT>::Hash> = err.into();
+
+					metrics.register_event(&event);
+
 					_ = sink.send(&event).await;
 				},
 			};
@@ -134,7 +161,7 @@ where
 /// Handle events generated by the transaction-pool and convert them
 /// to the new API expected state.
 #[inline]
-pub fn handle_event<Hash: Clone, BlockHash: Clone>(
+fn handle_event<Hash: Clone, BlockHash: Clone>(
 	event: TransactionStatus<Hash, BlockHash>,
 ) -> Option<TransactionEvent<BlockHash>> {
 	match event {

substrate/client/rpc-spec-v2/src/transaction/transaction.rs

@@ -515,6 +515,14 @@ where
 	let rpc_id_provider = config.rpc.id_provider.take();
 
 	// jsonrpsee RPC
+	// RPC-V2 specific metrics need to be registered before the RPC server is started,
+	// since we might have two instances running (one for the in-memory RPC and one for the network
+	// RPC).
+	let rpc_v2_metrics = config
+		.prometheus_registry()
+		.map(|registry| sc_rpc_spec_v2::transaction::TransactionMetrics::new(registry))
+		.transpose()?;
+
 	let gen_rpc_module = || {
 		gen_rpc_module(
 			task_manager.spawn_handle(),
@@ -529,6 +537,7 @@ where
 			config.blocks_pruning,
 			backend.clone(),
 			&*rpc_builder,
+			rpc_v2_metrics.clone(),
 		)
 	};
 
@@ -676,6 +685,7 @@ pub fn gen_rpc_module<TBl, TBackend, TCl, TRpc, TExPool>(
 	blocks_pruning: BlocksPruning,
 	backend: Arc<TBackend>,
 	rpc_builder: &(dyn Fn(SubscriptionTaskExecutor) -> Result<RpcModule<TRpc>, Error>),
+	metrics: Option<sc_rpc_spec_v2::transaction::TransactionMetrics>,
 ) -> Result<RpcModule<()>, Error>
 where
 	TBl: BlockT,
@@ -731,6 +741,7 @@ where
 		client.clone(),
 		transaction_pool.clone(),
 		task_executor.clone(),
+		metrics,
 	)
 	.into_rpc();