Fix transaction pool event sending

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