fatxpool: report_invalid is now aync (#8836)

[`TransactionPool::report_invalid`](https://github.com/paritytech/polkadot-sdk/blob/74dafaee5c600fd2c8a59a280f647f94ccf0a755/substrate/client/transaction-pool/api/src/lib.rs#L314)
is now `async`, function was typically used in async context, seems
right to be fully async.

_Note_: expected to be merged after: #8596

---------

Co-authored-by: cmd[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: michalkucharczyk

prdoc/pr_8836.prdoc

@@ -0,0 +1,19 @@
+title: '`fatxpool`: `report_invalid` is now aync'
+doc:
+- audience: Node Dev
+  description: '[`TransactionPool::report_invalid`](https://github.com/paritytech/polkadot-sdk/blob/74dafaee5c600fd2c8a59a280f647f94ccf0a755/substrate/client/transaction-pool/api/src/lib.rs#L314)
+    is now `async`, function was typically used in async context, seems right to be
+    fully async.'
+crates:
+- name: sc-basic-authorship
+  bump: major
+- name: sc-rpc-api
+  bump: major
+- name: sc-rpc-spec-v2
+  bump: major
+- name: sc-rpc
+  bump: major
+- name: sc-transaction-pool-api
+  bump: major
+- name: sc-transaction-pool
+  bump: major

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

@@ -271,7 +271,7 @@ impl sc_transaction_pool_api::TransactionPool for Transactions {
 		unimplemented!()
 	}
 
-	fn report_invalid(
+	async fn report_invalid(
 		&self,
 		_at: Option<Self::Hash>,
 		_invalid_tx_errors: TxInvalidityReportMap<TxHash<Self>>,

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

@@ -530,7 +530,9 @@ where
 			);
 		}
 
-		self.transaction_pool.report_invalid(Some(self.parent_hash), unqueue_invalid);
+		self.transaction_pool
+			.report_invalid(Some(self.parent_hash), unqueue_invalid)
+			.await;
 		Ok(end_reason)
 	}
 

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

@@ -61,7 +61,7 @@ pub trait AuthorApi<Hash, BlockHash> {
 
 	/// Remove given extrinsic from the pool and temporarily ban it to prevent reimporting.
 	#[method(name = "author_removeExtrinsic", with_extensions)]
-	fn remove_extrinsic(
+	async fn remove_extrinsic(
 		&self,
 		bytes_or_hash: Vec<hash::ExtrinsicOrHash<Hash>>,
 	) -> Result<Vec<Hash>, Error>;

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

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

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

@@ -226,20 +226,22 @@ where
 		let pool = self.pool.clone();
 		// The future expected by the executor must be `Future<Output = ()>` instead of
 		// `Future<Output = Result<(), Aborted>>`.
-		let fut = fut.map(move |result| {
-			// Connection space is cleaned when this object is dropped.
-			drop(reserved_identifier);
+		let fut = fut.then(move |result| {
+			async move {
+				// Connection space is cleaned when this object is dropped.
+				drop(reserved_identifier);
 
-			// Remove the entry from the broadcast IDs map.
-			let Some(broadcast_state) = broadcast_ids.write().remove(&drop_id) else { return };
+				// Remove the entry from the broadcast IDs map.
+				let Some(broadcast_state) = broadcast_ids.write().remove(&drop_id) else { return };
 
-			// The broadcast was not stopped.
-			if result.is_ok() {
-				return
-			}
+				// The broadcast was not stopped.
+				if result.is_ok() {
+					return
+				}
 
-			// Best effort pool removal (tx can already be finalized).
-			pool.report_invalid(None, [(broadcast_state.tx_hash, None)].into());
+				// Best effort pool removal (tx can already be finalized).
+				pool.report_invalid(None, [(broadcast_state.tx_hash, None)].into()).await;
+			}
 		});
 
 		// Keep track of this entry and the abortable handle.

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

@@ -153,7 +153,7 @@ where
 		Ok(self.pool.ready().map(|tx| tx.data().encode().into()).collect())
 	}
 
-	fn remove_extrinsic(
+	async fn remove_extrinsic(
 		&self,
 		ext: &Extensions,
 		bytes_or_hash: Vec<hash::ExtrinsicOrHash<TxHash<P>>>,
@@ -173,6 +173,7 @@ where
 		Ok(self
 			.pool
 			.report_invalid(None, hashes)
+			.await
 			.into_iter()
 			.map(|tx| tx.hash().clone())
 			.collect())

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

@@ -311,7 +311,7 @@ pub trait TransactionPool: Send + Sync {
 	/// occurred.
 	///
 	/// Function returns the transactions actually removed from the pool.
-	fn report_invalid(
+	async fn report_invalid(
 		&self,
 		at: Option<<Self::Block as BlockT>::Hash>,
 		invalid_tx_errors: TxInvalidityReportMap<TxHash<Self>>,

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

@@ -1005,7 +1005,7 @@ where
 	/// 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(
+	async fn report_invalid(
 		&self,
 		at: Option<<Self::Block as BlockT>::Hash>,
 		invalid_tx_errors: TxInvalidityReportMap<TxHash<Self>>,
@@ -1018,7 +1018,7 @@ where
 		let removed = self.view_store.report_invalid(at, invalid_tx_errors);
 
 		let removed_hashes = removed.iter().map(|tx| tx.hash).collect::<Vec<_>>();
-		self.mempool.clone().remove_transactions_sync(removed_hashes.clone());
+		self.mempool.remove_transactions(&removed_hashes).await;
 		self.import_notification_sink.clean_notified_items(&removed_hashes);
 
 		self.metrics

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

@@ -334,7 +334,7 @@ where
 		.map(|mut outcome| outcome.expect_watcher().into_stream().boxed())
 	}
 
-	fn report_invalid(
+	async fn report_invalid(
 		&self,
 		_at: Option<<Self::Block as BlockT>::Hash>,
 		invalid_tx_errors: TxInvalidityReportMap<TxHash<Self>>,