Add transaction pool to Aura and Babe import queue (#3225)

* Add transaction pool to babe import queue

* Add transaction pool to Babe check header

* Fix tests

* Add tx pool to Aura import_queue

* Fix tests, node-template

* Add comments regarding unused _transaction_pool

* Make tx pool optional in check_header

substrate/core/consensus/aura/src/lib.rs

@@ -390,18 +390,21 @@ fn find_pre_digest<B: BlockT, P: Pair>(header: &B::Header) -> Result<u64, String
 ///
 /// This digest item will always return `Some` when used with `as_aura_seal`.
 //
-// FIXME #1018 needs misbehavior types
-fn check_header<C, B: BlockT, P: Pair>(
+// FIXME #1018 needs misbehavior types. The `transaction_pool` parameter will be 
+// used to submit such misbehavior reports.
+fn check_header<C, B: BlockT, P: Pair, T>(
 	client: &C,
 	slot_now: u64,
 	mut header: B::Header,
 	hash: B::Hash,
 	authorities: &[AuthorityId<P>],
+	_transaction_pool: Option<&T>,
 ) -> Result<CheckedHeader<B::Header, (u64, DigestItemFor<B>)>, String> where
 	DigestItemFor<B>: CompatibleDigestItem<P>,
 	P::Signature: Decode,
 	C: client::backend::AuxStore,
 	P::Public: Encode + Decode + PartialEq + Clone,
+	T: Send + Sync + 'static,
 {
 	let seal = match header.digest_mut().pop() {
 		Some(x) => x,
@@ -451,13 +454,14 @@ fn check_header<C, B: BlockT, P: Pair>(
 }
 
 /// A verifier for Aura blocks.
-pub struct AuraVerifier<C, P> {
+pub struct AuraVerifier<C, P, T> {
 	client: Arc<C>,
 	phantom: PhantomData<P>,
 	inherent_data_providers: inherents::InherentDataProviders,
+	transaction_pool: Option<Arc<T>>,
 }
 
-impl<C, P> AuraVerifier<C, P>
+impl<C, P, T> AuraVerifier<C, P, T>
 	where P: Send + Sync + 'static
 {
 	fn check_inherents<B: BlockT>(
@@ -510,13 +514,14 @@ impl<C, P> AuraVerifier<C, P>
 }
 
 #[forbid(deprecated)]
-impl<B: BlockT, C, P> Verifier<B> for AuraVerifier<C, P> where
+impl<B: BlockT, C, P, T> Verifier<B> for AuraVerifier<C, P, T> where
 	C: ProvideRuntimeApi + Send + Sync + client::backend::AuxStore + ProvideCache<B> + BlockOf,
 	C::Api: BlockBuilderApi<B> + AuraApi<B, AuthorityId<P>>,
 	DigestItemFor<B>: CompatibleDigestItem<P>,
 	P: Pair + Send + Sync + 'static,
 	P::Public: Send + Sync + Hash + Eq + Clone + Decode + Encode + Debug + 'static,
 	P::Signature: Encode + Decode,
+	T: Send + Sync + 'static,
 {
 	fn verify(
 		&mut self,
@@ -536,12 +541,13 @@ impl<B: BlockT, C, P> Verifier<B> for AuraVerifier<C, P> where
 		// we add one to allow for some small drift.
 		// FIXME #1019 in the future, alter this queue to allow deferring of
 		// headers
-		let checked_header = check_header::<C, B, P>(
+		let checked_header = check_header::<C, B, P, T>(
 			&self.client,
 			slot_now + 1,
 			header,
 			hash,
 			&authorities[..],
+			self.transaction_pool.as_ref().map(|x| &**x),
 		)?;
 		match checked_header {
 			CheckedHeader::Checked(pre_header, (slot_num, seal)) => {
@@ -680,13 +686,14 @@ fn register_aura_inherent_data_provider(
 }
 
 /// Start an import queue for the Aura consensus algorithm.
-pub fn import_queue<B, C, P>(
+pub fn import_queue<B, C, P, T>(
 	slot_duration: SlotDuration,
 	block_import: BoxBlockImport<B>,
 	justification_import: Option<BoxJustificationImport<B>>,
 	finality_proof_import: Option<BoxFinalityProofImport<B>>,
 	client: Arc<C>,
 	inherent_data_providers: InherentDataProviders,
+	transaction_pool: Option<Arc<T>>,
 ) -> Result<AuraImportQueue<B>, consensus_common::Error> where
 	B: BlockT,
 	C: 'static + ProvideRuntimeApi + BlockOf + ProvideCache<B> + Send + Sync + AuxStore,
@@ -695,6 +702,7 @@ pub fn import_queue<B, C, P>(
 	P: Pair + Send + Sync + 'static,
 	P::Public: Clone + Eq + Send + Sync + Hash + Debug + Encode + Decode,
 	P::Signature: Encode + Decode,
+	T: Send + Sync + 'static,
 {
 	register_aura_inherent_data_provider(&inherent_data_providers, slot_duration.get())?;
 	initialize_authorities_cache(&*client)?;
@@ -703,6 +711,7 @@ pub fn import_queue<B, C, P>(
 		client: client.clone(),
 		inherent_data_providers,
 		phantom: PhantomData,
+		transaction_pool,
 	};
 	Ok(BasicQueue::new(
 		verifier,
@@ -773,7 +782,7 @@ mod tests {
 
 	impl TestNetFactory for AuraTestNet {
 		type Specialization = DummySpecialization;
-		type Verifier = AuraVerifier<PeersFullClient, AuthorityPair>;
+		type Verifier = AuraVerifier<PeersFullClient, AuthorityPair, ()>;
 		type PeerData = ();
 
 		/// Create new test network with peers and given config.
@@ -800,6 +809,7 @@ mod tests {
 					AuraVerifier {
 						client,
 						inherent_data_providers,
+						transaction_pool: Default::default(),
 						phantom: Default::default(),
 					}
 				},

substrate/core/consensus/babe/src/lib.rs

@@ -463,8 +463,9 @@ fn find_next_epoch_digest<B: BlockT>(header: &B::Header) -> Result<Option<Epoch>
 /// unsigned.  This is required for security and must not be changed.
 ///
 /// This digest item will always return `Some` when used with `as_babe_pre_digest`.
-// FIXME #1018 needs misbehavior types
-fn check_header<B: BlockT + Sized, C: AuxStore>(
+// FIXME #1018 needs misbehavior types. The `transaction_pool` parameter will be 
+// used to submit such misbehavior reports.
+fn check_header<B: BlockT + Sized, C: AuxStore, T>(
 	client: &C,
 	slot_now: u64,
 	mut header: B::Header,
@@ -473,8 +474,10 @@ fn check_header<B: BlockT + Sized, C: AuxStore>(
 	randomness: [u8; 32],
 	epoch_index: u64,
 	c: (u64, u64),
-) -> Result<CheckedHeader<B::Header, (DigestItemFor<B>, DigestItemFor<B>)>, String>
-	where DigestItemFor<B>: CompatibleDigestItem,
+	_transaction_pool: Option<&T>,
+) -> Result<CheckedHeader<B::Header, (DigestItemFor<B>, DigestItemFor<B>)>, String> where
+	DigestItemFor<B>: CompatibleDigestItem,
+	T: Send + Sync + 'static,
 {
 	trace!(target: "babe", "Checking header");
 	let seal = match header.digest_mut().pop() {
@@ -548,14 +551,15 @@ fn check_header<B: BlockT + Sized, C: AuxStore>(
 pub struct BabeLink(Arc<Mutex<(Option<Duration>, Vec<(Instant, u64)>)>>);
 
 /// A verifier for Babe blocks.
-pub struct BabeVerifier<C> {
+pub struct BabeVerifier<C, T> {
 	api: Arc<C>,
 	inherent_data_providers: inherents::InherentDataProviders,
 	config: Config,
 	time_source: BabeLink,
+	transaction_pool: Option<Arc<T>>,
 }
 
-impl<C> BabeVerifier<C> {
+impl<C, T> BabeVerifier<C, T> {
 	fn check_inherents<B: BlockT>(
 		&self,
 		block: B,
@@ -625,9 +629,10 @@ fn median_algorithm(
 	}
 }
 
-impl<B: BlockT, C> Verifier<B> for BabeVerifier<C> where
+impl<B: BlockT, C, T> Verifier<B> for BabeVerifier<C, T> where
 	C: ProvideRuntimeApi + Send + Sync + AuxStore + ProvideCache<B>,
 	C::Api: BlockBuilderApi<B> + BabeApi<B>,
+	T: Send + Sync + 'static,
 {
 	fn verify(
 		&mut self,
@@ -662,7 +667,7 @@ impl<B: BlockT, C> Verifier<B> for BabeVerifier<C> where
 
 		// We add one to allow for some small drift.
 		// FIXME #1019 in the future, alter this queue to allow deferring of headers
-		let checked_header = check_header::<B, C>(
+		let checked_header = check_header::<B, C, T>(
 			&self.api,
 			slot_now + 1,
 			header,
@@ -671,6 +676,7 @@ impl<B: BlockT, C> Verifier<B> for BabeVerifier<C> where
 			randomness,
 			epoch_index,
 			self.config.c(),
+			self.transaction_pool.as_ref().map(|x| &**x),
 		)?;
 
 		match checked_header {
@@ -1129,7 +1135,7 @@ impl<B, E, Block, I, RA, PRA> BlockImport<Block> for BabeBlockImport<B, E, Block
 /// authoring when importing its own blocks, and a future that must be run to
 /// completion and is responsible for listening to finality notifications and
 /// pruning the epoch changes tree.
-pub fn import_queue<B, E, Block: BlockT<Hash=H256>, I, RA, PRA>(
+pub fn import_queue<B, E, Block: BlockT<Hash=H256>, I, RA, PRA, T>(
 	config: Config,
 	block_import: I,
 	justification_import: Option<BoxJustificationImport<Block>>,
@@ -1137,6 +1143,7 @@ pub fn import_queue<B, E, Block: BlockT<Hash=H256>, I, RA, PRA>(
 	client: Arc<Client<B, E, Block, RA>>,
 	api: Arc<PRA>,
 	inherent_data_providers: InherentDataProviders,
+	transaction_pool: Option<Arc<T>>,
 ) -> ClientResult<(
 	BabeImportQueue<Block>,
 	BabeLink,
@@ -1150,6 +1157,7 @@ pub fn import_queue<B, E, Block: BlockT<Hash=H256>, I, RA, PRA>(
 	RA: Send + Sync + 'static,
 	PRA: ProvideRuntimeApi + ProvideCache<Block> + Send + Sync + AuxStore + 'static,
 	PRA::Api: BlockBuilderApi<Block> + BabeApi<Block>,
+	T: Send + Sync + 'static,
 {
 	register_babe_inherent_data_provider(&inherent_data_providers, config.get())?;
 	initialize_authorities_cache(&*api)?;
@@ -1159,6 +1167,7 @@ pub fn import_queue<B, E, Block: BlockT<Hash=H256>, I, RA, PRA>(
 		inherent_data_providers,
 		time_source: Default::default(),
 		config,
+		transaction_pool,
 	};
 
 	#[allow(deprecated)]

substrate/core/consensus/babe/src/tests.rs

@@ -88,7 +88,7 @@ type TestHeader = <TestBlock as BlockT>::Header;
 type TestExtrinsic = <TestBlock as BlockT>::Extrinsic;
 
 pub struct TestVerifier {
-	inner: BabeVerifier<PeersFullClient>,
+	inner: BabeVerifier<PeersFullClient, ()>,
 	mutator: Mutator,
 }
 
@@ -143,6 +143,7 @@ impl TestNetFactory for BabeTestNet {
 				inherent_data_providers,
 				config,
 				time_source: Default::default(),
+				transaction_pool : Default::default(),
 			},
 			mutator: MUTATOR.with(|s| s.borrow().clone()),
 		}

substrate/core/service/src/chain_ops.rs

@@ -146,7 +146,8 @@ pub fn import_blocks<F, E, R>(
 	let (mut queue, _) = components::FullComponents::<F>::build_import_queue(
 		&mut config,
 		client.clone(),
-		select_chain
+		select_chain,
+		None,
 	)?;
 
 	let (exit_send, exit_recv) = std::sync::mpsc::channel();

substrate/core/service/src/components.rs

@@ -381,6 +381,7 @@ pub trait ServiceFactory: 'static + Sized {
 		config: &mut FactoryFullConfiguration<Self>,
 		_client: Arc<FullClient<Self>>,
 		_select_chain: Self::SelectChain,
+		_transaction_pool: Option<Arc<TransactionPool<Self::FullTransactionPoolApi>>>,
 	) -> Result<Self::FullImportQueue, error::Error> {
 		if let Some(name) = config.chain_spec.consensus_engine() {
 			match name {
@@ -454,6 +455,7 @@ pub trait Components: Sized + 'static {
 		config: &mut FactoryFullConfiguration<Self::Factory>,
 		client: Arc<ComponentClient<Self>>,
 		select_chain: Option<Self::SelectChain>,
+		_transaction_pool: Option<Arc<TransactionPool<Self::TransactionPoolApi>>>,
 	) -> Result<(Self::ImportQueue, Option<BoxFinalityProofRequestBuilder<FactoryBlock<Self::Factory>>>), error::Error>;
 
 	/// Finality proof provider for serving network requests.
@@ -572,10 +574,11 @@ impl<Factory: ServiceFactory> Components for FullComponents<Factory> {
 		config: &mut FactoryFullConfiguration<Self::Factory>,
 		client: Arc<ComponentClient<Self>>,
 		select_chain: Option<Self::SelectChain>,
+		transaction_pool: Option<Arc<TransactionPool<Self::TransactionPoolApi>>>,
 	) -> Result<(Self::ImportQueue, Option<BoxFinalityProofRequestBuilder<FactoryBlock<Self::Factory>>>), error::Error> {
 		let select_chain = select_chain
 			.ok_or(error::Error::SelectChainRequired)?;
-		Factory::build_full_import_queue(config, client, select_chain)
+		Factory::build_full_import_queue(config, client, select_chain, transaction_pool)
 			.map(|queue| (queue, None))
 	}
 
@@ -695,6 +698,7 @@ impl<Factory: ServiceFactory> Components for LightComponents<Factory> {
 		config: &mut FactoryFullConfiguration<Self::Factory>,
 		client: Arc<ComponentClient<Self>>,
 		_select_chain: Option<Self::SelectChain>,
+		_transaction_pool: Option<Arc<TransactionPool<Self::TransactionPoolApi>>>,
 	) -> Result<(Self::ImportQueue, Option<BoxFinalityProofRequestBuilder<FactoryBlock<Self::Factory>>>), error::Error> {
 		Factory::build_light_import_queue(config, client)
 			.map(|(queue, builder)| (queue, Some(builder)))

substrate/core/service/src/lib.rs

@@ -173,10 +173,21 @@ impl<Components: components::Components> Service<Components> {
 
 		let (client, on_demand) = Components::build_client(&config, executor, Some(keystore.clone()))?;
 		let select_chain = Components::build_select_chain(&mut config, client.clone())?;
+
+		let transaction_pool = Arc::new(
+			Components::build_transaction_pool(config.transaction_pool.clone(), client.clone())?
+		);
+		let transaction_pool_adapter = Arc::new(TransactionPoolAdapter {
+			imports_external_transactions: !config.roles.is_light(),
+			pool: transaction_pool.clone(),
+			client: client.clone(),
+		});
+
 		let (import_queue, finality_proof_request_builder) = Components::build_import_queue(
 			&mut config,
 			client.clone(),
 			select_chain.clone(),
+			Some(transaction_pool.clone()),
 		)?;
 		let import_queue = Box::new(import_queue);
 		let finality_proof_provider = Components::build_finality_proof_provider(client.clone())?;
@@ -197,14 +208,6 @@ impl<Components: components::Components> Service<Components> {
 		);
 
 		let network_protocol = <Components::Factory>::build_network_protocol(&config)?;
-		let transaction_pool = Arc::new(
-			Components::build_transaction_pool(config.transaction_pool.clone(), client.clone())?
-		);
-		let transaction_pool_adapter = Arc::new(TransactionPoolAdapter {
-			imports_external_transactions: !config.roles.is_light(),
-			pool: transaction_pool.clone(),
-			client: client.clone(),
-		});
 
 		let protocol_id = {
 			let protocol_id_full = match config.chain_spec.protocol_id() {
@@ -971,7 +974,7 @@ where
 /// 		LightService = LightComponents<Self>
 /// 			{ |config| <LightComponents<Factory>>::new(config) },
 /// 		FullImportQueue = BasicQueue<Block>
-/// 			{ |_, client, _| Ok(BasicQueue::new(MyVerifier, Box::new(client), None, None)) },
+/// 			{ |_, client, _, _| Ok(BasicQueue::new(MyVerifier, Box::new(client), None, None)) },
 /// 		LightImportQueue = BasicQueue<Block>
 /// 			{ |_, client| {
 /// 				let fprb = Box::new(DummyFinalityProofRequestBuilder::default()) as Box<_>;
@@ -1064,9 +1067,10 @@ macro_rules! construct_service_factory {
 			fn build_full_import_queue(
 				config: &mut $crate::FactoryFullConfiguration<Self>,
 				client: $crate::Arc<$crate::FullClient<Self>>,
-				select_chain: Self::SelectChain
+				select_chain: Self::SelectChain,
+				transaction_pool: Option<Arc<$crate::TransactionPool<Self::FullTransactionPoolApi>>>,
 			) -> $crate::Result<Self::FullImportQueue, $crate::Error> {
-				( $( $full_import_queue_init )* ) (config, client, select_chain)
+				( $( $full_import_queue_init )* ) (config, client, select_chain, transaction_pool)
 			}
 
 			fn build_light_import_queue(

substrate/node-template/src/service.rs

@@ -95,14 +95,20 @@ construct_service_factory! {
 		FullImportQueue = AuraImportQueue<
 			Self::Block,
 		>
-			{ |config: &mut FactoryFullConfiguration<Self> , client: Arc<FullClient<Self>>, _select_chain: Self::SelectChain| {
-					import_queue::<_, _, aura_primitives::sr25519::AuthorityPair>(
+			{ |
+				config: &mut FactoryFullConfiguration<Self>,
+				client: Arc<FullClient<Self>>,
+				_select_chain: Self::SelectChain,
+				transaction_pool: Option<Arc<TransactionPool<Self::FullTransactionPoolApi>>>,
+			| {
+					import_queue::<_, _, aura_primitives::sr25519::AuthorityPair, _>(
 						SlotDuration::get_or_compute(&*client)?,
 						Box::new(client.clone()),
 						None,
 						None,
 						client,
 						config.custom.inherent_data_providers.clone(),
+						transaction_pool,
 					).map_err(Into::into)
 				}
 			},
@@ -111,13 +117,14 @@ construct_service_factory! {
 		>
 			{ |config: &mut FactoryFullConfiguration<Self>, client: Arc<LightClient<Self>>| {
 					let fprb = Box::new(DummyFinalityProofRequestBuilder::default()) as Box<_>;
-					import_queue::<_, _, AuraAuthorityPair>(
+					import_queue::<_, _, AuraAuthorityPair, TransactionPool<Self::FullTransactionPoolApi>>(
 						SlotDuration::get_or_compute(&*client)?,
 						Box::new(client.clone()),
 						None,
 						None,
 						client,
 						config.custom.inherent_data_providers.clone(),
+						None,
 					).map(|q| (q, fprb)).map_err(Into::into)
 				}
 			},

substrate/node/cli/src/service.rs

@@ -209,12 +209,15 @@ construct_service_factory! {
 		},
 		LightService = LightComponents<Self>
 			{ |config| <LightComponents<Factory>>::new(config) },
-		FullImportQueue = BabeImportQueue<Self::Block> {
-			|
-				config: &mut FactoryFullConfiguration<Self>,
-				client: Arc<FullClient<Self>>,
-				select_chain: Self::SelectChain
-			| {
+		FullImportQueue = BabeImportQueue<Self::Block>
+			{
+				|
+					config: &mut FactoryFullConfiguration<Self>,
+					client: Arc<FullClient<Self>>,
+					select_chain: Self::SelectChain,
+					transaction_pool: Option<Arc<TransactionPool<Self::FullTransactionPoolApi>>>,
+				|
+			{
 				let (block_import, link_half) =
 					grandpa::block_import::<_, _, _, RuntimeApi, FullClient<Self>, _>(
 						client.clone(), client.clone(), select_chain
@@ -229,6 +232,7 @@ construct_service_factory! {
 					client.clone(),
 					client,
 					config.custom.inherent_data_providers.clone(),
+					transaction_pool,
 				)?;
 
 				config.custom.import_setup = Some((babe_block_import.clone(), link_half, babe_link));
@@ -252,7 +256,7 @@ construct_service_factory! {
 					finality_proof_import.create_finality_proof_request_builder();
 
 				// FIXME: pruning task isn't started since light client doesn't do `AuthoritySetup`.
-				let (import_queue, ..) = import_queue(
+				let (import_queue, ..) = import_queue::<_, _, _, _, _, _, TransactionPool<Self::FullTransactionPoolApi>>(
 					Config::get_or_compute(&*client)?,
 					block_import,
 					None,
@@ -260,6 +264,7 @@ construct_service_factory! {
 					client.clone(),
 					client,
 					config.custom.inherent_data_providers.clone(),
+					None,
 				)?;
 
 				Ok((import_queue, finality_proof_request_builder))