txpool api: remove_invalid call improved

prdoc/pr_6661.prdoc

@@ -0,0 +1,18 @@
+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.
+
+    fixes: #6008
+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/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,26 @@ 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
+	/// dependent on invalid ones will also be removed.
+	///
+	/// 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/dropped_watcher.rs

@@ -53,11 +53,13 @@ pub struct DroppedTransaction<Hash> {
 }
 
 impl<Hash> DroppedTransaction<Hash> {
-	fn new_usurped(tx_hash: Hash, by: Hash) -> Self {
+	/// Creates an new instnance with reason set to `DroppedReason::Usurped(by)`.
+	pub fn new_usurped(tx_hash: Hash, by: Hash) -> Self {
 		Self { reason: DroppedReason::Usurped(by), tx_hash }
 	}
 
-	fn new_enforced_by_limts(tx_hash: Hash) -> Self {
+	/// Creates an new instnance with reason set to `DroppedReason::LimitsEnforced`.
+	pub fn new_enforced_by_limts(tx_hash: Hash) -> Self {
 		Self { reason: DroppedReason::LimitsEnforced, tx_hash }
 	}
 }
@@ -257,10 +259,12 @@ where
 			},
 			TransactionStatus::Ready | TransactionStatus::InBlock(..) => {
 				// note: if future transaction was once seens as the ready we may want to treat it
-				// as ready transactions. Unreferenced future transactions are more likely to be
-				// removed when the last referencing view is removed then ready transactions.
-				// Transcaction seen as ready is likely quite close to be included in some
-				// future fork.
+				// as ready transaction. The rationale behind this is as follows: we want to remove
+				// unreferenced future transactions when the last referencing view is removed (to
+				// avoid clogging mempool). For ready transactions we prefer to keep them in mempool
+				// even if no view is currently referencing them. Future transcaction once seen as
+				// ready is likely quite close to be included in some future fork (it is close to be
+				// ready, so we make exception and treat such transaction as ready).
 				if let Some(mut views) = self.future_transaction_views.remove(&tx_hash) {
 					views.insert(block_hash);
 					self.ready_transaction_views.insert(tx_hash, views);
@@ -329,14 +333,14 @@ where
 		let stream_map = futures::stream::unfold(ctx, |mut ctx| async move {
 			loop {
 				if let Some(dropped) = ctx.get_pending_dropped_transaction() {
-					debug!("dropped_watcher: sending out (pending): {dropped:?}");
+					trace!("dropped_watcher: sending out (pending): {dropped:?}");
 					return Some((dropped, ctx));
 				}
 				tokio::select! {
 					biased;
 					Some(event) = next_event(&mut ctx.stream_map) => {
 						if let Some(dropped) = ctx.handle_event(event.0, event.1) {
-							debug!("dropped_watcher: sending out: {dropped:?}");
+							trace!("dropped_watcher: sending out: {dropped:?}");
 							return Some((dropped, ctx));
 						}
 					},