Adds fork-awareness and finalization notifications to transaction pool watchers. (#4740)

* adds finalization support to sc-transaction-pool using MaintainedTransactionPool for finalization events

* adds TransactionStatus::Retracted, notify watchers of retracted blocks, finalized now finalizes, transactions for current finalized -> last finalized block

* adds last_finalized to ChainApi, use generic BlockT for ChainEvent

* fix tests

* Apply suggestions from code review

Co-Authored-By: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

* tests

* fix tests, docs, lazily dedupe pruned hashes

* fix tests, Cargo.lock

* Apply suggestions from code review

Co-Authored-By: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

* remove tree_route, last_finalized from ChainApi, add block hash to Finalization and Retracted events

* prune finality watchers

* fix tests

* remove HeaderBackend bound from FullChainApi

* code style nits, terminate stream in finality_timeout

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

substrate/Cargo.lock

@@ -6263,6 +6263,7 @@ dependencies = [
  "serde_json",
  "sp-core",
  "sp-rpc",
+ "sp-runtime",
  "sp-transaction-pool",
  "sp-version",
 ]
@@ -6424,11 +6425,13 @@ dependencies = [
  "criterion 0.3.1",
  "derive_more",
  "futures 0.3.4",
+ "linked-hash-map",
  "log 0.4.8",
  "parity-scale-codec",
  "parity-util-mem",
  "parking_lot 0.10.0",
  "serde",
+ "sp-blockchain",
  "sp-core",
  "sp-runtime",
  "sp-transaction-pool",
@@ -7644,6 +7647,7 @@ dependencies = [
  "parity-scale-codec",
  "parking_lot 0.10.0",
  "sc-transaction-graph",
+ "sp-blockchain",
  "sp-runtime",
  "sp-transaction-pool",
  "substrate-test-runtime-client",

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

@@ -198,7 +198,7 @@ pub async fn run_instant_seal<B, CB, E, A, C, T>(
 {
 	// instant-seal creates blocks as soon as transactions are imported
 	// into the transaction pool.
-	let seal_block_channel = pool.import_notification_stream()
+	let seal_block_channel = pool.validated_pool().import_notification_stream()
 		.map(|_| {
 			EngineCommand::SealNewBlock {
 				create_empty: false,
@@ -260,7 +260,7 @@ mod tests {
 		// this test checks that blocks are created as soon as transactions are imported into the pool.
 		let (sender, receiver) = futures::channel::oneshot::channel();
 		let mut sender = Arc::new(Some(sender));
-		let stream = pool.pool().import_notification_stream()
+		let stream = pool.pool().validated_pool().import_notification_stream()
 			.map(move |_| {
 				// we're only going to submit one tx so this fn will only be called once.
 				let mut_sender =  Arc::get_mut(&mut sender).unwrap();

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

@@ -92,7 +92,7 @@ pub async fn seal_new_block<B, SC, CB, E, T, P>(
 		SC: SelectChain<B>,
 {
 	let future = async {
-		if pool.status().ready == 0 && !create_empty {
+		if pool.validated_pool().status().ready == 0 && !create_empty {
 			return Err(Error::EmptyTransactionPool)
 		}
 

substrate/client/rpc-api/Cargo.toml

@@ -17,6 +17,7 @@ log = "0.4.8"
 parking_lot = "0.10.0"
 sp-core = { version = "2.0.0", path = "../../primitives/core" }
 sp-version = { version = "2.0.0", path = "../../primitives/version" }
+sp-runtime = { path = "../../primitives/runtime" }
 serde = { version = "1.0.101", features = ["derive"] }
 serde_json = "1.0.41"
 sp-transaction-pool = { version = "2.0.0", path = "../../primitives/transaction-pool" }

substrate/client/service/src/builder.rs

@@ -51,7 +51,7 @@ use std::{
 use wasm_timer::SystemTime;
 use sysinfo::{get_current_pid, ProcessExt, System, SystemExt};
 use sc_telemetry::{telemetry, SUBSTRATE_INFO};
-use sp_transaction_pool::MaintainedTransactionPool;
+use sp_transaction_pool::{MaintainedTransactionPool, ChainEvent};
 use sp_blockchain;
 use grafana_data_source::{self, record_metrics};
 
@@ -882,42 +882,52 @@ ServiceBuilder<
 			let network_state_info: Arc<dyn NetworkStateInfo + Send + Sync> = network.clone();
 			let is_validator = config.roles.is_authority();
 
-			let events = client.import_notification_stream()
-				.for_each(move |notification| {
-					let txpool = txpool.upgrade();
+			let (import_stream, finality_stream) = (
+				client.import_notification_stream().map(|n| ChainEvent::NewBlock {
+					id: BlockId::Hash(n.hash),
+					header: n.header,
+					retracted: n.retracted,
+					is_new_best: n.is_new_best,
+				}),
+				client.finality_notification_stream().map(|n| ChainEvent::Finalized {
+					hash: n.hash
+				})
+			);
+			let events = futures::stream::select(import_stream, finality_stream)
+				.for_each(move |event| {
+					// offchain worker is only interested in block import events
+					if let ChainEvent::NewBlock { ref header, is_new_best, .. } = event {
+						let offchain = offchain.as_ref().and_then(|o| o.upgrade());
+						match offchain {
+							Some(offchain) if is_new_best => {
+								let future = offchain.on_block_imported(
+									&header,
+									network_state_info.clone(),
+									is_validator,
+								);
+								let _ = to_spawn_tx_.unbounded_send((
+									Box::pin(future),
+									From::from("offchain-on-block"),
+								));
+							},
+							Some(_) => log::debug!(
+									target: "sc_offchain",
+									"Skipping offchain workers for non-canon block: {:?}",
+									header,
+								),
+							_ => {},
+						}
+					};
 
+					let txpool = txpool.upgrade();
 					if let Some(txpool) = txpool.as_ref() {
-						let future = txpool.maintain(
-							&BlockId::hash(notification.hash),
-							&notification.retracted,
-						);
+						let future = txpool.maintain(event);
 						let _ = to_spawn_tx_.unbounded_send((
 							Box::pin(future),
 							From::from("txpool-maintain")
 						));
 					}
 
-					let offchain = offchain.as_ref().and_then(|o| o.upgrade());
-					match offchain {
-						Some(offchain) if notification.is_new_best => {
-							let future = offchain.on_block_imported(
-								&notification.header,
-								network_state_info.clone(),
-								is_validator,
-							);
-							let _ = to_spawn_tx_.unbounded_send((
-									Box::pin(future),
-									From::from("offchain-on-block"),
-							));
-						},
-						Some(_) => log::debug!(
-							target: "sc_offchain",
-							"Skipping offchain workers for non-canon block: {:?}",
-							notification.header,
-						),
-						_ => {},
-					}
-
 					ready(())
 				});
 			let _ = to_spawn_tx.unbounded_send((

substrate/client/transaction-pool/graph/Cargo.toml

@@ -12,10 +12,12 @@ log = "0.4.8"
 parking_lot = "0.10.0"
 serde = { version = "1.0.101", features = ["derive"] }
 wasm-timer = "0.2"
+sp-blockchain = { version = "2.0.0", path = "../../../primitives/blockchain" }
 sp-core = { version = "2.0.0", path = "../../../primitives/core" }
 sp-runtime = { version = "2.0.0", path = "../../../primitives/runtime" }
 sp-transaction-pool = { version = "2.0.0", path = "../../../primitives/transaction-pool" }
 parity-util-mem = { version = "0.5.1", default-features = false, features = ["primitive-types"] }
+linked-hash-map = "0.5.2"
 
 [dev-dependencies]
 assert_matches = "1.3.0"

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

@@ -135,8 +135,8 @@ fn bench_configured(pool: Pool<TestApi>, number: u64) {
 	let res = block_on(futures::future::join_all(futures.into_iter()));
 	assert!(res.iter().all(Result::is_ok));
 
-	assert_eq!(pool.status().future, 0);
-	assert_eq!(pool.status().ready, number as usize);
+	assert_eq!(pool.validated_pool().status().future, 0);
+	assert_eq!(pool.validated_pool().status().ready, number as usize);
 
 	// Prune all transactions.
 	let block_num = 6;
@@ -147,8 +147,8 @@ fn bench_configured(pool: Pool<TestApi>, number: u64) {
 	)).expect("Prune failed");
 
 	// pool is empty
-	assert_eq!(pool.status().ready, 0);
-	assert_eq!(pool.status().future, 0);
+	assert_eq!(pool.validated_pool().status().ready, 0);
+	assert_eq!(pool.validated_pool().status().future, 0);
 }
 
 fn benchmark_main(c: &mut Criterion) {

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

@@ -16,30 +16,34 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use std::{
-	collections::HashMap,
-	fmt,
-	hash,
+	collections::HashMap, hash,
 };
+use linked_hash_map::LinkedHashMap;
 use serde::Serialize;
-use crate::watcher;
-use sp_runtime::traits;
+use crate::{watcher, ChainApi, BlockHash};
 use log::{debug, trace, warn};
+use sp_runtime::traits;
 
 /// Extrinsic pool default listener.
-pub struct Listener<H: hash::Hash + Eq, H2> {
-	watchers: HashMap<H, watcher::Sender<H, H2>>
+pub struct Listener<H: hash::Hash + Eq, C: ChainApi> {
+	watchers: HashMap<H, watcher::Sender<H, BlockHash<C>>>,
+	finality_watchers: LinkedHashMap<BlockHash<C>, Vec<H>>,
 }
 
-impl<H: hash::Hash + Eq, H2> Default for Listener<H, H2> {
+/// Maximum number of blocks awaiting finality at any time.
+const MAX_FINALITY_WATCHERS: usize = 512;
+
+impl<H: hash::Hash + Eq, C: ChainApi> Default for Listener<H, C> {
 	fn default() -> Self {
 		Listener {
 			watchers: Default::default(),
+			finality_watchers: Default::default(),
 		}
 	}
 }
 
-impl<H: hash::Hash + traits::Member + Serialize, H2: Clone + fmt::Debug> Listener<H, H2> {
-	fn fire<F>(&mut self, hash: &H, fun: F) where F: FnOnce(&mut watcher::Sender<H, H2>) {
+impl<H: hash::Hash + traits::Member + Serialize, C: ChainApi> Listener<H, C> {
+	fn fire<F>(&mut self, hash: &H, fun: F) where F: FnOnce(&mut watcher::Sender<H, BlockHash<C>>) {
 		let clean = if let Some(h) = self.watchers.get_mut(hash) {
 			fun(h);
 			h.is_done()
@@ -55,7 +59,7 @@ impl<H: hash::Hash + traits::Member + Serialize, H2: Clone + fmt::Debug> Listene
 	/// Creates a new watcher for given verified extrinsic.
 	///
 	/// The watcher can be used to subscribe to lifecycle events of that extrinsic.
-	pub fn create_watcher(&mut self, hash: H) -> watcher::Watcher<H, H2> {
+	pub fn create_watcher(&mut self, hash: H) -> watcher::Watcher<H, BlockHash<C>> {
 		let sender = self.watchers.entry(hash.clone()).or_insert_with(watcher::Sender::default);
 		sender.new_watcher(hash)
 	}
@@ -101,8 +105,34 @@ impl<H: hash::Hash + traits::Member + Serialize, H2: Clone + fmt::Debug> Listene
 	}
 
 	/// Transaction was pruned from the pool.
-	pub fn pruned(&mut self, header_hash: H2, tx: &H) {
-		debug!(target: "txpool", "[{:?}] Pruned at {:?}", tx, header_hash);
-		self.fire(tx, |watcher| watcher.in_block(header_hash))
+	pub fn pruned(&mut self, block_hash: BlockHash<C>, tx: &H) {
+		debug!(target: "txpool", "[{:?}] Pruned at {:?}", tx, block_hash);
+		self.fire(tx, |s| s.in_block(block_hash));
+		self.finality_watchers.entry(block_hash).or_insert(vec![]).push(tx.clone());
+
+		while self.finality_watchers.len() > MAX_FINALITY_WATCHERS {
+			if let Some((hash, txs)) = self.finality_watchers.pop_front() {
+				for tx in txs {
+					self.fire(&tx, |s| s.finality_timeout(hash.clone()));
+				}
+			}
+		}
+	}
+
+	/// The block this transaction was included in has been retracted.
+	pub fn retracted(&mut self, block_hash: BlockHash<C>) {
+		if let Some(hashes) = self.finality_watchers.remove(&block_hash) {
+			for hash in hashes {
+				self.fire(&hash, |s| s.retracted(block_hash))
+			}
+		}
+	}
+
+	/// Notify all watchers that transactions have been finalized
+	pub fn finalized(&mut self, block_hash: BlockHash<C>, txs: Vec<H>) {
+		self.finality_watchers.remove(&block_hash);
+		for h in txs {
+			self.fire(&h, |s| s.finalized(block_hash.clone()))
+		}
 	}
 }

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

@@ -33,7 +33,7 @@ use sp_runtime::{
 	traits::{self, SaturatedConversion},
 	transaction_validity::{TransactionValidity, TransactionTag as Tag, TransactionValidityError},
 };
-use sp_transaction_pool::{error, PoolStatus};
+use sp_transaction_pool::error;
 use wasm_timer::Instant;
 
 use crate::validated_pool::{ValidatedPool, ValidatedTransaction};
@@ -338,34 +338,6 @@ impl<B: ChainApi> Pool<B> {
 		)
 	}
 
-	/// Return an event stream of notifications for when transactions are imported to the pool.
-	///
-	/// Consumers of this stream should use the `ready` method to actually get the
-	/// pending transactions in the right order.
-	pub fn import_notification_stream(&self) -> EventStream<ExHash<B>> {
-		self.validated_pool.import_notification_stream()
-	}
-
-	/// Invoked when extrinsics are broadcasted.
-	pub fn on_broadcasted(&self, propagated: HashMap<ExHash<B>, Vec<String>>) {
-		self.validated_pool.on_broadcasted(propagated)
-	}
-
-	/// Remove invalid transactions from the pool.
-	pub fn remove_invalid(&self, hashes: &[ExHash<B>]) -> Vec<TransactionFor<B>> {
-		self.validated_pool.remove_invalid(hashes)
-	}
-
-	/// Get an iterator for ready transactions ordered by priority
-	pub fn ready(&self) -> impl Iterator<Item=TransactionFor<B>> {
-		self.validated_pool.ready()
-	}
-
-	/// Returns pool status.
-	pub fn status(&self) -> PoolStatus {
-		self.validated_pool.status()
-	}
-
 	/// Returns transaction hash
 	pub fn hash_of(&self, xt: &ExtrinsicFor<B>) -> ExHash<B> {
 		self.validated_pool.api().hash_and_length(xt).0
@@ -454,9 +426,9 @@ impl<B: ChainApi> Pool<B> {
 		(hash, validity)
 	}
 
-	/// Get ready transaction by hash, if it present in the pool.
-	pub fn ready_transaction(&self, hash: &ExHash<B>) -> Option<TransactionFor<B>> {
-		self.validated_pool.ready_by_hash(hash)
+	/// get a reference to the underlying validated pool.
+	pub fn validated_pool(&self) ->  &ValidatedPool<B> {
+		&self.validated_pool
 	}
 }
 
@@ -598,7 +570,7 @@ mod tests {
 		}))).unwrap();
 
 		// then
-		assert_eq!(pool.ready().map(|v| v.hash).collect::<Vec<_>>(), vec![hash]);
+		assert_eq!(pool.validated_pool().ready().map(|v| v.hash).collect::<Vec<_>>(), vec![hash]);
 	}
 
 	#[test]
@@ -615,8 +587,8 @@ mod tests {
 		// when
 		pool.validated_pool.rotator().ban(&Instant::now(), vec![pool.hash_of(&uxt)]);
 		let res = block_on(pool.submit_one(&BlockId::Number(0), uxt));
-		assert_eq!(pool.status().ready, 0);
-		assert_eq!(pool.status().future, 0);
+		assert_eq!(pool.validated_pool().status().ready, 0);
+		assert_eq!(pool.validated_pool().status().future, 0);
 
 		// then
 		assert_matches!(res.unwrap_err(), error::Error::TemporarilyBanned);
@@ -627,7 +599,7 @@ mod tests {
 		let stream = {
 			// given
 			let pool = pool();
-			let stream = pool.import_notification_stream();
+			let stream = pool.validated_pool().import_notification_stream();
 
 			// when
 			let _hash = block_on(pool.submit_one(&BlockId::Number(0), uxt(Transfer {
@@ -650,8 +622,8 @@ mod tests {
 				nonce: 3,
 			}))).unwrap();
 
-			assert_eq!(pool.status().ready, 2);
-			assert_eq!(pool.status().future, 1);
+			assert_eq!(pool.validated_pool().status().ready, 2);
+			assert_eq!(pool.validated_pool().status().future, 1);
 			stream
 		};
 
@@ -689,9 +661,9 @@ mod tests {
 		pool.validated_pool.clear_stale(&BlockId::Number(5)).unwrap();
 
 		// then
-		assert_eq!(pool.ready().count(), 0);
-		assert_eq!(pool.status().future, 0);
-		assert_eq!(pool.status().ready, 0);
+		assert_eq!(pool.validated_pool().ready().count(), 0);
+		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));
@@ -735,7 +707,7 @@ mod tests {
 			amount: 5,
 			nonce: 1,
 		}))).unwrap();
-		assert_eq!(pool.status().future, 1);
+		assert_eq!(pool.validated_pool().status().future, 1);
 
 		// when
 		let hash2 = block_on(pool.submit_one(&BlockId::Number(0), uxt(Transfer {
@@ -746,7 +718,7 @@ mod tests {
 		}))).unwrap();
 
 		// then
-		assert_eq!(pool.status().future, 1);
+		assert_eq!(pool.validated_pool().status().future, 1);
 		assert!(pool.validated_pool.rotator().is_banned(&hash1));
 		assert!(!pool.validated_pool.rotator().is_banned(&hash2));
 	}
@@ -773,8 +745,8 @@ mod tests {
 		}))).unwrap_err();
 
 		// then
-		assert_eq!(pool.status().ready, 0);
-		assert_eq!(pool.status().future, 0);
+		assert_eq!(pool.validated_pool().status().ready, 0);
+		assert_eq!(pool.validated_pool().status().future, 0);
 	}
 
 	#[test]
@@ -791,8 +763,8 @@ mod tests {
 		}))).unwrap_err();
 
 		// then
-		assert_eq!(pool.status().ready, 0);
-		assert_eq!(pool.status().future, 0);
+		assert_eq!(pool.validated_pool().status().ready, 0);
+		assert_eq!(pool.validated_pool().status().future, 0);
 		assert_matches!(err, error::Error::NoTagsProvided);
 	}
 
@@ -809,19 +781,18 @@ mod tests {
 				amount: 5,
 				nonce: 0,
 			}))).unwrap();
-			assert_eq!(pool.status().ready, 1);
-			assert_eq!(pool.status().future, 0);
+			assert_eq!(pool.validated_pool().status().ready, 1);
+			assert_eq!(pool.validated_pool().status().future, 0);
 
 			// when
 			block_on(pool.prune_tags(&BlockId::Number(2), vec![vec![0u8]], vec![])).unwrap();
-			assert_eq!(pool.status().ready, 0);
-			assert_eq!(pool.status().future, 0);
+			assert_eq!(pool.validated_pool().status().ready, 0);
+			assert_eq!(pool.validated_pool().status().future, 0);
 
 			// then
 			let mut stream = futures::executor::block_on_stream(watcher.into_stream());
 			assert_eq!(stream.next(), Some(TransactionStatus::Ready));
 			assert_eq!(stream.next(), Some(TransactionStatus::InBlock(H256::from_low_u64_be(2).into())));
-			assert_eq!(stream.next(), None);
 		}
 
 		#[test]
@@ -834,19 +805,18 @@ mod tests {
 				amount: 5,
 				nonce: 0,
 			}))).unwrap();
-			assert_eq!(pool.status().ready, 1);
-			assert_eq!(pool.status().future, 0);
+			assert_eq!(pool.validated_pool().status().ready, 1);
+			assert_eq!(pool.validated_pool().status().future, 0);
 
 			// when
 			block_on(pool.prune_tags(&BlockId::Number(2), vec![vec![0u8]], vec![2u64])).unwrap();
-			assert_eq!(pool.status().ready, 0);
-			assert_eq!(pool.status().future, 0);
+			assert_eq!(pool.validated_pool().status().ready, 0);
+			assert_eq!(pool.validated_pool().status().future, 0);
 
 			// then
 			let mut stream = futures::executor::block_on_stream(watcher.into_stream());
 			assert_eq!(stream.next(), Some(TransactionStatus::Ready));
 			assert_eq!(stream.next(), Some(TransactionStatus::InBlock(H256::from_low_u64_be(2).into())));
-			assert_eq!(stream.next(), None);
 		}
 
 		#[test]
@@ -859,8 +829,8 @@ mod tests {
 				amount: 5,
 				nonce: 1,
 			}))).unwrap();
-			assert_eq!(pool.status().ready, 0);
-			assert_eq!(pool.status().future, 1);
+			assert_eq!(pool.validated_pool().status().ready, 0);
+			assert_eq!(pool.validated_pool().status().future, 1);
 
 			// when
 			block_on(pool.submit_one(&BlockId::Number(0), uxt(Transfer {
@@ -869,7 +839,7 @@ mod tests {
 				amount: 5,
 				nonce: 0,
 			}))).unwrap();
-			assert_eq!(pool.status().ready, 2);
+			assert_eq!(pool.validated_pool().status().ready, 2);
 
 			// then
 			let mut stream = futures::executor::block_on_stream(watcher.into_stream());
@@ -888,7 +858,7 @@ mod tests {
 				nonce: 0,
 			});
 			let watcher = block_on(pool.submit_and_watch(&BlockId::Number(0), uxt)).unwrap();
-			assert_eq!(pool.status().ready, 1);
+			assert_eq!(pool.validated_pool().status().ready, 1);
 
 			// when
 			pool.validated_pool.remove_invalid(&[*watcher.hash()]);
@@ -912,13 +882,13 @@ mod tests {
 				nonce: 0,
 			});
 			let watcher = block_on(pool.submit_and_watch(&BlockId::Number(0), uxt)).unwrap();
-			assert_eq!(pool.status().ready, 1);
+			assert_eq!(pool.validated_pool().status().ready, 1);
 
 			// when
 			let mut map = HashMap::new();
 			let peers = vec!["a".into(), "b".into(), "c".into()];
 			map.insert(*watcher.hash(), peers.clone());
-			pool.on_broadcasted(map);
+			pool.validated_pool().on_broadcasted(map);
 
 
 			// then
@@ -947,7 +917,7 @@ mod tests {
 				nonce: 0,
 			});
 			let watcher = block_on(pool.submit_and_watch(&BlockId::Number(0), xt)).unwrap();
-			assert_eq!(pool.status().ready, 1);
+			assert_eq!(pool.validated_pool().status().ready, 1);
 
 			// when
 			let xt = uxt(Transfer {
@@ -957,7 +927,7 @@ mod tests {
 				nonce: 1,
 			});
 			block_on(pool.submit_one(&BlockId::Number(1), xt)).unwrap();
-			assert_eq!(pool.status().ready, 1);
+			assert_eq!(pool.validated_pool().status().ready, 1);
 
 			// then
 			let mut stream = futures::executor::block_on_stream(watcher.into_stream());
@@ -1000,11 +970,11 @@ mod tests {
 			// The tag the above transaction provides (TestApi is using just nonce as u8)
 			let provides = vec![0_u8];
 			block_on(pool.submit_one(&BlockId::Number(0), xt)).unwrap();
-			assert_eq!(pool.status().ready, 1);
+			assert_eq!(pool.validated_pool().status().ready, 1);
 
 			// Now block import happens before the second transaction is able to finish verification.
 			block_on(pool.prune_tags(&BlockId::Number(1), vec![provides], vec![])).unwrap();
-			assert_eq!(pool.status().ready, 0);
+			assert_eq!(pool.validated_pool().status().ready, 0);
 
 
 			// so when we release the verification of the previous one it will have
@@ -1014,8 +984,8 @@ mod tests {
 
 			// then
 			is_ready.recv().unwrap(); // wait for finish
-			assert_eq!(pool.status().ready, 1);
-			assert_eq!(pool.status().future, 0);
+			assert_eq!(pool.validated_pool().status().ready, 1);
+			assert_eq!(pool.validated_pool().status().future, 0);
 		}
 	}
 
@@ -1047,7 +1017,7 @@ mod tests {
 		let tx4 = transfer(4);
 		let hash4 = pool.validated_pool.api().hash_and_length(&tx4).0;
 		let watcher4 = block_on(pool.submit_and_watch(&BlockId::Number(0), tx4)).unwrap();
-		assert_eq!(pool.status().ready, 5);
+		assert_eq!(pool.validated_pool().status().ready, 5);
 
 		// when
 		pool.validated_pool.api().invalidate.lock().insert(hash3);
@@ -1064,7 +1034,7 @@ mod tests {
 		//
 		// events for hash3 are: Ready, Invalid
 		// events for hash4 are: Ready, Invalid
-		assert_eq!(pool.status().ready, 2);
+		assert_eq!(pool.validated_pool().status().ready, 2);
 		assert_eq!(
 			futures::executor::block_on_stream(watcher3.into_stream()).collect::<Vec<_>>(),
 			vec![TransactionStatus::Ready, TransactionStatus::Invalid],
@@ -1095,4 +1065,3 @@ mod tests {
 		);
 	}
 }
-

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

@@ -16,12 +16,11 @@
 
 use std::{
 	collections::{HashSet, HashMap},
-	fmt,
 	hash,
 	sync::Arc,
 };
 
-use crate::base_pool as base;
+use crate::{base_pool as base, BlockHash};
 use crate::listener::Listener;
 use crate::rotator::PoolRotator;
 use crate::watcher::Watcher;
@@ -39,7 +38,7 @@ use sp_transaction_pool::{error, PoolStatus};
 use wasm_timer::Instant;
 
 use crate::base_pool::PruneStatus;
-use crate::pool::{EventStream, Options, ChainApi, BlockHash, ExHash, ExtrinsicFor, TransactionFor};
+use crate::pool::{EventStream, Options, ChainApi, ExHash, ExtrinsicFor, TransactionFor};
 
 /// Pre-validated transaction. Validated pool only accepts transactions wrapped in this enum.
 #[derive(Debug)]
@@ -62,10 +61,10 @@ pub type ValidatedTransactionFor<B> = ValidatedTransaction<
 >;
 
 /// Pool that deals with validated transactions.
-pub(crate) struct ValidatedPool<B: ChainApi> {
+pub struct ValidatedPool<B: ChainApi> {
 	api: Arc<B>,
 	options: Options,
-	listener: RwLock<Listener<ExHash<B>, BlockHash<B>>>,
+	listener: RwLock<Listener<ExHash<B>, B>>,
 	pool: RwLock<base::BasePool<
 		ExHash<B>,
 		ExtrinsicFor<B>,
@@ -91,9 +90,9 @@ impl<B: ChainApi> ValidatedPool<B> {
 	pub fn new(options: Options, api: Arc<B>) -> Self {
 		let base_pool = base::BasePool::new(options.reject_future_transactions);
 		ValidatedPool {
-			api,
 			options,
 			listener: Default::default(),
+			api,
 			pool: RwLock::new(base_pool),
 			import_notification_sinks: Default::default(),
 			rotator: Default::default(),
@@ -138,13 +137,14 @@ impl<B: ChainApi> ValidatedPool<B> {
 				let imported = self.pool.write().import(tx)?;
 
 				if let base::Imported::Ready { ref hash, .. } = imported {
-					self.import_notification_sinks.lock().retain(|sink| sink.unbounded_send(hash.clone()).is_ok());
+					self.import_notification_sinks.lock()
+						.retain(|sink| sink.unbounded_send(hash.clone()).is_ok());
 				}
 
 				let mut listener = self.listener.write();
 				fire_events(&mut *listener, &imported);
 				Ok(imported.hash().clone())
-			}
+			},
 			ValidatedTransaction::Invalid(hash, err) => {
 				self.rotator.ban(&Instant::now(), std::iter::once(hash));
 				Err(err.into())
@@ -152,7 +152,7 @@ impl<B: ChainApi> ValidatedPool<B> {
 			ValidatedTransaction::Unknown(hash, err) => {
 				self.listener.write().invalid(&hash, false);
 				Err(err.into())
-			}
+			},
 		}
 	}
 
@@ -343,8 +343,7 @@ impl<B: ChainApi> ValidatedPool<B> {
 		self.pool.read().by_hashes(&hashes)
 			.into_iter()
 			.map(|existing_in_pool| existing_in_pool
-				.map(|transaction| transaction.provides.iter().cloned()
-				.collect()))
+				.map(|transaction| transaction.provides.iter().cloned().collect()))
 			.collect()
 	}
 
@@ -416,8 +415,14 @@ impl<B: ChainApi> ValidatedPool<B> {
 		let header_hash = self.api.block_id_to_hash(at)?
 			.ok_or_else(|| error::Error::InvalidBlockId(format!("{:?}", at)).into())?;
 		let mut listener = self.listener.write();
+		let mut set = HashSet::with_capacity(hashes.size_hint().0);
 		for h in hashes {
-			listener.pruned(header_hash, &h);
+			// `hashes` has possibly duplicate hashes.
+			// we'd like to send out the `InBlock` notification only once.
+			if !set.contains(&h) {
+				listener.pruned(header_hash, &h);
+				set.insert(h);
+			}
 		}
 		Ok(())
 	}
@@ -468,7 +473,10 @@ impl<B: ChainApi> ValidatedPool<B> {
 		&self.api
 	}
 
-	/// Return an event stream of transactions imported to the pool.
+	/// Return an event stream of notifications for when transactions are imported to the pool.
+	///
+	/// Consumers of this stream should use the `ready` method to actually get the
+	/// pending transactions in the right order.
 	pub fn import_notification_stream(&self) -> EventStream<ExHash<B>> {
 		let (sink, stream) = mpsc::unbounded();
 		self.import_notification_sinks.lock().push(sink);
@@ -492,7 +500,7 @@ impl<B: ChainApi> ValidatedPool<B> {
 	pub fn remove_invalid(&self, hashes: &[ExHash<B>]) -> Vec<TransactionFor<B>> {
 		// early exit in case there is no invalid transactions.
 		if hashes.is_empty() {
-			return vec![]
+			return vec![];
 		}
 
 		debug!(target: "txpool", "Removing invalid transactions: {:?}", hashes);
@@ -521,14 +529,34 @@ impl<B: ChainApi> ValidatedPool<B> {
 	pub fn status(&self) -> PoolStatus {
 		self.pool.read().status()
 	}
+
+	/// Notify all watchers that transactions in the block with hash have been finalized
+	pub async fn on_block_finalized(&self, block_hash: BlockHash<B>) -> Result<(), B::Error> {
+		debug!(target: "txpool", "Attempting to notify watchers of finalization for {}", block_hash);
+		// fetch all extrinsic hashes
+		if let Some(txs) = self.api.block_body(&BlockId::Hash(block_hash.clone())).await? {
+			let tx_hashes = txs.into_iter()
+				.map(|tx| self.api.hash_and_length(&tx).0)
+				.collect::<Vec<_>>();
+			// notify the watcher that these extrinsics have been finalized
+			self.listener.write().finalized(block_hash, tx_hashes);
+		}
+
+		Ok(())
+	}
+
+	/// Notify the listener of retracted blocks
+	pub fn on_block_retracted(&self, block_hash: BlockHash<B>) {
+		self.listener.write().retracted(block_hash)
+	}
 }
 
-fn fire_events<H, H2, Ex>(
-	listener: &mut Listener<H, H2>,
+fn fire_events<H, B, Ex>(
+	listener: &mut Listener<H, B>,
 	imported: &base::Imported<H, Ex>,
 ) where
 	H: hash::Hash + Eq + traits::Member + Serialize,
-	H2: Clone + fmt::Debug,
+	B: ChainApi,
 {
 	match *imported {
 		base::Imported::Ready { ref promoted, ref failed, ref removed, ref hash } => {

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

@@ -26,12 +26,12 @@ use sp_transaction_pool::TransactionStatus;
 ///
 /// Represents a stream of status updates for particular extrinsic.
 #[derive(Debug)]
-pub struct Watcher<H, H2> {
-	receiver: mpsc::UnboundedReceiver<TransactionStatus<H, H2>>,
+pub struct Watcher<H, BH> {
+	receiver: mpsc::UnboundedReceiver<TransactionStatus<H, BH>>,
 	hash: H,
 }
 
-impl<H, H2> Watcher<H, H2> {
+impl<H, BH> Watcher<H, BH> {
 	/// Returns the transaction hash.
 	pub fn hash(&self) -> &H {
 		&self.hash
@@ -40,30 +40,30 @@ impl<H, H2> Watcher<H, H2> {
 	/// Pipe the notifications to given sink.
 	///
 	/// Make sure to drive the future to completion.
-	pub fn into_stream(self) -> impl Stream<Item=TransactionStatus<H, H2>> {
+	pub fn into_stream(self) -> impl Stream<Item=TransactionStatus<H, BH>> {
 		self.receiver
 	}
 }
 
 /// Sender part of the watcher. Exposed only for testing purposes.
 #[derive(Debug)]
-pub struct Sender<H, H2> {
-	receivers: Vec<mpsc::UnboundedSender<TransactionStatus<H, H2>>>,
-	finalized: bool,
+pub struct Sender<H, BH> {
+	receivers: Vec<mpsc::UnboundedSender<TransactionStatus<H, BH>>>,
+	is_finalized: bool,
 }
 
-impl<H, H2> Default for Sender<H, H2> {
+impl<H, BH> Default for Sender<H, BH> {
 	fn default() -> Self {
 		Sender {
 			receivers: Default::default(),
-			finalized: false,
+			is_finalized: false,
 		}
 	}
 }
 
-impl<H: Clone, H2: Clone> Sender<H, H2> {
+impl<H: Clone, BH: Clone> Sender<H, BH> {
 	/// Add a new watcher to this sender object.
-	pub fn new_watcher(&mut self, hash: H) -> Watcher<H, H2> {
+	pub fn new_watcher(&mut self, hash: H) -> Watcher<H, BH> {
 		let (tx, receiver) = mpsc::unbounded();
 		self.receivers.push(tx);
 		Watcher {
@@ -85,26 +85,42 @@ impl<H: Clone, H2: Clone> Sender<H, H2> {
 	/// Some state change (perhaps another extrinsic was included) rendered this extrinsic invalid.
 	pub fn usurped(&mut self, hash: H) {
 		self.send(TransactionStatus::Usurped(hash));
-		self.finalized = true;
+		self.is_finalized = true;
 	}
 
 	/// Extrinsic has been included in block with given hash.
-	pub fn in_block(&mut self, hash: H2) {
+	pub fn in_block(&mut self, hash: BH) {
 		self.send(TransactionStatus::InBlock(hash));
-		self.finalized = true;
+	}
+
+	/// Extrinsic has been finalized by a finality gadget.
+	pub fn finalized(&mut self, hash: BH) {
+		self.send(TransactionStatus::Finalized(hash));
+		self.is_finalized = true;
+	}
+
+	/// The block this extrinsic was included in has been retracted
+	pub fn finality_timeout(&mut self, hash: BH) {
+		self.send(TransactionStatus::FinalityTimeout(hash));
+		self.is_finalized = true;
+	}
+
+	/// The block this extrinsic was included in has been retracted
+	pub fn retracted(&mut self, hash: BH) {
+		self.send(TransactionStatus::Retracted(hash));
 	}
 
 	/// Extrinsic has been marked as invalid by the block builder.
 	pub fn invalid(&mut self) {
 		self.send(TransactionStatus::Invalid);
 		// we mark as finalized as there are no more notifications
-		self.finalized = true;
+		self.is_finalized = true;
 	}
 
 	/// Transaction has been dropped from the pool because of the limit.
 	pub fn dropped(&mut self) {
 		self.send(TransactionStatus::Dropped);
-		self.finalized = true;
+		self.is_finalized = true;
 	}
 
 	/// The extrinsic has been broadcast to the given peers.
@@ -114,10 +130,10 @@ impl<H: Clone, H2: Clone> Sender<H, H2> {
 
 	/// Returns true if the are no more listeners for this extrinsic or it was finalized.
 	pub fn is_done(&self) -> bool {
-		self.finalized || self.receivers.is_empty()
+		self.is_finalized || self.receivers.is_empty()
 	}
 
-	fn send(&mut self, status: TransactionStatus<H, H2>) {
+	fn send(&mut self, status: TransactionStatus<H, BH>) {
 		self.receivers.retain(|sender| sender.unbounded_send(status.clone()).is_ok())
 	}
 }

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

@@ -64,7 +64,8 @@ impl<Client, Block> FullChainApi<Client, Block> where
 
 impl<Client, Block> sc_transaction_graph::ChainApi for FullChainApi<Client, Block> where
 	Block: BlockT,
-	Client: ProvideRuntimeApi<Block> + BlockBody<Block> + BlockIdTo<Block> + 'static + Send + Sync,
+	Client: ProvideRuntimeApi<Block> + BlockBody<Block> + BlockIdTo<Block>,
+	Client: Send + Sync + 'static,
 	Client::Api: TaggedTransactionQueue<Block>,
 	sp_api::ApiErrorFor<Client, Block>: Send,
 {

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

@@ -37,9 +37,8 @@ use sp_runtime::{
 	traits::{Block as BlockT, NumberFor, AtLeast32Bit, Extrinsic},
 };
 use sp_transaction_pool::{
-	TransactionPool, PoolStatus, ImportNotificationStream,
-	TxHash, TransactionFor, TransactionStatusStreamFor, BlockHash,
-	MaintainedTransactionPool, PoolFuture,
+	TransactionPool, PoolStatus, ImportNotificationStream, TxHash, TransactionFor,
+	TransactionStatusStreamFor, MaintainedTransactionPool, PoolFuture, ChainEvent,
 };
 use wasm_timer::Instant;
 
@@ -115,7 +114,6 @@ impl<PoolApi, Block> BasicPool<PoolApi, Block>
 				}
 			)),
 		}
-
 	}
 
 	/// Gets shared reference to the underlying pool.
@@ -174,19 +172,19 @@ impl<PoolApi, Block> TransactionPool for BasicPool<PoolApi, Block>
 	}
 
 	fn remove_invalid(&self, hashes: &[TxHash<Self>]) -> Vec<Arc<Self::InPoolTransaction>> {
-		self.pool.remove_invalid(hashes)
+		self.pool.validated_pool().remove_invalid(hashes)
 	}
 
 	fn status(&self) -> PoolStatus {
-		self.pool.status()
+		self.pool.validated_pool().status()
 	}
 
 	fn ready(&self) -> Box<dyn Iterator<Item=Arc<Self::InPoolTransaction>>> {
-		Box::new(self.pool.ready())
+		Box::new(self.pool.validated_pool().ready())
 	}
 
 	fn import_notification_stream(&self) -> ImportNotificationStream<TxHash<Self>> {
-		self.pool.import_notification_stream()
+		self.pool.validated_pool().import_notification_stream()
 	}
 
 	fn hash_of(&self, xt: &TransactionFor<Self>) -> TxHash<Self> {
@@ -194,11 +192,11 @@ impl<PoolApi, Block> TransactionPool for BasicPool<PoolApi, Block>
 	}
 
 	fn on_broadcasted(&self, propagations: HashMap<TxHash<Self>, Vec<String>>) {
-		self.pool.on_broadcasted(propagations)
+		self.pool.validated_pool().on_broadcasted(propagations)
 	}
 
 	fn ready_transaction(&self, hash: &TxHash<Self>) -> Option<Arc<Self::InPoolTransaction>> {
-		self.pool.ready_transaction(hash)
+		self.pool.validated_pool().ready_by_hash(hash)
 	}
 }
 
@@ -214,7 +212,7 @@ enum RevalidationStatus<N> {
 
 enum RevalidationStrategy<N> {
 	Always,
-	Light(RevalidationStatus<N>)
+	Light(RevalidationStatus<N>),
 }
 
 struct RevalidationAction {
@@ -241,7 +239,7 @@ impl<N: Clone + Copy + AtLeast32Bit> RevalidationStrategy<N> {
 				revalidate: status.next_required(
 					block,
 					revalidate_time_period,
-					revalidate_block_period
+					revalidate_block_period,
 				),
 				resubmit: false,
 				revalidate_amount: None,
@@ -275,7 +273,7 @@ impl<N: Clone + Copy + AtLeast32Bit> RevalidationStatus<N> {
 					revalidate_block_period.map(|period| block + period),
 				);
 				false
-			},
+			}
 			Self::Scheduled(revalidate_at_time, revalidate_at_block) => {
 				let is_required = revalidate_at_time.map(|at| Instant::now() >= at).unwrap_or(false)
 					|| revalidate_at_block.map(|at| block >= at).unwrap_or(false);
@@ -283,87 +281,105 @@ impl<N: Clone + Copy + AtLeast32Bit> RevalidationStatus<N> {
 					*self = Self::InProgress;
 				}
 				is_required
-			},
+			}
 			Self::InProgress => false,
 		}
 	}
 }
 
 impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
-where
-	Block: BlockT,
-	PoolApi: 'static + sc_transaction_graph::ChainApi<Block=Block, Hash=Block::Hash>,
+	where
+		Block: BlockT,
+		PoolApi: 'static + sc_transaction_graph::ChainApi<Block=Block, Hash=Block::Hash>,
 {
-	fn maintain(&self, id: &BlockId<Self::Block>, retracted: &[BlockHash<Self>])
-		-> Pin<Box<dyn Future<Output=()> + Send>>
-	{
-		let id = id.clone();
-		let pool = self.pool.clone();
-		let api = self.api.clone();
+	fn maintain(&self, event: ChainEvent<Self::Block>) -> Pin<Box<dyn Future<Output=()> + Send>> {
+		match event {
+			ChainEvent::NewBlock { id, retracted, .. } => {
+				let id = id.clone();
+				let pool = self.pool.clone();
+				let api = self.api.clone();
 
-		let block_number = match api.block_id_to_number(&id) {
-			Ok(Some(number)) => number,
-			_ => {
-				log::trace!(target: "txqueue", "Skipping chain event - no number for that block {:?}", id);
-				return Box::pin(ready(()));
+				let block_number = match api.block_id_to_number(&id) {
+					Ok(Some(number)) => number,
+					_ => {
+						log::trace!(target: "txqueue", "Skipping chain event - no number for that block {:?}", id);
+						return Box::pin(ready(()));
+					}
+				};
+
+				let next_action = self.revalidation_strategy.lock().next(
+					block_number,
+					Some(std::time::Duration::from_secs(60)),
+					Some(20.into()),
+				);
+				let revalidation_strategy = self.revalidation_strategy.clone();
+				let retracted = retracted.clone();
+
+				async move {
+					// We don't query block if we won't prune anything
+					if !pool.validated_pool().status().is_empty() {
+						let hashes = api.block_body(&id).await
+							.unwrap_or_else(|e| {
+								log::warn!("Prune known transactions: error request {:?}!", e);
+								None
+							})
+							.unwrap_or_default()
+							.into_iter()
+							.map(|tx| pool.hash_of(&tx))
+							.collect::<Vec<_>>();
+
+						if let Err(e) = pool.prune_known(&id, &hashes) {
+							log::error!("Cannot prune known in the pool {:?}!", e);
+						}
+					}
+
+					if next_action.resubmit {
+						let mut resubmit_transactions = Vec::new();
+
+						for retracted_hash in retracted {
+							// notify txs awaiting finality that it has been retracted
+							pool.validated_pool().on_block_retracted(retracted_hash.clone());
+
+							let block_transactions = api.block_body(&BlockId::hash(retracted_hash.clone())).await
+								.unwrap_or_else(|e| {
+									log::warn!("Failed to fetch block body {:?}!", e);
+									None
+								})
+								.unwrap_or_default()
+								.into_iter()
+								.filter(|tx| tx.is_signed().unwrap_or(true));
+
+							resubmit_transactions.extend(block_transactions);
+						}
+						if let Err(e) = pool.submit_at(&id, resubmit_transactions, true).await {
+							log::debug!(
+								target: "txpool",
+								"[{:?}] Error re-submitting transactions: {:?}", id, e
+							)
+						}
+					}
+
+					if next_action.revalidate {
+						if let Err(e) = pool.revalidate_ready(&id, next_action.revalidate_amount).await {
+							log::warn!("Revalidate ready failed {:?}", e);
+						}
+					}
+
+					revalidation_strategy.lock().clear();
+				}.boxed()
 			}
-		};
-
-		let next_action = self.revalidation_strategy.lock().next(
-			block_number,
-			Some(std::time::Duration::from_secs(60)),
-			Some(20.into()),
-		);
-		let revalidation_strategy = self.revalidation_strategy.clone();
-		let retracted = retracted.to_vec();
-
-		async move {
-			// We don't query block if we won't prune anything
-			if !pool.status().is_empty() {
-				let hashes = api.block_body(&id).await
-					.unwrap_or_else(|e| {
-						log::warn!("Prune known transactions: error request {:?}!", e);
-						None
-					})
-				.unwrap_or_default()
-				.into_iter()
-				.map(|tx| pool.hash_of(&tx))
-				.collect::<Vec<_>>();
-
-				if let Err(e) = pool.prune_known(&id, &hashes) {
-					log::error!("Cannot prune known in the pool {:?}!", e);
-				}
+			ChainEvent::Finalized { hash } => {
+				let pool = self.pool.clone();
+				async move {
+					if let Err(e) = pool.validated_pool().on_block_finalized(hash).await {
+						log::warn!(
+							target: "txpool",
+							"Error [{}] occurred while attempting to notify watchers of finalization {}",
+							e, hash
+						)
+					}
+				}.boxed()
 			}
-
-			if next_action.resubmit {
-				let mut resubmit_transactions = Vec::new();
-
-				for retracted_hash in retracted {
-					let block_transactions = api.block_body(&BlockId::hash(retracted_hash.clone())).await
-						.unwrap_or_else(|e| {
-							log::warn!("Failed to fetch block body {:?}!", e);
-							None
-						})
-						.unwrap_or_default()
-						.into_iter()
-						.filter(|tx| tx.is_signed().unwrap_or(true));
-
-					resubmit_transactions.extend(block_transactions);
-				}
-				if let Err(e) = pool.submit_at(&id, resubmit_transactions, true).await {
-					log::debug!(target: "txpool",
-						"[{:?}] Error re-submitting transactions: {:?}", id, e
-					)
-				}
-			}
-
-			if next_action.revalidate {
-				if let Err(e) = pool.revalidate_ready(&id, next_action.revalidate_amount).await {
-					log::warn!("Revalidate ready failed {:?}", e);
-				}
-			}
-
-			revalidation_strategy.lock().clear();
-		}.boxed()
+		}
 	}
 }

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

@@ -15,17 +15,18 @@
 // along with Substrate.  If not, see <http://www.gnu.org/licenses/>.
 
 use crate::*;
-use sc_transaction_graph::Pool;
 use futures::executor::block_on;
+use txpool::{self, Pool};
 use sp_runtime::{
 	generic::BlockId,
 	transaction_validity::ValidTransaction,
 };
 use substrate_test_runtime_client::{
-	runtime::{Block, Hash, Index},
+	runtime::{Block, Hash, Index, Header},
 	AccountKeyring::*,
 };
 use substrate_test_runtime_transaction_pool::{TestApi, uxt};
+use sp_transaction_pool::TransactionStatus;
 
 fn pool() -> Pool<TestApi> {
 	Pool::new(Default::default(), TestApi::with_alice_nonce(209).into())
@@ -35,12 +36,22 @@ fn maintained_pool() -> BasicPool<TestApi, Block> {
 	BasicPool::new(Default::default(), std::sync::Arc::new(TestApi::with_alice_nonce(209)))
 }
 
+fn header(number: u64) -> Header {
+	Header {
+		number,
+		digest: Default::default(),
+		extrinsics_root:  Default::default(),
+		parent_hash: Default::default(),
+		state_root: Default::default(),
+	}
+}
+
 #[test]
 fn submission_should_work() {
 	let pool = pool();
 	block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 209))).unwrap();
 
-	let pending: Vec<_> = pool.ready().map(|a| a.data.transfer().nonce).collect();
+	let pending: Vec<_> = pool.validated_pool().ready().map(|a| a.data.transfer().nonce).collect();
 	assert_eq!(pending, vec![209]);
 }
 
@@ -50,7 +61,7 @@ fn multiple_submission_should_work() {
 	block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 209))).unwrap();
 	block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 210))).unwrap();
 
-	let pending: Vec<_> = pool.ready().map(|a| a.data.transfer().nonce).collect();
+	let pending: Vec<_> = pool.validated_pool().ready().map(|a| a.data.transfer().nonce).collect();
 	assert_eq!(pending, vec![209, 210]);
 }
 
@@ -59,7 +70,7 @@ fn early_nonce_should_be_culled() {
 	let pool = pool();
 	block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 208))).unwrap();
 
-	let pending: Vec<_> = pool.ready().map(|a| a.data.transfer().nonce).collect();
+	let pending: Vec<_> = pool.validated_pool().ready().map(|a| a.data.transfer().nonce).collect();
 	assert_eq!(pending, Vec::<Index>::new());
 }
 
@@ -68,11 +79,11 @@ fn late_nonce_should_be_queued() {
 	let pool = pool();
 
 	block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 210))).unwrap();
-	let pending: Vec<_> = pool.ready().map(|a| a.data.transfer().nonce).collect();
+	let pending: Vec<_> = pool.validated_pool().ready().map(|a| a.data.transfer().nonce).collect();
 	assert_eq!(pending, Vec::<Index>::new());
 
 	block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 209))).unwrap();
-	let pending: Vec<_> = pool.ready().map(|a| a.data.transfer().nonce).collect();
+	let pending: Vec<_> = pool.validated_pool().ready().map(|a| a.data.transfer().nonce).collect();
 	assert_eq!(pending, vec![209, 210]);
 }
 
@@ -82,7 +93,7 @@ fn prune_tags_should_work() {
 	let hash209 = block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 209))).unwrap();
 	block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 210))).unwrap();
 
-	let pending: Vec<_> = pool.ready().map(|a| a.data.transfer().nonce).collect();
+	let pending: Vec<_> = pool.validated_pool().ready().map(|a| a.data.transfer().nonce).collect();
 	assert_eq!(pending, vec![209, 210]);
 
 	block_on(
@@ -93,7 +104,7 @@ fn prune_tags_should_work() {
 		)
 	).expect("Prune tags");
 
-	let pending: Vec<_> = pool.ready().map(|a| a.data.transfer().nonce).collect();
+	let pending: Vec<_> = pool.validated_pool().ready().map(|a| a.data.transfer().nonce).collect();
 	assert_eq!(pending, vec![210]);
 }
 
@@ -102,11 +113,11 @@ fn should_ban_invalid_transactions() {
 	let pool = pool();
 	let uxt = uxt(Alice, 209);
 	let hash = block_on(pool.submit_one(&BlockId::number(0), uxt.clone())).unwrap();
-	pool.remove_invalid(&[hash]);
+	pool.validated_pool().remove_invalid(&[hash]);
 	block_on(pool.submit_one(&BlockId::number(0), uxt.clone())).unwrap_err();
 
 	// when
-	let pending: Vec<_> = pool.ready().map(|a| a.data.transfer().nonce).collect();
+	let pending: Vec<_> = pool.validated_pool().ready().map(|a| a.data.transfer().nonce).collect();
 	assert_eq!(pending, Vec::<Index>::new());
 
 	// then
@@ -122,29 +133,29 @@ fn should_correctly_prune_transactions_providing_more_than_one_tag() {
 	let pool = Pool::new(Default::default(), api.clone());
 	let xt = uxt(Alice, 209);
 	block_on(pool.submit_one(&BlockId::number(0), xt.clone())).expect("1. Imported");
-	assert_eq!(pool.status().ready, 1);
+	assert_eq!(pool.validated_pool().status().ready, 1);
 
 	// remove the transaction that just got imported.
 	api.increment_nonce(Alice.into());
 	block_on(pool.prune_tags(&BlockId::number(1), vec![vec![209]], vec![])).expect("1. Pruned");
-	assert_eq!(pool.status().ready, 0);
+	assert_eq!(pool.validated_pool().status().ready, 0);
 	// it's re-imported to future
-	assert_eq!(pool.status().future, 1);
+	assert_eq!(pool.validated_pool().status().future, 1);
 
 	// so now let's insert another transaction that also provides the 155
 	api.increment_nonce(Alice.into());
 	let xt = uxt(Alice, 211);
 	block_on(pool.submit_one(&BlockId::number(2), xt.clone())).expect("2. Imported");
-	assert_eq!(pool.status().ready, 1);
-	assert_eq!(pool.status().future, 1);
-	let pending: Vec<_> = pool.ready().map(|a| a.data.transfer().nonce).collect();
+	assert_eq!(pool.validated_pool().status().ready, 1);
+	assert_eq!(pool.validated_pool().status().future, 1);
+	let pending: Vec<_> = pool.validated_pool().ready().map(|a| a.data.transfer().nonce).collect();
 	assert_eq!(pending, vec![211]);
 
 	// prune it and make sure the pool is empty
 	api.increment_nonce(Alice.into());
 	block_on(pool.prune_tags(&BlockId::number(3), vec![vec![155]], vec![])).expect("2. Pruned");
-	assert_eq!(pool.status().ready, 0);
-	assert_eq!(pool.status().future, 2);
+	assert_eq!(pool.validated_pool().status().ready, 0);
+	assert_eq!(pool.validated_pool().status().future, 2);
 }
 
 #[test]
@@ -158,7 +169,14 @@ fn should_prune_old_during_maintenance() {
 
 	pool.api.push_block(1, vec![xt.clone()]);
 
-	block_on(pool.maintain(&BlockId::number(1), &[]));
+	let event = ChainEvent::NewBlock {
+		id: BlockId::number(1),
+		is_new_best: true,
+		retracted: vec![],
+		header: header(1),
+	};
+
+	block_on(pool.maintain(event));
 	assert_eq!(pool.status().ready, 0);
 }
 
@@ -174,8 +192,14 @@ fn should_revalidate_during_maintenance() {
 	assert_eq!(pool.api.validation_requests().len(), 2);
 
 	pool.api.push_block(1, vec![xt1.clone()]);
+	let event = ChainEvent::NewBlock {
+		id: BlockId::number(1),
+		is_new_best: true,
+		retracted: vec![],
+		header: header(1),
+	};
 
-	block_on(pool.maintain(&BlockId::number(1), &[]));
+	block_on(pool.maintain(event));
 	assert_eq!(pool.status().ready, 1);
 	// test that pool revalidated transaction that left ready and not included in the block
 	assert_eq!(pool.api.validation_requests().len(), 3);
@@ -193,8 +217,14 @@ fn should_resubmit_from_retracted_during_maintaince() {
 
 	pool.api.push_block(1, vec![]);
 	pool.api.push_fork_block(retracted_hash, vec![xt.clone()]);
+	let event = ChainEvent::NewBlock {
+		id: BlockId::Number(1),
+		is_new_best: true,
+		header: header(1),
+		retracted: vec![retracted_hash]
+	};
 
-	block_on(pool.maintain(&BlockId::number(1), &[retracted_hash]));
+	block_on(pool.maintain(event));
 	assert_eq!(pool.status().ready, 1);
 }
 
@@ -212,7 +242,14 @@ fn should_not_retain_invalid_hashes_from_retracted() {
 	pool.api.push_fork_block(retracted_hash, vec![xt.clone()]);
 	pool.api.add_invalid(&xt);
 
-	block_on(pool.maintain(&BlockId::number(1), &[retracted_hash]));
+	let event = ChainEvent::NewBlock {
+		id: BlockId::Number(1),
+		is_new_best: true,
+		header: header(1),
+		retracted: vec![retracted_hash]
+	};
+
+	block_on(pool.maintain(event));
 	assert_eq!(pool.status().ready, 0);
 }
 
@@ -225,4 +262,188 @@ fn can_track_heap_size() {
 	block_on(pool.submit_one(&BlockId::number(0), uxt(Alice, 212))).expect("1. Imported");
 
 	assert!(parity_util_mem::malloc_size(&pool) > 3000);
-}
+}
+
+#[test]
+fn finalization() {
+	let xt = uxt(Alice, 209);
+	let api = TestApi::with_alice_nonce(209);
+	api.push_block(1, vec![]);
+	let pool = BasicPool::new(Default::default(), api.into());
+	let watcher = block_on(pool.submit_and_watch(&BlockId::number(1), xt.clone())).expect("1. Imported");
+	pool.api.push_block(2, vec![xt.clone()]);
+
+	let header = pool.api.chain().read().header_by_number.get(&2).cloned().unwrap();
+	let event = ChainEvent::NewBlock {
+		id: BlockId::Hash(header.hash()),
+		is_new_best: true,
+		header: header.clone(),
+		retracted: vec![]
+	};
+	block_on(pool.maintain(event));
+
+	let event = ChainEvent::Finalized { hash: header.hash() };
+	block_on(pool.maintain(event));
+
+	let mut stream = futures::executor::block_on_stream(watcher);
+	assert_eq!(stream.next(), Some(TransactionStatus::Ready));
+	assert_eq!(stream.next(), Some(TransactionStatus::InBlock(header.hash())));
+	assert_eq!(stream.next(), Some(TransactionStatus::Finalized(header.hash())));
+	assert_eq!(stream.next(), None);
+}
+
+#[test]
+fn fork_aware_finalization() {
+	let api = TestApi::empty();
+	// starting block A1 (last finalized.)
+	api.push_block(1, vec![]);
+
+	let pool = BasicPool::new(Default::default(), api.into());
+	let mut canon_watchers = vec![];
+
+	let from_alice = uxt(Alice, 1);
+	let from_dave = uxt(Dave, 1);
+	let from_bob = uxt(Bob, 1);
+	let from_charlie = uxt(Charlie, 1);
+	pool.api.increment_nonce(Alice.into());
+	pool.api.increment_nonce(Dave.into());
+	pool.api.increment_nonce(Charlie.into());
+	pool.api.increment_nonce(Bob.into());
+
+	let from_dave_watcher;
+	let from_bob_watcher;
+	let b1;
+	let d1;
+	let c2;
+	let d2;
+
+
+	// block B1
+	{
+		let watcher = block_on(pool.submit_and_watch(&BlockId::number(1), from_alice.clone())).expect("1. Imported");
+		let header = pool.api.push_block(2, vec![from_alice.clone()]);
+		canon_watchers.push((watcher, header.hash()));
+
+		let event = ChainEvent::NewBlock {
+			id: BlockId::Number(2),
+			is_new_best: true,
+			header: header.clone(),
+			retracted: vec![],
+		};
+		b1 = header.hash();
+		block_on(pool.maintain(event));
+		let event = ChainEvent::Finalized { hash: b1 };
+		block_on(pool.maintain(event));
+	}
+
+	// block C2
+	{
+		let header = pool.api.push_fork_block_with_parent(b1, vec![from_dave.clone()]);
+		from_dave_watcher = block_on(pool.submit_and_watch(&BlockId::number(1), from_dave.clone()))
+			.expect("1. Imported");
+		let event = ChainEvent::NewBlock {
+			id: BlockId::Hash(header.hash()),
+			is_new_best: true,
+			header: header.clone(),
+			retracted: vec![]
+		};
+		c2 = header.hash();
+		block_on(pool.maintain(event));
+	}
+
+	// block D2
+	{
+		from_bob_watcher = block_on(pool.submit_and_watch(&BlockId::number(1), from_bob.clone())).expect("1. Imported");
+		let header = pool.api.push_fork_block_with_parent(c2, vec![from_bob.clone()]);
+
+		let event = ChainEvent::NewBlock {
+			id: BlockId::Hash(header.hash()),
+			is_new_best: true,
+			header: header.clone(),
+			retracted: vec![]
+		};
+		d2 = header.hash();
+		block_on(pool.maintain(event));
+	}
+
+	// block C1
+	{
+		let watcher = block_on(pool.submit_and_watch(&BlockId::number(1), from_charlie.clone())).expect("1.Imported");
+		let header = pool.api.push_block(3, vec![from_charlie.clone()]);
+
+		canon_watchers.push((watcher, header.hash()));
+		let event = ChainEvent::NewBlock {
+			id: BlockId::Number(3),
+			is_new_best: true,
+			header: header.clone(),
+			retracted: vec![c2, d2],
+		};
+		block_on(pool.maintain(event));
+		let event = ChainEvent::Finalized { hash: header.hash() };
+		block_on(pool.maintain(event));
+	}
+
+	// block D1
+	{
+		let xt = uxt(Eve, 0);
+		let w = block_on(pool.submit_and_watch(&BlockId::number(1), xt.clone())).expect("1. Imported");
+		let header = pool.api.push_block(4, vec![xt.clone()]);
+		canon_watchers.push((w, header.hash()));
+
+		let event = ChainEvent::NewBlock {
+			id: BlockId::Hash(header.hash()),
+			is_new_best: true,
+			header: header.clone(),
+			retracted: vec![]
+		};
+		d1 = header.hash();
+		block_on(pool.maintain(event));
+		let event = ChainEvent::Finalized { hash: d1 };
+		block_on(pool.maintain(event));
+	}
+
+	let e1;
+
+	// block e1
+	{
+		let header = pool.api.push_block(5, vec![from_dave]);
+		e1 = header.hash();
+		let event = ChainEvent::NewBlock {
+			id: BlockId::Hash(header.hash()),
+			is_new_best: true,
+			header: header.clone(),
+			retracted: vec![]
+		};
+		block_on(pool.maintain(event));
+		block_on(pool.maintain(ChainEvent::Finalized { hash: e1 }));
+	}
+
+
+	for (canon_watcher, h) in canon_watchers {
+		let mut stream = futures::executor::block_on_stream(canon_watcher);
+		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(h.clone())));
+		assert_eq!(stream.next(), Some(TransactionStatus::Finalized(h)));
+		assert_eq!(stream.next(), None);
+	}
+
+
+	{
+		let mut stream= futures::executor::block_on_stream(from_dave_watcher);
+		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(c2.clone())));
+		assert_eq!(stream.next(), Some(TransactionStatus::Retracted(c2)));
+		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(e1)));
+		assert_eq!(stream.next(), Some(TransactionStatus::Finalized(e1.clone())));
+		assert_eq!(stream.next(), None);
+	}
+
+	{
+		let mut stream= futures::executor::block_on_stream(from_bob_watcher);
+		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(d2.clone())));
+		assert_eq!(stream.next(), Some(TransactionStatus::Retracted(d2)));
+	}
+
+}

substrate/primitives/blockchain/src/header_metadata.rs

@@ -151,7 +151,7 @@ pub fn tree_route<Block: BlockT, T: HeaderMetadata<Block>>(
 }
 
 /// Hash and number of a block.
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct HashAndNumber<Block: BlockT> {
 	/// The number of the block.
 	pub number: NumberFor<Block>,

substrate/primitives/transaction-pool/src/pool.rs

@@ -67,7 +67,7 @@ impl PoolStatus {
 /// 2. Inside `Ready` queue:
 ///		- `Broadcast`
 /// 3. Leaving the pool:
-///		- `InBlock`
+///		- `Finalized`
 ///		- `Invalid`
 ///		- `Usurped`
 ///		- `Dropped`
@@ -100,6 +100,13 @@ pub enum TransactionStatus<Hash, BlockHash> {
 	/// Transaction has been included in block with given hash.
 	#[serde(rename = "finalized")] // See #4438
 	InBlock(BlockHash),
+	/// The block this transaction was included in has been retracted.
+	Retracted(BlockHash),
+	/// Maximum number of finality watchers has been reached,
+	/// old watchers are being removed.
+	FinalityTimeout(BlockHash),
+	/// Transaction has been finalized by a finality-gadget, e.g GRANDPA
+	Finalized(BlockHash),
 	/// Transaction has been replaced in the pool, by another transaction
 	/// that provides the same tags. (e.g. same (sender, nonce)).
 	Usurped(Hash),
@@ -217,11 +224,30 @@ pub trait TransactionPool: Send + Sync {
 	fn ready_transaction(&self, hash: &TxHash<Self>) -> Option<Arc<Self::InPoolTransaction>>;
 }
 
+/// Events that the transaction pool listens for.
+pub enum ChainEvent<B: BlockT> {
+	/// New blocks have been added to the chain
+	NewBlock {
+		/// Is this the new best block.
+		is_new_best: bool,
+		/// Id of the just imported block.
+		id: BlockId<B>,
+		/// Header of the just imported block
+		header: B::Header,
+		/// List of retracted blocks ordered by block number.
+		retracted: Vec<B::Hash>,
+	},
+	/// An existing block has been finalzied.
+	Finalized {
+		/// Hash of just finalized block
+		hash: B::Hash,
+	},
+}
+
 /// Trait for transaction pool maintenance.
-pub trait MaintainedTransactionPool : TransactionPool {
+pub trait MaintainedTransactionPool: TransactionPool {
 	/// Perform maintenance
-	fn maintain(&self, block: &BlockId<Self::Block>, retracted: &[BlockHash<Self>])
-		-> Pin<Box<dyn Future<Output=()> + Send>>;
+	fn maintain(&self, event: ChainEvent<Self::Block>) -> Pin<Box<dyn Future<Output=()> + Send>>;
 }
 
 /// An abstraction for transaction pool.

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

@@ -9,6 +9,7 @@ license = "GPL-3.0"
 substrate-test-runtime-client = { version = "2.0.0", path = "../client" }
 parking_lot = "0.10.0"
 codec = { package = "parity-scale-codec", version = "1.0.0" }
+sp-blockchain = { version = "2.0.0", path = "../../../primitives/blockchain" }
 sp-runtime = { version = "2.0.0", path = "../../../primitives/runtime" }
 sp-transaction-pool = { version = "2.0.0", path = "../../../primitives/transaction-pool" }
 sc-transaction-graph = { version = "2.0.0", path = "../../../client/transaction-pool/graph" }

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

@@ -50,7 +50,7 @@ impl std::error::Error for Error {
 }
 
 #[derive(Default)]
-struct ChainState {
+pub struct ChainState {
 	pub block_by_number: HashMap<BlockNumber, Vec<Extrinsic>>,
 	pub block_by_hash: HashMap<Hash, Vec<Extrinsic>>,
 	pub header_by_number: HashMap<BlockNumber, Header>,
@@ -96,16 +96,24 @@ impl TestApi {
 	}
 
 	/// Push block as a part of canonical chain under given number.
-	pub fn push_block(&self, block_number: BlockNumber, xts: Vec<Extrinsic>) {
+	pub fn push_block(&self, block_number: BlockNumber, xts: Vec<Extrinsic>) -> Header {
 		let mut chain = self.chain.write();
-		chain.block_by_number.insert(block_number, xts);
-		chain.header_by_number.insert(block_number, Header {
+		chain.block_by_number.insert(block_number, xts.clone());
+		let header = Header {
 			number: block_number,
 			digest: Default::default(),
 			extrinsics_root:  Default::default(),
-			parent_hash: Default::default(),
+			parent_hash: block_number
+				.checked_sub(1)
+				.and_then(|num| {
+					chain.header_by_number.get(&num)
+						.cloned().map(|h| h.hash())
+				}).unwrap_or_default(),
 			state_root: Default::default(),
-		});
+		};
+		chain.block_by_hash.insert(header.hash(), xts);
+		chain.header_by_number.insert(block_number, header.clone());
+		header
 	}
 
 	/// Push a block without a number.
@@ -116,6 +124,20 @@ impl TestApi {
 		chain.block_by_hash.insert(block_hash, xts);
 	}
 
+	pub fn push_fork_block_with_parent(&self, parent: Hash, xts: Vec<Extrinsic>) -> Header {
+		let mut chain = self.chain.write();
+		let blocknum = chain.block_by_number.keys().max().expect("block_by_number shouldn't be empty");
+		let header = Header {
+			number: *blocknum,
+			digest: Default::default(),
+			extrinsics_root:  Default::default(),
+			parent_hash: parent,
+			state_root: Default::default(),
+		};
+		chain.block_by_hash.insert(header.hash(), xts);
+		header
+	}
+
 	fn hash_and_length_inner(ex: &Extrinsic) -> (Hash, usize) {
 		let encoded = ex.encode();
 		(BlakeTwo256::hash(&encoded), encoded.len())
@@ -136,6 +158,11 @@ impl TestApi {
 		self.validation_requests.read().clone()
 	}
 
+	/// get a reference to the chain state
+	pub fn chain(&self) -> &RwLock<ChainState> {
+		&self.chain
+	}
+
 	/// Increment nonce in the inner state.
 	pub fn increment_nonce(&self, account: AccountId) {
 		let mut chain = self.chain.write();
@@ -197,7 +224,12 @@ impl sc_transaction_graph::ChainApi for TestApi {
 	) -> Result<Option<sc_transaction_graph::BlockHash<Self>>, Error> {
 		Ok(match at {
 			generic::BlockId::Hash(x) => Some(x.clone()),
-			_ => Some(Default::default()),
+			generic::BlockId::Number(num) => {
+				self.chain.read()
+					.header_by_number.get(num)
+					.map(|h| h.hash())
+					.or_else(|| Some(Default::default()))
+			},
 		})
 	}
 
@@ -209,10 +241,9 @@ impl sc_transaction_graph::ChainApi for TestApi {
 	}
 
 	fn block_body(&self, id: &BlockId<Self::Block>) -> Self::BodyFuture {
-		futures::future::ready(Ok(if let BlockId::Number(num) = id {
-			self.chain.read().block_by_number.get(num).cloned()
-		} else {
-			None
+		futures::future::ready(Ok(match id {
+			BlockId::Number(num) => self.chain.read().block_by_number.get(num).cloned(),
+			BlockId::Hash(hash) => self.chain.read().block_by_hash.get(hash).cloned(),
 		}))
 	}
 }