txpool api: remove_invalid call improved

prdoc/pr_6661.prdoc

@@ -0,0 +1,17 @@
+title: '`txpool api`: `remove_invalid` call improved'
+doc:
+- audience: Node Dev
+  description: |-
+    Currently the transaction which is reported as invalid by a block builder (or `removed_invalid` by other components) is silently skipped. This PR improves this behavior. The transaction pool `report_invalid` function now accepts optional error associated with every reported transaction, and also the optional block hash which both provide hints how reported invalid transaction shall be handled.  Depending on error, the transaction pool can decide if transaction shall be removed from the view only or entirely from the pool. Invalid event will be dispatched if required.
+
+crates:
+- name: sc-transaction-pool-api
+  bump: minor
+- name: sc-transaction-pool
+  bump: minor
+- name: sc-rpc-spec-v2
+  bump: minor
+- name: sc-rpc
+  bump: minor
+- name: sc-basic-authorship
+  bump: minor

substrate/bin/node/bench/Cargo.toml

@@ -36,6 +36,7 @@ sc-transaction-pool = { workspace = true, default-features = true }
 sc-transaction-pool-api = { workspace = true, default-features = true }
 serde = { workspace = true, default-features = true }
 serde_json = { workspace = true, default-features = true }
+sp-blockchain = { workspace = true, default-features = true }
 sp-consensus = { workspace = true, default-features = true }
 sp-core = { workspace = true, default-features = true }
 sp-inherents = { workspace = true, default-features = true }

substrate/bin/node/bench/src/construct.rs

@@ -271,7 +271,11 @@ impl sc_transaction_pool_api::TransactionPool for Transactions {
 		unimplemented!()
 	}
 
-	fn remove_invalid(&self, _hashes: &[TxHash<Self>]) -> Vec<Arc<Self::InPoolTransaction>> {
+	fn report_invalid(
+		&self,
+		_at: Option<Self::Hash>,
+		_invalid_tx_errors: &[(TxHash<Self>, Option<sp_blockchain::Error>)],
+	) -> Vec<Arc<Self::InPoolTransaction>> {
 		Default::default()
 	}
 

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

@@ -512,7 +512,7 @@ where
 						target: LOG_TARGET,
 						"[{:?}] Invalid transaction: {} at: {}", pending_tx_hash, e, self.parent_hash
 					);
-					unqueue_invalid.push(pending_tx_hash);
+					unqueue_invalid.push((pending_tx_hash, Some(e)));
 				},
 			}
 		};
@@ -524,7 +524,7 @@ where
 			);
 		}
 
-		self.transaction_pool.remove_invalid(&unqueue_invalid);
+		self.transaction_pool.report_invalid(Some(self.parent_hash), &unqueue_invalid);
 		Ok(end_reason)
 	}
 

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

@@ -137,8 +137,12 @@ impl TransactionPool for MiddlewarePool {
 		Ok(watcher.boxed())
 	}
 
-	fn remove_invalid(&self, hashes: &[TxHash<Self>]) -> Vec<Arc<Self::InPoolTransaction>> {
-		self.inner_pool.remove_invalid(hashes)
+	fn report_invalid(
+		&self,
+		at: Option<<Self::Block as BlockT>::Hash>,
+		invalid_tx_errors: &[(TxHash<Self>, Option<sp_blockchain::Error>)],
+	) -> Vec<Arc<Self::InPoolTransaction>> {
+		self.inner_pool.report_invalid(at, invalid_tx_errors)
 	}
 
 	fn status(&self) -> PoolStatus {

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

@@ -228,7 +228,7 @@ where
 			}
 
 			// Best effort pool removal (tx can already be finalized).
-			pool.remove_invalid(&[broadcast_state.tx_hash]);
+			pool.report_invalid(None, &[(broadcast_state.tx_hash, None)]);
 		});
 
 		// Keep track of this entry and the abortable handle.

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

@@ -164,17 +164,17 @@ where
 		let hashes = bytes_or_hash
 			.into_iter()
 			.map(|x| match x {
-				hash::ExtrinsicOrHash::Hash(h) => Ok(h),
+				hash::ExtrinsicOrHash::Hash(h) => Ok((h, None)),
 				hash::ExtrinsicOrHash::Extrinsic(bytes) => {
 					let xt = Decode::decode(&mut &bytes[..])?;
-					Ok(self.pool.hash_of(&xt))
+					Ok((self.pool.hash_of(&xt), None))
 				},
 			})
 			.collect::<Result<Vec<_>>>()?;
 
 		Ok(self
 			.pool
-			.remove_invalid(&hashes)
+			.report_invalid(None, &hashes)
 			.into_iter()
 			.map(|tx| tx.hash().clone())
 			.collect())

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

@@ -290,8 +290,27 @@ pub trait TransactionPool: Send + Sync {
 	fn ready(&self) -> Box<dyn ReadyTransactions<Item = Arc<Self::InPoolTransaction>> + Send>;
 
 	// *** Block production
-	/// Remove transactions identified by given hashes (and dependent transactions) from the pool.
-	fn remove_invalid(&self, hashes: &[TxHash<Self>]) -> Vec<Arc<Self::InPoolTransaction>>;
+	/// Reports invalid transactions to the transaction pool.
+	///
+	/// This function accepts an array of tuples, each containing a transaction hash and an
+	/// optional error encountered during the transaction execution at a specific (also optional)
+	/// block.
+	///
+	/// The transaction pool implementation decides which transactions to remove. Transactions
+	/// removed from the pool will be notified with `TransactionStatus::Invalid` event (if
+	/// `submit_and_watch` was used for submission).
+	///
+	/// If the tuple's error is None, the transaction will be forcibly removed from the pool.
+	///
+	/// The optional `at` parameter provides additional context regarding the block where the error
+	/// occurred.
+	///
+	/// Function returns the transactions actually removed from the pool.
+	fn report_invalid(
+		&self,
+		at: Option<<Self::Block as BlockT>::Hash>,
+		invalid_tx_errors: &[(TxHash<Self>, Option<sp_blockchain::Error>)],
+	) -> Vec<Arc<Self::InPoolTransaction>>;
 
 	// *** logging
 	/// Get futures transaction list.

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

@@ -784,19 +784,34 @@ where
 			})
 	}
 
-	/// Intended to remove transactions identified by the given hashes, and any dependent
-	/// transactions, from the pool. In current implementation this function only outputs the error.
-	/// Seems that API change is needed here to make this call reasonable.
-	// todo [#5491]: api change? we need block hash here (assuming we need it at all - could be
-	// useful for verification for debugging purposes).
-	fn remove_invalid(&self, hashes: &[TxHash<Self>]) -> Vec<Arc<Self::InPoolTransaction>> {
-		if !hashes.is_empty() {
-			log::debug!(target: LOG_TARGET, "fatp::remove_invalid {}", hashes.len());
-			log_xt_trace!(target:LOG_TARGET, hashes, "[{:?}] fatp::remove_invalid");
-			self.metrics
-				.report(|metrics| metrics.removed_invalid_txs.inc_by(hashes.len() as _));
-		}
-		Default::default()
+	/// Reports invalid transactions to the transaction pool.
+	///
+	/// This function takes an array of tuples, each consisting of a transaction hash and the
+	/// corresponding error that occurred during transaction execution at given block.
+	///
+	/// The transaction pool implementation will determine which transactions should be
+	/// removed from the pool. Transactions that depend on invalid transactions will also
+	/// be removed.
+	fn report_invalid(
+		&self,
+		at: Option<<Self::Block as BlockT>::Hash>,
+		invalid_tx_errors: &[(TxHash<Self>, Option<sp_blockchain::Error>)],
+	) -> Vec<Arc<Self::InPoolTransaction>> {
+		log::debug!(target: LOG_TARGET, "fatp::report_invalid {}", invalid_tx_errors.len());
+		log_xt_trace!(data: tuple, target:LOG_TARGET, invalid_tx_errors, "[{:?}] fatp::report_invalid {:?}");
+		self.metrics
+			.report(|metrics| metrics.reported_invalid_txs.inc_by(invalid_tx_errors.len() as _));
+
+		let removed = self.view_store.report_invalid(at, invalid_tx_errors);
+
+		removed.iter().for_each(|tx| {
+			self.mempool.remove_transaction(&tx.hash);
+		});
+
+		self.metrics
+			.report(|metrics| metrics.removed_invalid_txs.inc_by(removed.len() as _));
+
+		removed
 	}
 
 	// todo [#5491]: api change?

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

@@ -40,6 +40,8 @@ pub struct Metrics {
 	/// Total number of unwatched transactions in txpool.
 	pub unwatched_txs: Gauge<U64>,
 	/// Total number of transactions reported as invalid.
+	pub reported_invalid_txs: Counter<U64>,
+	/// Total number of transactions removed as invalid.
 	pub removed_invalid_txs: Counter<U64>,
 	/// Total number of finalized transactions.
 	pub finalized_txs: Counter<U64>,
@@ -99,10 +101,17 @@ impl MetricsRegistrant for Metrics {
 				)?,
 				registry,
 			)?,
+			reported_invalid_txs: register(
+				Counter::new(
+					"substrate_sub_txpool_reported_invalid_txs_total",
+					"Total number of transactions reported as invalid.",
+				)?,
+				registry,
+			)?,
 			removed_invalid_txs: register(
 				Counter::new(
 					"substrate_sub_txpool_removed_invalid_txs_total",
-					"Total number of transactions reported as invalid.",
+					"Total number of transactions removed as invalid.",
 				)?,
 				registry,
 			)?,

substrate/client/transaction-pool/src/fork_aware_txpool/view.rs

@@ -28,21 +28,23 @@ use crate::{
 	common::log_xt::log_xt_trace,
 	graph::{
 		self, base_pool::TimedTransactionSource, watcher::Watcher, ExtrinsicFor, ExtrinsicHash,
-		IsValidator, ValidatedPoolSubmitOutcome, ValidatedTransaction, ValidatedTransactionFor,
+		IsValidator, TransactionFor, ValidatedPoolSubmitOutcome, ValidatedTransaction,
+		ValidatedTransactionFor,
 	},
 	LOG_TARGET,
 };
+use indexmap::IndexMap;
 use parking_lot::Mutex;
 use sc_transaction_pool_api::{error::Error as TxPoolError, PoolStatus};
 use sp_blockchain::HashAndNumber;
 use sp_runtime::{
 	generic::BlockId, traits::Block as BlockT, transaction_validity::TransactionValidityError,
 	SaturatedConversion,
 };
-use std::{collections::HashMap, sync::Arc, time::Instant};
+use std::{sync::Arc, time::Instant};
 
 pub(super) struct RevalidationResult<ChainApi: graph::ChainApi> {
-	revalidated: HashMap<ExtrinsicHash<ChainApi>, ValidatedTransactionFor<ChainApi>>,
+	revalidated: IndexMap<ExtrinsicHash<ChainApi>, ValidatedTransactionFor<ChainApi>>,
 	invalid_hashes: Vec<ExtrinsicHash<ChainApi>>,
 }
 
@@ -271,7 +273,7 @@ where
 		//todo: revalidate future, remove if invalid [#5496]
 
 		let mut invalid_hashes = Vec::new();
-		let mut revalidated = HashMap::new();
+		let mut revalidated = IndexMap::new();
 
 		let mut validation_results = vec![];
 		let mut batch_iter = batch.into_iter();
@@ -310,7 +312,7 @@ where
 			batch_len,
 			revalidation_duration
 		);
-		log_xt_trace!(data:tuple, target:LOG_TARGET, validation_results.iter().map(|x| (x.1, &x.0)), "[{:?}] view::revalidateresult: {:?}");
+		log_xt_trace!(data:tuple, target:LOG_TARGET, validation_results.iter().map(|x| (x.1, &x.0)), "[{:?}] view::revalidate result: {:?}");
 
 		for (validation_result, tx_hash, tx) in validation_results {
 			match validation_result {
@@ -455,6 +457,16 @@ where
 		}
 	}
 
+	/// Remove invalid transactions from the view.
+	///
+	/// Refer to [`crate::graph::ValidatedPool::remove_invalid`] for more details.
+	pub(crate) fn remove_invalid(
+		&self,
+		hashes: &[ExtrinsicHash<ChainApi>],
+	) -> Vec<TransactionFor<ChainApi>> {
+		self.pool.validated_pool().remove_invalid(hashes)
+	}
+
 	/// Returns true if the transaction with given hash is already imported into the view.
 	pub(super) fn is_imported(&self, tx_hash: &ExtrinsicHash<ChainApi>) -> bool {
 		const IGNORE_BANNED: bool = false;
@@ -466,12 +478,12 @@ where
 	/// Refer to [`crate::graph::ValidatedPool::remove_subtree`] for more details.
 	pub fn remove_subtree<F>(
 		&self,
-		tx_hash: ExtrinsicHash<ChainApi>,
+		hashes: &[ExtrinsicHash<ChainApi>],
 		listener_action: F,
-	) -> Vec<ExtrinsicHash<ChainApi>>
+	) -> Vec<TransactionFor<ChainApi>>
 	where
 		F: Fn(&mut crate::graph::Listener<ChainApi>, ExtrinsicHash<ChainApi>),
 	{
-		self.pool.validated_pool().remove_subtree(tx_hash, listener_action)
+		self.pool.validated_pool().remove_subtree(hashes, listener_action)
 	}
 }