Produce block always on updated transaction pool state

bin/node/cli/src/service.rs

@@ -397,6 +397,7 @@ mod tests {
 	use sc_service::AbstractService;
 	use crate::service::{new_full, new_light};
 	use sp_runtime::traits::IdentifyAccount;
+	use sp_transaction_pool::{MaintainedTransactionPool, ChainEvent};
 
 	type AccountPublic = <Signature as Verify>::Signer;
 
@@ -414,7 +415,21 @@ mod tests {
 		let dummy_runtime = ::tokio::runtime::Runtime::new().unwrap();
 		let block_factory = |service: &<Factory as service::ServiceFactory>::FullService| {
 			let block_id = BlockId::number(service.client().chain_info().best_number);
-			let parent_header = service.client().header(&block_id).unwrap().unwrap();
+			let parent_header = service.client().best_header(&block_id)
+				.expect("db error")
+				.expect("best block should exist");
+
+			futures::executor::block_on(
+				service.transaction_pool().maintain(
+					ChainEvent::NewBlock {
+						is_new_best: true,
+						id: block_id.clone(),
+						retracted: vec![],
+						header: parent_header,
+					},
+				)
+			);
+
 			let consensus_net = ConsensusNetwork::new(service.network(), service.client().clone());
 			let proposer_factory = consensus::ProposerFactory {
 				client: service.client().clone(),
@@ -464,6 +479,8 @@ mod tests {
 	}
 
 	#[test]
+	// It is "ignored", but the node-cli ignored tests are running on the CI.
+	// This can be run locally with `cargo test --release -p node-cli test_sync -- --ignored`.
 	#[ignore]
 	fn test_sync() {
 		let keystore_path = tempfile::tempdir().expect("Creates keystore path");
@@ -504,6 +521,18 @@ mod tests {
 				let parent_header = service.client().header(&parent_id).unwrap().unwrap();
 				let parent_hash = parent_header.hash();
 				let parent_number = *parent_header.number();
+
+				futures::executor::block_on(
+					service.transaction_pool().maintain(
+						ChainEvent::NewBlock {
+							is_new_best: true,
+							id: parent_id.clone(),
+							retracted: vec![],
+							header: parent_header.clone(),
+						},
+					)
+				);
+
 				let mut proposer_factory = sc_basic_authorship::ProposerFactory::new(
 					service.client(),
 					service.transaction_pool()

client/basic-authorship/Cargo.toml

@@ -23,6 +23,7 @@ sc-telemetry = { version = "2.0.0-alpha.2", path = "../telemetry" }
 sp-transaction-pool = { version = "2.0.0-alpha.2", path = "../../primitives/transaction-pool" }
 sc-block-builder = { version = "0.8.0-alpha.2", path = "../block-builder" }
 tokio-executor = { version = "0.2.0-alpha.6", features = ["blocking"] }
+futures-timer = "3.0.1"
 
 [dev-dependencies]
 sc-transaction-pool = { version = "2.0.0-alpha.2", path = "../../client/transaction-pool" }

client/basic-authorship/src/basic_authorship.rs

@@ -33,7 +33,7 @@ use sp_transaction_pool::{TransactionPool, InPoolTransaction};
 use sc_telemetry::{telemetry, CONSENSUS_INFO};
 use sc_block_builder::{BlockBuilderApi, BlockBuilderProvider};
 use sp_api::{ProvideRuntimeApi, ApiExt};
-use futures::prelude::*;
+use futures::{executor, future, future::Either};
 use sp_blockchain::{HeaderBackend, ApplyExtrinsicFailed};
 use std::marker::PhantomData;
 
@@ -210,7 +210,18 @@ impl<A, B, Block, C> ProposerInner<B, Block, C, A>
 		let mut is_first = true;
 		let mut skipped = 0;
 		let mut unqueue_invalid = Vec::new();
-		let pending_iterator = self.transaction_pool.ready();
+		let pending_iterator = match executor::block_on(future::select(
+			self.transaction_pool.ready_at(self.parent_number),
+			futures_timer::Delay::new((deadline - (self.now)()) / 8),
+		)) {
+			Either::Left((iterator, _)) => iterator,
+			Either::Right(_) => {
+				log::warn!(
+					"Timeout fired waiting for transaction pool to be ready. Proceeding to block production anyway.",
+				);
+				self.transaction_pool.ready()
+			}
+		};
 
 		debug!("Attempting to push transactions from the pool.");
 		debug!("Pool status: {:?}", self.transaction_pool.status());
@@ -304,10 +315,12 @@ mod tests {
 		prelude::*,
 		runtime::{Extrinsic, Transfer},
 	};
+	use sp_transaction_pool::{ChainEvent, MaintainedTransactionPool};
 	use sc_transaction_pool::{BasicPool, FullChainApi};
 	use sp_api::Core;
 	use backend::Backend;
 	use sp_blockchain::HeaderBackend;
+	use sp_runtime::traits::NumberFor;
 
 	fn extrinsic(nonce: u64) -> Extrinsic {
 		Transfer {
@@ -318,6 +331,17 @@ mod tests {
 		}.into_signed_tx()
 	}
 
+	fn chain_event<B: BlockT>(block_number: u64, header: B::Header) -> ChainEvent<B>
+		where NumberFor<B>: From<u64>
+	{
+		ChainEvent::NewBlock {
+			id: BlockId::Number(block_number.into()),
+			retracted: vec![],
+			is_new_best: true,
+			header: header,
+		}
+	}
+
 	#[test]
 	fn should_cease_building_block_when_deadline_is_reached() {
 		// given
@@ -330,16 +354,27 @@ mod tests {
 			txpool.submit_at(&BlockId::number(0), vec![extrinsic(0), extrinsic(1)])
 		).unwrap();
 
+		futures::executor::block_on(
+			txpool.maintain(chain_event(
+				0,
+				client.header(&BlockId::Number(0u64)).expect("header get error").expect("there should be header")
+			))
+		);
+
 		let mut proposer_factory = ProposerFactory::new(client.clone(), txpool.clone());
 
-		let cell = Mutex::new(time::Instant::now());
+		let cell = Mutex::new((false, time::Instant::now()));
 		let mut proposer = proposer_factory.init_with_now(
 			&client.header(&BlockId::number(0)).unwrap().unwrap(),
 			Box::new(move || {
 				let mut value = cell.lock();
-				let old = *value;
+				if !value.0 {
+					value.0 = true;
+					return value.1;
+				}
+				let old = value.1;
 				let new = old + time::Duration::from_secs(2);
-				*value = new;
+				*value = (true, new);
 				old
 			})
 		);
@@ -371,6 +406,13 @@ mod tests {
 			txpool.submit_at(&BlockId::number(0), vec![extrinsic(0)]),
 		).unwrap();
 
+		futures::executor::block_on(
+			txpool.maintain(chain_event(
+				0,
+				client.header(&BlockId::Number(0u64)).expect("header get error").expect("there should be header")
+			))
+		);
+
 		let mut proposer_factory = ProposerFactory::new(client.clone(), txpool.clone());
 
 		let mut proposer = proposer_factory.init_with_now(
@@ -459,15 +501,26 @@ mod tests {
 			block
 		};
 
+		futures::executor::block_on(
+			txpool.maintain(chain_event(
+				0,
+				client.header(&BlockId::Number(0u64)).expect("header get error").expect("there should be header")
+			))
+		);
+
 		// let's create one block and import it
 		let block = propose_block(&client, 0, 2, 7);
 		client.import(BlockOrigin::Own, block).unwrap();
 
-		// now let's make sure that we can still make some progress
+		futures::executor::block_on(
+			txpool.maintain(chain_event(
+				1,
+				client.header(&BlockId::Number(1)).expect("header get error").expect("there should be header")
+			))
+		);
 
-		// This is most likely incorrect, and caused by #5139
-		let tx_remaining = 0;
-		let block = propose_block(&client, 1, 2, tx_remaining);
+		// now let's make sure that we can still make some progress
+		let block = propose_block(&client, 1, 2, 5);
 		client.import(BlockOrigin::Own, block).unwrap();
 	}
 }

client/consensus/manual-seal/Cargo.toml

@@ -17,6 +17,7 @@ jsonrpc-derive = "14.0.5"
 log = "0.4.8"
 parking_lot = "0.10.0"
 serde = { version = "1.0", features=["derive"] }
+assert_matches = "1.3.0"
 
 sc-client = { path = "../../../client" , version = "0.8.0-alpha.2"}
 sc-client-api = { path = "../../../client/api" , version = "2.0.0-alpha.2"}

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

@@ -224,7 +224,7 @@ mod tests {
 		txpool::Options,
 	};
 	use substrate_test_runtime_transaction_pool::{TestApi, uxt};
-	use sp_transaction_pool::TransactionPool;
+	use sp_transaction_pool::{TransactionPool, MaintainedTransactionPool};
 	use sp_runtime::generic::BlockId;
 	use sp_blockchain::HeaderBackend;
 	use sp_consensus::ImportedAux;
@@ -432,14 +432,24 @@ mod tests {
 		assert!(backend.blockchain().header(BlockId::Number(0)).unwrap().is_some());
 		assert!(pool.submit_one(&BlockId::Number(1), uxt(Alice, 1)).await.is_ok());
 
+		pool.maintain(sp_transaction_pool::ChainEvent::NewBlock {
+			id: BlockId::Number(1),
+			header: backend.blockchain().header(BlockId::Number(1)).expect("db error").expect("imported above"),
+			is_new_best: true,
+			retracted: vec![],
+		}).await;
+
 		let (tx1, rx1) = futures::channel::oneshot::channel();
 		assert!(sink.send(EngineCommand::SealNewBlock {
-			parent_hash: Some(created_block.hash.clone()),
+			parent_hash: Some(created_block.hash),
 			sender: Some(tx1),
 			create_empty: false,
 			finalize: false,
 		}).await.is_ok());
-		assert!(rx1.await.unwrap().is_ok());
+		assert_matches::assert_matches!(
+			rx1.await.expect("should be no error receiving"),
+			Ok(_)
+		);
 		assert!(backend.blockchain().header(BlockId::Number(1)).unwrap().is_some());
 		pool_api.increment_nonce(Alice.into());
 

client/service/test/src/lib.rs

@@ -482,7 +482,11 @@ pub fn sync<G, E, Fb, F, Lb, L, B, ExF, U>(
 	let first_user_data = &network.full_nodes[0].2;
 	let best_block = BlockId::number(first_service.get().client().chain_info().best_number);
 	let extrinsic = extrinsic_factory(&first_service.get(), first_user_data);
-	futures::executor::block_on(first_service.get().transaction_pool().submit_one(&best_block, extrinsic)).unwrap();
+
+	futures::executor::block_on(
+		first_service.get().transaction_pool().submit_one(&best_block, extrinsic)
+	).expect("failed to submit extrinsic");
+
 	network.run_until_all_full(
 		|_index, service| service.get().transaction_pool().ready().count() == 1,
 		|_index, _service| true,

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

@@ -545,7 +545,7 @@ impl<B: ChainApi> ValidatedPool<B> {
 	}
 
 	/// Get an iterator for ready transactions ordered by priority
-	pub fn ready(&self) -> impl Iterator<Item=TransactionFor<B>> {
+	pub fn ready(&self) -> impl Iterator<Item=TransactionFor<B>> + Send {
 		self.pool.read().ready()
 	}