Finalized block event triggers tx maintanance

client/transaction-pool/src/lib.rs

@@ -563,153 +563,179 @@ async fn prune_known_txs_for_block<Block: BlockT, Api: graph::ChainApi<Block = B
 	hashes
 }
 
-impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
+impl<PoolApi, Block> BasicPool<PoolApi, Block>
 where
 	Block: BlockT,
 	PoolApi: 'static + graph::ChainApi<Block = Block>,
 {
-	fn maintain(&self, event: ChainEvent<Self::Block>) -> Pin<Box<dyn Future<Output = ()> + Send>> {
-		match event {
-			ChainEvent::NewBestBlock { hash, tree_route } => {
-				let pool = self.pool.clone();
-				let api = self.api.clone();
-
-				let id = BlockId::hash(hash);
-				let block_number = match api.block_id_to_number(&id) {
-					Ok(Some(number)) => number,
-					_ => {
-						log::trace!(
-							target: "txpool",
-							"Skipping chain event - no number for that block {:?}",
-							id,
-						);
-						return Box::pin(ready(()))
-					},
-				};
+	fn handle_blocks_reorg(
+		&self,
+		hash: Block::Hash,
+		enacted: Vec<Block::Hash>,
+		retracted: Vec<Block::Hash>,
+	) -> Pin<Box<dyn Future<Output = ()> + Send>> {
+		println!("xxx handle_blocks_reorg {hash:?} retracted: {retracted:?} enacted: {enacted:?}");
 
-				let next_action = self.revalidation_strategy.lock().next(
-					block_number,
-					Some(std::time::Duration::from_secs(60)),
-					Some(20u32.into()),
+		let pool = self.pool.clone();
+		let api = self.api.clone();
+
+		let id = BlockId::hash(hash);
+		let block_number = match api.block_id_to_number(&id) {
+			Ok(Some(number)) => number,
+			_ => {
+				log::trace!(
+				target: "txpool",
+				"Skipping chain event - no number for that block {:?}",
+				id,
 				);
-				let revalidation_strategy = self.revalidation_strategy.clone();
-				let revalidation_queue = self.revalidation_queue.clone();
-				let ready_poll = self.ready_poll.clone();
-				let metrics = self.metrics.clone();
+				return Box::pin(ready(()))
+			},
+		};
 
-				async move {
-					// We keep track of everything we prune so that later we won't add
-					// transactions with those hashes from the retracted blocks.
-					let mut pruned_log = HashSet::<ExtrinsicHash<PoolApi>>::new();
-
-					// If there is a tree route, we use this to prune known tx based on the enacted
-					// blocks. Before pruning enacted transactions, we inform the listeners about
-					// retracted blocks and their transactions. This order is important, because
-					// if we enact and retract the same transaction at the same time, we want to
-					// send first the retract and than the prune event.
-					if let Some(ref tree_route) = tree_route {
-						for retracted in tree_route.retracted() {
-							// notify txs awaiting finality that it has been retracted
-							pool.validated_pool().on_block_retracted(retracted.hash);
-						}
+		let next_action = self.revalidation_strategy.lock().next(
+			block_number,
+			Some(std::time::Duration::from_secs(60)),
+			Some(20u32.into()),
+		);
+		let revalidation_strategy = self.revalidation_strategy.clone();
+		let revalidation_queue = self.revalidation_queue.clone();
+		let ready_poll = self.ready_poll.clone();
+		let metrics = self.metrics.clone();
 
-						future::join_all(tree_route.enacted().iter().map(|h| {
-							prune_known_txs_for_block(BlockId::Hash(h.hash), &*api, &*pool)
-						}))
+		async move {
+			// We keep track of everything we prune so that later we won't add
+			// transactions with those hashes from the retracted blocks.
+			let mut pruned_log = HashSet::<ExtrinsicHash<PoolApi>>::new();
+
+			// If there is a tree route, we use this to prune known tx based on the enacted
+			// blocks. Before pruning enacted transactions, we inform the listeners about
+			// retracted blocks and their transactions. This order is important, because
+			// if we enact and retract the same transaction at the same time, we want to
+			// send first the retract and than the prune event.
+			for retracted_hash in retracted.iter() {
+				// notify txs awaiting finality that it has been retracted
+				pool.validated_pool().on_block_retracted(*retracted_hash);
+			}
+
+			future::join_all(
+				enacted
+					.iter()
+					.map(|hash| prune_known_txs_for_block(BlockId::Hash(*hash), &*api, &*pool)),
+			)
+			.await
+			.into_iter()
+			.for_each(|enacted_log| {
+				pruned_log.extend(enacted_log);
+			});
+
+			pruned_log.extend(prune_known_txs_for_block(id, &*api, &*pool).await);
+
+			metrics.report(|metrics| {
+				metrics.block_transactions_pruned.inc_by(pruned_log.len() as u64)
+			});
+
+			// is Some(tree_route) == (enacted.len != 0 || retracted.len != 0)
+			if let true = next_action.resubmit {
+				let mut resubmit_transactions = Vec::new();
+
+				for hash in retracted.iter() {
+					let block_transactions = api
+						.block_body(&BlockId::hash(*hash))
 						.await
-						.into_iter()
-						.for_each(|enacted_log| {
-							pruned_log.extend(enacted_log);
+						.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));
 
-					pruned_log.extend(prune_known_txs_for_block(id, &*api, &*pool).await);
+					let mut resubmitted_to_report = 0;
 
-					metrics.report(|metrics| {
-						metrics.block_transactions_pruned.inc_by(pruned_log.len() as u64)
-					});
+					resubmit_transactions.extend(block_transactions.into_iter().filter(|tx| {
+						let tx_hash = pool.hash_of(tx);
+						let contains = pruned_log.contains(&tx_hash);
 
-					if let (true, Some(tree_route)) = (next_action.resubmit, tree_route) {
-						let mut resubmit_transactions = Vec::new();
-
-						for retracted in tree_route.retracted() {
-							let hash = retracted.hash;
-
-							let block_transactions = api
-								.block_body(&BlockId::hash(hash))
-								.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));
-
-							let mut resubmitted_to_report = 0;
-
-							resubmit_transactions.extend(block_transactions.into_iter().filter(
-								|tx| {
-									let tx_hash = pool.hash_of(tx);
-									let contains = pruned_log.contains(&tx_hash);
-
-									// need to count all transactions, not just filtered, here
-									resubmitted_to_report += 1;
-
-									if !contains {
-										log::debug!(
-											target: "txpool",
-											"[{:?}]: Resubmitting from retracted block {:?}",
-											tx_hash,
-											hash,
-										);
-									}
-									!contains
-								},
-							));
-
-							metrics.report(|metrics| {
-								metrics.block_transactions_resubmitted.inc_by(resubmitted_to_report)
-							});
-						}
+						// need to count all transactions, not just filtered, here
+						resubmitted_to_report += 1;
 
-						if let Err(e) = pool
-							.resubmit_at(
-								&id,
-								// These transactions are coming from retracted blocks, we should
-								// simply consider them external.
-								TransactionSource::External,
-								resubmit_transactions,
-							)
-							.await
-						{
+						if !contains {
 							log::debug!(
-								target: "txpool",
-								"[{:?}] Error re-submitting transactions: {}",
-								id,
-								e,
-							)
+							target: "txpool",
+							"[{:?}]: Resubmitting from retracted block {:?}",
+							tx_hash,
+							hash,
+							);
 						}
-					}
+						!contains
+					}));
 
-					let extra_pool = pool.clone();
-					// After #5200 lands, this arguably might be moved to the
-					// handler of "all blocks notification".
-					ready_poll.lock().trigger(block_number, move || {
-						Box::new(extra_pool.validated_pool().ready())
+					metrics.report(|metrics| {
+						metrics.block_transactions_resubmitted.inc_by(resubmitted_to_report)
 					});
+				}
 
-					if next_action.revalidate {
-						let hashes = pool.validated_pool().ready().map(|tx| tx.hash).collect();
-						revalidation_queue.revalidate_later(block_number, hashes).await;
-
-						revalidation_strategy.lock().clear();
-					}
+				if let Err(e) = pool
+					.resubmit_at(
+						&id,
+						// These transactions are coming from retracted blocks, we should
+						// simply consider them external.
+						TransactionSource::External,
+						resubmit_transactions,
+					)
+					.await
+				{
+					log::debug!(
+					target: "txpool",
+					"[{:?}] Error re-submitting transactions: {}",
+					id,
+					e,
+					)
 				}
-				.boxed()
+			}
+
+			let extra_pool = pool.clone();
+			// After #5200 lands, this arguably might be moved to the
+			// handler of "all blocks notification".
+			ready_poll
+				.lock()
+				.trigger(block_number, move || Box::new(extra_pool.validated_pool().ready()));
+
+			if next_action.revalidate {
+				let hashes = pool.validated_pool().ready().map(|tx| tx.hash).collect();
+				revalidation_queue.revalidate_later(block_number, hashes).await;
+
+				revalidation_strategy.lock().clear();
+			}
+		}
+		.boxed()
+	}
+}
+
+impl<PoolApi, Block> MaintainedTransactionPool for BasicPool<PoolApi, Block>
+where
+	Block: BlockT,
+	PoolApi: 'static + graph::ChainApi<Block = Block>,
+{
+	fn maintain(&self, event: ChainEvent<Self::Block>) -> Pin<Box<dyn Future<Output = ()> + Send>> {
+		match event {
+			ChainEvent::NewBestBlock { hash, tree_route } => {
+				println!("xxx NewBestBlock {hash:?} tree_route: {tree_route:?}");
+				let (enacted, retracted) = if let Some(tree_route) = tree_route {
+					(
+						tree_route.enacted().iter().map(|block| block.hash).collect(),
+						tree_route.retracted().iter().map(|block| block.hash).collect(),
+					)
+				} else {
+					(vec![], vec![])
+				};
+
+				self.handle_blocks_reorg(hash, enacted, retracted)
 			},
 			ChainEvent::Finalized { hash, tree_route } => {
 				let pool = self.pool.clone();
+				println!("xxx Finalized {hash:?} {tree_route:?}");
+				self.handle_blocks_reorg(hash, tree_route.to_vec(), vec![]);
 				async move {
 					for hash in tree_route.iter().chain(&[hash]) {
 						if let Err(e) = pool.validated_pool().on_block_finalized(*hash).await {

client/transaction-pool/tests/pool.rs

@@ -457,6 +457,7 @@ fn finalization() {
 
 #[test]
 fn fork_aware_finalization() {
+	sp_tracing::try_init_simple();
 	let api = TestApi::empty();
 	// starting block A1 (last finalized.)
 	api.push_block(1, vec![], true);
@@ -997,3 +998,42 @@ fn stale_transactions_are_pruned() {
 	assert_eq!(pool.status().future, 0);
 	assert_eq!(pool.status().ready, 0);
 }
+
+#[test]
+fn playground() {
+	sp_tracing::try_init_simple();
+	println!("xxxx");
+	let xt = uxt(Alice, 209);
+
+	let (pool, api, _guard) = maintained_pool();
+
+	// block_on(pool.submit_one(&BlockId::number(0), SOURCE, xt.clone())).expect("1. Imported");
+	let watcher = block_on(pool.submit_and_watch(&BlockId::number(0), SOURCE, xt.clone()))
+		.expect("1. Imported");
+	assert_eq!(pool.status().ready, 1);
+
+	let header = api.push_block(1, vec![xt.clone()], true);
+
+	let ps = pool.pool().validated_pool().status();
+	println!("status: {ps:?}");
+	// block_on(pool.maintain(block_event(header.clone())));
+	// let ps = pool.pool().validated_pool().status();
+	// println!("status: {ps:?}");
+
+	let event =
+		ChainEvent::Finalized { hash: header.clone().hash(), tree_route: Arc::from(vec![]) };
+	block_on(pool.maintain(event));
+
+	let ps = pool.pool().validated_pool().status();
+	println!("status: {ps:?}");
+
+	assert_eq!(pool.status().ready, 0);
+
+	{
+		let mut stream = futures::executor::block_on_stream(watcher);
+		assert_eq!(stream.next(), Some(TransactionStatus::Ready));
+		assert_eq!(stream.next(), Some(TransactionStatus::InBlock(header.clone().hash())));
+		assert_eq!(stream.next(), Some(TransactionStatus::Finalized(header.hash())));
+		assert_eq!(stream.next(), None);
+	}
+}