From 5f119f711f5188173e9bf8b6f4feb9ad4b8b36db Mon Sep 17 00:00:00 2001 From: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com> Date: Fri, 20 Jun 2025 14:54:47 +0200 Subject: [PATCH 1/6] dropped stream: remove invalid txs --- .../src/fork_aware_txpool/fork_aware_txpool.rs | 16 ++++++++++++---- .../src/fork_aware_txpool/tx_mem_pool.rs | 14 ++++++++++++-- .../src/fork_aware_txpool/view_store.rs | 10 ++++++++++ 3 files changed, 34 insertions(+), 6 deletions(-) diff --git a/substrate/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs b/substrate/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs index 3553465668e3..504d680ba73b 100644 --- a/substrate/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs +++ b/substrate/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs @@ -238,8 +238,12 @@ where let (dropped_stream_controller, dropped_stream) = MultiViewDroppedWatcherController::::new(); - let view_store = - Arc::new(ViewStore::new(pool_api.clone(), listener, dropped_stream_controller)); + let view_store = Arc::new(ViewStore::new( + pool_api.clone(), + listener, + dropped_stream_controller, + import_notification_sink.clone(), + )); let dropped_monitor_task = Self::dropped_monitor_task( dropped_stream, @@ -372,8 +376,12 @@ where let (dropped_stream_controller, dropped_stream) = MultiViewDroppedWatcherController::::new(); - let view_store = - Arc::new(ViewStore::new(pool_api.clone(), listener, dropped_stream_controller)); + let view_store = Arc::new(ViewStore::new( + pool_api.clone(), + listener, + dropped_stream_controller, + import_notification_sink.clone(), + )); let dropped_monitor_task = Self::dropped_monitor_task( dropped_stream, diff --git a/substrate/client/transaction-pool/src/fork_aware_txpool/tx_mem_pool.rs b/substrate/client/transaction-pool/src/fork_aware_txpool/tx_mem_pool.rs index 559f11da4cdb..38aa31001476 100644 --- a/substrate/client/transaction-pool/src/fork_aware_txpool/tx_mem_pool.rs +++ b/substrate/client/transaction-pool/src/fork_aware_txpool/tx_mem_pool.rs @@ -605,8 +605,18 @@ where let revalidated_invalid_hashes_len = revalidated_invalid_hashes.len(); let invalid_hashes_subtrees_len = invalid_hashes_subtrees.len(); - self.listener - .transactions_invalidated(&invalid_hashes_subtrees.into_iter().collect::>()); + let invalid_hashes_subtrees = invalid_hashes_subtrees.into_iter().collect::>(); + + //note: here the consistency is assumed: it is expected that transaction will be + // actually removed from the listener with Invalid event. This means assumption that no view + // is referencing tx as ready. + self.listener.transactions_invalidated(&invalid_hashes_subtrees); + view_store + .import_notification_sink + .clean_notified_items(&invalid_hashes_subtrees); + view_store + .dropped_stream_controller + .remove_transactions(invalid_hashes_subtrees); trace!( target: LOG_TARGET, diff --git a/substrate/client/transaction-pool/src/fork_aware_txpool/view_store.rs b/substrate/client/transaction-pool/src/fork_aware_txpool/view_store.rs index bcf1e8d5ceef..6ccb708b631c 100644 --- a/substrate/client/transaction-pool/src/fork_aware_txpool/view_store.rs +++ b/substrate/client/transaction-pool/src/fork_aware_txpool/view_store.rs @@ -19,6 +19,7 @@ //! Transaction pool view store. Basically block hash to view map with some utility methods. use super::{ + import_notification_sink::MultiViewImportNotificationSink, multi_view_listener::{MultiViewListener, TxStatusStream}, view::{View, ViewPoolObserver}, }; @@ -171,6 +172,10 @@ where pub(super) most_recent_view: RwLock>, /// The controller of multi view dropped stream. pub(super) dropped_stream_controller: MultiViewDroppedWatcherController, + /// Util providing an aggregated stream of transactions that were imported to ready queue in + /// any view. Reference kept here for clean up purposes. + pub(super) import_notification_sink: + MultiViewImportNotificationSink>, /// The map used to synchronize replacement of transactions between maintain and dropped /// notifcication threads. It is meant to assure that replaced transaction is also removed from /// newly built views in maintain process. @@ -202,6 +207,10 @@ where api: Arc, listener: Arc>, dropped_stream_controller: MultiViewDroppedWatcherController, + import_notification_sink: MultiViewImportNotificationSink< + Block::Hash, + ExtrinsicHash, + >, ) -> Self { Self { api, @@ -210,6 +219,7 @@ where listener, most_recent_view: RwLock::from(None), dropped_stream_controller, + import_notification_sink, pending_txs_tasks: Default::default(), } } From eaa2a277397870048e34ff139cb98fa4357e61e4 Mon Sep 17 00:00:00 2001 From: "cmd[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Fri, 20 Jun 2025 13:07:58 +0000 Subject: [PATCH 2/6] Update from github-actions[bot] running command 'prdoc --bump minor --audience node_dev' --- prdoc/pr_8923.prdoc | 11 +++++++++++ 1 file changed, 11 insertions(+) create mode 100644 prdoc/pr_8923.prdoc diff --git a/prdoc/pr_8923.prdoc b/prdoc/pr_8923.prdoc new file mode 100644 index 000000000000..c2bb6982aa6f --- /dev/null +++ b/prdoc/pr_8923.prdoc @@ -0,0 +1,11 @@ +title: '`fatxpool`: fix: remove invalid txs from the dropped stream controller' +doc: +- audience: Node Dev + description: |- + While testing mortal transaction I encountered exactly the same problem as in #8490. + This PR should fix the problem. + + fixes: #8490 +crates: +- name: sc-transaction-pool + bump: minor From a643a2816a657945ff8dae41ff85760790ffcac9 Mon Sep 17 00:00:00 2001 From: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com> Date: Fri, 20 Jun 2025 18:46:20 +0200 Subject: [PATCH 3/6] test added --- .../fork_aware_txpool/fork_aware_txpool.rs | 7 +++ .../import_notification_sink.rs | 7 +++ .../src/graph/validated_pool.rs | 19 +++++--- .../client/transaction-pool/tests/fatp.rs | 45 +++++++++++++++++++ .../runtime/transaction-pool/src/lib.rs | 24 +++++++++- 5 files changed, 94 insertions(+), 8 deletions(-) diff --git a/substrate/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs b/substrate/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs index 504d680ba73b..0a2199356cc7 100644 --- a/substrate/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs +++ b/substrate/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs @@ -666,6 +666,13 @@ where ); (false, pending) } + + /// Number of notified items in import_notification_sink. + /// + /// Internal detail, exposed only for testing. + pub fn import_notification_sink_len(&self) -> usize { + self.import_notification_sink.notified_items_len() + } } /// Converts the input view-to-statuses map into the output vector of statuses. diff --git a/substrate/client/transaction-pool/src/fork_aware_txpool/import_notification_sink.rs b/substrate/client/transaction-pool/src/fork_aware_txpool/import_notification_sink.rs index 1ca287fa2371..a9a8eb811e4d 100644 --- a/substrate/client/transaction-pool/src/fork_aware_txpool/import_notification_sink.rs +++ b/substrate/client/transaction-pool/src/fork_aware_txpool/import_notification_sink.rs @@ -245,6 +245,13 @@ where already_notified_items.remove(i); }); } + + /// Lenght of the `already_notified_items` set. + /// + /// Exposed for testing only. + pub fn notified_items_len(&self) -> usize { + self.already_notified_items.read().len() + } } #[cfg(test)] diff --git a/substrate/client/transaction-pool/src/graph/validated_pool.rs b/substrate/client/transaction-pool/src/graph/validated_pool.rs index caccc3370a0a..0a1712dcb3b7 100644 --- a/substrate/client/transaction-pool/src/graph/validated_pool.rs +++ b/substrate/client/transaction-pool/src/graph/validated_pool.rs @@ -16,11 +16,6 @@ // You should have received a copy of the GNU General Public License // along with this program. If not, see . -use std::{ - collections::{HashMap, HashSet}, - sync::Arc, -}; - use crate::{common::tracing_log_xt::log_xt_trace, LOG_TARGET}; use futures::channel::mpsc::{channel, Sender}; use indexmap::IndexMap; @@ -31,8 +26,12 @@ use sp_runtime::{ traits::SaturatedConversion, transaction_validity::{TransactionTag as Tag, ValidTransaction}, }; -use std::time::Instant; -use tracing::{debug, trace, warn}; +use std::{ + collections::{HashMap, HashSet}, + sync::Arc, + time::{Duration, Instant}, +}; +use tracing::{debug, trace, warn, Level}; use super::{ base_pool::{self as base, PruneStatus}, @@ -704,6 +703,12 @@ impl> ValidatedPool { } hashes }; + debug!( + target:LOG_TARGET, + to_remove_len=to_remove.len(), + futures_to_remove_len=futures_to_remove.len(), + "clear_stale" + ); // removing old transactions self.remove_invalid(&to_remove); self.remove_invalid(&futures_to_remove); diff --git a/substrate/client/transaction-pool/tests/fatp.rs b/substrate/client/transaction-pool/tests/fatp.rs index c1411b29dafb..57d8b04fdd90 100644 --- a/substrate/client/transaction-pool/tests/fatp.rs +++ b/substrate/client/transaction-pool/tests/fatp.rs @@ -484,6 +484,51 @@ fn fatp_linear_old_ready_becoming_stale() { } } +#[test] +fn fatp_proper_cleanup_after_mortal_tx_becoming_invalid() { + sp_tracing::try_init_simple(); + + let (pool, api, _) = pool(); + + let xts = vec![uxt(Alice, 200), uxt(Alice, 201), uxt(Alice, 202)]; + + api.set_valid_till(&xts[0], 66); + api.set_valid_till(&xts[1], 66); + api.set_valid_till(&xts[2], 66); + + let header01 = api.push_block(1, vec![], true); + let event = new_best_block_event(&pool, None, header01.hash()); + block_on(pool.maintain(event)); + + xts.into_iter().for_each(|xt| { + block_on(pool.submit_one(invalid_hash(), SOURCE, xt)).unwrap(); + }); + assert_eq!(pool.status_all()[&header01.hash()].ready, 3); + assert_eq!(pool.status_all()[&header01.hash()].future, 0); + + // Import enough blocks to make our transactions stale (longevity is 64) + let mut prev_header = header01; + for n in 2..67 { + let header = api.push_block_with_parent(prev_header.hash(), vec![], true); + let event = new_best_block_event(&pool, Some(prev_header.hash()), header.hash()); + block_on(pool.maintain(event)); + + if n == 66 { + assert_eq!(pool.status_all()[&header.hash()].ready, 0); + assert_eq!(pool.status_all()[&header.hash()].future, 0); + } else { + assert_eq!(pool.status_all()[&header.hash()].ready, 3); + assert_eq!(pool.status_all()[&header.hash()].future, 0); + } + prev_header = header; + } + + let header = api.push_block_with_parent(prev_header.hash(), vec![], true); + let event = finalized_block_event(&pool, prev_header.hash(), header.hash()); + block_on(pool.maintain(event)); + assert_eq!(pool.import_notification_sink_len(), 0); +} + #[test] fn fatp_fork_reorg() { sp_tracing::try_init_simple(); diff --git a/substrate/test-utils/runtime/transaction-pool/src/lib.rs b/substrate/test-utils/runtime/transaction-pool/src/lib.rs index f88694fb1071..a4875deed496 100644 --- a/substrate/test-utils/runtime/transaction-pool/src/lib.rs +++ b/substrate/test-utils/runtime/transaction-pool/src/lib.rs @@ -85,6 +85,7 @@ pub struct ChainState { pub nonces: HashMap>, pub invalid_hashes: HashSet, pub priorities: HashMap, + pub valid_till_blocks: HashMap, } /// Test Api for transaction pool. @@ -269,6 +270,14 @@ impl TestApi { .insert(Self::hash_and_length_inner(xts).0, priority); } + /// Set a transaction mortality (block at which it will expire). + pub fn set_valid_till(&self, xts: &Extrinsic, valid_till: u64) { + self.chain + .write() + .valid_till_blocks + .insert(Self::hash_and_length_inner(xts).0, valid_till); + } + /// Query validation requests received. pub fn validation_requests(&self) -> Vec { self.validation_requests.read().clone() @@ -442,11 +451,24 @@ impl ChainApi for TestApi { } let priority = self.chain.read().priorities.get(&self.hash_and_length(&uxt).0).cloned(); + let longevity = self + .chain + .read() + .valid_till_blocks + .get(&self.hash_and_length(&uxt).0) + .cloned() + .map(|valid_till| valid_till.saturating_sub(block_number.unwrap())) + .unwrap_or(64); + + if longevity == 0 { + return Ok(Err(TransactionValidityError::Invalid(InvalidTransaction::AncientBirthBlock))) + } + let mut validity = ValidTransaction { priority: priority.unwrap_or(1), requires, provides, - longevity: 64, + longevity, propagate: true, }; From f16b9fcc4354aa4a75447bc447c595b18b1335f0 Mon Sep 17 00:00:00 2001 From: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com> Date: Sat, 21 Jun 2025 09:54:20 +0200 Subject: [PATCH 4/6] Update substrate/test-utils/runtime/transaction-pool/src/lib.rs --- substrate/test-utils/runtime/transaction-pool/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/substrate/test-utils/runtime/transaction-pool/src/lib.rs b/substrate/test-utils/runtime/transaction-pool/src/lib.rs index a4875deed496..27fd260569a0 100644 --- a/substrate/test-utils/runtime/transaction-pool/src/lib.rs +++ b/substrate/test-utils/runtime/transaction-pool/src/lib.rs @@ -461,7 +461,7 @@ impl ChainApi for TestApi { .unwrap_or(64); if longevity == 0 { - return Ok(Err(TransactionValidityError::Invalid(InvalidTransaction::AncientBirthBlock))) + return Ok(Err(TransactionValidityError::Invalid(InvalidTransaction::BadProof))) } let mut validity = ValidTransaction { From 79ca21a91834d00a2906022f8a921b9c8b46f985 Mon Sep 17 00:00:00 2001 From: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com> Date: Mon, 23 Jun 2025 08:32:32 +0200 Subject: [PATCH 5/6] fix --- substrate/client/transaction-pool/src/graph/validated_pool.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/substrate/client/transaction-pool/src/graph/validated_pool.rs b/substrate/client/transaction-pool/src/graph/validated_pool.rs index 0a1712dcb3b7..8d67ecd164d9 100644 --- a/substrate/client/transaction-pool/src/graph/validated_pool.rs +++ b/substrate/client/transaction-pool/src/graph/validated_pool.rs @@ -29,9 +29,9 @@ use sp_runtime::{ use std::{ collections::{HashMap, HashSet}, sync::Arc, - time::{Duration, Instant}, + time::Instant, }; -use tracing::{debug, trace, warn, Level}; +use tracing::{debug, trace, warn}; use super::{ base_pool::{self as base, PruneStatus}, From a17d9de499968bf368d47ece6d35944165c0e6df Mon Sep 17 00:00:00 2001 From: Michal Kucharczyk <1728078+michalkucharczyk@users.noreply.github.com> Date: Tue, 24 Jun 2025 22:49:45 +0200 Subject: [PATCH 6/6] trigger CI