Remove transaction-pool test-helpers feature (#10571)

* Remove transaction-pool `test-helpers` feature

`test-helpers` feature is a bad idea in general, because once the feature is enabled somewhere in
the workspace, it is enabled anywhere. While removing the feature, the tests were also rewritten to
get rid off other "only test" related code.

Contributes towards: https://github.com/paritytech/substrate/issues/9727

* Apply suggestions from code review

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

* Fix benches

Co-authored-by: André Silva <123550+andresilva@users.noreply.github.com>

substrate/Cargo.lock

@@ -1898,7 +1898,7 @@ checksum = "e8ac3ff5224ef91f3c97e03eb1de2db82743427e91aaa5ac635f454f0b164f5a"
 dependencies = [
  "either",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "log 0.4.14",
  "num-traits",
  "parity-scale-codec",
@@ -2391,12 +2391,6 @@ version = "0.3.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6ee7c6485c30167ce4dfb83ac568a849fe53274c831081476ee13e0dce1aad72"
 
-[[package]]
-name = "futures-timer"
-version = "2.0.2"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a1de7508b218029b0f01662ed8f61b1c964b3ae99d6f25462d0f55a595109df6"
-
 [[package]]
 name = "futures-timer"
 version = "3.0.2"
@@ -2974,16 +2968,6 @@ dependencies = [
  "num-traits",
 ]
 
-[[package]]
-name = "intervalier"
-version = "0.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "64fa110ec7b8f493f416eed552740d10e7030ad5f63b2308f82c9608ec2df275"
-dependencies = [
- "futures 0.3.16",
- "futures-timer 2.0.2",
-]
-
 [[package]]
 name = "io-lifetimes"
 version = "0.3.1"
@@ -3481,7 +3465,7 @@ dependencies = [
  "either",
  "fnv",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "lazy_static",
  "libsecp256k1",
  "log 0.4.14",
@@ -3745,7 +3729,7 @@ dependencies = [
  "asynchronous-codec 0.6.0",
  "bytes 1.1.0",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "libp2p-core",
  "libp2p-swarm",
  "log 0.4.14",
@@ -3834,7 +3818,7 @@ checksum = "7399c5b6361ef525d41c11fcf51635724f832baf5819b30d3d873eabb4fbae4b"
 dependencies = [
  "async-io",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "if-watch",
  "ipnet",
  "libc",
@@ -7625,7 +7609,7 @@ dependencies = [
  "async-trait",
  "derive_more",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "ip_network",
  "libp2p",
  "log 0.4.14",
@@ -7652,7 +7636,7 @@ name = "sc-basic-authorship"
 version = "0.10.0-dev"
 dependencies = [
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "log 0.4.14",
  "parity-scale-codec",
  "parking_lot 0.11.2",
@@ -7816,7 +7800,7 @@ version = "0.10.0-dev"
 dependencies = [
  "async-trait",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "libp2p",
  "log 0.4.14",
  "parking_lot 0.11.2",
@@ -8007,7 +7991,7 @@ dependencies = [
  "async-trait",
  "derive_more",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "log 0.4.14",
  "parity-scale-codec",
  "parking_lot 0.11.2",
@@ -8030,7 +8014,7 @@ version = "0.10.0-dev"
 dependencies = [
  "async-trait",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "log 0.4.14",
  "parity-scale-codec",
  "sc-client-api",
@@ -8163,7 +8147,7 @@ dependencies = [
  "finality-grandpa",
  "fork-tree",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "log 0.4.14",
  "parity-scale-codec",
  "parking_lot 0.11.2",
@@ -8228,7 +8212,7 @@ version = "0.10.0-dev"
 dependencies = [
  "ansi_term",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "log 0.4.14",
  "parity-util-mem",
  "sc-client-api",
@@ -8269,7 +8253,7 @@ dependencies = [
  "fnv",
  "fork-tree",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "hex",
  "ip_network",
  "libp2p",
@@ -8316,7 +8300,7 @@ version = "0.10.0-dev"
 dependencies = [
  "async-std",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "libp2p",
  "log 0.4.14",
  "lru 0.7.0",
@@ -8335,7 +8319,7 @@ dependencies = [
  "async-std",
  "async-trait",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "libp2p",
  "log 0.4.14",
  "parking_lot 0.11.2",
@@ -8362,7 +8346,7 @@ dependencies = [
  "bytes 1.1.0",
  "fnv",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "hex",
  "hyper 0.14.16",
  "hyper-rustls",
@@ -8512,7 +8496,7 @@ dependencies = [
  "directories",
  "exit-future",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "hash-db",
  "jsonrpc-core",
  "jsonrpc-pubsub",
@@ -8704,8 +8688,8 @@ dependencies = [
  "assert_matches",
  "criterion",
  "futures 0.3.16",
+ "futures-timer",
  "hex",
- "intervalier",
  "linked-hash-map",
  "log 0.4.14",
  "parity-scale-codec",
@@ -8749,7 +8733,7 @@ name = "sc-utils"
 version = "4.0.0-dev"
 dependencies = [
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "lazy_static",
  "prometheus",
 ]
@@ -9348,7 +9332,7 @@ version = "0.10.0-dev"
 dependencies = [
  "async-trait",
  "futures 0.3.16",
- "futures-timer 3.0.2",
+ "futures-timer",
  "log 0.4.14",
  "parity-scale-codec",
  "sp-core",
@@ -9910,7 +9894,7 @@ name = "sp-timestamp"
 version = "4.0.0-dev"
 dependencies = [
  "async-trait",
- "futures-timer 3.0.2",
+ "futures-timer",
  "log 0.4.14",
  "parity-scale-codec",
  "sp-api",

substrate/client/transaction-pool/Cargo.toml

@@ -16,7 +16,7 @@ targets = ["x86_64-unknown-linux-gnu"]
 codec = { package = "parity-scale-codec", version = "2.0.0" }
 thiserror = "1.0.30"
 futures = "0.3.16"
-intervalier = "0.4.0"
+futures-timer = "3.0.2"
 log = "0.4.8"
 parity-util-mem = { version = "0.10.2", default-features = false, features = ["primitive-types"] }
 parking_lot = "0.11.2"
@@ -48,6 +48,3 @@ criterion = "0.3"
 [[bench]]
 name = "basics"
 harness = false
-
-[features]
-test-helpers = []

substrate/client/transaction-pool/benches/basics.rs

@@ -23,11 +23,11 @@ use futures::{
 	executor::block_on,
 	future::{ready, Ready},
 };
-use sc_transaction_pool::{test_helpers::*, *};
+use sc_transaction_pool::*;
 use sp_core::blake2_256;
 use sp_runtime::{
 	generic::BlockId,
-	traits::Block as BlockT,
+	traits::{Block as BlockT, NumberFor},
 	transaction_validity::{
 		InvalidTransaction, TransactionSource, TransactionTag as Tag, TransactionValidity,
 		ValidTransaction,
@@ -63,7 +63,7 @@ impl ChainApi for TestApi {
 		&self,
 		at: &BlockId<Self::Block>,
 		_source: TransactionSource,
-		uxt: test_helpers::ExtrinsicFor<Self>,
+		uxt: <Self::Block as BlockT>::Extrinsic,
 	) -> Self::ValidationFuture {
 		let nonce = uxt.transfer().nonce;
 		let from = uxt.transfer().from.clone();
@@ -89,7 +89,7 @@ impl ChainApi for TestApi {
 	fn block_id_to_number(
 		&self,
 		at: &BlockId<Self::Block>,
-	) -> Result<Option<test_helpers::NumberFor<Self>>, Self::Error> {
+	) -> Result<Option<NumberFor<Self::Block>>, Self::Error> {
 		Ok(match at {
 			BlockId::Number(num) => Some(*num),
 			BlockId::Hash(_) => None,
@@ -99,14 +99,14 @@ impl ChainApi for TestApi {
 	fn block_id_to_hash(
 		&self,
 		at: &BlockId<Self::Block>,
-	) -> Result<Option<test_helpers::BlockHash<Self>>, Self::Error> {
+	) -> Result<Option<<Self::Block as BlockT>::Hash>, Self::Error> {
 		Ok(match at {
 			BlockId::Number(num) => Some(H256::from_low_u64_be(*num)).into(),
 			BlockId::Hash(_) => None,
 		})
 	}
 
-	fn hash_and_length(&self, uxt: &test_helpers::ExtrinsicFor<Self>) -> (H256, usize) {
+	fn hash_and_length(&self, uxt: &<Self::Block as BlockT>::Extrinsic) -> (H256, usize) {
 		let encoded = uxt.encode();
 		(blake2_256(&encoded).into(), encoded.len())
 	}

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

@@ -444,164 +444,17 @@ impl<B: ChainApi> Clone for Pool<B> {
 #[cfg(test)]
 mod tests {
 	use super::{super::base_pool::Limit, *};
+	use crate::tests::{pool, uxt, TestApi, INVALID_NONCE};
 	use assert_matches::assert_matches;
-	use codec::Encode;
 	use futures::executor::block_on;
 	use parking_lot::Mutex;
 	use sc_transaction_pool_api::TransactionStatus;
-	use sp_runtime::{
-		traits::Hash,
-		transaction_validity::{InvalidTransaction, TransactionSource, ValidTransaction},
-	};
-	use std::{
-		collections::{HashMap, HashSet},
-		time::Instant,
-	};
-	use substrate_test_runtime::{AccountId, Block, Extrinsic, Hashing, Transfer, H256};
+	use sp_runtime::transaction_validity::TransactionSource;
+	use std::{collections::HashMap, time::Instant};
+	use substrate_test_runtime::{AccountId, Extrinsic, Transfer, H256};
 
-	const INVALID_NONCE: u64 = 254;
 	const SOURCE: TransactionSource = TransactionSource::External;
 
-	#[derive(Clone, Debug, Default)]
-	struct TestApi {
-		delay: Arc<Mutex<Option<std::sync::mpsc::Receiver<()>>>>,
-		invalidate: Arc<Mutex<HashSet<H256>>>,
-		clear_requirements: Arc<Mutex<HashSet<H256>>>,
-		add_requirements: Arc<Mutex<HashSet<H256>>>,
-	}
-
-	impl ChainApi for TestApi {
-		type Block = Block;
-		type Error = error::Error;
-		type ValidationFuture = futures::future::Ready<error::Result<TransactionValidity>>;
-		type BodyFuture = futures::future::Ready<error::Result<Option<Vec<Extrinsic>>>>;
-
-		/// Verify extrinsic at given block.
-		fn validate_transaction(
-			&self,
-			at: &BlockId<Self::Block>,
-			_source: TransactionSource,
-			uxt: ExtrinsicFor<Self>,
-		) -> Self::ValidationFuture {
-			let hash = self.hash_and_length(&uxt).0;
-			let block_number = self.block_id_to_number(at).unwrap().unwrap();
-
-			let res = match uxt {
-				Extrinsic::Transfer { transfer, .. } => {
-					let nonce = transfer.nonce;
-
-					// This is used to control the test flow.
-					if nonce > 0 {
-						let opt = self.delay.lock().take();
-						if let Some(delay) = opt {
-							if delay.recv().is_err() {
-								println!("Error waiting for delay!");
-							}
-						}
-					}
-
-					if self.invalidate.lock().contains(&hash) {
-						InvalidTransaction::Custom(0).into()
-					} else if nonce < block_number {
-						InvalidTransaction::Stale.into()
-					} else {
-						let mut transaction = ValidTransaction {
-							priority: 4,
-							requires: if nonce > block_number {
-								vec![vec![nonce as u8 - 1]]
-							} else {
-								vec![]
-							},
-							provides: if nonce == INVALID_NONCE {
-								vec![]
-							} else {
-								vec![vec![nonce as u8]]
-							},
-							longevity: 3,
-							propagate: true,
-						};
-
-						if self.clear_requirements.lock().contains(&hash) {
-							transaction.requires.clear();
-						}
-
-						if self.add_requirements.lock().contains(&hash) {
-							transaction.requires.push(vec![128]);
-						}
-
-						Ok(transaction)
-					}
-				},
-				Extrinsic::IncludeData(_) => Ok(ValidTransaction {
-					priority: 9001,
-					requires: vec![],
-					provides: vec![vec![42]],
-					longevity: 9001,
-					propagate: false,
-				}),
-				Extrinsic::Store(_) => Ok(ValidTransaction {
-					priority: 9001,
-					requires: vec![],
-					provides: vec![vec![43]],
-					longevity: 9001,
-					propagate: false,
-				}),
-				_ => unimplemented!(),
-			};
-
-			futures::future::ready(Ok(res))
-		}
-
-		/// Returns a block number given the block id.
-		fn block_id_to_number(
-			&self,
-			at: &BlockId<Self::Block>,
-		) -> Result<Option<NumberFor<Self>>, Self::Error> {
-			Ok(match at {
-				BlockId::Number(num) => Some(*num),
-				BlockId::Hash(_) => None,
-			})
-		}
-
-		/// Returns a block hash given the block id.
-		fn block_id_to_hash(
-			&self,
-			at: &BlockId<Self::Block>,
-		) -> Result<Option<<Self::Block as BlockT>::Hash>, Self::Error> {
-			Ok(match at {
-				BlockId::Number(num) => Some(H256::from_low_u64_be(*num)).into(),
-				BlockId::Hash(_) => None,
-			})
-		}
-
-		/// Hash the extrinsic.
-		fn hash_and_length(&self, uxt: &ExtrinsicFor<Self>) -> (BlockHash<Self>, usize) {
-			let encoded = uxt.encode();
-			let len = encoded.len();
-			(Hashing::hash(&encoded), len)
-		}
-
-		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 {
-		let signature = TryFrom::try_from(&[0; 64][..]).unwrap();
-		Extrinsic::Transfer { transfer, signature, exhaust_resources_when_not_first: false }
-	}
-
-	fn pool() -> Pool<TestApi> {
-		Pool::new(Default::default(), true.into(), TestApi::default().into())
-	}
-
 	#[test]
 	fn should_validate_and_import_transaction() {
 		// given
@@ -636,7 +489,7 @@ mod tests {
 		});
 
 		// when
-		pool.validated_pool.rotator().ban(&Instant::now(), vec![pool.hash_of(&uxt)]);
+		pool.validated_pool.ban(&Instant::now(), vec![pool.hash_of(&uxt)]);
 		let res = block_on(pool.submit_one(&BlockId::Number(0), SOURCE, uxt));
 		assert_eq!(pool.validated_pool().status().ready, 0);
 		assert_eq!(pool.validated_pool().status().future, 0);
@@ -767,9 +620,9 @@ mod tests {
 		assert_eq!(pool.validated_pool().status().future, 0);
 		assert_eq!(pool.validated_pool().status().ready, 0);
 		// make sure they are temporarily banned as well
-		assert!(pool.validated_pool.rotator().is_banned(&hash1));
-		assert!(pool.validated_pool.rotator().is_banned(&hash2));
-		assert!(pool.validated_pool.rotator().is_banned(&hash3));
+		assert!(pool.validated_pool.is_banned(&hash1));
+		assert!(pool.validated_pool.is_banned(&hash2));
+		assert!(pool.validated_pool.is_banned(&hash3));
 	}
 
 	#[test]
@@ -792,7 +645,7 @@ mod tests {
 		block_on(pool.prune_tags(&BlockId::Number(1), vec![vec![0]], vec![hash1.clone()])).unwrap();
 
 		// then
-		assert!(pool.validated_pool.rotator().is_banned(&hash1));
+		assert!(pool.validated_pool.is_banned(&hash1));
 	}
 
 	#[test]
@@ -832,8 +685,8 @@ mod tests {
 
 		// then
 		assert_eq!(pool.validated_pool().status().future, 1);
-		assert!(pool.validated_pool.rotator().is_banned(&hash1));
-		assert!(!pool.validated_pool.rotator().is_banned(&hash2));
+		assert!(pool.validated_pool.is_banned(&hash1));
+		assert!(!pool.validated_pool.is_banned(&hash2));
 	}
 
 	#[test]

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

@@ -569,12 +569,6 @@ impl<B: ChainApi> ValidatedPool<B> {
 		Ok(())
 	}
 
-	/// Get rotator reference.
-	#[cfg(feature = "test-helpers")]
-	pub fn rotator(&self) -> &PoolRotator<ExtrinsicHash<B>> {
-		&self.rotator
-	}
-
 	/// Get api reference.
 	pub fn api(&self) -> &B {
 		&self.api

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

@@ -23,20 +23,12 @@
 #![warn(unused_extern_crates)]
 
 mod api;
+pub mod error;
 mod graph;
 mod metrics;
 mod revalidation;
-
-pub mod error;
-
-/// Common types for testing the transaction pool
-#[cfg(feature = "test-helpers")]
-pub mod test_helpers {
-	pub use super::{
-		graph::{BlockHash, ChainApi, ExtrinsicFor, NumberFor, Pool},
-		revalidation::RevalidationQueue,
-	};
-}
+#[cfg(test)]
+mod tests;
 
 pub use crate::api::FullChainApi;
 use futures::{
@@ -170,13 +162,10 @@ where
 	PoolApi: graph::ChainApi<Block = Block> + 'static,
 {
 	/// Create new basic transaction pool with provided api, for tests.
-	#[cfg(feature = "test-helpers")]
-	pub fn new_test(
-		pool_api: Arc<PoolApi>,
-	) -> (Self, Pin<Box<dyn Future<Output = ()> + Send>>, intervalier::BackSignalControl) {
+	pub fn new_test(pool_api: Arc<PoolApi>) -> (Self, Pin<Box<dyn Future<Output = ()> + Send>>) {
 		let pool = Arc::new(graph::Pool::new(Default::default(), true.into(), pool_api.clone()));
-		let (revalidation_queue, background_task, notifier) =
-			revalidation::RevalidationQueue::new_test(pool_api.clone(), pool.clone());
+		let (revalidation_queue, background_task) =
+			revalidation::RevalidationQueue::new_background(pool_api.clone(), pool.clone());
 		(
 			Self {
 				api: pool_api,
@@ -187,7 +176,6 @@ where
 				metrics: Default::default(),
 			},
 			background_task,
-			notifier,
 		)
 	}
 
@@ -237,7 +225,6 @@ where
 	}
 
 	/// Get access to the underlying api
-	#[cfg(feature = "test-helpers")]
 	pub fn api(&self) -> &PoolApi {
 		&self.api
 	}

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

@@ -35,10 +35,7 @@ use sp_runtime::{
 use futures::prelude::*;
 use std::time::Duration;
 
-#[cfg(not(feature = "test-helpers"))]
 const BACKGROUND_REVALIDATION_INTERVAL: Duration = Duration::from_millis(200);
-#[cfg(feature = "test-helpers")]
-pub const BACKGROUND_REVALIDATION_INTERVAL: Duration = Duration::from_millis(1);
 
 const MIN_BACKGROUND_REVALIDATION_BATCH_SIZE: usize = 20;
 
@@ -213,36 +210,25 @@ impl<Api: ChainApi> RevalidationWorker<Api> {
 	/// It does two things: periodically tries to process some transactions
 	/// from the queue and also accepts messages to enqueue some more
 	/// transactions from the pool.
-	pub async fn run<R: intervalier::IntoStream>(
+	pub async fn run(
 		mut self,
 		from_queue: TracingUnboundedReceiver<WorkerPayload<Api>>,
-		interval: R,
-	) where
-		R: Send,
-		R::Guard: Send,
-	{
-		let interval = interval.into_stream().fuse();
+		interval: Duration,
+	) {
+		let interval_fut = futures_timer::Delay::new(interval);
 		let from_queue = from_queue.fuse();
-		futures::pin_mut!(interval, from_queue);
+		futures::pin_mut!(interval_fut, from_queue);
 		let this = &mut self;
 
 		loop {
 			futures::select! {
-				_guard = interval.next() => {
+				// Using `fuse()` in here is okay, because we reset the interval when it has fired.
+				_ = (&mut interval_fut).fuse() => {
 					let next_batch = this.prepare_batch();
 					let batch_len = next_batch.len();
 
 					batch_revalidate(this.pool.clone(), this.api.clone(), this.best_block, next_batch).await;
 
-					#[cfg(feature = "test-helpers")]
-					{
-						use intervalier::Guard;
-						// only trigger test events if something was processed
-						if batch_len == 0 {
-							_guard.expect("Always some() in tests").skip();
-						}
-					}
-
 					if batch_len > 0 || this.len() > 0 {
 						log::debug!(
 							target: "txpool",
@@ -251,6 +237,8 @@ impl<Api: ChainApi> RevalidationWorker<Api> {
 							this.len(),
 						);
 					}
+
+					interval_fut.reset(interval);
 				},
 				workload = from_queue.next() => {
 					match workload {
@@ -298,15 +286,11 @@ where
 	}
 
 	/// New revalidation queue with background worker.
-	pub fn new_with_interval<R: intervalier::IntoStream>(
+	pub fn new_with_interval(
 		api: Arc<Api>,
 		pool: Arc<Pool<Api>>,
-		interval: R,
-	) -> (Self, Pin<Box<dyn Future<Output = ()> + Send>>)
-	where
-		R: Send + 'static,
-		R::Guard: Send,
-	{
+		interval: Duration,
+	) -> (Self, Pin<Box<dyn Future<Output = ()> + Send>>) {
 		let (to_worker, from_queue) = tracing_unbounded("mpsc_revalidation_queue");
 
 		let worker = RevalidationWorker::new(api.clone(), pool.clone());
@@ -321,24 +305,7 @@ where
 		api: Arc<Api>,
 		pool: Arc<Pool<Api>>,
 	) -> (Self, Pin<Box<dyn Future<Output = ()> + Send>>) {
-		Self::new_with_interval(
-			api,
-			pool,
-			intervalier::Interval::new(BACKGROUND_REVALIDATION_INTERVAL),
-		)
-	}
-
-	/// New revalidation queue with background worker and test signal.
-	#[cfg(feature = "test-helpers")]
-	pub fn new_test(
-		api: Arc<Api>,
-		pool: Arc<Pool<Api>>,
-	) -> (Self, Pin<Box<dyn Future<Output = ()> + Send>>, intervalier::BackSignalControl) {
-		let (interval, notifier) =
-			intervalier::BackSignalInterval::new(BACKGROUND_REVALIDATION_INTERVAL);
-		let (queue, background) = Self::new_with_interval(api, pool, interval);
-
-		(queue, background, notifier)
+		Self::new_with_interval(api, pool, BACKGROUND_REVALIDATION_INTERVAL)
 	}
 
 	/// Queue some transaction for later revalidation.
@@ -371,4 +338,41 @@ where
 }
 
 #[cfg(test)]
-mod tests {}
+mod tests {
+	use super::*;
+	use crate::{
+		graph::Pool,
+		tests::{uxt, TestApi},
+	};
+	use futures::executor::block_on;
+	use sc_transaction_pool_api::TransactionSource;
+	use sp_runtime::generic::BlockId;
+	use substrate_test_runtime::{AccountId, Transfer, H256};
+
+	#[test]
+	fn revalidation_queue_works() {
+		let api = Arc::new(TestApi::default());
+		let pool = Arc::new(Pool::new(Default::default(), true.into(), api.clone()));
+		let queue = Arc::new(RevalidationQueue::new(api.clone(), pool.clone()));
+
+		let uxt = uxt(Transfer {
+			from: AccountId::from_h256(H256::from_low_u64_be(1)),
+			to: AccountId::from_h256(H256::from_low_u64_be(2)),
+			amount: 5,
+			nonce: 0,
+		});
+		let uxt_hash = block_on(pool.submit_one(
+			&BlockId::number(0),
+			TransactionSource::External,
+			uxt.clone(),
+		))
+		.expect("Should be valid");
+
+		block_on(queue.revalidate_later(0, vec![uxt_hash]));
+
+		// revalidated in sync offload 2nd time
+		assert_eq!(api.validation_requests().len(), 2);
+		// number of ready
+		assert_eq!(pool.validated_pool().status().ready, 1);
+	}
+}

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

@@ -0,0 +1,185 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2018-2021 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+//! Testing related primitives for internal usage in this crate.
+
+use crate::graph::{BlockHash, ChainApi, ExtrinsicFor, NumberFor, Pool};
+use codec::Encode;
+use parking_lot::Mutex;
+use sc_transaction_pool_api::error;
+use sp_runtime::{
+	generic::BlockId,
+	traits::{Block as BlockT, Hash},
+	transaction_validity::{
+		InvalidTransaction, TransactionSource, TransactionValidity, ValidTransaction,
+	},
+};
+use std::{collections::HashSet, sync::Arc};
+use substrate_test_runtime::{Block, Extrinsic, Hashing, Transfer, H256};
+
+pub(crate) const INVALID_NONCE: u64 = 254;
+
+/// Test api that implements [`ChainApi`].
+#[derive(Clone, Debug, Default)]
+pub(crate) struct TestApi {
+	pub delay: Arc<Mutex<Option<std::sync::mpsc::Receiver<()>>>>,
+	pub invalidate: Arc<Mutex<HashSet<H256>>>,
+	pub clear_requirements: Arc<Mutex<HashSet<H256>>>,
+	pub add_requirements: Arc<Mutex<HashSet<H256>>>,
+	pub validation_requests: Arc<Mutex<Vec<Extrinsic>>>,
+}
+
+impl TestApi {
+	/// Query validation requests received.
+	pub fn validation_requests(&self) -> Vec<Extrinsic> {
+		self.validation_requests.lock().clone()
+	}
+}
+
+impl ChainApi for TestApi {
+	type Block = Block;
+	type Error = error::Error;
+	type ValidationFuture = futures::future::Ready<error::Result<TransactionValidity>>;
+	type BodyFuture = futures::future::Ready<error::Result<Option<Vec<Extrinsic>>>>;
+
+	/// Verify extrinsic at given block.
+	fn validate_transaction(
+		&self,
+		at: &BlockId<Self::Block>,
+		_source: TransactionSource,
+		uxt: ExtrinsicFor<Self>,
+	) -> Self::ValidationFuture {
+		self.validation_requests.lock().push(uxt.clone());
+		let hash = self.hash_and_length(&uxt).0;
+		let block_number = self.block_id_to_number(at).unwrap().unwrap();
+
+		let res = match uxt {
+			Extrinsic::Transfer { transfer, .. } => {
+				let nonce = transfer.nonce;
+
+				// This is used to control the test flow.
+				if nonce > 0 {
+					let opt = self.delay.lock().take();
+					if let Some(delay) = opt {
+						if delay.recv().is_err() {
+							println!("Error waiting for delay!");
+						}
+					}
+				}
+
+				if self.invalidate.lock().contains(&hash) {
+					InvalidTransaction::Custom(0).into()
+				} else if nonce < block_number {
+					InvalidTransaction::Stale.into()
+				} else {
+					let mut transaction = ValidTransaction {
+						priority: 4,
+						requires: if nonce > block_number {
+							vec![vec![nonce as u8 - 1]]
+						} else {
+							vec![]
+						},
+						provides: if nonce == INVALID_NONCE {
+							vec![]
+						} else {
+							vec![vec![nonce as u8]]
+						},
+						longevity: 3,
+						propagate: true,
+					};
+
+					if self.clear_requirements.lock().contains(&hash) {
+						transaction.requires.clear();
+					}
+
+					if self.add_requirements.lock().contains(&hash) {
+						transaction.requires.push(vec![128]);
+					}
+
+					Ok(transaction)
+				}
+			},
+			Extrinsic::IncludeData(_) => Ok(ValidTransaction {
+				priority: 9001,
+				requires: vec![],
+				provides: vec![vec![42]],
+				longevity: 9001,
+				propagate: false,
+			}),
+			Extrinsic::Store(_) => Ok(ValidTransaction {
+				priority: 9001,
+				requires: vec![],
+				provides: vec![vec![43]],
+				longevity: 9001,
+				propagate: false,
+			}),
+			_ => unimplemented!(),
+		};
+
+		futures::future::ready(Ok(res))
+	}
+
+	/// Returns a block number given the block id.
+	fn block_id_to_number(
+		&self,
+		at: &BlockId<Self::Block>,
+	) -> Result<Option<NumberFor<Self>>, Self::Error> {
+		Ok(match at {
+			BlockId::Number(num) => Some(*num),
+			BlockId::Hash(_) => None,
+		})
+	}
+
+	/// Returns a block hash given the block id.
+	fn block_id_to_hash(
+		&self,
+		at: &BlockId<Self::Block>,
+	) -> Result<Option<<Self::Block as BlockT>::Hash>, Self::Error> {
+		Ok(match at {
+			BlockId::Number(num) => Some(H256::from_low_u64_be(*num)).into(),
+			BlockId::Hash(_) => None,
+		})
+	}
+
+	/// Hash the extrinsic.
+	fn hash_and_length(&self, uxt: &ExtrinsicFor<Self>) -> (BlockHash<Self>, usize) {
+		let encoded = uxt.encode();
+		let len = encoded.len();
+		(Hashing::hash(&encoded), len)
+	}
+
+	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)
+	}
+}
+
+pub(crate) fn uxt(transfer: Transfer) -> Extrinsic {
+	let signature = TryFrom::try_from(&[0; 64][..]).unwrap();
+	Extrinsic::Transfer { transfer, signature, exhaust_resources_when_not_first: false }
+}
+
+pub(crate) fn pool() -> Pool<TestApi> {
+	Pool::new(Default::default(), true.into(), TestApi::default().into())
+}

substrate/client/transaction-pool/tests/pool.rs

@@ -17,6 +17,7 @@
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
 //! Tests for top-level transaction pool api
+
 use codec::Encode;
 use futures::{
 	executor::{block_on, block_on_stream},
@@ -25,7 +26,7 @@ use futures::{
 };
 use sc_block_builder::BlockBuilderProvider;
 use sc_client_api::client::BlockchainEvents;
-use sc_transaction_pool::{test_helpers::*, *};
+use sc_transaction_pool::*;
 use sc_transaction_pool_api::{
 	ChainEvent, MaintainedTransactionPool, TransactionPool, TransactionStatus,
 };
@@ -47,14 +48,13 @@ fn pool() -> Pool<TestApi> {
 	Pool::new(Default::default(), true.into(), TestApi::with_alice_nonce(209).into())
 }
 
-fn maintained_pool(
-) -> (BasicPool<TestApi, Block>, futures::executor::ThreadPool, intervalier::BackSignalControl) {
-	let (pool, background_task, notifier) =
-		BasicPool::new_test(Arc::new(TestApi::with_alice_nonce(209)));
+fn maintained_pool() -> (BasicPool<TestApi, Block>, Arc<TestApi>, futures::executor::ThreadPool) {
+	let api = Arc::new(TestApi::with_alice_nonce(209));
+	let (pool, background_task) = BasicPool::new_test(api.clone());
 
 	let thread_pool = futures::executor::ThreadPool::new().unwrap();
 	thread_pool.spawn_ok(background_task);
-	(pool, thread_pool, notifier)
+	(pool, api, thread_pool)
 }
 
 const SOURCE: TransactionSource = TransactionSource::External;
@@ -135,7 +135,7 @@ fn should_ban_invalid_transactions() {
 
 #[test]
 fn only_prune_on_new_best() {
-	let pool = maintained_pool().0;
+	let (pool, api, _) = maintained_pool();
 	let uxt = uxt(Alice, 209);
 
 	let _ = block_on(pool.submit_and_watch(&BlockId::number(0), SOURCE, uxt.clone()))
@@ -143,7 +143,7 @@ fn only_prune_on_new_best() {
 	pool.api().push_block(1, vec![uxt.clone()], true);
 	assert_eq!(pool.status().ready, 1);
 
-	let header = pool.api().push_block(2, vec![uxt], true);
+	let header = api.push_block(2, vec![uxt], true);
 	let event = ChainEvent::NewBestBlock { hash: header.hash(), tree_route: None };
 	block_on(pool.maintain(event));
 	assert_eq!(pool.status().ready, 0);
@@ -205,12 +205,12 @@ fn block_event_with_retracted(
 fn should_prune_old_during_maintenance() {
 	let xt = uxt(Alice, 209);
 
-	let (pool, _guard, _notifier) = maintained_pool();
+	let (pool, api, _guard) = 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![xt.clone()], true);
+	let header = api.push_block(1, vec![xt.clone()], true);
 
 	block_on(pool.maintain(block_event(header)));
 	assert_eq!(pool.status().ready, 0);
@@ -221,33 +221,38 @@ fn should_revalidate_during_maintenance() {
 	let xt1 = uxt(Alice, 209);
 	let xt2 = uxt(Alice, 210);
 
-	let (pool, _guard, mut notifier) = maintained_pool();
+	let (pool, api, _guard) = maintained_pool();
 	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("2. Imported");
+	let watcher = block_on(pool.submit_and_watch(&BlockId::number(0), SOURCE, xt2.clone()))
+		.expect("2. Imported");
 	assert_eq!(pool.status().ready, 2);
-	assert_eq!(pool.api().validation_requests().len(), 2);
+	assert_eq!(api.validation_requests().len(), 2);
 
-	let header = pool.api().push_block(1, vec![xt1.clone()], true);
+	let header = api.push_block(1, vec![xt1.clone()], true);
+
+	api.add_invalid(&xt2);
 
 	block_on(pool.maintain(block_event(header)));
 	assert_eq!(pool.status().ready, 1);
-	block_on(notifier.next());
 
 	// test that pool revalidated transaction that left ready and not included in the block
-	assert_eq!(pool.api().validation_requests().len(), 3);
+	assert_eq!(
+		futures::executor::block_on_stream(watcher).collect::<Vec<_>>(),
+		vec![TransactionStatus::Ready, TransactionStatus::Invalid],
+	);
 }
 
 #[test]
 fn should_resubmit_from_retracted_during_maintenance() {
 	let xt = uxt(Alice, 209);
 
-	let (pool, _guard, _notifier) = maintained_pool();
+	let (pool, api, _guard) = 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);
-	let fork_header = pool.api().push_block(1, vec![], false);
+	let header = api.push_block(1, vec![], true);
+	let fork_header = api.push_block(1, vec![], false);
 
 	let event = block_event_with_retracted(header, fork_header.hash(), &*pool.api());
 
@@ -259,13 +264,13 @@ fn should_resubmit_from_retracted_during_maintenance() {
 fn should_not_resubmit_from_retracted_during_maintenance_if_tx_is_also_in_enacted() {
 	let xt = uxt(Alice, 209);
 
-	let (pool, _guard, _notifier) = maintained_pool();
+	let (pool, api, _guard) = 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![xt.clone()], true);
-	let fork_header = pool.api().push_block(1, vec![xt], false);
+	let header = api.push_block(1, vec![xt.clone()], true);
+	let fork_header = api.push_block(1, vec![xt], false);
 
 	let event = block_event_with_retracted(header, fork_header.hash(), &*pool.api());
 
@@ -277,19 +282,23 @@ fn should_not_resubmit_from_retracted_during_maintenance_if_tx_is_also_in_enacte
 fn should_not_retain_invalid_hashes_from_retracted() {
 	let xt = uxt(Alice, 209);
 
-	let (pool, _guard, mut notifier) = maintained_pool();
+	let (pool, api, _guard) = maintained_pool();
 
-	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
+	let watcher = block_on(pool.submit_and_watch(&BlockId::number(0), SOURCE, xt.clone()))
+		.expect("1. Imported");
 	assert_eq!(pool.status().ready, 1);
 
-	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 header = api.push_block(1, vec![], true);
+	let fork_header = api.push_block(1, vec![xt.clone()], false);
+	api.add_invalid(&xt);
 
 	let event = block_event_with_retracted(header, fork_header.hash(), &*pool.api());
-
 	block_on(pool.maintain(event));
-	block_on(notifier.next());
+
+	assert_eq!(
+		futures::executor::block_on_stream(watcher).collect::<Vec<_>>(),
+		vec![TransactionStatus::Ready, TransactionStatus::Invalid],
+	);
 
 	assert_eq!(pool.status().ready, 0);
 }
@@ -300,26 +309,30 @@ fn should_revalidate_across_many_blocks() {
 	let xt2 = uxt(Alice, 210);
 	let xt3 = uxt(Alice, 211);
 
-	let (pool, _guard, mut notifier) = maintained_pool();
+	let (pool, api, _guard) = maintained_pool();
 
-	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt1.clone())).expect("1. Imported");
+	let watcher1 = block_on(pool.submit_and_watch(&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![], true);
+	let header = api.push_block(1, vec![], true);
 	block_on(pool.maintain(block_event(header)));
-	block_on(notifier.next());
 
 	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()], true);
-	block_on(pool.maintain(block_event(header)));
-	block_on(notifier.next());
+	let header = api.push_block(2, vec![xt1.clone()], true);
+	let block_hash = header.hash();
+	block_on(pool.maintain(block_event(header.clone())));
+
+	block_on(
+		watcher1
+			.take_while(|s| future::ready(*s != TransactionStatus::InBlock(block_hash)))
+			.collect::<Vec<_>>(),
+	);
 
 	assert_eq!(pool.status().ready, 2);
-	// xt1 and xt2 validated twice, then xt3 once, then xt2 and xt3 again
-	assert_eq!(pool.api().validation_requests().len(), 7);
 }
 
 #[test]
@@ -329,7 +342,7 @@ fn should_push_watchers_during_maintenance() {
 	}
 
 	// given
-	let (pool, _guard, mut notifier) = maintained_pool();
+	let (pool, api, _guard) = maintained_pool();
 
 	let tx0 = alice_uxt(0);
 	let watcher0 =
@@ -349,18 +362,16 @@ fn should_push_watchers_during_maintenance() {
 	assert_eq!(pool.status().ready, 5);
 
 	// when
-	pool.api().add_invalid(&tx3);
-	pool.api().add_invalid(&tx4);
+	api.add_invalid(&tx3);
+	api.add_invalid(&tx4);
 
 	// clear timer events if any
-	let header = pool.api().push_block(1, vec![], true);
+	let header = api.push_block(1, vec![], true);
 	block_on(pool.maintain(block_event(header)));
-	block_on(notifier.next());
 
 	// then
 	// hash3 is now invalid
 	// hash4 is now invalid
-	assert_eq!(pool.status().ready, 3);
 	assert_eq!(
 		futures::executor::block_on_stream(watcher3).collect::<Vec<_>>(),
 		vec![TransactionStatus::Ready, TransactionStatus::Invalid],
@@ -369,9 +380,10 @@ fn should_push_watchers_during_maintenance() {
 		futures::executor::block_on_stream(watcher4).collect::<Vec<_>>(),
 		vec![TransactionStatus::Ready, TransactionStatus::Invalid],
 	);
+	assert_eq!(pool.status().ready, 3);
 
 	// when
-	let header = pool.api().push_block(2, vec![tx0, tx1, tx2], true);
+	let header = api.push_block(2, vec![tx0, tx1, tx2], true);
 	let header_hash = header.hash();
 	block_on(pool.maintain(block_event(header)));
 
@@ -410,7 +422,7 @@ fn should_push_watchers_during_maintenance() {
 
 #[test]
 fn can_track_heap_size() {
-	let (pool, _guard, _notifier) = maintained_pool();
+	let (pool, _api, _guard) = maintained_pool();
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, uxt(Alice, 209))).expect("1. Imported");
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, uxt(Alice, 210))).expect("1. Imported");
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, uxt(Alice, 211))).expect("1. Imported");
@@ -424,7 +436,7 @@ fn finalization() {
 	let xt = uxt(Alice, 209);
 	let api = TestApi::with_alice_nonce(209);
 	api.push_block(1, vec![], true);
-	let (pool, _background, _) = BasicPool::new_test(api.into());
+	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()], true);
@@ -449,7 +461,7 @@ fn fork_aware_finalization() {
 	// starting block A1 (last finalized.)
 	api.push_block(1, vec![], true);
 
-	let (pool, _background, _) = BasicPool::new_test(api.into());
+	let (pool, _background) = BasicPool::new_test(api.into());
 	let mut canon_watchers = vec![];
 
 	let from_alice = uxt(Alice, 1);
@@ -597,7 +609,7 @@ fn prune_and_retract_tx_at_same_time() {
 	// starting block A1 (last finalized.)
 	api.push_block(1, vec![], true);
 
-	let (pool, _background, _) = BasicPool::new_test(api.into());
+	let (pool, _background) = BasicPool::new_test(api.into());
 
 	let from_alice = uxt(Alice, 1);
 	pool.api().increment_nonce(Alice.into());
@@ -663,7 +675,7 @@ fn resubmit_tx_of_fork_that_is_not_part_of_retracted() {
 	// starting block A1 (last finalized.)
 	api.push_block(1, vec![], true);
 
-	let (pool, _background, _) = BasicPool::new_test(api.into());
+	let (pool, _background) = BasicPool::new_test(api.into());
 
 	let tx0 = uxt(Alice, 1);
 	let tx1 = uxt(Dave, 2);
@@ -708,7 +720,7 @@ fn resubmit_from_retracted_fork() {
 	// starting block A1 (last finalized.)
 	api.push_block(1, vec![], true);
 
-	let (pool, _background, _) = BasicPool::new_test(api.into());
+	let (pool, _background) = BasicPool::new_test(api.into());
 
 	let tx0 = uxt(Alice, 1);
 	let tx1 = uxt(Dave, 2);
@@ -800,7 +812,7 @@ fn resubmit_from_retracted_fork() {
 
 #[test]
 fn ready_set_should_not_resolve_before_block_update() {
-	let (pool, _guard, _notifier) = maintained_pool();
+	let (pool, _api, _guard) = maintained_pool();
 	let xt1 = uxt(Alice, 209);
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt1.clone())).expect("1. Imported");
 
@@ -809,8 +821,8 @@ 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![], true);
+	let (pool, api, _guard) = maintained_pool();
+	let header = api.push_block(1, vec![], true);
 
 	let xt1 = uxt(Alice, 209);
 
@@ -822,8 +834,8 @@ 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![], true);
+	let (pool, api, _guard) = maintained_pool();
+	let header = api.push_block(1, vec![], true);
 
 	let xt1 = uxt(Alice, 209);
 
@@ -833,7 +845,7 @@ fn ready_set_should_eventually_resolve_when_block_update_arrives() {
 	let mut context = futures::task::Context::from_waker(&noop_waker);
 
 	let mut ready_set_future = pool.ready_at(1);
-	if let Poll::Ready(_) = ready_set_future.poll_unpin(&mut context) {
+	if ready_set_future.poll_unpin(&mut context).is_ready() {
 		panic!("Ready set should not be ready before block update!");
 	}
 
@@ -929,13 +941,13 @@ fn import_notification_to_pool_maintain_works() {
 // When we prune transactions, we need to make sure that we remove
 #[test]
 fn pruning_a_transaction_should_remove_it_from_best_transaction() {
-	let (pool, _guard, _notifier) = maintained_pool();
+	let (pool, api, _guard) = maintained_pool();
 
 	let xt1 = Extrinsic::IncludeData(Vec::new());
 
 	block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt1.clone())).expect("1. Imported");
 	assert_eq!(pool.status().ready, 1);
-	let header = pool.api().push_block(1, vec![xt1.clone()], true);
+	let header = api.push_block(1, vec![xt1.clone()], true);
 
 	// This will prune `xt1`.
 	block_on(pool.maintain(block_event(header)));
@@ -943,26 +955,6 @@ fn pruning_a_transaction_should_remove_it_from_best_transaction() {
 	assert_eq!(pool.status().ready, 0);
 }
 
-#[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);
-}
-
 #[test]
 fn stale_transactions_are_pruned() {
 	sp_tracing::try_init_simple();
@@ -974,7 +966,7 @@ fn stale_transactions_are_pruned() {
 		Transfer { from: Alice.into(), to: Bob.into(), nonce: 3, amount: 1 },
 	];
 
-	let (pool, _guard, _notifier) = maintained_pool();
+	let (pool, api, _guard) = maintained_pool();
 
 	xts.into_iter().for_each(|xt| {
 		block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.into_signed_tx()))
@@ -992,7 +984,7 @@ fn stale_transactions_are_pruned() {
 	];
 
 	// Import block
-	let header = pool.api().push_block(1, xts, true);
+	let header = api.push_block(1, xts, true);
 	block_on(pool.maintain(block_event(header)));
 	// The imported transactions have a different hash and should not evict our initial
 	// transactions.
@@ -1000,7 +992,7 @@ fn stale_transactions_are_pruned() {
 
 	// Import enough blocks to make our transactions stale
 	for n in 1..66 {
-		let header = pool.api().push_block(n, vec![], true);
+		let header = api.push_block(n, vec![], true);
 		block_on(pool.maintain(block_event(header)));
 	}
 

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

@@ -1,32 +0,0 @@
-use futures::executor::block_on;
-use sc_transaction_pool::test_helpers::{Pool, RevalidationQueue};
-use sc_transaction_pool_api::TransactionSource;
-use sp_runtime::generic::BlockId;
-use std::sync::Arc;
-use substrate_test_runtime_client::AccountKeyring::*;
-use substrate_test_runtime_transaction_pool::{uxt, TestApi};
-
-fn setup() -> (Arc<TestApi>, Pool<TestApi>) {
-	let test_api = Arc::new(TestApi::empty());
-	let pool = Pool::new(Default::default(), true.into(), test_api.clone());
-	(test_api, pool)
-}
-
-#[test]
-fn smoky() {
-	let (api, pool) = setup();
-	let pool = Arc::new(pool);
-	let queue = Arc::new(RevalidationQueue::new(api.clone(), pool.clone()));
-
-	let uxt = uxt(Alice, 0);
-	let uxt_hash =
-		block_on(pool.submit_one(&BlockId::number(0), TransactionSource::External, uxt.clone()))
-			.expect("Should be valid");
-
-	block_on(queue.revalidate_later(0, vec![uxt_hash]));
-
-	// revalidated in sync offload 2nd time
-	assert_eq!(api.validation_requests().len(), 2);
-	// number of ready
-	assert_eq!(pool.validated_pool().status().ready, 1);
-}

substrate/test-utils/runtime/transaction-pool/Cargo.toml

@@ -17,7 +17,7 @@ parking_lot = "0.11.2"
 codec = { package = "parity-scale-codec", version = "2.0.0" }
 sp-blockchain = { version = "4.0.0-dev", path = "../../../primitives/blockchain" }
 sp-runtime = { version = "4.0.0", path = "../../../primitives/runtime" }
-sc-transaction-pool = { version = "4.0.0-dev", path = "../../../client/transaction-pool", features = ["test-helpers"] }
+sc-transaction-pool = { version = "4.0.0-dev", path = "../../../client/transaction-pool" }
 sc-transaction-pool-api = { version = "4.0.0-dev", path = "../../../client/transaction-pool/api" }
 futures = "0.3.16"
 derive_more = "0.99.16"

substrate/test-utils/runtime/transaction-pool/src/lib.rs

@@ -25,7 +25,9 @@ use parking_lot::RwLock;
 use sp_blockchain::CachedHeaderMetadata;
 use sp_runtime::{
 	generic::{self, BlockId},
-	traits::{BlakeTwo256, Block as BlockT, Hash as HashT, Header as _, TrailingZeroInput},
+	traits::{
+		BlakeTwo256, Block as BlockT, Hash as HashT, Header as _, NumberFor, TrailingZeroInput,
+	},
 	transaction_validity::{
 		InvalidTransaction, TransactionSource, TransactionValidity, TransactionValidityError,
 		ValidTransaction,
@@ -227,7 +229,7 @@ impl TestApi {
 	}
 }
 
-impl sc_transaction_pool::test_helpers::ChainApi for TestApi {
+impl sc_transaction_pool::ChainApi for TestApi {
 	type Block = Block;
 	type Error = Error;
 	type ValidationFuture = futures::future::Ready<Result<TransactionValidity, Error>>;
@@ -237,7 +239,7 @@ impl sc_transaction_pool::test_helpers::ChainApi for TestApi {
 		&self,
 		at: &BlockId<Self::Block>,
 		_source: TransactionSource,
-		uxt: sc_transaction_pool::test_helpers::ExtrinsicFor<Self>,
+		uxt: <Self::Block as BlockT>::Extrinsic,
 	) -> Self::ValidationFuture {
 		self.validation_requests.write().push(uxt.clone());
 
@@ -297,7 +299,7 @@ impl sc_transaction_pool::test_helpers::ChainApi for TestApi {
 	fn block_id_to_number(
 		&self,
 		at: &BlockId<Self::Block>,
-	) -> Result<Option<sc_transaction_pool::test_helpers::NumberFor<Self>>, Error> {
+	) -> Result<Option<NumberFor<Self::Block>>, Error> {
 		Ok(match at {
 			generic::BlockId::Hash(x) =>
 				self.chain.read().block_by_hash.get(x).map(|b| *b.header.number()),
@@ -308,7 +310,7 @@ impl sc_transaction_pool::test_helpers::ChainApi for TestApi {
 	fn block_id_to_hash(
 		&self,
 		at: &BlockId<Self::Block>,
-	) -> Result<Option<sc_transaction_pool::test_helpers::BlockHash<Self>>, Error> {
+	) -> Result<Option<<Self::Block as BlockT>::Hash>, Error> {
 		Ok(match at {
 			generic::BlockId::Hash(x) => Some(x.clone()),
 			generic::BlockId::Number(num) =>
@@ -318,10 +320,7 @@ impl sc_transaction_pool::test_helpers::ChainApi for TestApi {
 		})
 	}
 
-	fn hash_and_length(
-		&self,
-		ex: &sc_transaction_pool::test_helpers::ExtrinsicFor<Self>,
-	) -> (Hash, usize) {
+	fn hash_and_length(&self, ex: &<Self::Block as BlockT>::Extrinsic) -> (Hash, usize) {
 		Self::hash_and_length_inner(ex)
 	}