Revalidate transactions only on latest best block (#6824)

* Revalidate transactions only on latest best block

We should revalidate transactions only on the latest best block and not
on any arbitrary block. The revalidation before failed when there were
multiple blocks on the height given to the revalidation function, but no
block was imported as best block.

* Update test-utils/runtime/transaction-pool/src/lib.rs

Co-authored-by: Jaco Greeff <jacogr@gmail.com>

* Fix tests

* Only process best blocks in the transaction pool

Co-authored-by: Jaco Greeff <jacogr@gmail.com>

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

@@ -34,7 +34,7 @@ pub mod testing;
 pub use sc_transaction_graph as txpool;
 pub use crate::api::{FullChainApi, LightChainApi};
 
-use std::{collections::{HashMap, HashSet}, sync::Arc, pin::Pin};
+use std::{collections::{HashMap, HashSet}, sync::Arc, pin::Pin, convert::TryInto};
 use futures::{prelude::*, future::{self, ready}, channel::oneshot};
 use parking_lot::Mutex;
 
@@ -549,7 +549,7 @@ impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
 {
 	fn maintain(&self, event: ChainEvent<Self::Block>) -> Pin<Box<dyn Future<Output=()> + Send>> {
 		match event {
-			ChainEvent::NewBlock { hash, tree_route, is_new_best, .. } => {
+			ChainEvent::NewBestBlock { hash, tree_route } => {
 				let pool = self.pool.clone();
 				let api = self.api.clone();
 
@@ -608,10 +608,7 @@ impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
 						})
 					}
 
-					// If this is a new best block, we need to prune its transactions from the pool.
-					if is_new_best {
-						pruned_log.extend(prune_known_txs_for_block(id.clone(), &*api, &*pool).await);
-					}
+					pruned_log.extend(prune_known_txs_for_block(id.clone(), &*api, &*pool).await);
 
 					metrics.report(
 						|metrics| metrics.block_transactions_pruned.inc_by(pruned_log.len() as u64)
@@ -690,9 +687,9 @@ impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
 							.map(|tx| tx.hash.clone())
 							.collect();
 						revalidation_queue.revalidate_later(block_number, hashes).await;
-					}
 
-					revalidation_strategy.lock().clear();
+						revalidation_strategy.lock().clear();
+					}
 				}.boxed()
 			}
 			ChainEvent::Finalized { hash } => {
@@ -721,7 +718,9 @@ pub async fn notification_future<Client, Pool, Block>(
 		Client: sc_client_api::BlockchainEvents<Block>,
 		Pool: MaintainedTransactionPool<Block=Block>,
 {
-	let import_stream = client.import_notification_stream().map(Into::into).fuse();
+	let import_stream = client.import_notification_stream()
+		.filter_map(|n| ready(n.try_into().ok()))
+		.fuse();
 	let finality_stream = client.finality_notification_stream()
 		.map(Into::into)
 		.fuse();

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

@@ -211,8 +211,7 @@ impl<Api: ChainApi> RevalidationWorker<Api> {
 		mut self,
 		from_queue: TracingUnboundedReceiver<WorkerPayload<Api>>,
 		interval: R,
-	) where R: Send, R::Guard: Send
-	{
+	) where R: Send, R::Guard: Send {
 		let interval = interval.into_stream().fuse();
 		let from_queue = from_queue.fuse();
 		futures::pin_mut!(interval, from_queue);
@@ -253,7 +252,7 @@ impl<Api: ChainApi> RevalidationWorker<Api> {
 							if this.members.len() > 0 {
 								log::debug!(
 									target: "txpool",
-									"Updated revalidation queue at {}. Transactions: {:?}",
+									"Updated revalidation queue at {:?}. Transactions: {:?}",
 									this.best_block,
 									this.members,
 								);
@@ -298,9 +297,7 @@ where
 		api: Arc<Api>,
 		pool: Arc<Pool<Api>>,
 		interval: R,
-	) -> (Self, Pin<Box<dyn Future<Output=()> + Send>>)
-	where R: Send + 'static, R::Guard: Send
-	{
+	) -> (Self, Pin<Box<dyn Future<Output=()> + Send>>) where R: Send + 'static, R::Guard: Send {
 		let (to_worker, from_queue) = tracing_unbounded("mpsc_revalidation_queue");
 
 		let worker = RevalidationWorker::new(api.clone(), pool.clone());
@@ -338,16 +335,22 @@ where
 	/// If queue configured with background worker, this will return immediately.
 	/// If queue configured without background worker, this will resolve after
 	/// revalidation is actually done.
-	pub async fn revalidate_later(&self, at: NumberFor<Api>, transactions: Vec<ExtrinsicHash<Api>>) {
+	pub async fn revalidate_later(
+		&self,
+		at: NumberFor<Api>,
+		transactions: Vec<ExtrinsicHash<Api>>,
+	) {
 		if transactions.len() > 0 {
-			log::debug!(target: "txpool", "Sent {} transactions to revalidation queue", transactions.len());
+			log::debug!(
+				target: "txpool", "Sent {} transactions to revalidation queue",
+				transactions.len(),
+			);
 		}
 
 		if let Some(ref to_worker) = self.background {
 			if let Err(e) = to_worker.unbounded_send(WorkerPayload { at, transactions }) {
 				log::warn!(target: "txpool", "Failed to update background worker: {:?}", e);
 			}
-			return;
 		} else {
 			let pool = self.pool.clone();
 			let api = self.api.clone();

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

@@ -106,7 +106,7 @@ fn prune_tags_should_work() {
 	let pending: Vec<_> = pool.validated_pool().ready().map(|a| a.data.transfer().nonce).collect();
 	assert_eq!(pending, vec![209, 210]);
 
-	pool.validated_pool().api().push_block(1, Vec::new());
+	pool.validated_pool().api().push_block(1, Vec::new(), true);
 	block_on(
 		pool.prune_tags(
 			&BlockId::number(1),
@@ -141,25 +141,14 @@ fn only_prune_on_new_best() {
 	let uxt = uxt(Alice, 209);
 
 	let _ = block_on(
-		pool.submit_and_watch(&BlockId::number(1), SOURCE, uxt.clone())
+		pool.submit_and_watch(&BlockId::number(0), SOURCE, uxt.clone())
 	).expect("1. Imported");
-	let header = pool.api.push_block(1, vec![uxt.clone()]);
+	pool.api.push_block(1, vec![uxt.clone()], true);
 	assert_eq!(pool.status().ready, 1);
 
-	let event = ChainEvent::NewBlock {
+	let header = pool.api.push_block(2, vec![uxt], true);
+	let event = ChainEvent::NewBestBlock {
 		hash: header.hash(),
-		is_new_best: false,
-		header: header.clone(),
-		tree_route: None,
-	};
-	block_on(pool.maintain(event));
-	assert_eq!(pool.status().ready, 1);
-
-	let header = pool.api.push_block(2, vec![uxt]);
-	let event = ChainEvent::NewBlock {
-		hash: header.hash(),
-		is_new_best: true,
-		header: header.clone(),
 		tree_route: None,
 	};
 	block_on(pool.maintain(event));
@@ -179,7 +168,7 @@ fn should_correctly_prune_transactions_providing_more_than_one_tag() {
 
 	// remove the transaction that just got imported.
 	api.increment_nonce(Alice.into());
-	api.push_block(1, Vec::new());
+	api.push_block(1, Vec::new(), true);
 	block_on(pool.prune_tags(&BlockId::number(1), vec![vec![209]], vec![])).expect("1. Pruned");
 	assert_eq!(pool.validated_pool().status().ready, 0);
 	// it's re-imported to future
@@ -187,7 +176,7 @@ fn should_correctly_prune_transactions_providing_more_than_one_tag() {
 
 	// so now let's insert another transaction that also provides the 155
 	api.increment_nonce(Alice.into());
-	api.push_block(2, Vec::new());
+	api.push_block(2, Vec::new(), true);
 	let xt = uxt(Alice, 211);
 	block_on(pool.submit_one(&BlockId::number(2), SOURCE, xt.clone())).expect("2. Imported");
 	assert_eq!(pool.validated_pool().status().ready, 1);
@@ -197,18 +186,16 @@ fn should_correctly_prune_transactions_providing_more_than_one_tag() {
 
 	// prune it and make sure the pool is empty
 	api.increment_nonce(Alice.into());
-	api.push_block(3, Vec::new());
+	api.push_block(3, Vec::new(), true);
 	block_on(pool.prune_tags(&BlockId::number(3), vec![vec![155]], vec![])).expect("2. Pruned");
 	assert_eq!(pool.validated_pool().status().ready, 0);
 	assert_eq!(pool.validated_pool().status().future, 2);
 }
 
 fn block_event(header: Header) -> ChainEvent<Block> {
-	ChainEvent::NewBlock {
+	ChainEvent::NewBestBlock {
 		hash: header.hash(),
-		is_new_best: true,
 		tree_route: None,
-		header,
 	}
 }
 
@@ -219,11 +206,9 @@ fn block_event_with_retracted(
 ) -> ChainEvent<Block> {
 	let tree_route = api.tree_route(retracted_start, header.parent_hash).expect("Tree route exists");
 
-	ChainEvent::NewBlock {
+	ChainEvent::NewBestBlock {
 		hash: header.hash(),
-		is_new_best: true,
 		tree_route: Some(Arc::new(tree_route)),
-		header,
 	}
 }
 
@@ -236,7 +221,7 @@ fn should_prune_old_during_maintenance() {
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
 	assert_eq!(pool.status().ready, 1);
 
-	let header = pool.api.push_block(1, vec![xt.clone()]);
+	let header = pool.api.push_block(1, vec![xt.clone()], true);
 
 	block_on(pool.maintain(block_event(header)));
 	assert_eq!(pool.status().ready, 0);
@@ -253,7 +238,7 @@ fn should_revalidate_during_maintenance() {
 	assert_eq!(pool.status().ready, 2);
 	assert_eq!(pool.api.validation_requests().len(), 2);
 
-	let header = pool.api.push_block(1, vec![xt1.clone()]);
+	let header = pool.api.push_block(1, vec![xt1.clone()], true);
 
 	block_on(pool.maintain(block_event(header)));
 	assert_eq!(pool.status().ready, 1);
@@ -272,8 +257,8 @@ fn should_resubmit_from_retracted_during_maintenance() {
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
 	assert_eq!(pool.status().ready, 1);
 
-	let header = pool.api.push_block(1, vec![]);
-	let fork_header = pool.api.push_block(1, vec![]);
+	let header = pool.api.push_block(1, vec![], true);
+	let fork_header = pool.api.push_block(1, vec![], false);
 
 	let event = block_event_with_retracted(header, fork_header.hash(), &*pool.api);
 
@@ -291,8 +276,8 @@ fn should_not_resubmit_from_retracted_during_maintenance_if_tx_is_also_in_enacte
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
 	assert_eq!(pool.status().ready, 1);
 
-	let header = pool.api.push_block(1, vec![xt.clone()]);
-	let fork_header = pool.api.push_block(1, vec![xt]);
+	let header = pool.api.push_block(1, vec![xt.clone()], true);
+	let fork_header = pool.api.push_block(1, vec![xt], false);
 
 	let event = block_event_with_retracted(header, fork_header.hash(), &*pool.api);
 
@@ -309,8 +294,8 @@ fn should_not_retain_invalid_hashes_from_retracted() {
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
 	assert_eq!(pool.status().ready, 1);
 
-	let header = pool.api.push_block(1, vec![]);
-	let fork_header = pool.api.push_block(1, vec![xt.clone()]);
+	let header = pool.api.push_block(1, vec![], true);
+	let fork_header = pool.api.push_block(1, vec![xt.clone()], false);
 	pool.api.add_invalid(&xt);
 
 	let event = block_event_with_retracted(header, fork_header.hash(), &*pool.api);
@@ -330,14 +315,14 @@ fn should_revalidate_transaction_multiple_times() {
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
 	assert_eq!(pool.status().ready, 1);
 
-	let header = pool.api.push_block(1, vec![xt.clone()]);
+	let header = pool.api.push_block(1, vec![xt.clone()], true);
 
 	block_on(pool.maintain(block_event(header)));
 
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
 	assert_eq!(pool.status().ready, 1);
 
-	let header = pool.api.push_block(2, vec![]);
+	let header = pool.api.push_block(2, vec![], true);
 	pool.api.add_invalid(&xt);
 
 	block_on(pool.maintain(block_event(header)));
@@ -354,18 +339,18 @@ fn should_revalidate_across_many_blocks() {
 
 	let (pool, _guard, mut notifier) = maintained_pool();
 
-	block_on(pool.submit_one(&BlockId::number(1), SOURCE, xt1.clone())).expect("1. Imported");
-	block_on(pool.submit_one(&BlockId::number(1), SOURCE, xt2.clone())).expect("1. Imported");
+	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt1.clone())).expect("1. Imported");
+	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt2.clone())).expect("1. Imported");
 	assert_eq!(pool.status().ready, 2);
 
-	let header = pool.api.push_block(1, vec![]);
+	let header = pool.api.push_block(1, vec![], true);
 	block_on(pool.maintain(block_event(header)));
 	block_on(notifier.next());
 
-	block_on(pool.submit_one(&BlockId::number(2), SOURCE, xt3.clone())).expect("1. Imported");
+	block_on(pool.submit_one(&BlockId::number(1), SOURCE, xt3.clone())).expect("1. Imported");
 	assert_eq!(pool.status().ready, 3);
 
-	let header = pool.api.push_block(2, vec![xt1.clone()]);
+	let header = pool.api.push_block(2, vec![xt1.clone()], true);
 	block_on(pool.maintain(block_event(header)));
 	block_on(notifier.next());
 
@@ -411,7 +396,7 @@ fn should_push_watchers_during_maintaince() {
 	pool.api.add_invalid(&tx4);
 
 	// clear timer events if any
-	let header = pool.api.push_block(1, vec![]);
+	let header = pool.api.push_block(1, vec![], true);
 	block_on(pool.maintain(block_event(header)));
 	block_on(notifier.next());
 
@@ -429,7 +414,7 @@ fn should_push_watchers_during_maintaince() {
 	);
 
 	// when
-	let header = pool.api.push_block(2, vec![tx0, tx1, tx2]);
+	let header = pool.api.push_block(2, vec![tx0, tx1, tx2], true);
 	let header_hash = header.hash();
 	block_on(pool.maintain(block_event(header)));
 
@@ -478,18 +463,16 @@ fn can_track_heap_size() {
 fn finalization() {
 	let xt = uxt(Alice, 209);
 	let api = TestApi::with_alice_nonce(209);
-	api.push_block(1, vec![]);
+	api.push_block(1, vec![], true);
 	let (pool, _background, _) = BasicPool::new_test(api.into());
 	let watcher = block_on(
 		pool.submit_and_watch(&BlockId::number(1), SOURCE, xt.clone())
 	).expect("1. Imported");
-	pool.api.push_block(2, vec![xt.clone()]);
+	pool.api.push_block(2, vec![xt.clone()], true);
 
-	let header = pool.api.chain().read().block_by_number.get(&2).unwrap()[0].header().clone();
-	let event = ChainEvent::NewBlock {
+	let header = pool.api.chain().read().block_by_number.get(&2).unwrap()[0].0.header().clone();
+	let event = ChainEvent::NewBestBlock {
 		hash: header.hash(),
-		is_new_best: true,
-		header: header.clone(),
 		tree_route: None,
 	};
 	block_on(pool.maintain(event));
@@ -508,7 +491,7 @@ fn finalization() {
 fn fork_aware_finalization() {
 	let api = TestApi::empty();
 	// starting block A1 (last finalized.)
-	api.push_block(1, vec![]);
+	api.push_block(1, vec![], true);
 
 	let (pool, _background, _) = BasicPool::new_test(api.into());
 	let mut canon_watchers = vec![];
@@ -534,14 +517,12 @@ fn fork_aware_finalization() {
 		let watcher = block_on(
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, from_alice.clone())
 		).expect("1. Imported");
-		let header = pool.api.push_block(2, vec![from_alice.clone()]);
+		let header = pool.api.push_block(2, vec![from_alice.clone()], true);
 		canon_watchers.push((watcher, header.hash()));
 		assert_eq!(pool.status().ready, 1);
 
-		let event = ChainEvent::NewBlock {
+		let event = ChainEvent::NewBestBlock {
 			hash: header.hash(),
-			is_new_best: true,
-			header: header.clone(),
 			tree_route: None,
 		};
 		b1 = header.hash();
@@ -553,15 +534,13 @@ fn fork_aware_finalization() {
 
 	// block C2
 	{
-		let header = pool.api.push_block_with_parent(b1, vec![from_dave.clone()]);
+		let header = pool.api.push_block_with_parent(b1, vec![from_dave.clone()], true);
 		from_dave_watcher = block_on(
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, from_dave.clone())
 		).expect("1. Imported");
 		assert_eq!(pool.status().ready, 1);
-		let event = ChainEvent::NewBlock {
+		let event = ChainEvent::NewBestBlock {
 			hash: header.hash(),
-			is_new_best: true,
-			header: header.clone(),
 			tree_route: None,
 		};
 		c2 = header.hash();
@@ -575,12 +554,10 @@ fn fork_aware_finalization() {
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, from_bob.clone())
 		).expect("1. Imported");
 		assert_eq!(pool.status().ready, 1);
-		let header = pool.api.push_block_with_parent(c2, vec![from_bob.clone()]);
+		let header = pool.api.push_block_with_parent(c2, vec![from_bob.clone()], true);
 
-		let event = ChainEvent::NewBlock {
+		let event = ChainEvent::NewBestBlock {
 			hash: header.hash(),
-			is_new_best: true,
-			header: header.clone(),
 			tree_route: None,
 		};
 		d2 = header.hash();
@@ -594,7 +571,7 @@ fn fork_aware_finalization() {
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, from_charlie.clone())
 		).expect("1.Imported");
 		assert_eq!(pool.status().ready, 1);
-		let header = pool.api.push_block(3, vec![from_charlie.clone()]);
+		let header = pool.api.push_block(3, vec![from_charlie.clone()], true);
 
 		canon_watchers.push((watcher, header.hash()));
 		let event = block_event_with_retracted(header.clone(), d2, &*pool.api);
@@ -612,13 +589,11 @@ fn fork_aware_finalization() {
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, xt.clone())
 		).expect("1. Imported");
 		assert_eq!(pool.status().ready, 3);
-		let header = pool.api.push_block(4, vec![xt.clone()]);
+		let header = pool.api.push_block(4, vec![xt.clone()], true);
 		canon_watchers.push((w, header.hash()));
 
-		let event = ChainEvent::NewBlock {
+		let event = ChainEvent::NewBestBlock {
 			hash: header.hash(),
-			is_new_best: true,
-			header: header.clone(),
 			tree_route: None,
 		};
 		d1 = header.hash();
@@ -632,12 +607,10 @@ fn fork_aware_finalization() {
 
 	// block e1
 	{
-		let header = pool.api.push_block(5, vec![from_dave, from_bob]);
+		let header = pool.api.push_block(5, vec![from_dave, from_bob], true);
 		e1 = header.hash();
-		let event = ChainEvent::NewBlock {
+		let event = ChainEvent::NewBestBlock {
 			hash: header.hash(),
-			is_new_best: true,
-			header: header.clone(),
 			tree_route: None,
 		};
 		block_on(pool.maintain(event));
@@ -684,7 +657,7 @@ fn fork_aware_finalization() {
 fn prune_and_retract_tx_at_same_time() {
 	let api = TestApi::empty();
 	// starting block A1 (last finalized.)
-	api.push_block(1, vec![]);
+	api.push_block(1, vec![], true);
 
 	let (pool, _background, _) = BasicPool::new_test(api.into());
 
@@ -697,13 +670,11 @@ fn prune_and_retract_tx_at_same_time() {
 
 	// Block B1
 	let b1 = {
-		let header = pool.api.push_block(2, vec![from_alice.clone()]);
+		let header = pool.api.push_block(2, vec![from_alice.clone()], true);
 		assert_eq!(pool.status().ready, 1);
 
-		let event = ChainEvent::NewBlock {
+		let event = ChainEvent::NewBestBlock {
 			hash: header.hash(),
-			is_new_best: true,
-			header: header.clone(),
 			tree_route: None,
 		};
 		block_on(pool.maintain(event));
@@ -713,7 +684,7 @@ fn prune_and_retract_tx_at_same_time() {
 
 	// Block B2
 	let b2 = {
-		let header = pool.api.push_block(2, vec![from_alice.clone()]);
+		let header = pool.api.push_block(2, vec![from_alice.clone()], false);
 		assert_eq!(pool.status().ready, 0);
 
 		let event = block_event_with_retracted(header.clone(), b1, &*pool.api);
@@ -757,7 +728,7 @@ fn prune_and_retract_tx_at_same_time() {
 fn resubmit_tx_of_fork_that_is_not_part_of_retracted() {
 	let api = TestApi::empty();
 	// starting block A1 (last finalized.)
-	api.push_block(1, vec![]);
+	api.push_block(1, vec![], true);
 
 	let (pool, _background, _) = BasicPool::new_test(api.into());
 
@@ -773,13 +744,11 @@ fn resubmit_tx_of_fork_that_is_not_part_of_retracted() {
 		let _ = block_on(
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, tx0.clone())
 		).expect("1. Imported");
-		let header = pool.api.push_block(2, vec![tx0.clone()]);
+		let header = pool.api.push_block(2, vec![tx0.clone()], true);
 		assert_eq!(pool.status().ready, 1);
 
-		let event = ChainEvent::NewBlock {
+		let event = ChainEvent::NewBestBlock {
 			hash: header.hash(),
-			is_new_best: true,
-			header: header.clone(),
 			tree_route: None,
 		};
 		d0 = header.hash();
@@ -792,23 +761,13 @@ fn resubmit_tx_of_fork_that_is_not_part_of_retracted() {
 		let _ = block_on(
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, tx1.clone())
 		).expect("1. Imported");
-		let header = pool.api.push_block(2, vec![tx1.clone()]);
-		assert_eq!(pool.status().ready, 1);
-		let event = ChainEvent::NewBlock {
-			hash: header.hash(),
-			is_new_best: false,
-			header: header.clone(),
-			tree_route: None,
-		};
-		block_on(pool.maintain(event));
-
-		// Only transactions from new best should be pruned
+		pool.api.push_block(2, vec![tx1.clone()], false);
 		assert_eq!(pool.status().ready, 1);
 	}
 
 	// Block D2
 	{
-		let header = pool.api.push_block(2, vec![]);
+		let header = pool.api.push_block(2, vec![], false);
 		let event = block_event_with_retracted(header, d0, &*pool.api);
 		block_on(pool.maintain(event));
 		assert_eq!(pool.status().ready, 2);
@@ -819,7 +778,7 @@ fn resubmit_tx_of_fork_that_is_not_part_of_retracted() {
 fn resubmit_from_retracted_fork() {
 	let api = TestApi::empty();
 	// starting block A1 (last finalized.)
-	api.push_block(1, vec![]);
+	api.push_block(1, vec![], true);
 
 	let (pool, _background, _) = BasicPool::new_test(api.into());
 
@@ -844,16 +803,10 @@ fn resubmit_from_retracted_fork() {
 		let _ = block_on(
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, tx0.clone())
 		).expect("1. Imported");
-		let header = pool.api.push_block(2, vec![tx0.clone()]);
+		let header = pool.api.push_block(2, vec![tx0.clone()], true);
 		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));
+		block_on(pool.maintain(block_event(header)));
 		assert_eq!(pool.status().ready, 0);
 	}
 
@@ -862,14 +815,8 @@ fn resubmit_from_retracted_fork() {
 		let _ = block_on(
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, tx1.clone())
 		).expect("1. Imported");
-		let header = pool.api.push_block(3, vec![tx1.clone()]);
-		let event = ChainEvent::NewBlock {
-			hash: header.hash(),
-			is_new_best: true,
-			header: header.clone(),
-			tree_route: None,
-		};
-		block_on(pool.maintain(event));
+		let header = pool.api.push_block(3, vec![tx1.clone()], true);
+		block_on(pool.maintain(block_event(header)));
 		assert_eq!(pool.status().ready, 0);
 	}
 
@@ -878,14 +825,8 @@ fn resubmit_from_retracted_fork() {
 		let _ = block_on(
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, tx2.clone())
 		).expect("1. Imported");
-		let header = pool.api.push_block(4, vec![tx2.clone()]);
-		let event = ChainEvent::NewBlock {
-			hash: header.hash(),
-			is_new_best: true,
-			header: header.clone(),
-			tree_route: None,
-		};
-		block_on(pool.maintain(event));
+		let header = pool.api.push_block(4, vec![tx2.clone()], true);
+		block_on(pool.maintain(block_event(header.clone())));
 		assert_eq!(pool.status().ready, 0);
 		header.hash()
 	};
@@ -895,14 +836,7 @@ fn resubmit_from_retracted_fork() {
 		let _ = block_on(
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, tx3.clone())
 		).expect("1. Imported");
-		let header = pool.api.push_block(2, vec![tx3.clone()]);
-		let event = ChainEvent::NewBlock {
-			hash: header.hash(),
-			is_new_best: false,
-			header: header.clone(),
-			tree_route: None,
-		};
-		block_on(pool.maintain(event));
+		let header = pool.api.push_block(2, vec![tx3.clone()], true);
 		assert_eq!(pool.status().ready, 1);
 		header.hash()
 	};
@@ -912,14 +846,7 @@ fn resubmit_from_retracted_fork() {
 		let _ = block_on(
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, tx4.clone())
 		).expect("1. Imported");
-		let header = pool.api.push_block_with_parent(d1.clone(), vec![tx4.clone()]);
-		let event = ChainEvent::NewBlock {
-			hash: header.hash(),
-			is_new_best: false,
-			header: header.clone(),
-			tree_route: None,
-		};
-		block_on(pool.maintain(event));
+		let header = pool.api.push_block_with_parent(d1.clone(), vec![tx4.clone()], true);
 		assert_eq!(pool.status().ready, 2);
 		header.hash()
 	};
@@ -929,7 +856,7 @@ fn resubmit_from_retracted_fork() {
 		let _ = block_on(
 			pool.submit_and_watch(&BlockId::number(1), SOURCE, tx5.clone())
 		).expect("1. Imported");
-		let header = pool.api.push_block_with_parent(e1.clone(), vec![tx5.clone()]);
+		let header = pool.api.push_block_with_parent(e1.clone(), vec![tx5.clone()], true);
 		// Don't announce the block event to the pool directly, because we will
 		// re-org to this block.
 		assert_eq!(pool.status().ready, 3);
@@ -953,7 +880,7 @@ fn resubmit_from_retracted_fork() {
 fn ready_set_should_not_resolve_before_block_update() {
 	let (pool, _guard, _notifier) = maintained_pool();
 	let xt1 = uxt(Alice, 209);
-	block_on(pool.submit_one(&BlockId::number(1), SOURCE, xt1.clone())).expect("1. Imported");
+	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt1.clone())).expect("1. Imported");
 
 	assert!(pool.ready_at(1).now_or_never().is_none());
 }
@@ -961,7 +888,7 @@ fn ready_set_should_not_resolve_before_block_update() {
 #[test]
 fn ready_set_should_resolve_after_block_update() {
 	let (pool, _guard, _notifier) = maintained_pool();
-	let header = pool.api.push_block(1, vec![]);
+	let header = pool.api.push_block(1, vec![], true);
 
 	let xt1 = uxt(Alice, 209);
 
@@ -974,7 +901,7 @@ fn ready_set_should_resolve_after_block_update() {
 #[test]
 fn ready_set_should_eventually_resolve_when_block_update_arrives() {
 	let (pool, _guard, _notifier) = maintained_pool();
-	let header = pool.api.push_block(1, vec![]);
+	let header = pool.api.push_block(1, vec![], true);
 
 	let xt1 = uxt(Alice, 209);
 
@@ -1063,7 +990,7 @@ fn import_notification_to_pool_maintain_works() {
 	// Get the notification of the block import and maintain the pool with it,
 	// Now, the pool should not contain any transactions.
 	let evt = import_stream.next().expect("Importing a block leads to an event");
-	block_on(pool.maintain(evt.into()));
+	block_on(pool.maintain(evt.try_into().expect("Imported as new best block")));
 	assert_eq!(pool.status().ready, 0);
 }
 
@@ -1075,7 +1002,7 @@ fn pruning_a_transaction_should_remove_it_from_best_transaction() {
 	let xt1 = Extrinsic::IncludeData(Vec::new());
 
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt1.clone())).expect("1. Imported");
-	let header = pool.api.push_block(1, vec![xt1.clone()]);
+	let header = pool.api.push_block(1, vec![xt1.clone()], true);
 
 	// This will prune `xt1`.
 	block_on(pool.maintain(block_event(header)));
@@ -1091,3 +1018,23 @@ fn pruning_a_transaction_should_remove_it_from_best_transaction() {
 	// returned a second time by the iterator.
 	assert!(iterator.next().is_none());
 }
+
+#[test]
+fn only_revalidate_on_best_block() {
+	let xt = uxt(Alice, 209);
+
+	let (pool, _guard, mut notifier) = maintained_pool();
+
+	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
+	assert_eq!(pool.status().ready, 1);
+
+	let header = pool.api.push_block(1, vec![], true);
+
+	pool.api.push_block(2, vec![], false);
+	pool.api.push_block(2, vec![], false);
+
+	block_on(pool.maintain(block_event(header)));
+	block_on(notifier.next());
+
+	assert_eq!(pool.status().ready, 1);
+}