Transaction pool: Ensure that we prune transactions properly (#8963)

* Transaction pool: Ensure that we prune transactions properly

There was a bug in the transaction pool that we didn't pruned
transactions properly because we called `prune_known`, instead of `prune`.

This bug was introduced by:
https://github.com/paritytech/substrate/pull/4629

This is required to have stale extrinsics being removed properly, so
that they don't fill up the tx pool.

* Fix compilation

* Fix benches

* ...

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

@@ -23,7 +23,7 @@ use std::{
 
 use linked_hash_map::LinkedHashMap;
 use serde::Serialize;
-use log::{debug, trace, warn};
+use log::{debug, trace};
 use sp_runtime::traits;
 
 use crate::{watcher, ChainApi, ExtrinsicHash, BlockHash};
@@ -99,12 +99,8 @@ impl<H: hash::Hash + traits::Member + Serialize, C: ChainApi> Listener<H, C> {
 	}
 
 	/// Transaction was removed as invalid.
-	pub fn invalid(&mut self, tx: &H, warn: bool) {
-		if warn {
-			warn!(target: "txpool", "[{:?}] Extrinsic invalid", tx);
-		} else {
-			debug!(target: "txpool", "[{:?}] Extrinsic invalid", tx);
-		}
+	pub fn invalid(&mut self, tx: &H) {
+		debug!(target: "txpool", "[{:?}] Extrinsic invalid", tx);
 		self.fire(tx, |watcher| watcher.invalid());
 	}
 

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

@@ -95,6 +95,12 @@ pub trait ChainApi: Send + Sync {
 
 	/// Returns a block body given the block id.
 	fn block_body(&self, at: &BlockId<Self::Block>) -> Self::BodyFuture;
+
+	/// Returns a block header given the block id.
+	fn block_header(
+		&self,
+		at: &BlockId<Self::Block>,
+	) -> Result<Option<<Self::Block as BlockT>::Header>, Self::Error>;
 }
 
 /// Pool configuration options.
@@ -237,7 +243,7 @@ impl<B: ChainApi> Pool<B> {
 	) -> Result<(), B::Error> {
 		// Get details of all extrinsics that are already in the pool
 		let in_pool_tags = self.validated_pool.extrinsics_tags(hashes)
-			.into_iter().filter_map(|x| x).flat_map(|x| x);
+			.into_iter().filter_map(|x| x).flatten();
 
 		// Prune all transactions that provide given tags
 		let prune_status = self.validated_pool.prune_tags(in_pool_tags)?;
@@ -579,6 +585,13 @@ mod tests {
 		fn block_body(&self, _id: &BlockId<Self::Block>) -> Self::BodyFuture {
 			futures::future::ready(Ok(None))
 		}
+
+		fn block_header(
+			&self,
+			_: &BlockId<Self::Block>,
+		) -> Result<Option<<Self::Block as BlockT>::Header>, Self::Error> {
+			Ok(None)
+		}
 	}
 
 	fn uxt(transfer: Transfer) -> Extrinsic {

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

@@ -230,7 +230,7 @@ impl<B: ChainApi> ValidatedPool<B> {
 				Err(err)
 			},
 			ValidatedTransaction::Unknown(hash, err) => {
-				self.listener.write().invalid(&hash, false);
+				self.listener.write().invalid(&hash);
 				Err(err)
 			},
 		}
@@ -415,7 +415,7 @@ impl<B: ChainApi> ValidatedPool<B> {
 					Status::Future => listener.future(&hash),
 					Status::Ready => listener.ready(&hash, None),
 					Status::Dropped => listener.dropped(&hash, None),
-					Status::Failed => listener.invalid(&hash, initial_status.is_some()),
+					Status::Failed => listener.invalid(&hash),
 				}
 			}
 		}
@@ -423,10 +423,12 @@ impl<B: ChainApi> ValidatedPool<B> {
 
 	/// For each extrinsic, returns tags that it provides (if known), or None (if it is unknown).
 	pub fn extrinsics_tags(&self, hashes: &[ExtrinsicHash<B>]) -> Vec<Option<Vec<Tag>>> {
-		self.pool.read().by_hashes(&hashes)
+		self.pool.read()
+			.by_hashes(&hashes)
 			.into_iter()
-			.map(|existing_in_pool| existing_in_pool
-				.map(|transaction| transaction.provides.to_vec()))
+			.map(|existing_in_pool|
+				existing_in_pool.map(|transaction| transaction.provides.to_vec())
+			)
 			.collect()
 	}
 
@@ -599,7 +601,7 @@ impl<B: ChainApi> ValidatedPool<B> {
 
 		let mut listener = self.listener.write();
 		for tx in &invalid {
-			listener.invalid(&tx.hash, true);
+			listener.invalid(&tx.hash);
 		}
 
 		invalid
@@ -645,15 +647,9 @@ fn fire_events<H, B, Ex>(
 	match *imported {
 		base::Imported::Ready { ref promoted, ref failed, ref removed, ref hash } => {
 			listener.ready(hash, None);
-			for f in failed {
-				listener.invalid(f, true);
-			}
-			for r in removed {
-				listener.dropped(&r.hash, Some(hash));
-			}
-			for p in promoted {
-				listener.ready(p, None);
-			}
+			failed.into_iter().for_each(|f| listener.invalid(f));
+			removed.into_iter().for_each(|r| listener.dropped(&r.hash, Some(hash)));
+			promoted.into_iter().for_each(|p| listener.ready(p, None));
 		},
 		base::Imported::Future { ref hash } => {
 			listener.future(hash)