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.

substrate/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