Improve transaction submission (#6599)

* Improve transaction submission

Before this pr the transaction pool validated each transaction, even if
the transaction was already known to the pool. This pr changes the
behavior to first check if we are already aware of a transaction and
thus, to only validate them if we don't know them yet. However, there is
still the possibility that a given transaction is validated multiple
times. This can happen if the transaction is added the first time, but
is not yet validated and added to the validated pool.

Besides that, this pr fixes the wrong metrics of gossiped transactions
in the network. It also moves some metrics to the transaction pool api,
to better track when a transaction actually is scheduled for validation.

* Make sure we don't submit the same transaction twice from the network concurrently

* Remove added listener call

* Feedback

* Ignore banned on resubmit

substrate/bin/node-template/node/src/service.rs

@@ -47,6 +47,7 @@ macro_rules! new_full_start {
 			.with_transaction_pool(|builder| {
 				let pool_api = sc_transaction_pool::FullChainApi::new(
 					builder.client().clone(),
+					None,
 				);
 				Ok(sc_transaction_pool::BasicPool::new(
 					builder.config().transaction_pool.clone(),

substrate/bin/node/cli/src/service.rs

@@ -61,6 +61,7 @@ macro_rules! new_full_start {
 			.with_transaction_pool(|builder| {
 				let pool_api = sc_transaction_pool::FullChainApi::new(
 					builder.client().clone(),
+					builder.prometheus_registry(),
 				);
 				let config = builder.config();
 

substrate/client/basic-authorship/src/basic_authorship.rs

@@ -361,7 +361,7 @@ mod tests {
 		let txpool = Arc::new(
 			BasicPool::new(
 				Default::default(),
-				Arc::new(FullChainApi::new(client.clone())),
+				Arc::new(FullChainApi::new(client.clone(), None)),
 				None,
 			).0
 		);
@@ -414,7 +414,7 @@ mod tests {
 		let txpool = Arc::new(
 			BasicPool::new(
 				Default::default(),
-				Arc::new(FullChainApi::new(client.clone())),
+				Arc::new(FullChainApi::new(client.clone(), None)),
 				None,
 			).0
 		);
@@ -449,7 +449,7 @@ mod tests {
 		let txpool = Arc::new(
 			BasicPool::new(
 				Default::default(),
-				Arc::new(FullChainApi::new(client.clone())),
+				Arc::new(FullChainApi::new(client.clone(), None)),
 				None,
 			).0
 		);
@@ -511,7 +511,7 @@ mod tests {
 		let txpool = Arc::new(
 			BasicPool::new(
 				Default::default(),
-				Arc::new(FullChainApi::new(client.clone())),
+				Arc::new(FullChainApi::new(client.clone(), None)),
 				None,
 			).0
 		);

substrate/client/basic-authorship/src/lib.rs

@@ -31,7 +31,11 @@
 //! # };
 //! # use sc_transaction_pool::{BasicPool, FullChainApi};
 //! # let client = Arc::new(substrate_test_runtime_client::new());
-//! # let txpool = Arc::new(BasicPool::new(Default::default(), Arc::new(FullChainApi::new(client.clone())), None).0);
+//! # let txpool = Arc::new(BasicPool::new(
+//! #     Default::default(),
+//! #     Arc::new(FullChainApi::new(client.clone(), None)),
+//! #     None).0,
+//! # );
 //! // The first step is to create a `ProposerFactory`.
 //! let mut proposer_factory = ProposerFactory::new(client.clone(), txpool.clone(), None);
 //!

substrate/client/network/src/protocol.rs

@@ -51,7 +51,7 @@ use message::generic::{Message as GenericMessage, ConsensusMessage, Roles};
 use prometheus_endpoint::{Registry, Gauge, Counter, GaugeVec, HistogramVec, PrometheusError, Opts, register, U64};
 use sync::{ChainSync, SyncState};
 use std::borrow::Cow;
-use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
+use std::collections::{BTreeMap, HashMap, HashSet, VecDeque, hash_map::Entry};
 use std::sync::Arc;
 use std::fmt::Write;
 use std::{cmp, io, num::NonZeroUsize, pin::Pin, task::Poll, time};
@@ -199,18 +199,21 @@ impl Metrics {
 	}
 }
 
-struct PendingTransaction {
+#[pin_project::pin_project]
+struct PendingTransaction<H> {
+	#[pin]
 	validation: TransactionImportFuture,
-	peer_id: PeerId,
+	tx_hash: H,
 }
 
-impl Future for PendingTransaction {
-	type Output = (PeerId, TransactionImport);
+impl<H: ExHashT> Future for PendingTransaction<H> {
+	type Output = (H, TransactionImport);
 
 	fn poll(self: Pin<&mut Self>, cx: &mut std::task::Context<'_>) -> Poll<Self::Output> {
-		let this = Pin::into_inner(self);
-		if let Poll::Ready(import_result) = this.validation.poll_unpin(cx) {
-			return Poll::Ready((this.peer_id.clone(), import_result));
+		let mut this = self.project();
+
+		if let Poll::Ready(import_result) = Pin::new(&mut this.validation).poll_unpin(cx) {
+			return Poll::Ready((this.tx_hash.clone(), import_result));
 		}
 
 		Poll::Pending
@@ -226,7 +229,12 @@ pub struct Protocol<B: BlockT, H: ExHashT> {
 	/// Pending list of messages to return from `poll` as a priority.
 	pending_messages: VecDeque<CustomMessageOutcome<B>>,
 	/// Pending transactions verification tasks.
-	pending_transactions: FuturesUnordered<PendingTransaction>,
+	pending_transactions: FuturesUnordered<PendingTransaction<H>>,
+	/// As multiple peers can send us the same transaction, we group
+	/// these peers using the transaction hash while the transaction is
+	/// imported. This prevents that we import the same transaction
+	/// multiple times concurrently.
+	pending_transactions_peers: HashMap<H, Vec<PeerId>>,
 	config: ProtocolConfig,
 	genesis_hash: B::Hash,
 	sync: ChainSync<B>,
@@ -452,6 +460,7 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 			propagate_timeout: Box::pin(interval(PROPAGATE_TIMEOUT)),
 			pending_messages: VecDeque::new(),
 			pending_transactions: FuturesUnordered::new(),
+			pending_transactions_peers: HashMap::new(),
 			config,
 			context_data: ContextData {
 				peers: HashMap::new(),
@@ -1162,7 +1171,7 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 	fn on_transactions(
 		&mut self,
 		who: PeerId,
-		transactions: message::Transactions<B::Extrinsic>
+		transactions: message::Transactions<B::Extrinsic>,
 	) {
 		// sending transaction to light node is considered a bad behavior
 		if !self.config.roles.is_full() {
@@ -1191,14 +1200,22 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 				}
 
 				let hash = self.transaction_pool.hash_of(&t);
-				peer.known_transactions.insert(hash);
+				peer.known_transactions.insert(hash.clone());
 
 				self.peerset_handle.report_peer(who.clone(), rep::ANY_TRANSACTION);
 
-				self.pending_transactions.push(PendingTransaction {
-					peer_id: who.clone(),
-					validation: self.transaction_pool.import(t),
-				});
+				match self.pending_transactions_peers.entry(hash.clone()) {
+					Entry::Vacant(entry) => {
+						self.pending_transactions.push(PendingTransaction {
+							validation: self.transaction_pool.import(t),
+							tx_hash: hash,
+						});
+						entry.insert(vec![who.clone()]);
+					},
+					Entry::Occupied(mut entry) => {
+						entry.get_mut().push(who.clone());
+					}
+				}
 			}
 		}
 	}
@@ -1232,7 +1249,9 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 		&mut self,
 		transactions: &[(H, B::Extrinsic)],
 	) -> HashMap<H, Vec<String>> {
-		let mut propagated_to = HashMap::new();
+		let mut propagated_to = HashMap::<_, Vec<_>>::new();
+		let mut propagated_transactions = 0;
+
 		for (who, peer) in self.context_data.peers.iter_mut() {
 			// never send transactions to the light node
 			if !peer.info.roles.is_full() {
@@ -1245,11 +1264,13 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 				.cloned()
 				.unzip();
 
+			propagated_transactions += hashes.len();
+
 			if !to_send.is_empty() {
 				for hash in hashes {
 					propagated_to
 						.entry(hash)
-						.or_insert_with(Vec::new)
+						.or_default()
 						.push(who.to_base58());
 				}
 				trace!(target: "sync", "Sending {} transactions to {}", to_send.len(), who);
@@ -1264,10 +1285,8 @@ impl<B: BlockT, H: ExHashT> Protocol<B, H> {
 			}
 		}
 
-		if propagated_to.len() > 0 {
-			if let Some(ref metrics) = self.metrics {
-				metrics.propagated_transactions.inc();
-			}
+		if let Some(ref metrics) = self.metrics {
+			metrics.propagated_transactions.inc_by(propagated_transactions as _)
 		}
 
 		propagated_to
@@ -2017,8 +2036,12 @@ impl<B: BlockT, H: ExHashT> NetworkBehaviour for Protocol<B, H> {
 			};
 			self.pending_messages.push_back(event);
 		}
-		if let Poll::Ready(Some((peer_id, result))) = self.pending_transactions.poll_next_unpin(cx) {
-			self.on_handle_transaction_import(peer_id, result);
+		if let Poll::Ready(Some((tx_hash, result))) = self.pending_transactions.poll_next_unpin(cx) {
+			if let Some(peers) = self.pending_transactions_peers.remove(&tx_hash) {
+				peers.into_iter().for_each(|p| self.on_handle_transaction_import(p, result));
+			} else {
+				warn!(target: "sub-libp2p", "Inconsistent state, no peers for pending transaction!");
+			}
 		}
 		if let Some(message) = self.pending_messages.pop_front() {
 			return Poll::Ready(NetworkBehaviourAction::GenerateEvent(message));

substrate/client/offchain/src/lib.rs

@@ -250,7 +250,7 @@ mod tests {
 		let client = Arc::new(substrate_test_runtime_client::new());
 		let pool = Arc::new(TestPool(BasicPool::new(
 			Default::default(),
-			Arc::new(FullChainApi::new(client.clone())),
+			Arc::new(FullChainApi::new(client.clone(), None)),
 			None,
 		).0));
 		client.execution_extensions()

substrate/client/rpc/src/author/tests.rs

@@ -63,7 +63,7 @@ impl Default for TestSetup {
 
 		let pool = Arc::new(BasicPool::new(
 			Default::default(),
-			Arc::new(FullChainApi::new(client.clone())),
+			Arc::new(FullChainApi::new(client.clone(), None)),
 			None,
 		).0);
 		TestSetup {

substrate/client/service/src/lib.rs

@@ -579,7 +579,7 @@ mod tests {
 		let client = Arc::new(client);
 		let pool = Arc::new(BasicPool::new(
 			Default::default(),
-			Arc::new(FullChainApi::new(client.clone())),
+			Arc::new(FullChainApi::new(client.clone(), None)),
 			None,
 		).0);
 		let source = sp_runtime::transaction_validity::TransactionSource::External;

substrate/client/transaction-pool/graph/src/base_pool.rs

@@ -261,6 +261,11 @@ impl<Hash: hash::Hash + Member + Serialize, Ex: std::fmt::Debug> BasePool<Hash,
 		return_value
 	}
 
+	/// Returns if the transaction for the given hash is already imported.
+	pub fn is_imported(&self, tx_hash: &Hash) -> bool {
+		self.future.contains(tx_hash) || self.ready.contains(tx_hash)
+	}
+
 	/// Imports transaction to the pool.
 	///
 	/// The pool consists of two parts: Future and Ready.
@@ -272,7 +277,7 @@ impl<Hash: hash::Hash + Member + Serialize, Ex: std::fmt::Debug> BasePool<Hash,
 		&mut self,
 		tx: Transaction<Hash, Ex>,
 	) -> error::Result<Imported<Hash, Ex>> {
-		if self.future.contains(&tx.hash) || self.ready.contains(&tx.hash) {
+		if self.is_imported(&tx.hash) {
 			return Err(error::Error::AlreadyImported(Box::new(tx.hash.clone())))
 		}
 

substrate/client/transaction-pool/graph/src/pool.rs

@@ -23,7 +23,7 @@ use std::{
 
 use crate::{base_pool as base, watcher::Watcher};
 
-use futures::{Future, FutureExt};
+use futures::Future;
 use sp_runtime::{
 	generic::BlockId,
 	traits::{self, SaturatedConversion, Block as BlockT},
@@ -125,6 +125,14 @@ impl Default for Options {
 	}
 }
 
+/// Should we check that the transaction is banned
+/// in the pool, before we verify it?
+#[derive(Copy, Clone)]
+enum CheckBannedBeforeVerify {
+	Yes,
+	No,
+}
+
 /// Extrinsics pool that performs validation.
 pub struct Pool<B: ChainApi> {
 	validated_pool: Arc<ValidatedPool<B>>,
@@ -149,23 +157,29 @@ impl<B: ChainApi> Pool<B> {
 	}
 
 	/// Imports a bunch of unverified extrinsics to the pool
-	pub async fn submit_at<T>(
+	pub async fn submit_at(
 		&self,
 		at: &BlockId<B::Block>,
 		source: TransactionSource,
-		xts: T,
-		force: bool,
-	) -> Result<Vec<Result<ExtrinsicHash<B>, B::Error>>, B::Error> where
-		T: IntoIterator<Item=ExtrinsicFor<B>>,
-	{
-		let validated_pool = self.validated_pool.clone();
+		xts: impl IntoIterator<Item=ExtrinsicFor<B>>,
+	) -> Result<Vec<Result<ExtrinsicHash<B>, B::Error>>, B::Error> {
 		let xts = xts.into_iter().map(|xt| (source, xt));
-		self.verify(at, xts, force)
-			.map(move |validated_transactions| validated_transactions
-				.map(|validated_transactions| validated_pool.submit(validated_transactions
-					.into_iter()
-					.map(|(_, tx)| tx))))
-			.await
+		let validated_transactions = self.verify(at, xts, CheckBannedBeforeVerify::Yes).await?;
+		Ok(self.validated_pool.submit(validated_transactions.into_iter().map(|(_, tx)| tx)))
+	}
+
+	/// Resubmit the given extrinsics to the pool.
+	///
+	/// This does not check if a transaction is banned, before we verify it again.
+	pub async fn resubmit_at(
+		&self,
+		at: &BlockId<B::Block>,
+		source: TransactionSource,
+		xts: impl IntoIterator<Item=ExtrinsicFor<B>>,
+	) -> Result<Vec<Result<ExtrinsicHash<B>, B::Error>>, B::Error> {
+		let xts = xts.into_iter().map(|xt| (source, xt));
+		let validated_transactions = self.verify(at, xts, CheckBannedBeforeVerify::No).await?;
+		Ok(self.validated_pool.submit(validated_transactions.into_iter().map(|(_, tx)| tx)))
 	}
 
 	/// Imports one unverified extrinsic to the pool
@@ -175,12 +189,8 @@ impl<B: ChainApi> Pool<B> {
 		source: TransactionSource,
 		xt: ExtrinsicFor<B>,
 	) -> Result<ExtrinsicHash<B>, B::Error> {
-		self.submit_at(at, source, std::iter::once(xt), false)
-			.map(|import_result| import_result.and_then(|mut import_result| import_result
-				.pop()
-				.expect("One extrinsic passed; one result returned; qed")
-			))
-			.await
+		let res = self.submit_at(at, source, std::iter::once(xt)).await?.pop();
+		res.expect("One extrinsic passed; one result returned; qed")
 	}
 
 	/// Import a single extrinsic and starts to watch their progress in the pool.
@@ -192,7 +202,11 @@ impl<B: ChainApi> Pool<B> {
 	) -> Result<Watcher<ExtrinsicHash<B>, ExtrinsicHash<B>>, B::Error> {
 		let block_number = self.resolve_block_number(at)?;
 		let (_, tx) = self.verify_one(
-			at, block_number, source, xt, false
+			at,
+			block_number,
+			source,
+			xt,
+			CheckBannedBeforeVerify::Yes,
 		).await;
 		self.validated_pool.submit_and_watch(tx)
 	}
@@ -328,7 +342,11 @@ impl<B: ChainApi> Pool<B> {
 			.into_iter()
 			.map(|tx| (tx.source, tx.data.clone()));
 
-		let reverified_transactions = self.verify(at, pruned_transactions, false).await?;
+		let reverified_transactions = self.verify(
+			at,
+			pruned_transactions,
+			CheckBannedBeforeVerify::Yes,
+		).await?;
 
 		log::trace!(target: "txpool", "Pruning at {:?}. Resubmitting transactions.", at);
 		// And finally - submit reverified transactions back to the pool
@@ -358,23 +376,17 @@ impl<B: ChainApi> Pool<B> {
 		&self,
 		at: &BlockId<B::Block>,
 		xts: impl IntoIterator<Item=(TransactionSource, ExtrinsicFor<B>)>,
-		force: bool,
+		check: CheckBannedBeforeVerify,
 	) -> Result<HashMap<ExtrinsicHash<B>, ValidatedTransactionFor<B>>, B::Error> {
 		// we need a block number to compute tx validity
 		let block_number = self.resolve_block_number(at)?;
-		let mut result = HashMap::new();
 
-		for (hash, validated_tx) in
-			futures::future::join_all(
-				xts.into_iter()
-					.map(|(source, xt)| self.verify_one(at, block_number, source, xt, force))
-			)
-			.await
-		{
-			result.insert(hash, validated_tx);
-		}
+		let res = futures::future::join_all(
+			xts.into_iter()
+				.map(|(source, xt)| self.verify_one(at, block_number, source, xt, check))
+		).await.into_iter().collect::<HashMap<_, _>>();
 
-		Ok(result)
+		Ok(res)
 	}
 
 	/// Returns future that validates single transaction at given block.
@@ -384,14 +396,13 @@ impl<B: ChainApi> Pool<B> {
 		block_number: NumberFor<B>,
 		source: TransactionSource,
 		xt: ExtrinsicFor<B>,
-		force: bool,
+		check: CheckBannedBeforeVerify,
 	) -> (ExtrinsicHash<B>, ValidatedTransactionFor<B>) {
 		let (hash, bytes) = self.validated_pool.api().hash_and_length(&xt);
-		if !force && self.validated_pool.is_banned(&hash) {
-			return (
-				hash.clone(),
-				ValidatedTransaction::Invalid(hash, error::Error::TemporarilyBanned.into()),
-			)
+
+		let ignore_banned = matches!(check, CheckBannedBeforeVerify::No);
+		if let Err(err) = self.validated_pool.check_is_known(&hash, ignore_banned) {
+			return (hash.clone(), ValidatedTransaction::Invalid(hash, err.into()))
 		}
 
 		let validation_result = self.validated_pool.api().validate_transaction(

substrate/client/transaction-pool/graph/src/validated_pool.rs

@@ -137,10 +137,30 @@ impl<B: ChainApi> ValidatedPool<B> {
 		self.rotator.is_banned(hash)
 	}
 
+	/// A fast check before doing any further processing of a transaction, like validation.
+	///
+	/// If `ingore_banned` is `true`, it will not check if the transaction is banned.
+	///
+	/// It checks if the transaction is already imported or banned. If so, it returns an error.
+	pub fn check_is_known(
+		&self,
+		tx_hash: &ExtrinsicHash<B>,
+		ignore_banned: bool,
+	) -> Result<(), B::Error> {
+		if !ignore_banned && self.is_banned(tx_hash) {
+			Err(error::Error::TemporarilyBanned.into())
+		} else if self.pool.read().is_imported(tx_hash) {
+			Err(error::Error::AlreadyImported(Box::new(tx_hash.clone())).into())
+		} else {
+			Ok(())
+		}
+	}
+
 	/// Imports a bunch of pre-validated transactions to the pool.
-	pub fn submit<T>(&self, txs: T) -> Vec<Result<ExtrinsicHash<B>, B::Error>> where
-		T: IntoIterator<Item=ValidatedTransactionFor<B>>
-	{
+	pub fn submit(
+		&self,
+		txs: impl IntoIterator<Item=ValidatedTransactionFor<B>>,
+	) -> Vec<Result<ExtrinsicHash<B>, B::Error>> {
 		let results = txs.into_iter()
 			.map(|validated_tx| self.submit_one(validated_tx))
 			.collect::<Vec<_>>();

substrate/client/transaction-pool/src/api.rs

@@ -33,19 +33,38 @@ use sp_runtime::{
 };
 use sp_transaction_pool::runtime_api::TaggedTransactionQueue;
 use sp_api::{ProvideRuntimeApi, ApiExt};
+use prometheus_endpoint::Registry as PrometheusRegistry;
 
-use crate::error::{self, Error};
+use crate::{metrics::{ApiMetrics, ApiMetricsExt}, error::{self, Error}};
 
 /// The transaction pool logic for full client.
 pub struct FullChainApi<Client, Block> {
 	client: Arc<Client>,
 	pool: ThreadPool,
 	_marker: PhantomData<Block>,
+	metrics: Option<Arc<ApiMetrics>>,
 }
 
 impl<Client, Block> FullChainApi<Client, Block> {
 	/// Create new transaction pool logic.
-	pub fn new(client: Arc<Client>) -> Self {
+	pub fn new(
+		client: Arc<Client>,
+		prometheus: Option<&PrometheusRegistry>,
+	) -> Self {
+		let metrics = prometheus.map(ApiMetrics::register).and_then(|r| {
+			match r {
+				Err(err) => {
+					log::warn!(
+						target: "txpool",
+						"Failed to register transaction pool api prometheus metrics: {:?}",
+						err,
+					);
+					None
+				},
+				Ok(api) => Some(Arc::new(api))
+			}
+		});
+
 		FullChainApi {
 			client,
 			pool: ThreadPoolBuilder::new()
@@ -54,6 +73,7 @@ impl<Client, Block> FullChainApi<Client, Block> {
 				.create()
 				.expect("Failed to spawn verifier threads, that are critical for node operation."),
 			_marker: Default::default(),
+			metrics,
 		}
 	}
 }
@@ -87,6 +107,9 @@ where
 		let client = self.client.clone();
 		let at = at.clone();
 
+		let metrics = self.metrics.clone();
+		metrics.report(|m| m.validations_scheduled.inc());
+
 		self.pool.spawn_ok(futures_diagnose::diagnose(
 			"validate-transaction",
 			async move {
@@ -94,6 +117,7 @@ where
 				if let Err(e) = tx.send(res) {
 					log::warn!("Unable to send a validate transaction result: {:?}", e);
 				}
+				metrics.report(|m| m.validations_finished.inc());
 			},
 		));
 

substrate/client/transaction-pool/src/lib.rs

@@ -248,15 +248,9 @@ impl<PoolApi, Block> TransactionPool for BasicPool<PoolApi, Block>
 		let pool = self.pool.clone();
 		let at = *at;
 
-		self.metrics.report(|metrics| metrics.validations_scheduled.inc_by(xts.len() as u64));
+		self.metrics.report(|metrics| metrics.submitted_transactions.inc_by(xts.len() as u64));
 
-		let metrics = self.metrics.clone();
-		async move {
-			let tx_count = xts.len();
-			let res = pool.submit_at(&at, source, xts, false).await;
-			metrics.report(|metrics| metrics.validations_finished.inc_by(tx_count as u64));
-			res
-		}.boxed()
+		async move { pool.submit_at(&at, source, xts).await }.boxed()
 	}
 
 	fn submit_one(
@@ -268,16 +262,9 @@ impl<PoolApi, Block> TransactionPool for BasicPool<PoolApi, Block>
 		let pool = self.pool.clone();
 		let at = *at;
 
-		self.metrics.report(|metrics| metrics.validations_scheduled.inc());
+		self.metrics.report(|metrics| metrics.submitted_transactions.inc());
 
-		let metrics = self.metrics.clone();
-		async move {
-			let res = pool.submit_one(&at, source, xt).await;
-
-			metrics.report(|metrics| metrics.validations_finished.inc());
-			res
-
-		}.boxed()
+		async move { pool.submit_one(&at, source, xt).await }.boxed()
 	}
 
 	fn submit_and_watch(
@@ -289,17 +276,12 @@ impl<PoolApi, Block> TransactionPool for BasicPool<PoolApi, Block>
 		let at = *at;
 		let pool = self.pool.clone();
 
-		self.metrics.report(|metrics| metrics.validations_scheduled.inc());
+		self.metrics.report(|metrics| metrics.submitted_transactions.inc());
 
-		let metrics = self.metrics.clone();
 		async move {
-			let result = pool.submit_and_watch(&at, source, xt)
+			pool.submit_and_watch(&at, source, xt)
 				.map(|result| result.map(|watcher| Box::new(watcher.into_stream()) as _))
-				.await;
-
-			metrics.report(|metrics| metrics.validations_finished.inc());
-
-			result
+				.await
 		}.boxed()
 	}
 
@@ -632,13 +614,12 @@ impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
 							);
 						}
 
-						if let Err(e) = pool.submit_at(
+						if let Err(e) = pool.resubmit_at(
 							&id,
 							// These transactions are coming from retracted blocks, we should
 							// simply consider them external.
 							TransactionSource::External,
 							resubmit_transactions,
-							true,
 						).await {
 							log::debug!(
 								target: "txpool",

substrate/client/transaction-pool/src/metrics.rs

@@ -45,8 +45,7 @@ impl MetricsLink {
 
 /// Transaction pool Prometheus metrics.
 pub struct Metrics {
-	pub validations_scheduled: Counter<U64>,
-	pub validations_finished: Counter<U64>,
+	pub submitted_transactions: Counter<U64>,
 	pub validations_invalid: Counter<U64>,
 	pub block_transactions_pruned: Counter<U64>,
 	pub block_transactions_resubmitted: Counter<U64>,
@@ -55,17 +54,10 @@ pub struct Metrics {
 impl Metrics {
 	pub fn register(registry: &Registry) -> Result<Self, PrometheusError> {
 		Ok(Self {
-			validations_scheduled: register(
+			submitted_transactions: register(
 				Counter::new(
-					"sub_txpool_validations_scheduled",
-					"Total number of transactions scheduled for validation",
-				)?,
-				registry,
-			)?,
-			validations_finished: register(
-				Counter::new(
-					"sub_txpool_validations_finished",
-					"Total number of transactions that finished validation",
+					"sub_txpool_submitted_transactions",
+					"Total number of transactions submitted",
 				)?,
 				registry,
 			)?,
@@ -93,3 +85,45 @@ impl Metrics {
 		})
 	}
 }
+
+/// Transaction pool api Prometheus metrics.
+pub struct ApiMetrics {
+	pub validations_scheduled: Counter<U64>,
+	pub validations_finished: Counter<U64>,
+}
+
+impl ApiMetrics {
+	/// Register the metrics at the given Prometheus registry.
+	pub fn register(registry: &Registry) -> Result<Self, PrometheusError> {
+		Ok(Self {
+			validations_scheduled: register(
+				Counter::new(
+					"sub_txpool_validations_scheduled",
+					"Total number of transactions scheduled for validation",
+				)?,
+				registry,
+			)?,
+			validations_finished: register(
+				Counter::new(
+					"sub_txpool_validations_finished",
+					"Total number of transactions that finished validation",
+				)?,
+				registry,
+			)?,
+		})
+	}
+}
+
+/// An extension trait for [`ApiMetrics`].
+pub trait ApiMetricsExt {
+	/// Report an event to the metrics.
+	fn report(&self, report: impl FnOnce(&ApiMetrics));
+}
+
+impl ApiMetricsExt for Option<Arc<ApiMetrics>> {
+	fn report(&self, report: impl FnOnce(&ApiMetrics)) {
+		if let Some(metrics) = self.as_ref() {
+			report(metrics)
+		}
+	}
+}

substrate/client/transaction-pool/src/testing/pool.rs

@@ -1008,7 +1008,7 @@ fn should_not_accept_old_signatures() {
 	let client = Arc::new(substrate_test_runtime_client::new());
 
 	let pool = Arc::new(
-		BasicPool::new_test(Arc::new(FullChainApi::new(client))).0
+		BasicPool::new_test(Arc::new(FullChainApi::new(client, None))).0
 	);
 
 	let transfer = Transfer {
@@ -1044,7 +1044,7 @@ fn import_notification_to_pool_maintain_works() {
 	let mut client = Arc::new(substrate_test_runtime_client::new());
 
 	let pool = Arc::new(
-		BasicPool::new_test(Arc::new(FullChainApi::new(client.clone()))).0
+		BasicPool::new_test(Arc::new(FullChainApi::new(client.clone(), None))).0
 	);
 
 	// Prepare the extrisic, push it to the pool and check that it was added.

substrate/utils/frame/rpc/system/src/lib.rs

@@ -301,7 +301,7 @@ mod tests {
 		let pool = Arc::new(
 			BasicPool::new(
 				Default::default(),
-				Arc::new(FullChainApi::new(client.clone())),
+				Arc::new(FullChainApi::new(client.clone(), None)),
 				None,
 			).0
 		);
@@ -340,7 +340,7 @@ mod tests {
 		let pool = Arc::new(
 			BasicPool::new(
 				Default::default(),
-				Arc::new(FullChainApi::new(client.clone())),
+				Arc::new(FullChainApi::new(client.clone(), None)),
 				None,
 			).0
 		);
@@ -363,7 +363,7 @@ mod tests {
 		let pool = Arc::new(
 			BasicPool::new(
 				Default::default(),
-				Arc::new(FullChainApi::new(client.clone())),
+				Arc::new(FullChainApi::new(client.clone(), None)),
 				None,
 			).0
 		);
@@ -395,7 +395,7 @@ mod tests {
 		let pool = Arc::new(
 			BasicPool::new(
 				Default::default(),
-				Arc::new(FullChainApi::new(client.clone())),
+				Arc::new(FullChainApi::new(client.clone(), None)),
 				None,
 			).0
 		);