Fix transaction pool event sending (#6341)

This pr fixes a bug with the transaction pool not sending certain events
like finalized and also fixes the order of events. The problem with the
finalized event was that we did not extracted pruned extrinsics if there
were not ready transactions in the pool. However this is wrong, if we
have a re-org, a tx is clearly not ready anymore and we still need to
send a pruned event for it because it is in a new block included. This
also lead to sending "ready" events and tx being re-validated. The
listener also only send the "finalized" event if it has seen a block as
being included, which did not happen before with the old code.

The second fix of the pr is the order of events. If we prune and retract the
same transaction in the same block, we first need to send the "retract"
event and after that the "pruned" event, because finalization takes
longer and this would lead to the UI showing "retract" while it actually
is included.

client/transaction-pool/src/lib.rs

@@ -495,11 +495,6 @@ async fn prune_known_txs_for_block<Block: BlockT, Api: ChainApi<Block = Block>>(
 	api: &Api,
 	pool: &sc_transaction_graph::Pool<Api>,
 ) -> Vec<ExtrinsicHash<Api>> {
-	// We don't query block if we won't prune anything
-	if pool.validated_pool().status().is_empty() {
-		return Vec::new();
-	}
-
 	let hashes = api.block_body(&block_id).await
 		.unwrap_or_else(|e| {
 			log::warn!("Prune known transactions: error request {:?}!", e);
@@ -559,8 +554,16 @@ impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
 					let mut pruned_log = HashSet::<ExtrinsicHash<PoolApi>>::new();
 
 					// If there is a tree route, we use this to prune known tx based on the enacted
-					// blocks.
+					// blocks. Before pruning enacted transactions, we inform the listeners about
+					// retracted blocks and their transactions. This order is important, because
+					// if we enact and retract the same transaction at the same time, we want to
+					// send first the retract and than the prune event.
 					if let Some(ref tree_route) = tree_route {
+						for retracted in tree_route.retracted() {
+							// notify txs awaiting finality that it has been retracted
+							pool.validated_pool().on_block_retracted(retracted.hash.clone());
+						}
+
 						future::join_all(
 							tree_route
 								.enacted()
@@ -592,9 +595,6 @@ impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
 						for retracted in tree_route.retracted() {
 							let hash = retracted.hash.clone();
 
-							// notify txs awaiting finality that it has been retracted
-							pool.validated_pool().on_block_retracted(hash.clone());
-
 							let block_transactions = api.block_body(&BlockId::hash(hash))
 								.await
 								.unwrap_or_else(|e| {

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

@@ -678,6 +678,66 @@ fn fork_aware_finalization() {
 	}
 }
 
+/// Tests that when pruning and retracing a tx by the same event, we generate
+/// the correct events in the correct order.
+#[test]
+fn prune_and_retract_tx_at_same_time() {
+	let api = TestApi::empty();
+	// starting block A1 (last finalized.)
+	api.push_block(1, vec![]);
+
+	let (pool, _background, _) = BasicPool::new_test(api.into());
+
+	let from_alice = uxt(Alice, 1);
+	pool.api.increment_nonce(Alice.into());
+
+	let watcher = block_on(
+		pool.submit_and_watch(&BlockId::number(1), SOURCE, from_alice.clone())
+	).expect("1. Imported");
+
+	// Block B1
+	let b1 = {
+		let header = pool.api.push_block(2, vec![from_alice.clone()]);
+		assert_eq!(pool.status().ready, 1);
+
+		let event = ChainEvent::NewBlock {
+			hash: header.hash(),
+			is_new_best: true,
+			header: header.clone(),
+			tree_route: None,
+		};
+		block_on(pool.maintain(event));
+		assert_eq!(pool.status().ready, 0);
+		header.hash()
+	};
+
+	// Block B2
+	let b2 = {
+		let header = pool.api.push_block(2, vec![from_alice.clone()]);
+		assert_eq!(pool.status().ready, 0);
+
+		let event = block_event_with_retracted(header.clone(), b1, &*pool.api);
+		block_on(pool.maintain(event));
+		assert_eq!(pool.status().ready, 0);
+
+		let event = ChainEvent::Finalized { hash: header.hash() };
+		block_on(pool.maintain(event));
+
+		header.hash()
+	};
+
+	{
+		let mut stream = futures::executor::block_on_stream(watcher);
+		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(b1.clone())));
+		assert_eq!(stream.next(), Some(TransactionStatus::Retracted(b1)));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(b2.clone())));
+		assert_eq!(stream.next(), Some(TransactionStatus::Finalized(b2)));
+		assert_eq!(stream.next(), None);
+	}
+}
+
+
 /// This test ensures that transactions from a fork are re-submitted if
 /// the forked block is not part of the retracted blocks. This happens as the
 /// retracted block list only contains the route from the old best to the new