Make transaction pool prune transactions only of canonical blocks (#6123

)

* Make tx pool aware of retracted fork blocks

* Make it compile

* Update client/transaction-pool/src/lib.rs

Co-authored-by: Nikolay Volf <nikvolf@gmail.com>

* Fix doc test

* Simplify the implementation

* Send tree route as arc to prevent heavy clones

* Switch to use `ExtrinsicHash` to make it more clear

* Fix benchmark

Co-authored-by: Nikolay Volf <nikvolf@gmail.com>

bin/node-template/node/src/service.rs

@@ -34,14 +34,22 @@ macro_rules! new_full_start {
 		let inherent_data_providers = sp_inherents::InherentDataProviders::new();
 
 		let builder = sc_service::ServiceBuilder::new_full::<
-			node_template_runtime::opaque::Block, node_template_runtime::RuntimeApi, crate::service::Executor
+			node_template_runtime::opaque::Block,
+			node_template_runtime::RuntimeApi,
+			crate::service::Executor
 		>($config)?
 			.with_select_chain(|_config, backend| {
 				Ok(sc_consensus::LongestChain::new(backend.clone()))
 			})?
-			.with_transaction_pool(|config, client, _fetcher, prometheus_registry| {
-				let pool_api = sc_transaction_pool::FullChainApi::new(client.clone());
-				Ok(sc_transaction_pool::BasicPool::new(config, std::sync::Arc::new(pool_api), prometheus_registry))
+			.with_transaction_pool(|builder| {
+				let pool_api = sc_transaction_pool::FullChainApi::new(
+					builder.client().clone(),
+				);
+				Ok(sc_transaction_pool::BasicPool::new(
+					builder.config().transaction_pool.clone(),
+					std::sync::Arc::new(pool_api),
+					builder.prometheus_registry(),
+				))
 			})?
 			.with_import_queue(|
 				_config,
@@ -199,13 +207,19 @@ pub fn new_light(config: Configuration) -> Result<impl AbstractService, ServiceE
 		.with_select_chain(|_config, backend| {
 			Ok(LongestChain::new(backend.clone()))
 		})?
-		.with_transaction_pool(|config, client, fetcher, prometheus_registry| {
-			let fetcher = fetcher
+		.with_transaction_pool(|builder| {
+			let fetcher = builder.fetcher()
 				.ok_or_else(|| "Trying to start light transaction pool without active fetcher")?;
 
-			let pool_api = sc_transaction_pool::LightChainApi::new(client.clone(), fetcher.clone());
+			let pool_api = sc_transaction_pool::LightChainApi::new(
+				builder.client().clone(),
+				fetcher.clone(),
+			);
 			let pool = sc_transaction_pool::BasicPool::with_revalidation_type(
-				config, Arc::new(pool_api), prometheus_registry, sc_transaction_pool::RevalidationType::Light,
+				builder.config().transaction_pool.clone(),
+				Arc::new(pool_api),
+				builder.prometheus_registry(),
+				sc_transaction_pool::RevalidationType::Light,
 			);
 			Ok(pool)
 		})?

bin/node/cli/src/service.rs

@@ -53,12 +53,16 @@ macro_rules! new_full_start {
 			.with_select_chain(|_config, backend| {
 				Ok(sc_consensus::LongestChain::new(backend.clone()))
 			})?
-			.with_transaction_pool(|config, client, _fetcher, prometheus_registry| {
-				let pool_api = sc_transaction_pool::FullChainApi::new(client.clone());
+			.with_transaction_pool(|builder| {
+				let pool_api = sc_transaction_pool::FullChainApi::new(
+					builder.client().clone(),
+				);
+				let config = builder.config();
+
 				Ok(sc_transaction_pool::BasicPool::new(
-					config,
+					config.transaction_pool.clone(),
 					std::sync::Arc::new(pool_api),
-					prometheus_registry,
+					builder.prometheus_registry(),
 				))
 			})?
 			.with_import_queue(|
@@ -323,12 +327,18 @@ pub fn new_light(config: Configuration)
 		.with_select_chain(|_config, backend| {
 			Ok(LongestChain::new(backend.clone()))
 		})?
-		.with_transaction_pool(|config, client, fetcher, prometheus_registry| {
-			let fetcher = fetcher
+		.with_transaction_pool(|builder| {
+			let fetcher = builder.fetcher()
 				.ok_or_else(|| "Trying to start light transaction pool without active fetcher")?;
-			let pool_api = sc_transaction_pool::LightChainApi::new(client.clone(), fetcher.clone());
+			let pool_api = sc_transaction_pool::LightChainApi::new(
+				builder.client().clone(),
+				fetcher,
+			);
 			let pool = sc_transaction_pool::BasicPool::with_revalidation_type(
-				config, Arc::new(pool_api), prometheus_registry, sc_transaction_pool::RevalidationType::Light,
+				builder.config().transaction_pool.clone(),
+				Arc::new(pool_api),
+				builder.prometheus_registry(),
+				sc_transaction_pool::RevalidationType::Light,
 			);
 			Ok(pool)
 		})?
@@ -481,7 +491,7 @@ mod tests {
 						ChainEvent::NewBlock {
 							is_new_best: true,
 							id: parent_id.clone(),
-							retracted: vec![],
+							tree_route: None,
 							header: parent_header.clone(),
 						},
 					)

client/api/src/backend.rs

@@ -67,8 +67,10 @@ pub struct ImportSummary<Block: BlockT> {
 	pub is_new_best: bool,
 	/// Optional storage changes.
 	pub storage_changes: Option<(StorageCollection, ChildStorageCollection)>,
-	/// Blocks that got retracted because of this one got imported.
-	pub retracted: Vec<Block::Hash>,
+	/// Tree route from old best to new best.
+	///
+	/// If `None`, there was no re-org while importing.
+	pub tree_route: Option<sp_blockchain::TreeRoute<Block>>,
 }
 
 /// Import operation wrapper

client/api/src/client.rs

@@ -16,7 +16,7 @@
 
 //! A set of APIs supported by the client along with their primitives.
 
-use std::{fmt, collections::HashSet};
+use std::{fmt, collections::HashSet, sync::Arc};
 use sp_core::storage::StorageKey;
 use sp_runtime::{
 	traits::{Block as BlockT, NumberFor},
@@ -234,8 +234,10 @@ pub struct BlockImportNotification<Block: BlockT> {
 	pub header: Block::Header,
 	/// Is this the new best block.
 	pub is_new_best: bool,
-	/// List of retracted blocks ordered by block number.
-	pub retracted: Vec<Block::Hash>,
+	/// Tree route from old best to new best.
+	///
+	/// If `None`, there was no re-org while importing.
+	pub tree_route: Option<Arc<sp_blockchain::TreeRoute<Block>>>,
 }
 
 /// Summary of a finalized block.

client/basic-authorship/src/basic_authorship.rs

@@ -331,15 +331,14 @@ mod tests {
 	use parking_lot::Mutex;
 	use sp_consensus::{BlockOrigin, Proposer};
 	use substrate_test_runtime_client::{
-		prelude::*,
-		runtime::{Extrinsic, Transfer},
+		prelude::*, TestClientBuilder, runtime::{Extrinsic, Transfer}, TestClientBuilderExt,
 	};
 	use sp_transaction_pool::{ChainEvent, MaintainedTransactionPool, TransactionSource};
 	use sc_transaction_pool::{BasicPool, FullChainApi};
 	use sp_api::Core;
-	use backend::Backend;
 	use sp_blockchain::HeaderBackend;
 	use sp_runtime::traits::NumberFor;
+	use sc_client_api::Backend;
 
 	const SOURCE: TransactionSource = TransactionSource::External;
 
@@ -357,7 +356,7 @@ mod tests {
 	{
 		ChainEvent::NewBlock {
 			id: BlockId::Number(block_number.into()),
-			retracted: vec![],
+			tree_route: None,
 			is_new_best: true,
 			header,
 		}
@@ -452,8 +451,7 @@ mod tests {
 
 	#[test]
 	fn proposed_storage_changes_should_match_execute_block_storage_changes() {
-		let (client, backend) = substrate_test_runtime_client::TestClientBuilder::new()
-			.build_with_backend();
+		let (client, backend) = TestClientBuilder::new().build_with_backend();
 		let client = Arc::new(client);
 		let txpool = Arc::new(
 			BasicPool::new(
@@ -473,7 +471,9 @@ mod tests {
 		futures::executor::block_on(
 			txpool.maintain(chain_event(
 				0,
-				client.header(&BlockId::Number(0u64)).expect("header get error").expect("there should be header")
+				client.header(&BlockId::Number(0u64))
+					.expect("header get error")
+					.expect("there should be header"),
 			))
 		);
 
@@ -500,8 +500,11 @@ mod tests {
 			backend.changes_trie_storage(),
 		).unwrap();
 
-		let storage_changes = api.into_storage_changes(&state, changes_trie_state.as_ref(), genesis_hash)
-			.unwrap();
+		let storage_changes = api.into_storage_changes(
+			&state,
+			changes_trie_state.as_ref(),
+			genesis_hash,
+		).unwrap();
 
 		assert_eq!(
 			proposal.storage_changes.transaction_storage_root,

client/basic-authorship/src/lib.rs

@@ -25,7 +25,10 @@
 //! # use sp_consensus::{Environment, Proposer, RecordProof};
 //! # use sp_runtime::generic::BlockId;
 //! # use std::{sync::Arc, time::Duration};
-//! # use substrate_test_runtime_client::{self, runtime::{Extrinsic, Transfer}, AccountKeyring};
+//! # use substrate_test_runtime_client::{
+//! #     runtime::{Extrinsic, Transfer}, AccountKeyring,
+//! #     DefaultTestClientBuilderExt, TestClientBuilderExt,
+//! # };
 //! # use sc_transaction_pool::{BasicPool, FullChainApi};
 //! # let client = Arc::new(substrate_test_runtime_client::new());
 //! # let txpool = Arc::new(BasicPool::new(Default::default(), Arc::new(FullChainApi::new(client.clone())), None).0);

client/consensus/manual-seal/src/lib.rs

@@ -98,7 +98,7 @@ pub async fn run_manual_seal<B, CB, E, C, A, SC, S, T>(
 	inherent_data_providers: InherentDataProviders,
 )
 	where
-		A: txpool::ChainApi<Block=B, Hash=<B as BlockT>::Hash> + 'static,
+		A: txpool::ChainApi<Block=B> + 'static,
 		B: BlockT + 'static,
 		C: HeaderBackend<B> + Finalizer<B, CB> + 'static,
 		CB: ClientBackend<B> + 'static,
@@ -158,7 +158,7 @@ pub async fn run_instant_seal<B, CB, E, C, A, SC, T>(
 	inherent_data_providers: InherentDataProviders,
 )
 	where
-		A: txpool::ChainApi<Block=B, Hash=<B as BlockT>::Hash> + 'static,
+		A: txpool::ChainApi<Block=B> + 'static,
 		B: BlockT + 'static,
 		C: HeaderBackend<B> + Finalizer<B, CB> + 'static,
 		CB: ClientBackend<B> + 'static,
@@ -417,7 +417,7 @@ mod tests {
 			id: BlockId::Number(1),
 			header: client.header(&BlockId::Number(1)).expect("db error").expect("imported above"),
 			is_new_best: true,
-			retracted: vec![],
+			tree_route: None,
 		}).await;
 
 		let (tx1, rx1) = futures::channel::oneshot::channel();

client/consensus/manual-seal/src/seal_new_block.rs

@@ -87,7 +87,7 @@ pub async fn seal_new_block<B, SC, HB, E, T, P>(
 		E: Environment<B>,
 		<E as Environment<B>>::Error: std::fmt::Display,
 		<E::Proposer as Proposer<B>>::Error: std::fmt::Display,
-		P: txpool::ChainApi<Block=B, Hash=<B as BlockT>::Hash>,
+		P: txpool::ChainApi<Block=B>,
 		SC: SelectChain<B>,
 {
 	let future = async {

client/finality-grandpa/src/until_imported.rs

@@ -585,7 +585,7 @@ mod tests {
 				origin: BlockOrigin::File,
 				header,
 				is_new_best: false,
-				retracted: vec![],
+				tree_route: None,
 			}).unwrap();
 		}
 	}

client/offchain/src/lib.rs

@@ -166,7 +166,7 @@ mod tests {
 	use super::*;
 	use std::sync::Arc;
 	use sc_network::{Multiaddr, PeerId};
-	use substrate_test_runtime_client::runtime::Block;
+	use substrate_test_runtime_client::{TestClient, runtime::Block};
 	use sc_transaction_pool::{BasicPool, FullChainApi};
 	use sp_transaction_pool::{TransactionPool, InPoolTransaction};
 	use sc_client_api::ExecutorProvider;
@@ -183,7 +183,9 @@ mod tests {
 		}
 	}
 
-	struct TestPool(BasicPool<FullChainApi<substrate_test_runtime_client::TestClient, Block>, Block>);
+	struct TestPool(
+		BasicPool<FullChainApi<TestClient, Block>, Block>
+	);
 
 	impl sp_transaction_pool::OffchainSubmitTransaction<Block> for TestPool {
 		fn submit_at(
@@ -200,8 +202,8 @@ mod tests {
 
 	#[test]
 	fn should_call_into_runtime_and_produce_extrinsic() {
-		// given
 		let _ = env_logger::try_init();
+
 		let client = Arc::new(substrate_test_runtime_client::new());
 		let pool = Arc::new(TestPool(BasicPool::new(
 			Default::default(),

client/rpc/src/author/tests.rs

@@ -58,11 +58,9 @@ struct TestSetup {
 impl Default for TestSetup {
 	fn default() -> Self {
 		let keystore = KeyStore::new();
-		let client = Arc::new(
-			substrate_test_runtime_client::TestClientBuilder::new()
-				.set_keystore(keystore.clone())
-				.build()
-		);
+		let client_builder = substrate_test_runtime_client::TestClientBuilder::new();
+		let client = Arc::new(client_builder.set_keystore(keystore.clone()).build());
+
 		let pool = Arc::new(BasicPool::new(
 			Default::default(),
 			Arc::new(FullChainApi::new(client.clone())),