diff --git a/Cargo.lock b/Cargo.lock index 9d9cfe530..0f108719a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7444,6 +7444,7 @@ dependencies = [ "sp-core", "sp-runtime", "thiserror 2.0.18", + "zeroize", ] [[package]] diff --git a/client/cli/src/params/transaction_pool_params.rs b/client/cli/src/params/transaction_pool_params.rs index b4dc17f13..e7d1f785d 100644 --- a/client/cli/src/params/transaction_pool_params.rs +++ b/client/cli/src/params/transaction_pool_params.rs @@ -43,11 +43,15 @@ impl Into for TransactionPoolType { #[derive(Debug, Clone, Args)] pub struct TransactionPoolParams { /// Maximum number of transactions in the transaction pool. - #[arg(long, value_name = "COUNT", default_value_t = 8192)] + /// + /// Default sized for Quantus PQ signatures (~7300 bytes/tx) within ~268 MiB. + #[arg(long, value_name = "COUNT", default_value_t = 36772)] pub pool_limit: usize, /// Maximum number of kilobytes of all transactions stored in the pool. - #[arg(long, value_name = "COUNT", default_value_t = 20480)] + /// + /// Default is 262144 KiB (256 MiB), matching the previous hardcoded node sizing. + #[arg(long, value_name = "COUNT", default_value_t = 262144)] pub pool_kbytes: usize, /// How long a transaction is banned for. @@ -57,7 +61,10 @@ pub struct TransactionPoolParams { pub tx_ban_seconds: Option, /// The type of transaction pool to be instantiated. - #[arg(long, value_enum, default_value_t = TransactionPoolType::ForkAware)] + /// + /// Defaults to `single-state` to preserve prior Quantus node behavior; pass + /// `--pool-type fork-aware` to exercise the fork-aware pool. + #[arg(long, value_enum, default_value_t = TransactionPoolType::SingleState)] pub pool_type: TransactionPoolType, } diff --git a/client/transaction-pool/src/common/api.rs b/client/transaction-pool/src/common/api.rs index b77c39007..c7d56cf05 100644 --- a/client/transaction-pool/src/common/api.rs +++ b/client/transaction-pool/src/common/api.rs @@ -63,11 +63,34 @@ pub struct FullChainApi { validate_transaction_maintained_stats: DurationSlidingStats, } +/// Boxed validation job scheduled on a worker lane. +type ValidationTask = Pin + Send>>; + +/// Receive the next validation job, preferring the Maintained lane when both are ready. +/// +/// `biased` is required: an unbiased `tokio::select!` polls branches in random order, so +/// `ValidateTransactionPriority::Maintained` would not actually outrank Submitted work. +async fn recv_next_validation_task( + receiver_normal: Arc>>, + receiver_maintained: Arc>>, +) -> Option { + tokio::select! { + biased; + Some(task) = async { + receiver_maintained.lock().await.recv().await + } => Some(task), + Some(task) = async { + receiver_normal.lock().await.recv().await + } => Some(task), + else => None, + } +} + /// Spawn a validation task that will be used by the transaction pool to validate transactions. fn spawn_validation_pool_task( name: &'static str, - receiver_normal: Arc + Send>>>>>, - receiver_maintained: Arc + Send>>>>>, + receiver_normal: Arc>>, + receiver_maintained: Arc>>, spawner: &impl SpawnEssentialNamed, stats: DurationSlidingStats, blocking_stats: DurationSlidingStats, @@ -79,20 +102,11 @@ fn spawn_validation_pool_task( loop { let start = Instant::now(); - let task = { - let receiver_maintained = receiver_maintained.clone(); - let receiver_normal = receiver_normal.clone(); - tokio::select! { - Some(task) = async { - receiver_maintained.lock().await.recv().await - } => { task } - Some(task) = async { - receiver_normal.lock().await.recv().await - } => { task } - else => { - return - } - } + let Some(task) = + recv_next_validation_task(receiver_normal.clone(), receiver_maintained.clone()) + .await + else { + return; }; let blocking_duration = { @@ -389,3 +403,55 @@ where ); result } + +#[cfg(test)] +mod validation_lane_tests { + use super::*; + use std::sync::atomic::{AtomicU8, Ordering}; + + /// When both lanes have a ready job, Maintained must always win. + #[tokio::test] + async fn recv_next_validation_task_prefers_maintained_lane() { + const TRIALS: usize = 32; + for _ in 0..TRIALS { + let (tx_normal, rx_normal) = mpsc::channel(1); + let (tx_maintained, rx_maintained) = mpsc::channel(1); + let rx_normal = Arc::new(Mutex::new(rx_normal)); + let rx_maintained = Arc::new(Mutex::new(rx_maintained)); + + let which = Arc::new(AtomicU8::new(0)); + let which_normal = which.clone(); + let which_maintained = which.clone(); + + tx_normal + .send( + async move { + which_normal.store(1, Ordering::SeqCst); + } + .boxed(), + ) + .await + .unwrap(); + tx_maintained + .send( + async move { + which_maintained.store(2, Ordering::SeqCst); + } + .boxed(), + ) + .await + .unwrap(); + + let task = recv_next_validation_task(rx_normal, rx_maintained) + .await + .expect("both lanes have work"); + task.await; + + assert_eq!( + which.load(Ordering::SeqCst), + 2, + "Maintained lane must be preferred when both lanes are ready" + ); + } + } +} diff --git a/client/transaction-pool/src/common/enactment_state.rs b/client/transaction-pool/src/common/enactment_state.rs index 13c7809c2..16c28c7a4 100644 --- a/client/transaction-pool/src/common/enactment_state.rs +++ b/client/transaction-pool/src/common/enactment_state.rs @@ -104,11 +104,15 @@ where let new_hash = event.hash(); let finalized = event.is_finalized(); - // do not proceed with txpool maintain if block distance is too high + // do not proceed with txpool maintain if block distance is too high. The + // absolute height distance is a lower bound on the tree route length, so this + // cheaply rejects deep forward jumps (full sync) and deep backward reorgs + // without computing the route. let skip_maintenance = match (hash_to_number(new_hash), hash_to_number(self.recent_best_block)) { (Ok(Some(new)), Ok(Some(current))) => - new.saturating_sub(current) > SKIP_MAINTENANCE_THRESHOLD.into(), + new.saturating_sub(current).max(current.saturating_sub(new)) > + SKIP_MAINTENANCE_THRESHOLD.into(), _ => true, }; @@ -127,6 +131,18 @@ where // compute actual tree route from best_block to notified block, and use // it instead of tree_route provided with event let tree_route = tree_route(self.recent_best_block, new_hash)?; + + // The block-number check above cannot see deep same-height fork transitions + // (the height delta is zero while the route through the common ancestor may be + // arbitrarily long), so also bound the length of the actual route before the + // pools do per-block prune/resubmit work on it. + let route_len = tree_route.enacted().len() + tree_route.retracted().len(); + if route_len > SKIP_MAINTENANCE_THRESHOLD as usize { + debug!(target: LOG_TARGET, route_len, "skip maintain: tree_route too long"); + self.force_update(event); + return Ok(EnactmentAction::Skip); + } + trace!( target: LOG_TARGET, ?new_hash, @@ -699,3 +715,148 @@ mod enactment_state_tests { assert_es_eq(&es, x1(), b1()); } } + +/// Tests for routes that are long even though the block-number distance between the +/// current best block and the notified block is small (or zero): deep same-height +/// fork switches and deep backward reorgs. +/// +/// Unlike `enactment_state_tests` above, this module does not depend on +/// `substrate_test_runtime_client` (unavailable in this vendored workspace), so it runs +/// as part of the regular `#[cfg(test)]` suite. +#[cfg(test)] +mod long_route_tests { + use super::{EnactmentAction, EnactmentState}; + use crate::common::mock_api::TestBlock as Block; + use sc_transaction_pool_api::ChainEvent; + use sp_blockchain::{HashAndNumber, TreeRoute}; + use sp_runtime::traits::NumberFor; + + type Hash = ::Hash; + + /// Common ancestor of the two forks, at height 1. + fn ancestor() -> HashAndNumber { + HashAndNumber { number: 1, hash: Hash::from([0xFF; 32]) } + } + + /// `i`-th block (1-based) on fork `fork_id`, at height `1 + i`. + fn fork_block(fork_id: u8, i: u64) -> HashAndNumber { + let mut bytes = [fork_id; 32]; + bytes[8..16].copy_from_slice(&i.to_be_bytes()); + HashAndNumber { number: 1 + i, hash: Hash::from(bytes) } + } + + fn fork(fork_id: u8, len: u64) -> Vec> { + (1..=len).map(|i| fork_block(fork_id, i)).collect() + } + + /// Tree route for switching from the tip of fork 1 to the tip of fork 2 + /// (both of length `len`), through the common ancestor. + fn fork_switch_route(len: u64) -> TreeRoute { + let mut route: Vec<_> = fork(1, len).into_iter().rev().collect(); + route.push(ancestor()); + route.extend(fork(2, len)); + TreeRoute::new(route, len as usize).unwrap() + } + + fn hash_to_number(len: u64) -> impl Fn(Hash) -> Result>, String> { + move |hash| { + let mut chain = vec![ancestor()]; + chain.extend(fork(1, len)); + chain.extend(fork(2, len)); + Ok(chain.iter().find(|x| x.hash == hash).map(|x| x.number)) + } + } + + fn assert_es_eq( + es: &EnactmentState, + expected_best_block: HashAndNumber, + expected_finalized_block: HashAndNumber, + ) { + assert_eq!(es.recent_best_block, expected_best_block.hash); + assert_eq!(es.recent_finalized_block, expected_finalized_block.hash); + } + + /// Switching between two same-height fork tips must be skipped when the actual + /// route exceeds the maintenance threshold, even though the block-number + /// distance between the tips is zero. + #[test] + fn test_enactment_skip_long_same_height_fork_route() { + sp_tracing::try_init_simple(); + const LEN: u64 = 25; + + let f1_tip = fork_block(1, LEN); + let f2_tip = fork_block(2, LEN); + let mut es = EnactmentState::new(f1_tip.hash, ancestor().hash); + + let tree_route_fn = + |_: Hash, _: Hash| -> Result, String> { Ok(fork_switch_route(LEN)) }; + + let result = es + .update( + &ChainEvent::NewBestBlock { hash: f2_tip.hash, tree_route: None }, + &tree_route_fn, + &hash_to_number(LEN), + ) + .unwrap(); + + // 25 retracted + 25 enacted blocks: way above the threshold. + assert!(matches!(result, EnactmentAction::Skip)); + assert_es_eq(&es, f2_tip, ancestor()); + } + + /// A new-best notification for a deep ancestor of the current best block must be + /// skipped: the forward block-number delta is zero but the route retracts many + /// blocks. + #[test] + fn test_enactment_skip_long_backward_route() { + sp_tracing::try_init_simple(); + const LEN: u64 = 25; + + let f1_tip = fork_block(1, LEN); + let f1_first = fork_block(1, 1); + let mut es = EnactmentState::new(f1_tip.hash, ancestor().hash); + + let tree_route_fn = |_: Hash, _: Hash| -> Result, String> { + // f1 tip down to f1 first block: 24 retracted blocks. + let route: Vec<_> = fork(1, LEN).into_iter().rev().collect(); + Ok(TreeRoute::new(route, (LEN - 1) as usize).unwrap()) + }; + + let result = es + .update( + &ChainEvent::NewBestBlock { hash: f1_first.hash, tree_route: None }, + &tree_route_fn, + &hash_to_number(LEN), + ) + .unwrap(); + + assert!(matches!(result, EnactmentAction::Skip)); + assert_es_eq(&es, f1_first, ancestor()); + } + + /// Guard: a same-height fork switch whose route is exactly at the threshold must + /// still be maintained. + #[test] + fn test_enactment_proceed_with_same_height_fork_route_at_threshold() { + sp_tracing::try_init_simple(); + const LEN: u64 = 10; // 10 retracted + 10 enacted = threshold (20) + + let f1_tip = fork_block(1, LEN); + let f2_tip = fork_block(2, LEN); + let mut es = EnactmentState::new(f1_tip.hash, ancestor().hash); + + let tree_route_fn = + |_: Hash, _: Hash| -> Result, String> { Ok(fork_switch_route(LEN)) }; + + let result = es + .update( + &ChainEvent::NewBestBlock { hash: f2_tip.hash, tree_route: None }, + &tree_route_fn, + &hash_to_number(LEN), + ) + .unwrap(); + + assert!(matches!(result, EnactmentAction::HandleEnactment { .. })); + assert_es_eq(&es, f2_tip, ancestor()); + } +} diff --git a/client/transaction-pool/src/common/mock_api.rs b/client/transaction-pool/src/common/mock_api.rs new file mode 100644 index 000000000..983eaf498 --- /dev/null +++ b/client/transaction-pool/src/common/mock_api.rs @@ -0,0 +1,165 @@ +// This file is part of Substrate. + +// Copyright (C) Parity Technologies (UK) Ltd. +// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0 + +// This program is free software: you can redistribute it and/or modify +// it under the terms of the GNU General Public License as published by +// the Free Software Foundation, either version 3 of the License, or +// (at your option) any later version. + +// This program is distributed in the hope that it will be useful, +// but WITHOUT ANY WARRANTY; without even the implied warranty of +// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +// GNU General Public License for more details. + +// You should have received a copy of the GNU General Public License +// along with this program. If not, see . + +//! Minimal mock of [`ChainApi`] for unit tests that cannot rely on the substrate test +//! runtime (not available in this vendored crate). + +use crate::{ + graph::{self, ExtrinsicFor, ExtrinsicHash, RawExtrinsicFor}, + ValidateTransactionPriority, +}; +use async_trait::async_trait; +use codec::Encode; +use sc_transaction_pool_api::error::Error as TxPoolApiError; +use sp_blockchain::TreeRoute; +use sp_core::H256; +use sp_runtime::{ + generic::BlockId, + traits::{BlakeTwo256, Block as BlockT, Hash as _}, + transaction_validity::{ + InvalidTransaction, TransactionSource, TransactionValidity, ValidTransaction, + }, +}; +use std::{ + sync::atomic::{AtomicUsize, Ordering}, + time::Duration, +}; + +/// Extrinsic type used with [`MockChainApi`]. +pub(crate) type Extrinsic = sp_runtime::testing::TestXt; + +/// Block type used with [`MockChainApi`]. +pub(crate) type TestBlock = sp_runtime::testing::Block; + +/// Transactions with a call value at or above this threshold are reported as invalid by +/// [`MockChainApi`]. +pub(crate) const INVALID_CALL_THRESHOLD: u64 = 1000; + +/// Creates a test extrinsic with the given call value. +pub(crate) fn xt(value: u64) -> Extrinsic { + Extrinsic::new_bare(sp_runtime::testing::MockCallU64(value)) +} + +/// Minimal `ChainApi` mock: treats every block id as existing and validates transactions +/// based on their call value only. +#[derive(Default)] +pub(crate) struct MockChainApi { + /// Number of performed `validate_transaction` calls. + validation_count: AtomicUsize, + /// Optional artificial delay applied inside `validate_transaction`. + validation_delay: Option, +} + +impl MockChainApi { + /// Creates a mock that sleeps for `delay` on every `validate_transaction` call. + pub(crate) fn with_validation_delay(delay: Duration) -> Self { + Self { validation_delay: Some(delay), ..Default::default() } + } + + /// Returns the number of performed `validate_transaction` calls. + pub(crate) fn validation_count(&self) -> usize { + self.validation_count.load(Ordering::Relaxed) + } +} + +#[async_trait] +impl graph::ChainApi for MockChainApi { + type Block = TestBlock; + type Error = TxPoolApiError; + + async fn validate_transaction( + &self, + _at: ::Hash, + _source: TransactionSource, + uxt: ExtrinsicFor, + _priority: ValidateTransactionPriority, + ) -> Result { + if let Some(delay) = self.validation_delay { + tokio::time::sleep(delay).await; + } + self.validation_count.fetch_add(1, Ordering::Relaxed); + let value = uxt.function.0; + Ok(if value >= INVALID_CALL_THRESHOLD { + Err(InvalidTransaction::Custom(0).into()) + } else { + Ok(ValidTransaction { + priority: 4, + requires: vec![], + provides: vec![value.encode()], + longevity: 64, + propagate: true, + }) + }) + } + + fn validate_transaction_blocking( + &self, + _at: ::Hash, + _source: TransactionSource, + _uxt: ExtrinsicFor, + ) -> Result { + unimplemented!() + } + + fn block_id_to_number( + &self, + at: &BlockId, + ) -> Result>, Self::Error> { + Ok(match at { + BlockId::Number(num) => Some(*num), + BlockId::Hash(hash) => Some(hash.to_low_u64_be()), + }) + } + + fn block_id_to_hash( + &self, + at: &BlockId, + ) -> Result::Hash>, Self::Error> { + Ok(match at { + BlockId::Number(num) => Some(H256::from_low_u64_be(*num)), + BlockId::Hash(hash) => Some(*hash), + }) + } + + fn hash_and_length(&self, uxt: &RawExtrinsicFor) -> (ExtrinsicHash, usize) { + let encoded = uxt.encode(); + (BlakeTwo256::hash(&encoded), encoded.len()) + } + + async fn block_body( + &self, + _at: ::Hash, + ) -> Result::Extrinsic>>, Self::Error> { + Ok(None) + } + + fn block_header( + &self, + _at: ::Hash, + ) -> Result::Header>, Self::Error> { + Ok(None) + } + + fn tree_route( + &self, + _from: ::Hash, + _to: ::Hash, + ) -> Result, Self::Error> { + unimplemented!() + } +} diff --git a/client/transaction-pool/src/common/mod.rs b/client/transaction-pool/src/common/mod.rs index bffb03db3..3ad234264 100644 --- a/client/transaction-pool/src/common/mod.rs +++ b/client/transaction-pool/src/common/mod.rs @@ -22,6 +22,8 @@ pub(crate) mod api; pub(crate) mod enactment_state; pub(crate) mod error; pub(crate) mod metrics; +#[cfg(test)] +pub(crate) mod mock_api; pub(crate) mod sliding_stat; #[cfg(all(test, feature = "test-helpers"))] pub(crate) mod tests; diff --git a/client/transaction-pool/src/fork_aware_txpool/dropped_watcher.rs b/client/transaction-pool/src/fork_aware_txpool/dropped_watcher.rs index 92764af5a..43526c856 100644 --- a/client/transaction-pool/src/fork_aware_txpool/dropped_watcher.rs +++ b/client/transaction-pool/src/fork_aware_txpool/dropped_watcher.rs @@ -289,6 +289,11 @@ where if let Some(mut views_keeping_tx_valid) = self.transaction_views(tx_hash) { views_keeping_tx_valid.get_mut().remove(&block_hash); if views_keeping_tx_valid.get().is_empty() { + // The consumer removes the reported transaction from the mempool, + // so drop the (now empty) map entry as well. Keeping it would leak + // it forever: `Command::RemoveTransactions` is only sent for + // finalized transactions. + views_keeping_tx_valid.remove_entry(); return Some(DroppedTransaction::new_enforced_by_limts(tx_hash)); } } else { @@ -296,12 +301,19 @@ where return Some(DroppedTransaction::new_enforced_by_limts(tx_hash)); } }, - TransactionStatus::Usurped(by) => - return Some(DroppedTransaction::new_usurped(tx_hash, by)), + TransactionStatus::Usurped(by) => { + // The usurped transaction is removed from the mempool by the consumer, + // so its view-tracking entries must be dropped here as well. + self.ready_transaction_views.remove(&tx_hash); + self.future_transaction_views.remove(&tx_hash); + return Some(DroppedTransaction::new_usurped(tx_hash, by)) + }, TransactionStatus::Invalid => { if let Some(mut views_keeping_tx_valid) = self.transaction_views(tx_hash) { views_keeping_tx_valid.get_mut().remove(&block_hash); if views_keeping_tx_valid.get().is_empty() { + // See the comment in the `Dropped` arm above. + views_keeping_tx_valid.remove_entry(); return Some(DroppedTransaction::new_invalid(tx_hash)); } } else { @@ -436,6 +448,93 @@ where } } +/// Tests that per-transaction view-map entries are reclaimed when transactions leave the +/// pool for reasons other than finalization (dropped / invalid / usurped). +/// +/// Unlike `dropped_watcher_tests` below, this module does not depend on the +/// `test-helpers` feature (unavailable in this vendored workspace), so it runs as part +/// of the regular `#[cfg(test)]` suite. It exercises `handle_event` directly on the +/// private context. +#[cfg(test)] +mod map_cleanup_tests { + use super::*; + use crate::common::mock_api::MockChainApi; + use sp_core::H256; + + fn ctx() -> MultiViewDropWatcherContext { + let (_sender, command_receiver) = + mpsc::tracing_unbounded::>("test-cmd-stream", 16); + MultiViewDropWatcherContext { + stream_map: StreamMap::new(), + command_receiver, + ready_transaction_views: Default::default(), + future_transaction_views: Default::default(), + pending_dropped_transactions: Default::default(), + } + } + + #[test] + fn dropped_event_removes_emptied_ready_map_entry() { + let mut ctx = ctx(); + let view = H256::repeat_byte(0x01); + let tx = H256::repeat_byte(0x0a); + + ctx.handle_event(view, (tx, TransactionStatus::Ready)); + assert!(ctx.ready_transaction_views.contains_key(&tx)); + + let dropped = ctx.handle_event(view, (tx, TransactionStatus::Dropped)); + assert_eq!(dropped, Some(DroppedTransaction::new_enforced_by_limts(tx))); + assert!(!ctx.ready_transaction_views.contains_key(&tx)); + assert!(!ctx.future_transaction_views.contains_key(&tx)); + } + + #[test] + fn dropped_event_keeps_entry_while_other_views_reference_tx() { + let mut ctx = ctx(); + let view_a = H256::repeat_byte(0x01); + let view_b = H256::repeat_byte(0x02); + let tx = H256::repeat_byte(0x0a); + + ctx.handle_event(view_a, (tx, TransactionStatus::Ready)); + ctx.handle_event(view_b, (tx, TransactionStatus::Ready)); + + let dropped = ctx.handle_event(view_a, (tx, TransactionStatus::Dropped)); + assert_eq!(dropped, None); + assert!(ctx.ready_transaction_views.contains_key(&tx)); + } + + #[test] + fn invalid_event_removes_emptied_future_map_entry() { + let mut ctx = ctx(); + let view = H256::repeat_byte(0x01); + let tx = H256::repeat_byte(0x0a); + + ctx.handle_event(view, (tx, TransactionStatus::Future)); + assert!(ctx.future_transaction_views.contains_key(&tx)); + + let dropped = ctx.handle_event(view, (tx, TransactionStatus::Invalid)); + assert_eq!(dropped, Some(DroppedTransaction::new_invalid(tx))); + assert!(!ctx.ready_transaction_views.contains_key(&tx)); + assert!(!ctx.future_transaction_views.contains_key(&tx)); + } + + #[test] + fn usurped_event_removes_map_entries() { + let mut ctx = ctx(); + let view = H256::repeat_byte(0x01); + let tx = H256::repeat_byte(0x0a); + let by = H256::repeat_byte(0x0b); + + ctx.handle_event(view, (tx, TransactionStatus::Ready)); + assert!(ctx.ready_transaction_views.contains_key(&tx)); + + let dropped = ctx.handle_event(view, (tx, TransactionStatus::Usurped(by))); + assert_eq!(dropped, Some(DroppedTransaction::new_usurped(tx, by))); + assert!(!ctx.ready_transaction_views.contains_key(&tx)); + assert!(!ctx.future_transaction_views.contains_key(&tx)); + } +} + #[cfg(all(test, feature = "test-helpers"))] mod dropped_watcher_tests { use super::*; diff --git a/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs b/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs index 604efaa95..10f71930d 100644 --- a/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs +++ b/client/transaction-pool/src/fork_aware_txpool/fork_aware_txpool.rs @@ -257,7 +257,10 @@ where mempool_max_transactions_count: usize, finality_timeout_threshold: Option, ) -> (Self, [ForkAwareTxPoolTask; 2]) { - let (listener, listener_task) = MultiViewListener::new_with_worker(Default::default()); + let (events_metrics_collector, event_metrics_task) = + EventsMetricsCollector::::new_with_worker(Default::default()); + let (listener, listener_task) = + MultiViewListener::new_with_worker(events_metrics_collector.clone()); let listener = Arc::new(listener); let (import_notification_sink, import_notification_sink_task) = @@ -293,7 +296,8 @@ where tokio::select! { _ = listener_task => {}, _ = import_notification_sink_task => {}, - _ = dropped_monitor_task => {} + _ = dropped_monitor_task => {}, + _ = event_metrics_task => {}, } } .boxed(); @@ -315,7 +319,7 @@ where options, is_validator: false.into(), metrics: Default::default(), - events_metrics_collector: EventsMetricsCollector::default(), + events_metrics_collector, finality_timeout_threshold: finality_timeout_threshold .unwrap_or(FINALITY_TIMEOUT_THRESHOLD), included_transactions: Default::default(), @@ -762,6 +766,9 @@ where match self.view_store.submit_and_watch(at, insertion.source, xt).await { Err(e) => { self.mempool.remove_transactions(&[insertion.hash]).await; + // The transaction never reached any view and no status events will be + // reported for it, so drop its entry from the events metrics collector. + self.events_metrics_collector.report_submission_rejected(insertion.hash); Err(e.into()) }, Ok(mut outcome) => { @@ -877,6 +884,10 @@ where }, Err(e) => { mempool.remove_transactions(&[hash]).await; + // The transaction never reached any view and no status events will + // be reported for it, so drop its entry from the events metrics + // collector. + self.events_metrics_collector.report_submission_rejected(hash); final_results.push(Err(e)); }, }, @@ -2260,3 +2271,57 @@ mod reduce_multiview_result_tests { ); } } + +#[cfg(test)] +mod submission_metrics_tests { + use super::*; + use crate::common::mock_api::{xt, MockChainApi, TestBlock, INVALID_CALL_THRESHOLD}; + use sp_core::H256; + + fn test_pool() -> ForkAwareTxPool { + let genesis = H256::from_low_u64_be(0); + let (pool, [combined_task, _mempool_task]) = + ForkAwareTxPool::new_test(Arc::new(MockChainApi::default()), genesis, genesis, None); + tokio::spawn(combined_task); + pool + } + + /// A transaction which is accepted by the mempool but rejected during the view + /// submission must not leave a stale entry in the events metrics collector. Otherwise an + /// external party could grow the collector's memory without bounds by submitting unique + /// invalid transactions. + #[tokio::test] + async fn transaction_rejected_by_view_is_removed_from_event_metrics() { + let pool = test_pool(); + let genesis = H256::from_low_u64_be(0); + + // Control check: a valid transaction shall be tracked by the collector. + let results = pool + .submit_at(genesis, TransactionSource::External, vec![xt(1)]) + .await + .expect("submit_at succeeds"); + assert!(results[0].is_ok()); + assert_eq!(pool.events_metrics_collector.tracked_txs_count().await, 1); + + // Transaction rejected by the view shall not leak a collector entry. + let results = pool + .submit_at(genesis, TransactionSource::External, vec![xt(INVALID_CALL_THRESHOLD)]) + .await + .expect("submit_at succeeds"); + assert!(results[0].is_err()); + assert_eq!(pool.events_metrics_collector.tracked_txs_count().await, 1); + } + + /// Same as above, for the watched (`submit_and_watch`) submission path. + #[tokio::test] + async fn watched_transaction_rejected_by_view_is_removed_from_event_metrics() { + let pool = test_pool(); + let genesis = H256::from_low_u64_be(0); + + let result = pool + .submit_and_watch(genesis, TransactionSource::External, xt(INVALID_CALL_THRESHOLD)) + .await; + assert!(result.is_err()); + assert_eq!(pool.events_metrics_collector.tracked_txs_count().await, 0); + } +} diff --git a/client/transaction-pool/src/fork_aware_txpool/import_notification_sink.rs b/client/transaction-pool/src/fork_aware_txpool/import_notification_sink.rs index a9a8eb811..f2b7c050b 100644 --- a/client/transaction-pool/src/fork_aware_txpool/import_notification_sink.rs +++ b/client/transaction-pool/src/fork_aware_txpool/import_notification_sink.rs @@ -198,7 +198,12 @@ where %error, "import_sink_worker sending message failed" ); - false + // A full channel is transient backpressure: drop only this + // notification but keep the subscriber. Dropping the sink + // would permanently close the subscriber's stream (used + // e.g. for transaction propagation). Only remove sinks + // whose receiver is gone. + error.is_full() } else { true } @@ -383,6 +388,52 @@ mod tests { futures::future::join_all(vec![j0, j1, j2, j3]).await; } + /// A subscriber whose channel temporarily fills up must lose at most the overflowing + /// notifications, not its subscription: a full channel is transient backpressure, + /// only a disconnected channel means the subscriber is gone. + #[tokio::test] + async fn transient_backlog_does_not_permanently_drop_subscriber() { + sp_tracing::try_init_simple(); + + let (ctrl, runnable) = MultiViewImportNotificationSink::::new_with_worker(); + let j0 = tokio::spawn(runnable); + + let mut stream = ctrl.event_stream(); + + // Flood more events than the external channel buffer (1024) while the + // subscriber is not consuming. + const FLOOD: i32 = 1500; + ctrl.add_view(1000, futures::stream::iter(0..FLOOD).boxed()); + + // Wait until the worker has dispatched the whole flood. + tokio::time::timeout(Duration::from_secs(5), async { + while ctrl.notified_items_len() < FLOOD as usize { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + + // Drain whatever was buffered. The stream must still be open afterwards. + let mut terminated = false; + while let Some(item) = stream.next().now_or_never() { + if item.is_none() { + terminated = true; + break; + } + } + assert!(!terminated, "subscriber stream must survive a transient backlog"); + + // New notifications must still reach the (now draining) subscriber. + ctrl.add_view(2000, futures::stream::iter(5000..5001).boxed()); + let next = tokio::time::timeout(Duration::from_secs(5), stream.next()).await.unwrap(); + assert_eq!(next, Some(5000)); + + drop(ctrl); + drop(stream); + j0.await.unwrap(); + } + #[tokio::test] async fn many_output_streams_are_supported() { sp_tracing::try_init_simple(); diff --git a/client/transaction-pool/src/fork_aware_txpool/metrics.rs b/client/transaction-pool/src/fork_aware_txpool/metrics.rs index 4ad000f7a..dc1212d21 100644 --- a/client/transaction-pool/src/fork_aware_txpool/metrics.rs +++ b/client/transaction-pool/src/fork_aware_txpool/metrics.rs @@ -456,6 +456,13 @@ enum EventMetricsMessage { /// Message indicating the new status of a transaction, including the timestamp and transaction /// hash. Status(Instant, Hash, TransactionStatus), + /// Message indicating that a previously submitted transaction was rejected before reaching + /// any view (e.g. it failed validation during submission) and no status events will ever be + /// reported for it, so its entry shall be removed from the collector's state. + SubmissionRejected(Hash), + /// Test-only query for the number of transactions currently tracked by the collector task. + #[cfg(test)] + TrackedTxsCount(futures::channel::oneshot::Sender), } /// Collects metrics related to transaction events. @@ -513,6 +520,35 @@ impl EventsMetricsCollector { } }); } + + /// Reports that a submitted transaction was rejected before being included in any view. + /// + /// Since no status events will ever be reported for such a transaction, this message + /// removes the transaction's entry from the collector's internal state, preventing + /// unbounded growth of the state for transactions that fail submission. + pub fn report_submission_rejected(&self, tx_hash: ExtrinsicHash) { + self.metrics_message_sink.as_ref().map(|sink| { + if let Err(error) = + sink.unbounded_send(EventMetricsMessage::SubmissionRejected(tx_hash)) + { + trace!(target: LOG_TARGET, %error, "tx submission rejected metrics message send failed") + } + }); + } + + /// Returns the number of transactions currently tracked by the metrics collector task. + /// + /// Intended for testing only. + #[cfg(test)] + pub(crate) async fn tracked_txs_count(&self) -> usize { + let (tx, rx) = futures::channel::oneshot::channel(); + self.metrics_message_sink + .as_ref() + .expect("metrics message sink is set in tests") + .unbounded_send(EventMetricsMessage::TrackedTxsCount(tx)) + .expect("metrics collector task is running"); + rx.await.expect("metrics collector task answers count queries") + } } /// A type alias for a asynchronous task that collects metrics related to events. @@ -651,6 +687,13 @@ where &metrics, ); }, + Some(EventMetricsMessage::SubmissionRejected(hash)) => { + submitted_timestamp_map.remove(&hash); + }, + #[cfg(test)] + Some(EventMetricsMessage::TrackedTxsCount(sender)) => { + let _ = sender.send(submitted_timestamp_map.len()); + }, None => { return; /* ? */ }, diff --git a/client/transaction-pool/src/fork_aware_txpool/mod.rs b/client/transaction-pool/src/fork_aware_txpool/mod.rs index b778042d3..8ba598707 100644 --- a/client/transaction-pool/src/fork_aware_txpool/mod.rs +++ b/client/transaction-pool/src/fork_aware_txpool/mod.rs @@ -242,16 +242,19 @@ //! //! ### Background tasks //! The [maintain](#maintain) procedure shall be as quick as possible, so heavy revalidation job is -//! delegated to the background worker. These includes view and *mempool* revalidation which are -//! both handled by the [`RevalidationQueue`] which simply sends revalidation requests to the -//! background thread. +//! delegated to background workers. View and *mempool* revalidation are both handled by the +//! [`RevalidationQueue`], which dispatches each kind of job to its own worker loop. Keeping these +//! queues separate ensures that an uncancellable mempool batch cannot stall +//! [`finish_background_revalidations`] on the maintain critical path. The view worker uses a +//! bounded queue (inline fallback when full); mempool work is latest-wins coalesced so +//! finalization bursts cannot accumulate an unbounded backlog of revalidation payloads. //! //! #### View revalidation -//! View revalidation is performed in the background thread. Revalidation is executed for every +//! View revalidation is performed on the view background worker. Revalidation is executed for every //! view. All the transaction from the view are [revalidated][`view::revalidate`]. //! -//! The fork-aware pool utilizes two threads to execute maintain and revalidation process -//! exclusively, ensuring maintain performance without overlapping with revalidation. +//! The fork-aware pool keeps maintain off the revalidation workers so maintain performance is not +//! overlapped with long-running validation work. //! //! The view revalidation process is [triggered][`start_background_revalidation`] at the very end of //! the [maintain][`maintain`] process, and [stopped][`finish_background_revalidations`] at the @@ -259,9 +262,10 @@ //! results from the revalidation are immediately applied once the revalidation is //! [terminated][crate::fork_aware_txpool::view::View::finish_revalidation]. //! ```text -//! time: ----------------------> -//! maintenance thread: M----M------M--M-M--- -//! revalidation thread: -RRRR-RR-----RR-R-RRR +//! time: ----------------------> +//! maintenance thread: M----M------M--M-M--- +//! view revalidation worker: -RRRR-RR-----RR-R-RRR +//! mempool revalidation worker: --MMMMMM----MM--MMMM //! ``` //! //! #### Mempool pruning/revalidation diff --git a/client/transaction-pool/src/fork_aware_txpool/multi_view_listener.rs b/client/transaction-pool/src/fork_aware_txpool/multi_view_listener.rs index 9a6d190df..7718a4f2b 100644 --- a/client/transaction-pool/src/fork_aware_txpool/multi_view_listener.rs +++ b/client/transaction-pool/src/fork_aware_txpool/multi_view_listener.rs @@ -26,10 +26,15 @@ use crate::{ graph::{self, BlockHash, ExtrinsicHash}, LOG_TARGET, }; -use futures::{Future, FutureExt, Stream, StreamExt}; -use parking_lot::RwLock; +use futures::{ + channel::mpsc::{ + channel, unbounded, Receiver as BoundedReceiver, Sender as BoundedSender, + UnboundedReceiver, UnboundedSender, + }, + Future, FutureExt, Stream, StreamExt, +}; +use parking_lot::{Mutex, RwLock}; use sc_transaction_pool_api::{TransactionStatus, TransactionStatusStream, TxIndex}; -use sc_utils::mpsc; use sp_runtime::traits::Block as BlockT; use std::{ collections::{hash_map::Entry, HashMap, HashSet}, @@ -37,23 +42,34 @@ use std::{ sync::Arc, }; use tokio_stream::StreamMap; -use tracing::trace; +use tracing::{trace, warn}; use super::{ dropped_watcher::{DroppedReason, DroppedTransaction}, metrics::EventsMetricsCollector, }; +/// Bound for the MultiViewListener task controller and per-transaction external watcher queues. +/// +/// Matches the import-notification external sink capacity. A full channel drops the overflowing +/// notification (or closes the watcher if a *final* status cannot be delivered) instead of growing +/// without bound when RPC / downstream consumers stop polling. +/// +/// Terminal pool statuses (`Finalized`, `Dropped`, `Invalid`, `FinalityTimeout`) and view stream +/// management commands bypass this bound on a dedicated loss-less lane — see +/// [`MultiViewListener::send_controller_command`]. +pub(super) const STATUS_CHANNEL_CAPACITY: usize = 1024; + /// A side channel allowing to control the external stream instance (one per transaction) with /// [`ControllerCommand`]. /// /// Set of instances of [`Controller`] lives within the [`MultiViewListener`]. -type Controller = mpsc::TracingUnboundedSender; +type Controller = BoundedSender; /// A receiver of [`ControllerCommand`] instances allowing to control the external stream. /// /// Lives within the [`ExternalWatcherContext`] instance. -type CommandReceiver = mpsc::TracingUnboundedReceiver; +type CommandReceiver = BoundedReceiver; /// The stream of the transaction events. /// @@ -233,6 +249,26 @@ where tx_hash, block_hash, )) } + + /// Whether this command must not be dropped under backlog. + /// + /// Such commands are routed via the loss-less lane instead of the bounded, drop-on-full + /// controller channel: + /// - terminal transaction statuses must reach the external watcher (or close it), + /// - view stream management is intrinsically rare (per block, not per transaction) and losing + /// `AddViewStream` would silence all of that view's events for its lifetime. + /// + /// Only the flood-prone `Broadcast` traffic (per transaction, driven by gossip) remains + /// droppable. + fn requires_lossless_delivery(&self) -> bool { + match self { + Self::TransactionStatusRequest(request) => { + let status: TransactionStatus<_, _> = request.into(); + status.is_final() + }, + Self::AddViewStream(..) | Self::RemoveViewStream(..) => true, + } + } } /// This struct allows to create and control listener for multiple transactions. @@ -248,7 +284,21 @@ where /// invalid, broadcast) independently of the view's stream. pub struct MultiViewListener { /// Provides the controller for sending control commands to the listener's task. - controller: Controller>, + /// + /// Mutex avoids cloning the sender on every command (clones raise `futures::mpsc` + /// effective capacity). + controller: Mutex>>, + + /// Loss-less lane for commands that must not be dropped under backlog: terminal transaction + /// statuses (`Finalized`, `Dropped`, `Invalid`, `FinalityTimeout`, `Usurped`) and view + /// stream management (`AddViewStream`/`RemoveViewStream`). + /// + /// Unbounded on purpose: each transaction known to the pool emits at most a handful of + /// terminal events, the pool itself is capacity-limited, and view management occurs a + /// couple of times per block — so this lane is bounded in practice. The flood-prone + /// `Broadcast` traffic stays on the bounded `controller` channel where dropping is + /// acceptable. + lossless_controller: UnboundedSender>, /// The map containing the sinks of the streams representing the external listeners of /// the individual transactions. Hash of the transaction is used as a map's key. A map is @@ -300,6 +350,100 @@ enum ExternalWatcherCommand { RemoveView(BlockHash), } +impl std::fmt::Debug for ExternalWatcherCommand { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::PoolTransactionStatus(request) => write!(f, "PoolTransactionStatus({request:?})"), + Self::ViewTransactionStatus(h, status) => + write!(f, "ViewTransactionStatus({h:?},{status:?})"), + Self::AddView(h) => write!(f, "AddView({h:?})"), + Self::RemoveView(h) => write!(f, "RemoveView({h:?})"), + } + } +} + +impl ExternalWatcherCommand { + /// Whether failing to deliver this command should close the external watcher. + /// + /// Intermediate statuses / view membership updates are dropped on a full channel so a slow + /// consumer cannot OOM the node. Final statuses close the watcher instead of leaving the + /// subscription hung without a terminal event. + fn is_final(&self) -> bool { + match self { + Self::PoolTransactionStatus(request) => { + let status: TransactionStatus<_, _> = request.into(); + status.is_final() + }, + Self::ViewTransactionStatus(_, status) => status.is_final(), + Self::AddView(_) | Self::RemoveView(_) => false, + } + } +} + +/// Attempts to enqueue a command for an external watcher. +/// +/// Returns `true` if the watcher controller should be retained. +fn try_send_external_watcher_command( + tx_hash: ExtrinsicHash, + ctrl: &mut Controller>, + command: ExternalWatcherCommand, +) -> bool { + match ctrl.try_send(command) { + Ok(()) => true, + Err(error) if error.is_full() => { + let command = error.into_inner(); + if command.is_final() { + warn!( + target: LOG_TARGET, + ?tx_hash, + ?command, + "external watcher channel full while delivering final status; closing watcher" + ); + false + } else { + warn!( + target: LOG_TARGET, + ?tx_hash, + ?command, + "external watcher channel full; dropping notification" + ); + true + } + }, + Err(error) => { + trace!( + target: LOG_TARGET, + ?tx_hash, + %error, + "external watcher send failed (receiver gone)" + ); + false + }, + } +} + +/// Attempts to enqueue a controller command for the listener task. +fn try_send_controller_command( + controller: &Mutex>>, + command: ControllerCommand, +) { + if let Err(error) = controller.lock().try_send(command) { + if error.is_full() { + warn!( + target: LOG_TARGET, + command = ?error.into_inner(), + "multi-view-listener controller channel full; dropping command" + ); + } else { + trace!( + target: LOG_TARGET, + command = ?error.into_inner(), + "multi-view-listener controller send failed (task gone)" + ); + } + } +} + impl ExternalWatcherContext where <::Block as BlockT>::Hash: Unpin, @@ -452,14 +596,26 @@ where RwLock, Controller>>>, >, mut command_receiver: CommandReceiver>, + mut lossless_command_receiver: UnboundedReceiver>, events_metrics_collector: EventsMetricsCollector, ) { let mut aggregated_streams_map: StreamMap, ViewStatusStream> = Default::default(); loop { + // `biased`: lossless lane first so a permanently-ready aggregated view + // stream cannot starve terminal statuses / view management. Droppable + // broadcasts remain last. tokio::select! { biased; + Some(cmd) = lossless_command_receiver.next() => { + Self::dispatch_command( + &mut aggregated_streams_map, + &external_watchers_tx_hash_map, + &events_metrics_collector, + cmd, + ); + }, Some((view_hash, (tx_hash, status))) = next_event(&mut aggregated_streams_map) => { events_metrics_collector.report_status(tx_hash, status.clone()); if let Entry::Occupied(mut ctrl) = external_watchers_tx_hash_map.write().entry(tx_hash) { @@ -470,60 +626,79 @@ where ?status, "aggregated_stream_map event", ); - if let Err(error) = ctrl - .get_mut() - .unbounded_send(ExternalWatcherCommand::ViewTransactionStatus(view_hash, status)) - { - trace!(target: LOG_TARGET, ?tx_hash, ?error, "send status failed"); + if !try_send_external_watcher_command( + tx_hash, + ctrl.get_mut(), + ExternalWatcherCommand::ViewTransactionStatus(view_hash, status), + ) { ctrl.remove(); } } }, cmd = command_receiver.next() => { - match cmd { - Some(ControllerCommand::AddViewStream(h,stream)) => { - aggregated_streams_map.insert(h,stream); - // //todo: aysnc and join all? - external_watchers_tx_hash_map.write().retain(|tx_hash, ctrl| { - ctrl.unbounded_send(ExternalWatcherCommand::AddView(h)) - .inspect_err(|error| { - trace!(target: LOG_TARGET, ?tx_hash, ?error, "add_view: send message failed"); - }) - .is_ok() - }) - }, - Some(ControllerCommand::RemoveViewStream(h)) => { - aggregated_streams_map.remove(&h); - //todo: aysnc and join all? - external_watchers_tx_hash_map.write().retain(|tx_hash, ctrl| { - ctrl.unbounded_send(ExternalWatcherCommand::RemoveView(h)) - .inspect_err(|error| { - trace!(target: LOG_TARGET, ?tx_hash, ?error, "remove_view: send message failed"); - }) - .is_ok() - }) - }, - - Some(ControllerCommand::TransactionStatusRequest(request)) => { - let tx_hash = request.hash(); - events_metrics_collector.report_status(tx_hash, (&request).into()); - if let Entry::Occupied(mut ctrl) = external_watchers_tx_hash_map.write().entry(tx_hash) { - if let Err(error) = ctrl - .get_mut() - .unbounded_send(ExternalWatcherCommand::PoolTransactionStatus(request)) - { - trace!(target: LOG_TARGET, ?tx_hash, ?error, "send message failed"); - ctrl.remove(); - } - } - }, - None => {} + if let Some(cmd) = cmd { + Self::dispatch_command( + &mut aggregated_streams_map, + &external_watchers_tx_hash_map, + &events_metrics_collector, + cmd, + ); } }, }; } } + /// Applies a single [`ControllerCommand`] within the listener's task. + /// + /// Shared by the bounded controller lane and the loss-less lane. + fn dispatch_command( + aggregated_streams_map: &mut StreamMap, ViewStatusStream>, + external_watchers_tx_hash_map: &RwLock< + HashMap, Controller>>, + >, + events_metrics_collector: &EventsMetricsCollector, + cmd: ControllerCommand, + ) { + match cmd { + ControllerCommand::AddViewStream(h, stream) => { + aggregated_streams_map.insert(h, stream); + external_watchers_tx_hash_map.write().retain(|tx_hash, ctrl| { + try_send_external_watcher_command( + *tx_hash, + ctrl, + ExternalWatcherCommand::AddView(h), + ) + }) + }, + ControllerCommand::RemoveViewStream(h) => { + aggregated_streams_map.remove(&h); + external_watchers_tx_hash_map.write().retain(|tx_hash, ctrl| { + try_send_external_watcher_command( + *tx_hash, + ctrl, + ExternalWatcherCommand::RemoveView(h), + ) + }) + }, + ControllerCommand::TransactionStatusRequest(request) => { + let tx_hash = request.hash(); + events_metrics_collector.report_status(tx_hash, (&request).into()); + if let Entry::Occupied(mut ctrl) = + external_watchers_tx_hash_map.write().entry(tx_hash) + { + if !try_send_external_watcher_command( + tx_hash, + ctrl.get_mut(), + ExternalWatcherCommand::PoolTransactionStatus(request), + ) { + ctrl.remove(); + } + } + }, + } + } + /// Creates a new [`MultiViewListener`] instance along with its associated worker task. /// /// This function instantiates the new `MultiViewListener` and provides the worker task that @@ -544,14 +719,38 @@ where Controller>, >::default())); - const CONTROLLER_QUEUE_WARN_SIZE: usize = 100_000; - let (tx, rx) = mpsc::tracing_unbounded( - "txpool-multi-view-listener-task-controller", - CONTROLLER_QUEUE_WARN_SIZE, - ); - let task = Self::task(external_controllers.clone(), rx, events_metrics_collector); + let (tx, rx) = channel(STATUS_CHANNEL_CAPACITY); + let (lossless_tx, lossless_rx) = unbounded(); + let task = + Self::task(external_controllers.clone(), rx, lossless_rx, events_metrics_collector); - (Self { external_controllers, controller: tx }, task.boxed()) + ( + Self { + external_controllers, + controller: Mutex::new(tx), + lossless_controller: lossless_tx, + }, + task.boxed(), + ) + } + + /// Routes a controller command to the appropriate lane. + /// + /// Commands carrying terminal statuses or managing view streams go through the loss-less + /// unbounded lane so they cannot be dropped under backlog; everything else (`Broadcast`) + /// goes through the bounded, drop-on-full controller channel. + fn send_controller_command(&self, command: ControllerCommand) { + if command.requires_lossless_delivery() { + if let Err(error) = self.lossless_controller.unbounded_send(command) { + trace!( + target: LOG_TARGET, + command = ?error.into_inner(), + "multi-view-listener loss-less lane send failed (task gone)" + ); + } + } else { + try_send_controller_command(&self.controller, command); + } } /// Creates an external tstream of events for given transaction. @@ -571,11 +770,7 @@ where let external_ctx = match self.external_controllers.write().entry(tx_hash) { Entry::Occupied(_) => return None, Entry::Vacant(entry) => { - const EXT_CONTROLLER_QUEUE_WARN_THRESHOLD: usize = 128; - let (tx, rx) = mpsc::tracing_unbounded( - "txpool-multi-view-listener", - EXT_CONTROLLER_QUEUE_WARN_THRESHOLD, - ); + let (tx, rx) = channel(STATUS_CHANNEL_CAPACITY); entry.insert(tx); ExternalWatcherContext::new(tx_hash, rx) }, @@ -646,17 +841,7 @@ where stream: ViewStatusStream, ) { trace!(target: LOG_TARGET, ?block_hash, "mvl::add_view_aggregated_stream"); - if let Err(error) = self - .controller - .unbounded_send(ControllerCommand::AddViewStream(block_hash, stream)) - { - trace!( - target: LOG_TARGET, - ?block_hash, - %error, - "add_view_aggregated_stream: send message failed" - ); - } + self.send_controller_command(ControllerCommand::AddViewStream(block_hash, stream)); } /// Removes a view's stream associated with a specific view hash. @@ -665,16 +850,7 @@ where /// dispatched to the external watcher context for every watched transaction. pub(crate) fn remove_view(&self, block_hash: BlockHash) { trace!(target: LOG_TARGET, ?block_hash, "mvl::remove_view"); - if let Err(error) = - self.controller.unbounded_send(ControllerCommand::RemoveViewStream(block_hash)) - { - trace!( - target: LOG_TARGET, - ?block_hash, - %error, - "remove_view: send message failed" - ); - } + self.send_controller_command(ControllerCommand::RemoveViewStream(block_hash)); } /// Invalidate given transaction. @@ -687,16 +863,7 @@ where pub(crate) fn transactions_invalidated(&self, invalid_hashes: &[ExtrinsicHash]) { log_xt_trace!(target: LOG_TARGET, invalid_hashes, "transactions_invalidated"); for tx_hash in invalid_hashes { - if let Err(error) = - self.controller.unbounded_send(ControllerCommand::new_invalidated(*tx_hash)) - { - trace!( - target: LOG_TARGET, - ?tx_hash, - %error, - "transactions_invalidated: send message failed" - ); - } + self.send_controller_command(ControllerCommand::new_invalidated(*tx_hash)); } } @@ -709,17 +876,7 @@ where propagated: HashMap, Vec>, ) { for (tx_hash, peers) in propagated { - if let Err(error) = self - .controller - .unbounded_send(ControllerCommand::new_broadcasted(tx_hash, peers)) - { - trace!( - target: LOG_TARGET, - ?tx_hash, - %error, - "transactions_broadcasted: send message failed" - ); - } + self.send_controller_command(ControllerCommand::new_broadcasted(tx_hash, peers)); } } @@ -730,16 +887,7 @@ where pub(crate) fn transaction_dropped(&self, dropped: DroppedTransaction>) { let DroppedTransaction { tx_hash, reason } = dropped; trace!(target: LOG_TARGET, ?tx_hash, ?reason, "transaction_dropped"); - if let Err(error) = - self.controller.unbounded_send(ControllerCommand::new_dropped(tx_hash, reason)) - { - trace!( - target: LOG_TARGET, - ?tx_hash, - %error, - "transaction_dropped: send message failed" - ); - } + self.send_controller_command(ControllerCommand::new_dropped(tx_hash, reason)); } /// Send `Finalized` event for given transaction at given block. @@ -752,17 +900,7 @@ where idx: TxIndex, ) { trace!(target: LOG_TARGET, ?tx_hash, "transaction_finalized"); - if let Err(error) = self - .controller - .unbounded_send(ControllerCommand::new_finalized(tx_hash, block, idx)) - { - trace!( - target: LOG_TARGET, - ?tx_hash, - %error, - "transaction_finalized: send message failed" - ); - }; + self.send_controller_command(ControllerCommand::new_finalized(tx_hash, block, idx)); } /// Send `FinalityTimeout` event for given transactions at given block. @@ -775,17 +913,7 @@ where ) { for tx_hash in tx_hashes { trace!(target: LOG_TARGET, ?tx_hash, "transaction_finality_timeout"); - if let Err(error) = self - .controller - .unbounded_send(ControllerCommand::new_finality_timeout(*tx_hash, block)) - { - trace!( - target: LOG_TARGET, - ?tx_hash, - %error, - "transaction_finality_timeout: send message failed" - ); - }; + self.send_controller_command(ControllerCommand::new_finality_timeout(*tx_hash, block)); } } @@ -1102,3 +1230,241 @@ mod tests { let _ = listener_task.await.unwrap(); } } + +#[cfg(test)] +mod backlog_tests { + use super::*; + use crate::common::mock_api::MockChainApi; + use futures::{stream, FutureExt, StreamExt}; + use sp_core::H256; + use std::time::Duration; + use tokio::select; + + /// An unread external watcher must lose at most overflowing notifications, not its + /// subscription: a full channel is transient backpressure. After the backlog drains, + /// later statuses (including a final one) must still be deliverable. + #[tokio::test] + async fn external_watcher_survives_transient_backlog() { + sp_tracing::try_init_simple(); + + let (listener, listener_task) = + MultiViewListener::::new_with_worker(Default::default()); + let (terminate_tx, terminate_rx) = tokio::sync::oneshot::channel(); + let listener_handle = tokio::spawn(async move { + select! { + _ = listener_task => {}, + _ = terminate_rx => {}, + } + }); + + let tx_hash = H256::repeat_byte(0x0a); + let mut watcher = listener.create_external_watcher_for_tx(tx_hash).unwrap(); + + let block_hash = H256::repeat_byte(0x01); + // Flood more commands than the external watcher buffer while the subscriber + // is not consuming. Each Ready still occupies a queue slot even though the + // watcher context coalesces duplicates when drained. + const FLOOD: usize = STATUS_CHANNEL_CAPACITY + 200; + let events: Vec<_> = (0..FLOOD).map(|_| (tx_hash, TransactionStatus::Ready)).collect(); + // Keep the view stream alive (same pattern as other MVL tests) so the + // listener task does not observe a fully empty StreamMap. + listener.add_view_aggregated_stream( + block_hash, + stream::iter(events).chain(stream::pending()).boxed(), + ); + + // Allow the listener task to dispatch the flood into the per-tx channel. + tokio::time::sleep(Duration::from_millis(100)).await; + + // Drain whatever was buffered. The stream must still be open afterwards. + let mut terminated = false; + while let Some(item) = watcher.next().now_or_never() { + if item.is_none() { + terminated = true; + break; + } + } + assert!(!terminated, "external watcher must survive a transient backlog"); + + // A subsequent final status must still reach the (now draining) subscriber. + listener.transaction_finalized(tx_hash, block_hash, 0); + let next = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("final status should arrive") + .expect("watcher stream must still be open"); + assert_eq!(next, TransactionStatus::Finalized((block_hash, 0))); + + let _ = terminate_tx.send(()); + listener_handle.await.unwrap(); + } + + /// Terminal statuses must not be discarded when the bounded controller channel is + /// saturated; they travel on a dedicated loss-less lane. Regression test for + /// `Finalized`/`Dropped`/`Invalid`/`FinalityTimeout` being dropped on a full controller, + /// which left `submit_and_watch` streams open forever without a terminal event. + #[tokio::test] + async fn final_status_survives_full_controller_channel() { + sp_tracing::try_init_simple(); + + let (listener, listener_task) = + MultiViewListener::::new_with_worker(Default::default()); + + let tx_hash = H256::repeat_byte(0x0a); + let mut watcher = listener.create_external_watcher_for_tx(tx_hash).unwrap(); + + // Saturate the bounded controller channel with non-final commands for unrelated + // transactions while the listener task is not yet running (so nothing drains it). + const FLOOD: usize = STATUS_CHANNEL_CAPACITY + 200; + for i in 0..FLOOD { + let unrelated = H256::from_low_u64_be(i as u64 + 1); + listener.transactions_broadcasted( + std::iter::once((unrelated, vec!["peer".to_string()])).collect(), + ); + } + + // With the controller channel full, a terminal status must still be enqueued. + let block_hash = H256::repeat_byte(0x01); + listener.transaction_finalized(tx_hash, block_hash, 0); + + // Only now start the task, so the flood above could not have been drained early. + let (terminate_tx, terminate_rx) = tokio::sync::oneshot::channel(); + let listener_handle = tokio::spawn(async move { + select! { + _ = listener_task => {}, + _ = terminate_rx => {}, + } + }); + + let next = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("final status must arrive despite a full controller channel") + .expect("watcher stream must still be open"); + assert_eq!(next, TransactionStatus::Finalized((block_hash, 0))); + + let _ = terminate_tx.send(()); + listener_handle.await.unwrap(); + } + + /// `AddViewStream` must not be discarded when the bounded controller channel is saturated + /// (e.g. by a gossip `Broadcast` flood): losing it would silence all of that view's events + /// (`Ready`/`InBlock`/...) for every watcher for the view's whole lifetime. View stream + /// management is intrinsically rare (per block), so it rides the loss-less lane. + #[tokio::test] + async fn add_view_stream_survives_full_controller_channel() { + sp_tracing::try_init_simple(); + + let (listener, listener_task) = + MultiViewListener::::new_with_worker(Default::default()); + + let tx_hash = H256::repeat_byte(0x0a); + let mut watcher = listener.create_external_watcher_for_tx(tx_hash).unwrap(); + + // Saturate the bounded controller channel with broadcasts for unrelated transactions + // while the listener task is not yet running (so nothing drains it). + const FLOOD: usize = STATUS_CHANNEL_CAPACITY + 200; + for i in 0..FLOOD { + let unrelated = H256::from_low_u64_be(i as u64 + 1); + listener.transactions_broadcasted( + std::iter::once((unrelated, vec!["peer".to_string()])).collect(), + ); + } + + // With the controller channel full, adding a view must still be enqueued. + let block_hash = H256::repeat_byte(0x01); + listener.add_view_aggregated_stream( + block_hash, + stream::iter(vec![(tx_hash, TransactionStatus::Ready)]) + .chain(stream::pending()) + .boxed(), + ); + + // Only now start the task, so the flood above could not have been drained early. + let (terminate_tx, terminate_rx) = tokio::sync::oneshot::channel(); + let listener_handle = tokio::spawn(async move { + select! { + _ = listener_task => {}, + _ = terminate_rx => {}, + } + }); + + let next = tokio::time::timeout(Duration::from_secs(5), watcher.next()) + .await + .expect("view events must arrive despite a full controller channel") + .expect("watcher stream must still be open"); + assert_eq!(next, TransactionStatus::Ready); + + let _ = terminate_tx.send(()); + listener_handle.await.unwrap(); + } + + /// A permanently-ready aggregated view stream must not starve the lossless + /// lane under `biased` select. Regression for finals sitting forever behind + /// a hot `next_event` branch, leaving external watchers open indefinitely. + /// + /// Multi-thread runtime: a tight `stream::repeat` keeps the listener task + /// Ready on every poll; on the current-thread scheduler that would also + /// starve this test's timeout. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn final_status_not_starved_by_ready_view_stream() { + use std::{ + sync::{ + atomic::{AtomicBool, Ordering}, + Arc, + }, + task::Poll, + }; + + sp_tracing::try_init_simple(); + + let (listener, listener_task) = + MultiViewListener::::new_with_worker(Default::default()); + let listener_handle = tokio::spawn(async move { + listener_task.await; + }); + + let tx_hash = H256::repeat_byte(0x0a); + let mut watcher = listener.create_external_watcher_for_tx(tx_hash).unwrap(); + + let block_hash = H256::repeat_byte(0x01); + // Permanently ready view stream for an *unrelated* hash — keeps the + // aggregated-view branch ready (and would starve the lossless lane if it + // were polled first) without filling this watcher's channel. `stop` ends + // the stream so the listener can park and the runtime can shut down; a + // pure `stream::repeat` never yields and ignores task abort. + let flood_tx = H256::repeat_byte(0x0b); + let stop = Arc::new(AtomicBool::new(false)); + let stop_stream = stop.clone(); + listener.add_view_aggregated_stream( + block_hash, + stream::poll_fn(move |_cx| { + if stop_stream.load(Ordering::Relaxed) { + Poll::Ready(None) + } else { + Poll::Ready(Some((flood_tx, TransactionStatus::Ready))) + } + }) + .boxed(), + ); + + // Let the view stream start flooding before the final is queued. + tokio::time::sleep(Duration::from_millis(50)).await; + listener.transaction_finalized(tx_hash, block_hash, 0); + + let finalized = tokio::time::timeout(Duration::from_secs(2), async { + loop { + match watcher.next().await { + Some(TransactionStatus::Finalized(v)) => return v, + Some(_) => continue, + None => panic!("watcher closed before Finalized"), + } + } + }) + .await + .expect("Finalized must not be starved by a permanently-ready view stream"); + assert_eq!(finalized, (block_hash, 0)); + + stop.store(true, Ordering::Relaxed); + listener_handle.abort(); + let _ = listener_handle.await; + } +} diff --git a/client/transaction-pool/src/fork_aware_txpool/revalidation_worker.rs b/client/transaction-pool/src/fork_aware_txpool/revalidation_worker.rs index 64782588c..bf3da96df 100644 --- a/client/transaction-pool/src/fork_aware_txpool/revalidation_worker.rs +++ b/client/transaction-pool/src/fork_aware_txpool/revalidation_worker.rs @@ -18,34 +18,76 @@ //! The background worker for the [`View`] and [`TxMemPool`] revalidation. //! +//! View and mempool revalidation run on **separate** worker loops so that +//! [`View::finish_revalidation`] (awaited on the maintain critical path) cannot +//! stall behind an uncancellable [`TxMemPool::revalidate`] batch. +//! +//! Queues are capacity-limited: +//! - view jobs use a bounded channel (inline revalidation when full) +//! - mempool jobs use a latest-wins slot so finalization bursts cannot accumulate unbounded +//! `RevalidateMempool` payloads +//! //! The [*Background tasks*](../index.html#background-tasks) section provides some extra details on //! revalidation process. use std::{marker::PhantomData, pin::Pin, sync::Arc}; use crate::{graph::ChainApi, LOG_TARGET}; -use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender}; +use futures::{ + channel::mpsc::{channel, Receiver, Sender}, + prelude::*, +}; +use parking_lot::Mutex; use sp_blockchain::HashAndNumber; use sp_runtime::traits::Block as BlockT; - -use super::{tx_mem_pool::TxMemPool, view_store::ViewStore}; -use futures::prelude::*; use tracing::{debug, warn}; -use super::view::{FinishRevalidationWorkerChannels, View}; +use super::{ + tx_mem_pool::TxMemPool, + view::{FinishRevalidationWorkerChannels, View}, + view_store::ViewStore, +}; + +/// Bound for the view-revalidation job channel. +/// +/// Maintain enqueues at most one job per active view tip per cycle and waits on +/// completion, so a modest bound is enough. When full, [`RevalidationQueue::revalidate_view`] +/// falls back to inline revalidation so the finish protocol cannot hang. +const VIEW_REVALIDATION_QUEUE_CAPACITY: usize = 64; + +/// Request to revalidate a [`View`]. +/// +/// Communication channels with the maintain thread are also provided. +struct RevalidateViewPayload +where + Api: ChainApi + 'static, +{ + view: Arc>, + worker_channels: FinishRevalidationWorkerChannels, +} + +/// Request to revalidate a [`TxMemPool`] at the provided block hash. +struct RevalidateMempoolPayload +where + Block: BlockT, + Api: ChainApi + 'static, +{ + mempool: Arc>, + view_store: Arc>, + finalized_hash: HashAndNumber, +} -/// Revalidation request payload sent from the queue to the worker. -enum WorkerPayload +/// Latest-wins mempool revalidation slot. +/// +/// Producers overwrite `pending` and wake the worker. Only one payload is retained +/// while the worker is busy, so finalization bursts cannot grow a queue of +/// `Arc` / `Arc` jobs. +struct MempoolPendingSlot where Block: BlockT, Api: ChainApi + 'static, { - /// Request to revalidated the given instance of the [`View`] - /// - /// Communication channels with maintain thread are also provided. - RevalidateView(Arc>, FinishRevalidationWorkerChannels), - /// Request to revalidated the given instance of the [`TxMemPool`] at provided block hash. - RevalidateMempool(Arc>, Arc>, HashAndNumber), + pending: Mutex>>, } /// The background revalidation worker. @@ -63,43 +105,61 @@ where Self { _phantom: Default::default() } } - /// A background worker main loop. - /// - /// Waits for and dispatches the [`WorkerPayload`] messages sent from the - /// [`RevalidationQueue`]. - pub async fn run + 'static>( + /// Worker loop for view revalidation payloads. + async fn run_view + 'static>( self, - from_queue: TracingUnboundedReceiver>, + from_queue: Receiver>, ) { let mut from_queue = from_queue.fuse(); loop { let Some(payload) = from_queue.next().await else { - // R.I.P. worker! break; }; - match payload { - WorkerPayload::RevalidateView(view, worker_channels) => - view.revalidate(worker_channels).await, - WorkerPayload::RevalidateMempool( - mempool, - view_store, - finalized_hash_and_number, - ) => mempool.revalidate(view_store, finalized_hash_and_number).await, + payload.view.revalidate(payload.worker_channels).await; + } + } + + /// Worker loop for mempool revalidation: always process the latest pending job. + async fn run_mempool + 'static>( + self, + slot: Arc>, + wake_rx: Receiver<()>, + ) { + let mut wake_rx = wake_rx.fuse(); + + loop { + let Some(()) = wake_rx.next().await else { + break; }; + // Drain whatever is pending; producers may overwrite while we work, so + // loop until the slot stays empty after a job completes. + loop { + let payload = { slot.pending.lock().take() }; + let Some(payload) = payload else { + break; + }; + payload.mempool.revalidate(payload.view_store, payload.finalized_hash).await; + } } } } /// A Revalidation queue. /// -/// Allows to send the revalidation requests to the [`RevalidationWorker`]. +/// Allows to send the revalidation requests to the background workers. +/// +/// View and mempool jobs use independent channels so maintain's +/// [`View::finish_revalidation`] wait is not coupled to mempool batch work. pub struct RevalidationQueue where Api: ChainApi + 'static, Block: BlockT, { - background: Option>>, + view_background: Option>>>, + mempool_pending: Option>>, + /// Capacity-1 wake signal owned by the queue; drop shuts the mempool worker down. + mempool_wake: Option>>, } impl RevalidationQueue @@ -112,20 +172,44 @@ where /// /// All validation requests will be blocking. pub fn new() -> Self { - Self { background: None } + Self { view_background: None, mempool_pending: None, mempool_wake: None } } - /// New revalidation queue with background worker. + /// New revalidation queue with background workers. /// - /// All validation requests will be executed in the background. + /// View and mempool revalidation each run on their own worker loop so that a long + /// mempool batch cannot delay cancellation / completion of view revalidation. + /// + /// Mempool jobs coalesce to a latest-wins slot (no unbounded FIFO backlog). pub fn new_with_worker() -> (Self, Pin + Send>>) { - let (to_worker, from_queue) = tracing_unbounded("mpsc_revalidation_queue", 100_000); - (Self { background: Some(to_worker) }, RevalidationWorker::new().run(from_queue).boxed()) + let (to_view_worker, from_view_queue) = channel(VIEW_REVALIDATION_QUEUE_CAPACITY); + let (wake_tx, wake_rx) = channel(1); + let mempool_slot = Arc::new(MempoolPendingSlot { pending: Mutex::new(None) }); + let mempool_slot_worker = mempool_slot.clone(); + + let worker = async move { + futures::future::join( + RevalidationWorker::new().run_view(from_view_queue), + RevalidationWorker::new().run_mempool(mempool_slot_worker, wake_rx), + ) + .await; + }; + + ( + Self { + view_background: Some(Mutex::new(to_view_worker)), + mempool_pending: Some(mempool_slot), + mempool_wake: Some(Mutex::new(wake_tx)), + }, + worker.boxed(), + ) } /// Queue the view for later revalidation. /// - /// If the queue is configured with background worker, this will return immediately. + /// If the queue is configured with background worker, this will return immediately + /// unless the bounded queue is full, in which case revalidation runs inline so + /// maintain's finish protocol cannot hang. /// If the queue is configured without background worker, this will resolve after /// revalidation is actually done. /// @@ -141,16 +225,33 @@ where "revalidation_queue::revalidate_view: Sending view to revalidation queue" ); - if let Some(ref to_worker) = self.background { - if let Err(error) = to_worker.unbounded_send(WorkerPayload::RevalidateView( - view, - finish_revalidation_worker_channels, - )) { - warn!( - target: LOG_TARGET, - ?error, - "revalidation_queue::revalidate_view: Failed to update background worker" - ); + if let Some(ref to_worker) = self.view_background { + let payload = RevalidateViewPayload { + view: view.clone(), + worker_channels: finish_revalidation_worker_channels, + }; + // Drop the lock before any `.await` (parking_lot guards are not `Send`). + let send_result = { to_worker.lock().try_send(payload) }; + match send_result { + Ok(()) => {}, + Err(error) => { + let is_full = error.is_full(); + let payload = error.into_inner(); + if is_full { + warn!( + target: LOG_TARGET, + view_at_hash = ?view.at.hash, + "revalidation_queue::revalidate_view: queue full, running inline" + ); + } else { + warn!( + target: LOG_TARGET, + view_at_hash = ?view.at.hash, + "revalidation_queue::revalidate_view: worker gone, running inline" + ); + } + payload.view.revalidate(payload.worker_channels).await; + }, } } else { view.revalidate(finish_revalidation_worker_channels).await @@ -159,7 +260,8 @@ where /// Revalidates the given mempool instance. /// - /// If queue configured with background worker, this will return immediately. + /// If queue configured with background worker, this schedules a latest-wins job + /// (overwriting any not-yet-started mempool revalidation) and returns immediately. /// If queue configured without background worker, this will resolve after /// revalidation is actually done. /// @@ -176,22 +278,34 @@ where "Sent mempool to revalidation queue" ); - if let Some(ref to_worker) = self.background { - if let Err(error) = to_worker.unbounded_send(WorkerPayload::RevalidateMempool( - mempool, - view_store, - finalized_hash, - )) { - warn!( - target: LOG_TARGET, - ?error, - "Failed to update background worker" - ); + if let (Some(slot), Some(wake)) = (&self.mempool_pending, &self.mempool_wake) { + *slot.pending.lock() = + Some(RevalidateMempoolPayload { mempool, view_store, finalized_hash }); + // Capacity-1 wake: ignore full (worker already notified). + let wake_result = { wake.lock().try_send(()) }; + if let Err(error) = wake_result { + if error.is_disconnected() { + warn!( + target: LOG_TARGET, + "mempool revalidation worker gone; dropping job" + ); + } } } else { mempool.revalidate(view_store, finalized_hash).await } } + + /// Number of mempool revalidation jobs currently waiting in the latest-wins slot. + /// + /// Test-only helper (0 or 1). + #[cfg(test)] + pub(super) fn mempool_pending_jobs(&self) -> usize { + self.mempool_pending + .as_ref() + .map(|slot| usize::from(slot.pending.lock().is_some())) + .unwrap_or(0) + } } #[cfg(all(test, feature = "test-helpers"))] @@ -250,3 +364,156 @@ mod tests { assert_eq!(view.status().ready, 1); } } + +#[cfg(test)] +mod concurrency_tests { + use super::{ + super::{ + dropped_watcher::MultiViewDroppedWatcherController, + import_notification_sink::MultiViewImportNotificationSink, + multi_view_listener::MultiViewListener, + tx_mem_pool::{TxMemPool, TXMEMPOOL_REVALIDATION_PERIOD}, + view::View, + view_store::ViewStore, + }, + *, + }; + use crate::common::mock_api::{xt, MockChainApi, TestBlock}; + use sp_core::H256; + use sp_runtime::transaction_validity::TransactionSource; + use std::time::Duration; + + fn setup_mempool_and_view_store( + api: Arc, + ) -> ( + Arc>, + Arc>, + impl Future + Send, + impl Future + Send, + ) { + let (listener, listener_task) = MultiViewListener::new_with_worker(Default::default()); + let listener = Arc::new(listener); + let (import_notification_sink, import_notification_sink_task) = + MultiViewImportNotificationSink::new_with_worker(); + let (dropped_stream_controller, _dropped_stream) = + MultiViewDroppedWatcherController::::new(); + + let view_store = Arc::new(ViewStore::new( + api.clone(), + listener.clone(), + dropped_stream_controller, + import_notification_sink, + )); + // Only async mempool APIs are used below; the sync-bridge task is unused. + let (mempool, _mempool_task) = + TxMemPool::new(api, listener, Default::default(), 1024, usize::MAX); + let mempool = Arc::new(mempool); + + (mempool, view_store, listener_task, import_notification_sink_task) + } + + /// `finish_revalidation` (maintain critical path) must not stall behind an uncancellable + /// mempool revalidation batch that was enqueued earlier on the shared worker infrastructure. + #[tokio::test] + async fn finish_view_revalidation_not_blocked_by_mempool_revalidation() { + let mempool_validation_delay = Duration::from_millis(500); + let finish_budget = Duration::from_millis(100); + + let api = Arc::new(MockChainApi::with_validation_delay(mempool_validation_delay)); + let (mempool, view_store, listener_task, import_notification_sink_task) = + setup_mempool_and_view_store(api.clone()); + + let (queue, worker_task) = RevalidationQueue::::new_with_worker(); + let queue = Arc::new(queue); + + tokio::spawn(listener_task); + tokio::spawn(import_notification_sink_task); + tokio::spawn(worker_task); + + // Mempool txs are due for revalidation at finalized height > PERIOD. + let xts = vec![Arc::from(xt(1)), Arc::from(xt(2))]; + let _ = mempool.extend_unwatched(TransactionSource::External, 0, &xts).await; + + let finalized = HashAndNumber { + hash: H256::from_low_u64_be(TXMEMPOOL_REVALIDATION_PERIOD + 1), + number: TXMEMPOOL_REVALIDATION_PERIOD + 1, + }; + queue.revalidate_mempool(mempool.clone(), view_store.clone(), finalized).await; + + // Ensure the mempool job is dequeued before the view job is enqueued. + tokio::time::sleep(Duration::from_millis(20)).await; + + let view_at = HashAndNumber { hash: H256::from_low_u64_be(1), number: 1 }; + let view = Arc::new( + View::new(api, view_at, Default::default(), Default::default(), false.into()).0, + ); + View::start_background_revalidation(view.clone(), queue).await; + + let finish = tokio::time::timeout(finish_budget, view.finish_revalidation()).await; + assert!( + finish.is_ok(), + "finish_revalidation stalled behind mempool revalidation \ + (budget {:?}, mempool validation delay {:?})", + finish_budget, + mempool_validation_delay + ); + } + + /// Finalization bursts must not enqueue unbounded mempool revalidation work: the slot + /// holds at most one pending job, and only the latest finalized tip is applied. + #[tokio::test] + async fn mempool_revalidation_latest_wins_under_finalization_burst() { + let api = Arc::new(MockChainApi::with_validation_delay(Duration::from_millis(50))); + let (mempool, view_store, listener_task, import_notification_sink_task) = + setup_mempool_and_view_store(api.clone()); + + let (queue, worker_task) = RevalidationQueue::::new_with_worker(); + + tokio::spawn(listener_task); + tokio::spawn(import_notification_sink_task); + + let xts = vec![Arc::from(xt(1)), Arc::from(xt(2))]; + let _ = mempool.extend_unwatched(TransactionSource::External, 0, &xts).await; + + const BURST: u64 = 40; + // Enqueue before the worker runs so a FIFO would retain all jobs. + for i in 1..=BURST { + let number = TXMEMPOOL_REVALIDATION_PERIOD + i; + queue + .revalidate_mempool( + mempool.clone(), + view_store.clone(), + HashAndNumber { hash: H256::from_low_u64_be(number), number }, + ) + .await; + assert_eq!( + queue.mempool_pending_jobs(), + 1, + "pending mempool revalidation jobs must stay latest-wins (at most one)" + ); + } + + tokio::spawn(worker_task); + + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if queue.mempool_pending_jobs() == 0 && api.validation_count() > 0 { + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await + .expect("mempool revalidation did not complete"); + + // All BURST enqueues coalesced into a single job before the worker started, so only + // one batch runs (one validation per mempool tx). + assert_eq!( + api.validation_count(), + xts.len(), + "expected a single coalesced revalidation batch, got {} validations for burst of {}", + api.validation_count(), + BURST + ); + } +} diff --git a/client/transaction-pool/src/fork_aware_txpool/tx_mem_pool.rs b/client/transaction-pool/src/fork_aware_txpool/tx_mem_pool.rs index 762e28665..92e71542f 100644 --- a/client/transaction-pool/src/fork_aware_txpool/tx_mem_pool.rs +++ b/client/transaction-pool/src/fork_aware_txpool/tx_mem_pool.rs @@ -259,6 +259,19 @@ where } } +/// Priority used to order mempool entries for eviction (`try_insert_with_replacement`). +/// +/// `None` (priority not yet known) deliberately ranks *above* any concrete priority, making +/// unvalidated entries un-evictable. This matches upstream polkadot-sdk and is a conscious +/// trade-off ("don't evict what you can't compare yet"): every transaction is `None` between +/// mempool insertion and its first successful view submission, and ranking `None` lowest +/// would let any prioritized replacement evict legitimate fresh transactions in steady state. +/// +/// Security note (reviewed, accepted risk): while no views exist (startup/major sync) all +/// entries stay `None`, so a saturated mempool rejects replacement-path insertions with +/// `ImmediatelyDropped` regardless of the incoming priority. The condition is bounded (the +/// mempool itself is limited) and self-heals once a view exists: resubmission assigns +/// concrete priorities or drops invalid entries, restoring normal eviction. #[derive(Debug, Clone, Copy, Eq, PartialEq)] struct MempoolTxPriority(pub Option); @@ -431,6 +444,25 @@ where } } + /// Like [`Self::new_test`], but drops the sync-bridge receiver so `_sync` calls fail. + #[cfg(test)] + fn new_test_without_sync_bridge( + api: Arc, + max_transactions_count: usize, + max_transactions_total_bytes: usize, + ) -> Self { + let (sync_channel, _rx) = sync_bridge_channel(); + Self { + api, + listener: Arc::from(MultiViewListener::new_with_worker(Default::default()).0), + transactions: Default::default(), + metrics: Default::default(), + sync_channel, + max_transactions_count, + max_transactions_total_bytes, + } + } + /// Retrieves a transaction by its hash if it exists in the memory pool. pub(super) async fn get_by_hash( &self, @@ -992,12 +1024,20 @@ where xts: Vec>, ) -> Vec>, sc_transaction_pool_api::error::Error>> { + // Preserve the async `extend_unwatched` contract: one result per input, in order. + let input_len = xts.len(); let (response, request) = TxMemPoolSyncRequest::extend_unwatched(self.clone(), source, validated_at, xts); let _ = self.sync_channel.send(request); response.recv().unwrap_or_else(|_| { error!(target: LOG_TARGET, "extend_unwatched_sync: {}", SYNC_BRIDGE_EXPECT); - Vec::new() + (0..input_len) + .map(|_| { + Err(sc_transaction_pool_api::error::Error::InvalidBlockId( + SYNC_BRIDGE_EXPECT.into(), + )) + }) + .collect() }) } @@ -1401,3 +1441,37 @@ mod tx_mem_pool_tests { )); } } + +#[cfg(test)] +mod sync_bridge_tests { + use super::*; + use crate::common::mock_api::{xt, MockChainApi, TestBlock}; + use sp_runtime::transaction_validity::TransactionSource; + + /// Bridge failure must still return one `Err` per input so callers like + /// `submit_local` (`.remove(0)`) never panic on an empty vector. + #[test] + fn extend_unwatched_sync_returns_one_error_per_input_when_bridge_dead() { + let api = Arc::new(MockChainApi::default()); + let mempool = Arc::new(TxMemPool::::new_test_without_sync_bridge( + api, + 1024, + usize::MAX, + )); + let xts = vec![Arc::from(xt(1)), Arc::from(xt(2)), Arc::from(xt(3))]; + let input_len = xts.len(); + + let results = mempool.extend_unwatched_sync(TransactionSource::Local, 0, xts); + + assert_eq!(results.len(), input_len); + assert!( + results.iter().all(|r| matches!( + r, + Err(sc_transaction_pool_api::error::Error::InvalidBlockId(_)) + )), + "expected per-input InvalidBlockId on dead bridge, got {results:?}" + ); + // Mimic submit_local indexing: must not panic. + let _ = results.into_iter().next().expect("one result per input"); + } +} diff --git a/client/transaction-pool/src/fork_aware_txpool/view.rs b/client/transaction-pool/src/fork_aware_txpool/view.rs index ea5714218..6e0ce6047 100644 --- a/client/transaction-pool/src/fork_aware_txpool/view.rs +++ b/client/transaction-pool/src/fork_aware_txpool/view.rs @@ -33,17 +33,30 @@ use crate::{ }, LOG_TARGET, }; +use futures::{ + channel::mpsc::{ + channel, unbounded, Receiver as StatusStreamReceiver, Sender as StatusStreamSink, + UnboundedSender, + }, + StreamExt, +}; use indexmap::IndexMap; use parking_lot::Mutex; use sc_transaction_pool_api::{error::Error as TxPoolError, PoolStatus, TransactionStatus}; -use sc_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender}; use sp_blockchain::HashAndNumber; use sp_runtime::{ generic::BlockId, traits::Block as BlockT, transaction_validity::TransactionValidityError, SaturatedConversion, }; -use std::{sync::Arc, time::Instant}; -use tracing::{debug, instrument, trace, Level}; +use std::{ + pin::Pin, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, + time::Instant, +}; +use tracing::{debug, instrument, trace, warn, Level}; pub(super) struct RevalidationResult { revalidated: IndexMap, ValidatedTransactionFor>, @@ -113,14 +126,36 @@ impl FinishRevalidationWorkerChannels { /// Single event used in aggregated stream. Tuple containing hash of transactions and its status. pub(super) type TransactionStatusEvent = (H, TransactionStatus); -/// Warning threshold for (unbounded) channel used in aggregated view's streams. -const VIEW_STREAM_WARN_THRESHOLD: usize = 100_000; + +/// Capacity of the bounded per-view status channels (aggregated + dropped monitoring). +/// +/// A full channel drops the overflowing notification but keeps the stream open so a slow +/// consumer cannot drive unbounded heap growth through `EventHandler` fan-out. +/// +/// For the dropped-monitoring stream this bound applies only to droppable statuses +/// (`Ready`/`Future`); actionable statuses (`Dropped`/`Usurped`/`Invalid`) are always delivered — +/// see [`ViewPoolObserver::send_to_dropped_stream_sink`]. +pub(super) const VIEW_STATUS_CHANNEL_CAPACITY: usize = 1024; /// Stream of events providing statuses of all the transactions within the pool. -pub(super) type AggregatedStream = TracingUnboundedReceiver>; +pub(super) type AggregatedStream = StatusStreamReceiver>; /// Type alias for a stream of events intended to track dropped transactions. -type DroppedMonitoringStream = TracingUnboundedReceiver>; +type DroppedMonitoringStream = + Pin> + Send>>; + +/// Whether the dropped-monitoring watcher must never lose this status. +/// +/// `Dropped`, `Usurped` and `Invalid` drive mempool removal and view-tracking cleanup in +/// [`super::dropped_watcher`]; losing one leaves the transaction stuck in the mempool and in the +/// view-tracking maps. `Ready`/`Future` only refresh view membership and may be dropped under +/// backpressure. +fn requires_lossless_delivery(status: &TransactionStatus) -> bool { + matches!( + status, + TransactionStatus::Dropped | TransactionStatus::Usurped(_) | TransactionStatus::Invalid + ) +} /// Notification handler for transactions updates triggered in `ValidatedPool`. /// @@ -131,19 +166,29 @@ pub(super) struct ViewPoolObserver { /// The sink used to notify dropped by enforcing limits or by being usurped, or invalid /// transactions. /// + /// The channel is unbounded so actionable statuses (`Dropped`/`Usurped`/`Invalid`) can never + /// be lost; the flood-prone `Ready`/`Future` traffic is bounded via + /// `droppable_events_in_flight`. Actionable events correspond to transactions leaving the + /// view's pool, so their backlog is bounded in practice by the pool's own capacity limits. + /// /// Note: Ready and future statuses are alse communicated through this channel, enabling the /// stream consumer to track views that reference the transaction. - dropped_stream_sink: TracingUnboundedSender< - TransactionStatusEvent, BlockHash>, - >, + dropped_stream_sink: + UnboundedSender, BlockHash>>, + + /// Number of droppable (`Ready`/`Future`) events currently buffered in + /// `dropped_stream_sink`, used to bound that traffic at [`VIEW_STATUS_CHANNEL_CAPACITY`]. + /// + /// Incremented on send, decremented by the receiving stream as events are consumed. + droppable_events_in_flight: Arc, /// The sink of the single, merged stream providing updates for all the transactions in the /// associated pool. /// /// Note: some of the events which are currently ignored on the other side of this channel /// (external watcher) are not relayed. - aggregated_stream_sink: TracingUnboundedSender< - TransactionStatusEvent, BlockHash>, + aggregated_stream_sink: Mutex< + StatusStreamSink, BlockHash>>, >, } @@ -206,33 +251,102 @@ impl ViewPoolObserver { DroppedMonitoringStream, BlockHash>, AggregatedStream, BlockHash>, ) { - let (dropped_stream_sink, dropped_stream) = - tracing_unbounded("mpsc_txpool_watcher", VIEW_STREAM_WARN_THRESHOLD); - let (aggregated_stream_sink, aggregated_stream) = - tracing_unbounded("mpsc_txpool_aggregated_stream", VIEW_STREAM_WARN_THRESHOLD); + let (dropped_stream_sink, dropped_stream) = unbounded(); + let (aggregated_stream_sink, aggregated_stream) = channel(VIEW_STATUS_CHANNEL_CAPACITY); + + let droppable_events_in_flight = Arc::new(AtomicUsize::new(0)); + let dropped_stream = { + let in_flight = droppable_events_in_flight.clone(); + dropped_stream + .map(move |event: TransactionStatusEvent<_, _>| { + if !requires_lossless_delivery(&event.1) { + in_flight.fetch_sub(1, Ordering::Relaxed); + } + event + }) + .boxed() + }; - (Self { dropped_stream_sink, aggregated_stream_sink }, dropped_stream, aggregated_stream) + ( + Self { + dropped_stream_sink, + droppable_events_in_flight, + aggregated_stream_sink: Mutex::new(aggregated_stream_sink), + }, + dropped_stream, + aggregated_stream, + ) } /// Sends given event to the `dropped_stream_sink`. + /// + /// Actionable statuses (`Dropped`/`Usurped`/`Invalid`) are always delivered — losing them + /// would leave transactions stuck in the mempool and view-tracking maps. Droppable statuses + /// (`Ready`/`Future`) are dropped once [`VIEW_STATUS_CHANNEL_CAPACITY`] of them are already + /// buffered (slow consumer). A disconnected receiver is ignored. fn send_to_dropped_stream_sink( &self, tx: ExtrinsicHash, status: TransactionStatus, BlockHash>, ) { - if let Err(e) = self.dropped_stream_sink.unbounded_send((tx, status.clone())) { - trace!(target: LOG_TARGET, "[{:?}] dropped_sink: {:?} send message failed: {:?}", tx, status, e); + if requires_lossless_delivery(&status) { + if self.dropped_stream_sink.unbounded_send((tx, status.clone())).is_err() { + trace!( + target: LOG_TARGET, + ?tx, + ?status, + "dropped_sink: send message failed (receiver gone)" + ); + } + return; + } + + if self.droppable_events_in_flight.load(Ordering::Relaxed) >= VIEW_STATUS_CHANNEL_CAPACITY { + warn!( + target: LOG_TARGET, + ?tx, + ?status, + "dropped_sink: channel full, dropping status notification" + ); + return; + } + self.droppable_events_in_flight.fetch_add(1, Ordering::Relaxed); + if self.dropped_stream_sink.unbounded_send((tx, status.clone())).is_err() { + self.droppable_events_in_flight.fetch_sub(1, Ordering::Relaxed); + trace!( + target: LOG_TARGET, + ?tx, + ?status, + "dropped_sink: send message failed (receiver gone)" + ); } } /// Sends given event to the `aggregated_stream_sink`. + /// + /// A full channel drops only this notification (slow consumer); a disconnected + /// receiver is ignored. fn send_to_aggregated_stream_sink( &self, tx: ExtrinsicHash, status: TransactionStatus, BlockHash>, ) { - if let Err(e) = self.aggregated_stream_sink.unbounded_send((tx, status.clone())) { - trace!(target: LOG_TARGET, "[{:?}] aggregated_stream {:?} send message failed: {:?}", tx, status, e); + if let Err(error) = self.aggregated_stream_sink.lock().try_send((tx, status.clone())) { + if error.is_full() { + warn!( + target: LOG_TARGET, + ?tx, + ?status, + "aggregated_stream: channel full, dropping status notification" + ); + } else { + trace!( + target: LOG_TARGET, + ?tx, + ?status, + "aggregated_stream: send message failed (receiver gone)" + ); + } } } } @@ -693,3 +807,96 @@ where .remove_subtree(hashes, ban_transactions, listener_action) } } + +#[cfg(test)] +mod status_channel_tests { + use super::*; + use crate::{common::mock_api::MockChainApi, graph::EventHandler}; + use futures::{FutureExt, StreamExt}; + use sp_core::H256; + + /// Flooding a non-draining view status stream must not grow without bound: at most + /// `VIEW_STATUS_CHANNEL_CAPACITY` notifications are retained, and subsequent sends + /// after a drain still succeed (the stream stays open through backpressure). + #[tokio::test] + async fn view_status_streams_bound_backlog_and_stay_open() { + let (observer, mut dropped_rx, mut aggregated_rx) = ViewPoolObserver::::new(); + + let flood = VIEW_STATUS_CHANNEL_CAPACITY + 500; + for i in 0..flood { + observer.ready(H256::from_low_u64_be(i as u64)); + } + + let mut dropped_count = 0usize; + while dropped_rx.next().now_or_never().flatten().is_some() { + dropped_count += 1; + } + let mut aggregated_count = 0usize; + while aggregated_rx.next().now_or_never().flatten().is_some() { + aggregated_count += 1; + } + + // `futures::mpsc` capacity is `buffer + num_senders` (one guaranteed slot per sender). + let max_retained = VIEW_STATUS_CHANNEL_CAPACITY + 1; + assert!( + dropped_count <= max_retained, + "dropped stream retained {dropped_count} > capacity {max_retained}" + ); + assert!( + aggregated_count <= max_retained, + "aggregated stream retained {aggregated_count} > capacity {max_retained}" + ); + assert!(dropped_count > 0); + assert!(aggregated_count > 0); + + // After draining, the sinks must still accept notifications. + observer.ready(H256::repeat_byte(0xff)); + assert!(dropped_rx.next().now_or_never().flatten().is_some()); + assert!(aggregated_rx.next().now_or_never().flatten().is_some()); + } + + /// Actionable statuses (`Dropped`/`Usurped`/`Invalid`) drive mempool removal and + /// view-tracking cleanup in the dropped watcher and must never be lost, even while the + /// dropped-monitoring stream is saturated by a `Ready` flood. Regression test for + /// actionable events being discarded on a full sink, which left transactions stuck in + /// the mempool and view-tracking maps. + #[tokio::test] + async fn dropped_sink_never_loses_actionable_events_under_ready_flood() { + let (observer, mut dropped_rx, _aggregated_rx) = ViewPoolObserver::::new(); + + // Saturate the dropped-monitoring stream with droppable events while nothing drains. + let flood = VIEW_STATUS_CHANNEL_CAPACITY + 500; + for i in 0..flood { + observer.ready(H256::from_low_u64_be(i as u64 + 1_000_000)); + } + + // Actionable events arriving on the saturated sink must still be delivered. + let dropped_tx = H256::repeat_byte(0xaa); + let usurped_tx = H256::repeat_byte(0xbb); + let usurper = H256::repeat_byte(0xbc); + let invalid_tx = H256::repeat_byte(0xcc); + observer.limits_enforced(dropped_tx); + observer.usurped(usurped_tx, usurper); + observer.invalid(invalid_tx); + + let mut actionable = Vec::new(); + while let Some((tx, status)) = dropped_rx.next().now_or_never().flatten() { + if matches!( + status, + TransactionStatus::Dropped | + TransactionStatus::Usurped(_) | + TransactionStatus::Invalid + ) { + actionable.push((tx, status)); + } + } + assert_eq!( + actionable, + vec![ + (dropped_tx, TransactionStatus::Dropped), + (usurped_tx, TransactionStatus::Usurped(usurper)), + (invalid_tx, TransactionStatus::Invalid), + ] + ); + } +} diff --git a/client/transaction-pool/src/graph/base_pool.rs b/client/transaction-pool/src/graph/base_pool.rs index 318643481..8a5663530 100644 --- a/client/transaction-pool/src/graph/base_pool.rs +++ b/client/transaction-pool/src/graph/base_pool.rs @@ -369,6 +369,31 @@ impl BasePool BasePool BasePool Some(current.clone()), Some(worst) => Some( - match (worst.transaction.source.timestamp, current.transaction.source.timestamp) - { - (Some(worst_timestamp), Some(current_timestamp)) => { - if worst_timestamp > current_timestamp { - current.clone() - } else { - worst - } - }, - _ => - if worst.imported_at > current.imported_at { - current.clone() - } else { - worst + // Prefer to evict the lowest-priority transaction first, matching the + // ready-queue policy and this function's documented behavior. Without + // this, age-only eviction lets an attacker flood the future queue with + // low-priority transactions to force out a victim's older, higher-priority + // future transaction (which `ValidatedPool` then bans), turning queue + // pressure into a priority-bypassing censorship path. Age is only used to + // break ties between equal priorities (evict the one waiting longest). + match worst.transaction.priority.cmp(¤t.transaction.priority) { + Ordering::Less => worst, + Ordering::Greater => current.clone(), + Ordering::Equal => match ( + worst.transaction.source.timestamp, + current.transaction.source.timestamp, + ) { + (Some(worst_timestamp), Some(current_timestamp)) => { + if worst_timestamp > current_timestamp { + current.clone() + } else { + worst + } }, + _ => + if worst.imported_at > current.imported_at { + current.clone() + } else { + worst + }, + }, }, ), }); @@ -545,10 +584,39 @@ impl BasePool Vec>> { let mut removed = self.ready.remove_subtree(hashes); + // Future txs only track tags that were missing at import. Tags that were + // satisfied by a ready provider which we just removed must be re-marked as + // missing, otherwise a later provider of the remaining tags can falsely + // promote an incompletely dependent future transaction into ready. + // Skip provides still covered by another ready tx or by recently pruned + // (included) tags — those remain satisfied. + let tags_to_unsatisfy = self.lost_provides(removed.iter().map(|tx| tx.as_ref())); + self.future.unsatisfy_tags(tags_to_unsatisfy); removed.extend(self.future.remove(hashes)); removed } + /// Tags provided by `removed` that are no longer satisfied by the ready pool or + /// by recently pruned (included) tags. + fn lost_provides<'a>( + &self, + removed: impl Iterator>, + ) -> HashSet + where + Hash: 'a, + Ex: 'a, + { + let provided = self.ready.provided_tags(); + removed + .flat_map(|tx| tx.provides.iter()) + .filter(|tag| { + !provided.contains_key(*tag) && + !self.recently_pruned.iter().any(|set| set.contains(*tag)) + }) + .cloned() + .collect() + } + /// Removes and returns all transactions from the future queue. pub fn clear_future(&mut self) -> Vec>> { self.future.clear() @@ -1089,6 +1157,116 @@ mod tests { assert_eq!(pool.future.len(), 0); } + /// Importing a higher-priority replacement can unlock a future (moving it into the + /// promotion queue) and simultaneously replace the ready tx that uniquely provided + /// one of that future's other requirements. The future has already left `self.future` + /// via `satisfy_tags`, so a repair that only scans the future map misses it and the + /// next loop iteration would admit it to Ready with an unsatisfied requirement. + #[test] + fn replace_during_promotion_keeps_incompletely_unlocked_tx_in_future() { + let mut pool = pool(); + let tag_conflict = vec![0xAAu8]; + let tag_lost = vec![0xBBu8]; + let tag_unlock = vec![0xCCu8]; + let tag_f = vec![0xDDu8]; + + // Ready R provides {conflict, lost}. + pool.import(Transaction { + data: vec![1u8], + hash: 1, + priority: 1u64, + provides: vec![tag_conflict.clone(), tag_lost.clone()], + ..default_tx() + }) + .unwrap(); + + // Future F requires {lost, unlock}; at import, lost is satisfied by R. + pool.import(Transaction { + data: vec![2u8], + hash: 2, + requires: vec![tag_lost.clone(), tag_unlock.clone()], + provides: vec![tag_f], + ..default_tx() + }) + .unwrap(); + assert_eq!(pool.ready().count(), 1); + assert_eq!(pool.future.len(), 1); + + // Higher-priority A provides {conflict, unlock}: unlocks F into to_import, then + // replaces R (losing `lost`). F must not enter Ready incomplete. + pool.import(Transaction { + data: vec![3u8], + hash: 3, + priority: 100u64, + provides: vec![tag_conflict, tag_unlock], + ..default_tx() + }) + .unwrap(); + + assert!(pool.ready().any(|tx| tx.hash == 3), "replacement A must be ready"); + assert!( + !pool.ready().any(|tx| tx.hash == 2), + "F must not be admitted to Ready without `lost`" + ); + assert_eq!(pool.future.len(), 1, "F must remain in future until `lost` is provided again"); + assert!(pool.futures().any(|tx| tx.hash == 2)); + } + + /// A future that required tags X and Y, imported while X was already ready, must not be + /// promoted once Y appears if the ready provider of X was removed in the meantime. + /// `missing_tags` only tracked Y at import; removing X's provider must re-mark X as missing. + #[test] + fn remove_subtree_re_marks_future_deps_on_removed_ready_provides() { + let mut pool = pool(); + let tag_x = vec![0x11]; + let tag_y = vec![0x22]; + let tag_f = vec![0x33]; + + // Ready provider of X. + pool.import(Transaction { + data: vec![1u8], + hash: 1, + provides: vec![tag_x.clone()], + ..default_tx() + }) + .unwrap(); + + // Future requiring X and Y: X is satisfied at import, so missing_tags = {Y}. + pool.import(Transaction { + data: vec![2u8], + hash: 2, + requires: vec![tag_x.clone(), tag_y.clone()], + provides: vec![tag_f], + ..default_tx() + }) + .unwrap(); + assert_eq!(pool.ready().count(), 1); + assert_eq!(pool.future.len(), 1); + + // Remove the ready provider of X (e.g. invalidation / limit eviction). + pool.remove_subtree(&[1]); + assert_eq!(pool.ready().count(), 0); + assert_eq!(pool.future.len(), 1); + + // A new transaction provides Y. Without re-marking X as missing, the future would + // be falsely promoted to ready with an incomplete dependency set. + pool.import(Transaction { + data: vec![3u8], + hash: 3, + provides: vec![tag_y], + ..default_tx() + }) + .unwrap(); + + assert_eq!(pool.ready().count(), 1, "only the Y provider should be ready"); + assert_eq!(pool.future.len(), 1, "future still waiting for X"); + assert!( + pool.ready().all(|tx| tx.hash == 3), + "future requiring X must not enter ready after X's provider was removed" + ); + assert!(pool.futures().any(|tx| tx.hash == 2)); + } + #[test] fn should_prune_ready_transactions() { // given @@ -1251,6 +1429,60 @@ source: TimedTransactionSource { source: TransactionSource::External, timestamp: assert_eq!(pool.future.len(), 0); } + #[test] + fn future_limit_enforcement_evicts_lowest_priority_first() { + use std::time::Duration; + + let mut pool = pool(); + + // Older, high-priority future transaction (submitted first). + let t_old = Instant::now(); + pool.import(Transaction { + data: vec![0u8].into(), + hash: 0x10, + priority: 1_000u64, + requires: vec![vec![10]], + provides: vec![vec![11]], + source: TimedTransactionSource { + source: TransactionSource::External, + timestamp: Some(t_old), + }, + ..default_tx().clone() + }) + .unwrap(); + + // Newer, low-priority future transaction (submitted later). + pool.import(Transaction { + data: vec![1u8].into(), + hash: 0x20, + priority: 1u64, + requires: vec![vec![20]], + provides: vec![vec![21]], + source: TimedTransactionSource { + source: TransactionSource::External, + timestamp: Some(t_old + Duration::from_secs(1)), + }, + ..default_tx().clone() + }) + .unwrap(); + + assert_eq!(pool.future.len(), 2); + + // Enforce a future limit that only leaves room for a single transaction. + let removed = pool.enforce_limits( + &Limit { count: 100, total_bytes: 100 }, + &Limit { count: 1, total_bytes: 100 }, + ); + + // The low-priority transaction must be evicted even though it is newer; the + // older but higher-priority transaction must survive. Age-only eviction would + // (incorrectly) drop the older high-priority one. + assert_eq!(removed.len(), 1); + assert_eq!(removed[0].hash, 0x20); + let remaining = pool.futures().map(|tx| tx.hash).collect::>(); + assert_eq!(remaining, vec![0x10]); + } + #[test] fn should_accept_future_transactions_when_explicitly_asked_to() { // given diff --git a/client/transaction-pool/src/graph/future.rs b/client/transaction-pool/src/graph/future.rs index 805c12976..56a0ddf98 100644 --- a/client/transaction-pool/src/graph/future.rs +++ b/client/transaction-pool/src/graph/future.rs @@ -221,6 +221,36 @@ impl became_ready } + /// Re-marks tags as missing on future transactions that require them. + /// + /// `WaitingTransaction::missing_tags` only records requirements that were unsatisfied at + /// import time. When a ready provider of a previously satisfied tag is removed (via + /// [`crate::graph::base_pool::BasePool::remove_subtree`]), those tags must be restored here + /// so a later `satisfy_tags` cannot promote an incompletely dependent transaction. + /// + /// Complexity is `O(future_entries × requirements_per_tx)` against a hashed tag set — + /// not `O(lost_tags × futures × requirements)`. Large ready-subtree removals can supply + /// tens of thousands of provides while thousands of futures sit in the pool; the + /// previous per-tag full scan was a DoS vector on the pool mutation path. + pub fn unsatisfy_tags(&mut self, tags: impl IntoIterator) { + let tags: HashSet = tags.into_iter().collect(); + if tags.is_empty() { + return; + } + + let mut newly_missing = Vec::new(); + for (hash, waiting) in self.waiting.iter_mut() { + for req in waiting.transaction.requires.iter() { + if tags.contains(req) && waiting.missing_tags.insert(req.clone()) { + newly_missing.push((hash.clone(), req.clone())); + } + } + } + for (hash, tag) in newly_missing { + self.wanted_tags.entry(tag).or_insert_with(HashSet::new).insert(hash); + } + } + /// Removes transactions for given list of hashes. /// /// Returns a list of actually removed transactions. diff --git a/client/transaction-pool/src/graph/listener.rs b/client/transaction-pool/src/graph/listener.rs index cc0e7d90c..37ef764d3 100644 --- a/client/transaction-pool/src/graph/listener.rs +++ b/client/transaction-pool/src/graph/listener.rs @@ -129,6 +129,25 @@ impl> EventDispatcher, C, L> { sender.new_watcher(hash) } + /// Reclaim watcher map state after a `Watcher` was dropped without a lifecycle event. + /// + /// `submit_and_watch` registers a watcher before import; if import fails + /// (`AlreadyImported`, `TooLowPriority`, …) the returned `Watcher` is dropped while + /// its sender would otherwise remain until a later `fire`. Prune closed receivers + /// and remove the map entry when nothing remains — without notifying any still-live + /// watchers for the same hash. + pub fn reclaim_closed_watcher(&mut self, hash: &ExtrinsicHash) { + let remove = if let Some(sender) = self.watchers.get_mut(hash) { + sender.prune_closed(); + sender.is_done() + } else { + false + }; + if remove { + self.watchers.remove(hash); + } + } + /// Notify the listeners about the extrinsic broadcast. pub fn broadcasted(&mut self, tx_hash: &ExtrinsicHash, peers: Vec) { trace!( @@ -237,7 +256,8 @@ impl> EventDispatcher, C, L> { if let Some((hash, txs)) = self.finality_watchers.pop_front() { for tx in txs { self.fire(&tx, |watcher| watcher.finality_timeout(hash)); - self.event_handler.as_ref().map(|l| l.finality_timeout(tx, block_hash)); + // Use the evicted block hash (`hash`), not the block currently being pruned. + self.event_handler.as_ref().map(|l| l.finality_timeout(tx, hash)); } } } @@ -274,3 +294,88 @@ impl> EventDispatcher, C, L> { self.watchers.keys() } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::common::mock_api::MockChainApi; + use sp_core::H256; + use std::{cell::RefCell, rc::Rc}; + + type Dispatcher = EventDispatcher; + + /// Records `finality_timeout` notifications for assertion. + struct RecordingFinalityHandler { + timeouts: Rc>>, + } + + impl EventHandler for RecordingFinalityHandler { + fn finality_timeout(&self, tx: ExtrinsicHash, hash: BlockHash) { + self.timeouts.borrow_mut().push((tx, hash)); + } + } + + #[test] + fn finality_timeout_event_handler_uses_evicted_block_hash() { + let timeouts = Rc::new(RefCell::new(Vec::new())); + let mut dispatcher = EventDispatcher::::new_with_event_handler( + Some(RecordingFinalityHandler { timeouts: timeouts.clone() }), + ); + + // Fill past the finality-watcher limit so the oldest block is evicted with a timeout. + for i in 0..=MAX_FINALITY_WATCHERS { + let block = H256::from_low_u64_be(i as u64); + let tx = H256::from_low_u64_be(10_000 + i as u64); + let _watcher = dispatcher.create_watcher(tx); + dispatcher.pruned(block, &tx); + } + + let events = timeouts.borrow(); + assert_eq!(events.len(), 1, "exactly one eviction timeout expected, got {events:?}"); + let (tx, timed_out_block) = events[0]; + assert_eq!(tx, H256::from_low_u64_be(10_000)); + assert_eq!( + timed_out_block, + H256::from_low_u64_be(0), + "handler must report the evicted block, not the block currently being pruned" + ); + } + + #[test] + fn reclaim_closed_watcher_removes_orphaned_entry() { + let mut dispatcher = Dispatcher::default(); + let hash = H256::repeat_byte(0x11); + + let watcher = dispatcher.create_watcher(hash); + assert_eq!(dispatcher.watched_transactions().count(), 1); + + // Mimic submit_and_watch error path: Watcher dropped, no lifecycle event fired. + drop(watcher); + assert_eq!( + dispatcher.watched_transactions().count(), + 1, + "without reclaim, closed watcher still occupies the map" + ); + + dispatcher.reclaim_closed_watcher(&hash); + assert_eq!(dispatcher.watched_transactions().count(), 0); + } + + #[test] + fn reclaim_closed_watcher_keeps_live_watchers() { + let mut dispatcher = Dispatcher::default(); + let hash = H256::repeat_byte(0x22); + + let live = dispatcher.create_watcher(hash); + let orphan = dispatcher.create_watcher(hash); + drop(orphan); + + dispatcher.reclaim_closed_watcher(&hash); + assert_eq!(dispatcher.watched_transactions().count(), 1); + assert_eq!(live.hash(), &hash); + + // Live watcher must still receive events after reclaim. + dispatcher.ready(&hash, None); + drop(live); + } +} diff --git a/client/transaction-pool/src/graph/ready.rs b/client/transaction-pool/src/graph/ready.rs index 4c0343cb2..88e2fa98e 100644 --- a/client/transaction-pool/src/graph/ready.rs +++ b/client/transaction-pool/src/graph/ready.rs @@ -469,6 +469,14 @@ impl ReadyTransactions { let new_provides = tx.provides.iter().cloned().collect::>(); let removed = self.remove_subtree_with_tag_filter(to_remove, Some(new_provides)); + // `unlocks` was collected before the tag-filtered removal. Descendants that + // depended on tags the replacement does not provide are gone from the ready + // map; keep only edges to transactions that actually survived so BestIterator + // never treats a stale hash as a satisfied dependency. + let ready = self.ready.read(); + let unlocks = + unlocks.into_iter().filter(|hash| ready.contains_key(hash)).collect::>(); + Ok((removed, unlocks)) } @@ -826,4 +834,65 @@ mod tests { assert!(!tx1_unlocks.contains(&tx3.hash)); assert!(tx1_unlocks.contains(&tx4.hash)); } + + /// Partial replacement (overlapping but non-identical provides) must not leave + /// unlock edges pointing at descendants that were dropped because they required + /// tags the replacement does not provide. Otherwise BestIterator can promote a + /// later re-imported descendant before its real provider is yielded. + #[test] + fn partial_replace_does_not_keep_stale_unlock_edges() { + let mut ready = ReadyTransactions::default(); + let tag_x = vec![0x11]; + let tag_y = vec![0x22]; + let tag_z = vec![0x33]; + + let mut tx_a = tx(1); + tx_a.requires.clear(); + tx_a.provides = vec![tag_x.clone(), tag_y.clone()]; + tx_a.priority = 1; + + let mut tx_d = tx(2); + tx_d.requires = vec![tag_y.clone()]; + tx_d.provides = vec![tag_z.clone()]; + tx_d.priority = 1; + + import(&mut ready, tx_a).unwrap(); + import(&mut ready, tx_d.clone()).unwrap(); + assert_eq!(ready.get().count(), 2); + + // Replacement provides only X (drops Y). Higher priority so it replaces tx_a. + let mut tx_c = tx(3); + tx_c.requires.clear(); + tx_c.provides = vec![tag_x]; + tx_c.priority = 10; + import(&mut ready, tx_c).unwrap(); + + // Descendant requiring Y must be gone, and the replacement must not retain + // a stale unlock edge to it. + assert!(!ready.contains(&tx_d.hash)); + { + let lock = ready.ready.read(); + let unlocks = &lock.get(&3).expect("replacement in ready").unlocks; + assert!( + !unlocks.contains(&tx_d.hash), + "stale unlock edge to removed descendant: {unlocks:?}" + ); + } + + // Real provider of Y, then re-submit the descendant. + let mut tx_y = tx(4); + tx_y.requires.clear(); + tx_y.provides = vec![tag_y]; + tx_y.priority = 1; + import(&mut ready, tx_y).unwrap(); + import(&mut ready, tx_d).unwrap(); + + let order: Vec<_> = ready.get().map(|tx| tx.hash).collect(); + let pos_y = order.iter().position(|h| *h == 4).expect("Y provider in ready set"); + let pos_d = order.iter().position(|h| *h == 2).expect("descendant in ready set"); + assert!( + pos_y < pos_d, + "descendant must not be ordered before its required-tag provider; order={order:?}" + ); + } } diff --git a/client/transaction-pool/src/graph/rotator.rs b/client/transaction-pool/src/graph/rotator.rs index 03215ac53..b7b9fc6ed 100644 --- a/client/transaction-pool/src/graph/rotator.rs +++ b/client/transaction-pool/src/graph/rotator.rs @@ -49,11 +49,22 @@ pub struct PoolRotator { expected_size: usize, } -impl Clone for PoolRotator { +impl Clone for PoolRotator { fn clone(&self) -> Self { + // Fork-aware views clone the rotator when creating a new view. Drop expired + // entries here so a stale ban cannot be propagated into later views after + // `ban_time` has elapsed (even if `clear_timeouts` was never called). + let now = Instant::now(); + let active = self + .banned_until + .read() + .iter() + .filter(|(_, until)| **until >= now) + .map(|(hash, until)| (hash.clone(), *until)) + .collect(); Self { ban_time: self.ban_time, - banned_until: RwLock::new(self.banned_until.read().clone()), + banned_until: RwLock::new(active), expected_size: self.expected_size, } } @@ -81,8 +92,35 @@ impl PoolRotator { } /// Returns `true` if extrinsic hash is currently banned. + /// + /// Respects the stored expiry: an entry whose ban time has elapsed is treated as + /// not banned (and lazily removed), so callers do not need to have run + /// [`clear_timeouts`] beforehand. Previously only key presence was checked, which + /// meant expired bans stayed authoritative until a separate cleanup ran — and + /// fork-aware view clones could keep them alive indefinitely. pub fn is_banned(&self, hash: &Hash) -> bool { - self.banned_until.read().contains_key(hash) + let now = Instant::now(); + { + let banned = self.banned_until.read(); + match banned.get(hash) { + Some(until) if *until >= now => return true, + Some(_) => {}, + None => return false, + } + } + // Expired under the read lock — reclaim only if still expired under the + // write lock. A concurrent re-ban (e.g. from `remove_invalid` after a + // failed revalidation) can land in the gap; an unconditional remove + // would drop that fresh ban and waste another revalidation cycle. + let mut banned = self.banned_until.write(); + match banned.get(hash) { + Some(until) if *until >= Instant::now() => true, + Some(_) => { + banned.remove(hash); + false + }, + None => false, + } } /// Bans given set of hashes. @@ -201,6 +239,41 @@ mod tests { assert!(!rotator.is_banned(&hash)); } + /// An expired ban must not reject the transaction even if `clear_timeouts` was never + /// called — otherwise correctness depends on every caller running maintenance before + /// every check, and fork-aware view clones can keep stale bans alive indefinitely. + #[test] + fn expired_ban_is_not_considered_banned_without_clear_timeouts() { + let (hash, _) = tx(); + let rotator = rotator(); + // Insert a ban whose expiry is already in the past. + let past = Instant::now() - rotator.ban_time - Duration::from_millis(1); + rotator.ban(&past, iter::once(hash)); + + assert!( + !rotator.is_banned(&hash), + "expired ban must not be treated as active without an explicit clear_timeouts" + ); + } + + /// Cloning a rotator (as fork-aware views do) must not resurrect expired bans into + /// the new view's rotator map. + #[test] + fn clone_does_not_propagate_expired_bans() { + let (hash, _) = tx(); + let rotator = rotator(); + let past = Instant::now() - rotator.ban_time - Duration::from_millis(1); + rotator.ban(&past, iter::once(hash)); + assert!(rotator.banned_until.read().contains_key(&hash)); + + let cloned = rotator.clone(); + assert!(!cloned.is_banned(&hash)); + assert!( + !cloned.banned_until.read().contains_key(&hash), + "clone must drop expired ban entries rather than copy them into the new view" + ); + } + #[test] fn should_garbage_collect() { // given diff --git a/client/transaction-pool/src/graph/validated_pool.rs b/client/transaction-pool/src/graph/validated_pool.rs index 7ea3d8ed7..060cf258a 100644 --- a/client/transaction-pool/src/graph/validated_pool.rs +++ b/client/transaction-pool/src/graph/validated_pool.rs @@ -259,6 +259,11 @@ impl> ValidatedPool { } } + /// Returns the pool configuration options. + pub fn options(&self) -> &Options { + &self.options + } + /// Bans given set of hashes. pub fn ban(&self, now: &Instant, hashes: impl IntoIterator>) { self.rotator.ban(now, hashes) @@ -449,11 +454,24 @@ impl> ValidatedPool { let mut results = self.submit(std::iter::once(ValidatedTransaction::Valid(tx))); // We submitted exactly one transaction, so we get exactly one result match results.pop() { - Some(result) => result.map(|outcome| outcome.with_watcher(watcher)), - None => Err(sc_transaction_pool_api::error::Error::InvalidBlockId( - "submit returned no results".into(), - ) - .into()), + Some(Ok(outcome)) => Ok(outcome.with_watcher(watcher)), + Some(Err(err)) => { + // Import failed after `create_watcher`. Drop the unused watcher and + // reclaim its map entry / closed sender — otherwise eventless + // failures (AlreadyImported, TooLowPriority, …) leak permanently + // because cleanup only happens inside `fire`. + drop(watcher); + self.event_dispatcher.write().reclaim_closed_watcher(&hash); + Err(err) + }, + None => { + drop(watcher); + self.event_dispatcher.write().reclaim_closed_watcher(&hash); + Err(sc_transaction_pool_api::error::Error::InvalidBlockId( + "submit returned no results".into(), + ) + .into()) + }, } }, ValidatedTransaction::Invalid(hash, err) => { @@ -899,3 +917,97 @@ fn fire_events( base::Imported::Future { ref hash } => event_dispatcher.future(hash), } } + +#[cfg(test)] +mod watcher_reclaim_tests { + use super::*; + use crate::common::mock_api::{xt, MockChainApi}; + use codec::Encode; + use sc_transaction_pool_api::error::Error as TxPoolError; + + fn pool() -> ValidatedPool { + ValidatedPool::new(Options::default(), true.into(), Arc::new(MockChainApi::default())) + } + + fn valid_tx( + api: &MockChainApi, + value: u64, + priority: TransactionPriority, + provides: Tag, + ) -> ValidatedTransactionFor { + let data = Arc::new(xt(value)); + let (hash, bytes) = api.hash_and_length(&data); + ValidatedTransaction::valid_at( + 0, + hash, + base::TimedTransactionSource::new_external(false), + data, + bytes, + ValidTransaction { + priority, + requires: vec![], + provides: vec![provides], + longevity: 64, + propagate: true, + }, + ) + } + + /// `submit` then `submit_and_watch` of the same hash fails with AlreadyImported and + /// must not leave an orphaned watcher map entry. + #[test] + fn submit_and_watch_already_imported_does_not_leak_watcher() { + let pool = pool(); + let api = pool.api(); + let tag = 1u64.encode(); + let tx = valid_tx(api, 1, 10, tag.clone()); + + assert!(pool.submit(std::iter::once(tx)).pop().unwrap().is_ok()); + assert!(pool.watched_transactions().is_empty()); + + let duplicate = valid_tx(api, 1, 10, tag); + let err = pool.submit_and_watch(duplicate).map(|_| ()).unwrap_err(); + assert!(matches!(err, TxPoolError::AlreadyImported(_))); + assert!( + pool.watched_transactions().is_empty(), + "AlreadyImported after create_watcher must reclaim watcher state" + ); + } + + /// A lower-priority conflicting replacement fails with TooLowPriority and must not + /// leave an orphaned watcher map entry. + #[test] + fn submit_and_watch_too_low_priority_does_not_leak_watcher() { + let pool = pool(); + let api = pool.api(); + // Distinct extrinsic hashes that compete for the same provides tag. + let tag = b"shared-tag".to_vec(); + + assert!(pool + .submit(std::iter::once(valid_tx(api, 7, 100, tag.clone()))) + .pop() + .unwrap() + .is_ok()); + + let low = valid_tx(api, 8, 1, tag); + let err = pool.submit_and_watch(low).map(|_| ()).unwrap_err(); + assert!(matches!(err, TxPoolError::TooLowPriority { .. })); + assert!( + pool.watched_transactions().is_empty(), + "TooLowPriority after create_watcher must reclaim watcher state" + ); + } + + /// Successful submit_and_watch still registers a watcher. + #[test] + fn submit_and_watch_success_keeps_watcher() { + let pool = pool(); + let api = pool.api(); + let mut outcome = pool + .submit_and_watch(valid_tx(api, 3, 10, 3u64.encode())) + .map_err(|e| e.to_string()) + .unwrap(); + assert!(outcome.take_watcher().is_some()); + assert_eq!(pool.watched_transactions().len(), 1); + } +} diff --git a/client/transaction-pool/src/graph/watcher.rs b/client/transaction-pool/src/graph/watcher.rs index 2fd31e772..c8c0f6f68 100644 --- a/client/transaction-pool/src/graph/watcher.rs +++ b/client/transaction-pool/src/graph/watcher.rs @@ -134,6 +134,14 @@ impl Sender { self.is_finalized || self.receivers.is_empty() } + /// Drop receivers whose `Watcher` has already been dropped (channel closed). + /// + /// Used to reclaim watcher state after a failed `submit_and_watch` without firing a + /// lifecycle event that would incorrectly notify any remaining live watchers. + pub fn prune_closed(&mut self) { + self.receivers.retain(|sender| !sender.is_closed()); + } + fn send(&mut self, status: TransactionStatus) { self.receivers.retain(|sender| sender.unbounded_send(status.clone()).is_ok()) } diff --git a/client/transaction-pool/src/single_state_txpool/single_state_txpool.rs b/client/transaction-pool/src/single_state_txpool/single_state_txpool.rs index 22df59fed..539d60157 100644 --- a/client/transaction-pool/src/single_state_txpool/single_state_txpool.rs +++ b/client/transaction-pool/src/single_state_txpool/single_state_txpool.rs @@ -273,21 +273,50 @@ where xts: Vec>, ) -> Result, Self::Error>>, Self::Error> { let pool = self.pool.clone(); - let xts = xts - .into_iter() - .map(|xt| { - (TimedTransactionSource::from_transaction_source(source, false), Arc::from(xt)) - }) - .collect::>(); + + // Pre-validation admission control: accept only as many transactions as the pool + // could possibly hold (count and cumulative bytes) and reject the excess up front. + // This bounds the validation work and the memory retained by a single batch before + // the pool storage limits (which are enforced only after validation) can engage. + let options = pool.validated_pool().options(); + let max_count = options.total_count(); + let max_bytes = options.ready.total_bytes.saturating_add(options.future.total_bytes); + + let total_count = xts.len(); + let mut cumulative_bytes = 0usize; + let mut accepted = Vec::with_capacity(total_count.min(max_count)); + for xt in xts { + cumulative_bytes = cumulative_bytes.saturating_add(self.api.hash_and_length(&xt).1); + if accepted.len() >= max_count || cumulative_bytes > max_bytes { + break; + } + accepted.push(( + TimedTransactionSource::from_transaction_source(source, false), + Arc::from(xt), + )); + } + let rejected_count = total_count - accepted.len(); + if rejected_count > 0 { + warn!( + target: LOG_TARGET, + total_count, + rejected_count, + "submit_at: rejecting batch transactions exceeding pool capacity" + ); + } let number = self.api.resolve_block_number(at); let at = HashAndNumber { hash: at, number: number? }; - let results = pool - .submit_at(&at, xts, ValidateTransactionPriority::Submitted) + let mut results = pool + .submit_at(&at, accepted, ValidateTransactionPriority::Submitted) .await .into_iter() .map(|result| result.map(|outcome| outcome.hash())) .collect::>(); + results.extend( + std::iter::repeat_with(|| Err(TxPoolError::ImmediatelyDropped.into())) + .take(rejected_count), + ); let success_count = results.iter().filter(|result| result.is_ok()).count() as u64; if success_count > 0 { self.metrics @@ -842,3 +871,83 @@ where } } } + +#[cfg(test)] +mod batch_admission_tests { + use super::*; + use crate::common::mock_api::{xt, MockChainApi, TestBlock}; + use codec::Encode; + use sp_core::H256; + + fn test_pool( + ready: graph::base_pool::Limit, + future: graph::base_pool::Limit, + ) -> (BasicPool, Arc) { + let api = Arc::new(MockChainApi::default()); + let genesis = H256::from_low_u64_be(0); + let options = graph::Options { ready, future, ..Default::default() }; + let (pool, background_task) = BasicPool::new_test(api.clone(), genesis, genesis, options); + tokio::spawn(background_task); + (pool, api) + } + + /// A batch with more transactions than the pool could ever hold must not have its + /// excess validated: validation work has to be bounded by the pool capacity before + /// any runtime validation is scheduled. Otherwise an attacker can saturate runtime + /// validation and retain arbitrarily large batches in memory before pool limits engage. + #[tokio::test] + async fn oversized_batch_count_is_capped_before_validation() { + let (pool, api) = test_pool( + graph::base_pool::Limit { count: 4, total_bytes: 1024 * 1024 }, + graph::base_pool::Limit { count: 4, total_bytes: 1024 * 1024 }, + ); + let genesis = H256::from_low_u64_be(0); + + let batch = (0..100).map(xt).collect::>(); + let results = pool + .submit_at(genesis, TransactionSource::External, batch) + .await + .expect("submit_at succeeds"); + + // Every transaction gets a result, in order. + assert_eq!(results.len(), 100); + // Only transactions that could possibly fit into the pool were validated. + assert!( + api.validation_count() <= 8, + "expected at most 8 validations, got {}", + api.validation_count() + ); + // The excess was rejected up front. + for result in &results[8..] { + assert!(matches!(result, Err(TxPoolError::ImmediatelyDropped))); + } + } + + /// Same as above, for the cumulative bytes limit. + #[tokio::test] + async fn oversized_batch_bytes_is_capped_before_validation() { + let tx_bytes = xt(0).encoded_size(); + // Byte budget fits 4 transactions in total (2 ready + 2 future), counts are large. + let (pool, api) = test_pool( + graph::base_pool::Limit { count: 1024, total_bytes: 2 * tx_bytes }, + graph::base_pool::Limit { count: 1024, total_bytes: 2 * tx_bytes }, + ); + let genesis = H256::from_low_u64_be(0); + + let batch = (0..50).map(xt).collect::>(); + let results = pool + .submit_at(genesis, TransactionSource::External, batch) + .await + .expect("submit_at succeeds"); + + assert_eq!(results.len(), 50); + assert!( + api.validation_count() <= 4, + "expected at most 4 validations, got {}", + api.validation_count() + ); + for result in &results[4..] { + assert!(matches!(result, Err(TxPoolError::ImmediatelyDropped))); + } + } +} diff --git a/node/src/service.rs b/node/src/service.rs index 34c50139f..b1d5f4373 100644 --- a/node/src/service.rs +++ b/node/src/service.rs @@ -25,8 +25,6 @@ use codec::Encode; use jsonrpsee::tokio; use quantus_miner_api::{ApiResponseStatus, MiningRequest, MiningResult}; use sc_basic_authorship::ProposerFactory; -use sc_cli::TransactionPoolType; -use sc_transaction_pool::TransactionPoolOptions; use sp_api::ProvideRuntimeApi; use sp_consensus::SyncOracle; use sp_consensus_qpow::QPoWApi; @@ -612,21 +610,15 @@ pub fn new_partial(config: &Configuration) -> Result { telemetry }); - let pool_options = TransactionPoolOptions::new_with_params( - 36772, /* each tx is about 7300 bytes so if we have 268MB for the pool we can fit this - * many txs */ - 268_435_456, - None, - TransactionPoolType::SingleState.into(), - false, - ); + // Pool type/limits come from CLI (`--pool-type`, `--pool-limit`, `--pool-kbytes`, …) + // via `Configuration::transaction_pool`. Builder logs the selected type at create time. let transaction_pool = Arc::from( sc_transaction_pool::Builder::new( task_manager.spawn_essential_handle(), client.clone(), config.role.is_authority().into(), ) - .with_options(pool_options) + .with_options(config.transaction_pool.clone()) .with_prometheus(config.prometheus_registry()) .build(), ); diff --git a/pallets/reversible-transfers/src/benchmarking.rs b/pallets/reversible-transfers/src/benchmarking.rs index 80e1d17a2..c0d15cbb0 100644 --- a/pallets/reversible-transfers/src/benchmarking.rs +++ b/pallets/reversible-transfers/src/benchmarking.rs @@ -286,6 +286,16 @@ mod benchmarks { // upper bound for any mix of successful and failed releases. Do not change // this to a cheaper (e.g. all-failing) path without re-deriving the weight // model, or failed-release recoveries would be undercharged. + // + // Scheduler worst case: each pending transfer's `dispatch_time` is derived + // from the current block plus the configured delay (`DefaultDelay` is + // block-based). Submitting one transfer per block therefore spreads the + // sender's pending set across `n` distinct `Scheduler::Agenda` keys. + // `cancel_named` mutates and cleans up the agenda entry for each `when`, so + // the benchmark must advance the block between every schedule — not cluster + // many transfers into a few agenda buckets. Clustering under-measures Agenda + // DB/proof work. Advancing every iteration also keeps each agenda at a + // single task, so `MaxScheduledPerBlock` is never a constraint here. #[benchmark] fn recover_funds(n: Linear<0, 16>) -> Result<(), BenchmarkError> { assert_eq!( @@ -306,7 +316,8 @@ mod benchmarks { let transfer_amount: BalanceOf = 100u128.into(); for i in 0..n { - if i > 0 && i.is_multiple_of(8) { + if i > 0 { + // One transfer per block => `n` distinct Agenda keys on cancel. let bn = frame_system::Pallet::::block_number(); frame_system::Pallet::::set_block_number(bn + BlockNumberFor::::one()); } diff --git a/pallets/reversible-transfers/src/lib.rs b/pallets/reversible-transfers/src/lib.rs index 343ffb0da..f726e54dc 100644 --- a/pallets/reversible-transfers/src/lib.rs +++ b/pallets/reversible-transfers/src/lib.rs @@ -533,6 +533,15 @@ pub mod pallet { /// It cancels all pending transfers first (applying volume fees), then transfers /// the remaining free balance to the guardian. /// + /// # Cancel vs recovery authority + /// + /// Per-transfer `cancel` freezes authority in `pending.guardian` at schedule time + /// (so a later `set_high_security` cannot rewrite cancel rights on pre-enrollment + /// one-time transfers). `recover_funds` does **not** use that freeze: it authorizes + /// against the *live* high-security guardian and seizes every pending hold on the + /// account (volume fee applied). That asymmetry is intentional — recovery is + /// seize-the-account, not a batch of frozen cancel policies. + /// /// # Repeated Recovery /// /// This function can be called multiple times on the same account. The high-security @@ -912,12 +921,17 @@ pub mod pallet { /// Cancels a previously scheduled transaction. Internal logic used by `cancel` extrinsic. fn cancel_transfer(who: &T::AccountId, tx_id: T::Hash) -> DispatchResult { let pending = PendingTransfers::::get(tx_id).ok_or(Error::::PendingTxNotFound)?; - let high_security_account_data = HighSecurityAccounts::::get(&pending.from); - // Determine recipient and apply fee based on account type - let (recipient, apply_fee) = if let Some(ref data) = high_security_account_data { - ensure!(who == &data.guardian, Error::::InvalidReverser); - (data.guardian.clone(), true) + // Authority, recipient, and fee policy are frozen at schedule time in + // `pending.guardian` (high-security schedules store the configured guardian; + // one-time schedules store the sender). Do not re-read live + // `HighSecurityAccounts` here — a later `set_high_security` would otherwise + // retroactively rewrite cancel rights and burn a volume fee for transfers + // the owner scheduled while still a regular account. + let apply_fee = pending.guardian != pending.from; + let (recipient, apply_fee) = if apply_fee { + ensure!(who == &pending.guardian, Error::::InvalidReverser); + (pending.guardian.clone(), true) } else { ensure!(who == &pending.from, Error::::NotOwner); (pending.from.clone(), false) diff --git a/pallets/reversible-transfers/src/tests/test_high_security_account.rs b/pallets/reversible-transfers/src/tests/test_high_security_account.rs index ef479ed60..85df8a184 100644 --- a/pallets/reversible-transfers/src/tests/test_high_security_account.rs +++ b/pallets/reversible-transfers/src/tests/test_high_security_account.rs @@ -241,3 +241,84 @@ fn too_many_pending_transactions_error() { ); }); } + +/// Mirrors the `recover_funds` benchmark's worst-case setup: one scheduled +/// transfer per block so cancellations touch `n` distinct `Scheduler::Agenda` +/// keys (not a handful of clustered buckets). +#[test] +fn recover_funds_cancels_across_distinct_agenda_buckets() { + use pallet_scheduler::Agenda; + use qp_scheduler::{BlockNumberOrTimestamp, ScheduleNamed}; + use std::collections::BTreeSet; + + new_test_ext().execute_with(|| { + System::set_block_number(1); + let hs_user = alice(); + let guardian = bob(); + let dest = charlie(); + let amount = 100u128; + let n = MaxPendingPerAccount::get(); + + for i in 0..n { + if i > 0 { + System::set_block_number(System::block_number() + 1); + } + assert_ok!(ReversibleTransfers::schedule_transfer( + RuntimeOrigin::signed(hs_user.clone()), + dest.clone(), + amount + )); + } + + let pending = crate::PendingTransfersBySender::::get(&hs_user); + assert_eq!(pending.len() as u32, n); + + let mut agenda_keys = BTreeSet::new(); + for tx_id in pending.iter() { + let schedule_id = ReversibleTransfers::make_schedule_id(tx_id).expect("schedule id"); + let when = >::next_dispatch_time( + schedule_id, + ) + .expect("named task is scheduled"); + let agenda_key = BlockNumberOrTimestamp::BlockNumber(when); + agenda_keys.insert(agenda_key); + assert!( + !Agenda::::get(agenda_key).is_empty(), + "expected a non-empty agenda at {agenda_key:?}" + ); + } + assert_eq!( + agenda_keys.len() as u32, + n, + "one schedule per block must produce n distinct Agenda keys; \ + clustering would under-weight recover_funds cancellations" + ); + + assert_ok!(ReversibleTransfers::recover_funds( + RuntimeOrigin::signed(guardian), + hs_user.clone() + )); + assert!(crate::PendingTransfersBySender::::get(&hs_user).is_empty()); + for when in agenda_keys { + assert!( + Agenda::::get(when).is_empty(), + "recover_funds must clear agenda bucket at {when:?}" + ); + } + }); +} + +#[test] +fn recover_funds_weight_charges_agenda_per_pending_transfer() { + use crate::weights::WeightInfo; + + let w0 = <() as WeightInfo>::recover_funds(0); + let w1 = <() as WeightInfo>::recover_funds(1); + let step = w1.saturating_sub(w0); + // Matches the unmodified FRAME-generated per-n proof component (Agenda MEL). + assert_eq!( + step.proof_size(), + 12493, + "recover_funds(n) must charge one Scheduler::Agenda proof per pending transfer" + ); +} diff --git a/pallets/reversible-transfers/src/tests/test_reversible_transfers.rs b/pallets/reversible-transfers/src/tests/test_reversible_transfers.rs index fecbe4315..c939004a3 100644 --- a/pallets/reversible-transfers/src/tests/test_reversible_transfers.rs +++ b/pallets/reversible-transfers/src/tests/test_reversible_transfers.rs @@ -671,6 +671,61 @@ fn no_volume_fee_for_regular_reversible_accounts() { }); } +/// A one-time schedule freezes *cancel* authority in `pending.guardian` (= sender). +/// Later `set_high_security` must not rewrite that: the owner keeps full-refund cancel +/// rights, and the new guardian must not be able to cancel/seize via `cancel`. +/// +/// This does **not** constrain `recover_funds`: once the account is high-security, the +/// live guardian may still seize these holds through recovery (account-level seize). +#[test] +fn set_high_security_does_not_retroactively_reclassify_pending_one_time_cancel() { + new_test_ext().execute_with(|| { + System::set_block_number(1); + let owner = charlie(); // regular account + let recipient = dave(); + let new_guardian = eve(); + let amount = 10_000u128; + + let initial_owner = Balances::free_balance(&owner); + let initial_guardian = Balances::free_balance(&new_guardian); + let initial_issuance = pallet_balances::TotalIssuance::::get(); + + let call = transfer_call(recipient.clone(), amount); + let tx_id = calculate_tx_id::(owner.clone(), &call); + + assert_ok!(ReversibleTransfers::schedule_transfer_with_delay( + RuntimeOrigin::signed(owner.clone()), + recipient, + amount, + BlockNumberOrTimestamp::BlockNumber(10), + )); + + // Policy frozen at schedule time: guardian is the owner themselves. + let pending = ReversibleTransfers::pending_dispatches(tx_id).expect("pending"); + assert_eq!(pending.guardian, owner); + + // Owner later enrolls in high security with a different guardian. + assert_ok!(ReversibleTransfers::set_high_security( + RuntimeOrigin::signed(owner.clone()), + BlockNumberOrTimestamp::BlockNumber(10), + new_guardian.clone(), + )); + + // New guardian must not be able to cancel / seize the pre-existing one-time transfer. + assert_err!( + ReversibleTransfers::cancel(RuntimeOrigin::signed(new_guardian.clone()), tx_id), + Error::::NotOwner + ); + + // Owner still cancels with a full refund and no volume fee. + assert_ok!(ReversibleTransfers::cancel(RuntimeOrigin::signed(owner.clone()), tx_id)); + assert!(ReversibleTransfers::pending_dispatches(tx_id).is_none()); + assert_eq!(Balances::free_balance(&owner), initial_owner); + assert_eq!(Balances::free_balance(&new_guardian), initial_guardian); + assert_eq!(pallet_balances::TotalIssuance::::get(), initial_issuance); + }); +} + #[test] fn cancel_dispatch_fails_not_owner() { new_test_ext().execute_with(|| { diff --git a/pallets/reversible-transfers/src/weights.rs b/pallets/reversible-transfers/src/weights.rs index 7a7d0baa6..bb642dc51 100644 --- a/pallets/reversible-transfers/src/weights.rs +++ b/pallets/reversible-transfers/src/weights.rs @@ -194,24 +194,28 @@ impl WeightInfo for SubstrateWeight { /// Proof: `System::Account` (`max_values`: None, `max_size`: Some(128), added: 2603, mode: `MaxEncodedLen`) /// Storage: `Scheduler::Lookup` (r:16 w:16) /// Proof: `Scheduler::Lookup` (`max_values`: None, `max_size`: Some(53), added: 2528, mode: `MaxEncodedLen`) - /// Storage: `Scheduler::Agenda` (r:2 w:2) + /// Storage: `Scheduler::Agenda` (r:16 w:16) /// Proof: `Scheduler::Agenda` (`max_values`: None, `max_size`: Some(10018), added: 12493, mode: `MaxEncodedLen`) /// Storage: `Scheduler::Retries` (r:0 w:16) /// Proof: `Scheduler::Retries` (`max_values`: None, `max_size`: Some(40), added: 2515, mode: `MaxEncodedLen`) /// The range of component `n` is `[0, 16]`. + /// + /// Regenerated 2026-08-03 (`--steps=50 --repeat=20`) against the distinct- + /// Agenda-bucket `recover_funds` setup (one schedule per block). Values below + /// are the unmodified FRAME CLI output for both ref-time and proof size. fn recover_funds(n: u32, ) -> Weight { // Proof Size summary in bytes: - // Measured: `522 + n * (364 ±0)` - // Estimated: `9482 + n * (2640 ±32)` - // Minimum execution time: 52_000_000 picoseconds. - Weight::from_parts(55_250_404, 9482) - // Standard Error: 70_515 - .saturating_add(Weight::from_parts(56_328_935, 0).saturating_mul(n.into())) + // Measured: `514 + n * (383 ±0)` + // Estimated: `4026 + n * (12493 ±0)` + // Minimum execution time: 62_000_000 picoseconds. + Weight::from_parts(65_862_538, 4026) + // Standard Error: 134_621 + .saturating_add(Weight::from_parts(64_986_364, 0).saturating_mul(n.into())) .saturating_add(T::DbWeight::get().reads(4_u64)) - .saturating_add(T::DbWeight::get().reads((2_u64).saturating_mul(n.into()))) + .saturating_add(T::DbWeight::get().reads((3_u64).saturating_mul(n.into()))) .saturating_add(T::DbWeight::get().writes(3_u64)) - .saturating_add(T::DbWeight::get().writes((3_u64).saturating_mul(n.into()))) - .saturating_add(Weight::from_parts(0, 2640).saturating_mul(n.into())) + .saturating_add(T::DbWeight::get().writes((4_u64).saturating_mul(n.into()))) + .saturating_add(Weight::from_parts(0, 12493).saturating_mul(n.into())) } } @@ -350,23 +354,27 @@ impl WeightInfo for () { /// Proof: `System::Account` (`max_values`: None, `max_size`: Some(128), added: 2603, mode: `MaxEncodedLen`) /// Storage: `Scheduler::Lookup` (r:16 w:16) /// Proof: `Scheduler::Lookup` (`max_values`: None, `max_size`: Some(53), added: 2528, mode: `MaxEncodedLen`) - /// Storage: `Scheduler::Agenda` (r:2 w:2) + /// Storage: `Scheduler::Agenda` (r:16 w:16) /// Proof: `Scheduler::Agenda` (`max_values`: None, `max_size`: Some(10018), added: 12493, mode: `MaxEncodedLen`) /// Storage: `Scheduler::Retries` (r:0 w:16) /// Proof: `Scheduler::Retries` (`max_values`: None, `max_size`: Some(40), added: 2515, mode: `MaxEncodedLen`) /// The range of component `n` is `[0, 16]`. + /// + /// Regenerated 2026-08-03 (`--steps=50 --repeat=20`) against the distinct- + /// Agenda-bucket `recover_funds` setup (one schedule per block). Values below + /// are the unmodified FRAME CLI output for both ref-time and proof size. fn recover_funds(n: u32, ) -> Weight { // Proof Size summary in bytes: - // Measured: `522 + n * (364 ±0)` - // Estimated: `9482 + n * (2640 ±32)` - // Minimum execution time: 52_000_000 picoseconds. - Weight::from_parts(55_250_404, 9482) - // Standard Error: 70_515 - .saturating_add(Weight::from_parts(56_328_935, 0).saturating_mul(n.into())) + // Measured: `514 + n * (383 ±0)` + // Estimated: `4026 + n * (12493 ±0)` + // Minimum execution time: 62_000_000 picoseconds. + Weight::from_parts(65_862_538, 4026) + // Standard Error: 134_621 + .saturating_add(Weight::from_parts(64_986_364, 0).saturating_mul(n.into())) .saturating_add(RocksDbWeight::get().reads(4_u64)) - .saturating_add(RocksDbWeight::get().reads((2_u64).saturating_mul(n.into()))) + .saturating_add(RocksDbWeight::get().reads((3_u64).saturating_mul(n.into()))) .saturating_add(RocksDbWeight::get().writes(3_u64)) - .saturating_add(RocksDbWeight::get().writes((3_u64).saturating_mul(n.into()))) - .saturating_add(Weight::from_parts(0, 2640).saturating_mul(n.into())) + .saturating_add(RocksDbWeight::get().writes((4_u64).saturating_mul(n.into()))) + .saturating_add(Weight::from_parts(0, 12493).saturating_mul(n.into())) } } diff --git a/pallets/transaction-payment-rpc/src/lib.rs b/pallets/transaction-payment-rpc/src/lib.rs index 050c7fb89..c6fc1fe78 100644 --- a/pallets/transaction-payment-rpc/src/lib.rs +++ b/pallets/transaction-payment-rpc/src/lib.rs @@ -37,6 +37,30 @@ use sp_runtime::traits::{Block as BlockT, MaybeDisplay}; pub use pallet_transaction_payment_rpc_runtime_api::TransactionPaymentApi as TransactionPaymentRuntimeApi; +/// Maximum encoded extrinsic length accepted by fee-query RPCs before SCALE decode. +/// +/// Aligned with this chain's maximum block length (`RuntimeBlockLength` = 5 MiB). Inputs +/// larger than this cannot be included on-chain; without a local cap, the JSON-RPC +/// handler would still decode them into a full `Block::Extrinsic` (e.g. a huge +/// `remark` payload) and burn RPC worker CPU/heap for free. +pub const MAX_ENCODED_EXTRINSIC_LEN: usize = 5 * 1024 * 1024; + +/// Rejects fee-query inputs that exceed `max_len` before any extrinsic decode work. +fn ensure_query_extrinsic_size(encoded_xt: &[u8], max_len: usize) -> RpcResult<()> { + if encoded_xt.len() > max_len { + return Err(ErrorObject::owned( + ErrorCode::InvalidParams.code(), + format!( + "Encoded extrinsic length ({}) exceeds maximum allowed for fee query ({})", + encoded_xt.len(), + max_len + ), + None::<()>, + )); + } + Ok(()) +} + #[rpc(client, server)] pub trait TransactionPaymentApi { #[method(name = "payment_queryInfo")] @@ -54,13 +78,23 @@ pub trait TransactionPaymentApi { pub struct TransactionPayment { /// Shared reference to the client. client: Arc, + /// Cap on `encoded_xt` accepted by fee-query RPCs before SCALE decode. + max_encoded_extrinsic_len: usize, _marker: std::marker::PhantomData

, } impl TransactionPayment { /// Creates a new instance of the TransactionPayment Rpc helper. pub fn new(client: Arc) -> Self { - Self { client, _marker: Default::default() } + Self::new_with_max_encoded_extrinsic_len(client, MAX_ENCODED_EXTRINSIC_LEN) + } + + /// Creates a new instance with an explicit pre-decode size cap for `encoded_xt`. + pub fn new_with_max_encoded_extrinsic_len( + client: Arc, + max_encoded_extrinsic_len: usize, + ) -> Self { + Self { client, max_encoded_extrinsic_len, _marker: Default::default() } } } @@ -97,6 +131,8 @@ where encoded_xt: Bytes, at: Option, ) -> RpcResult> { + ensure_query_extrinsic_size(&encoded_xt, self.max_encoded_extrinsic_len)?; + let api = self.client.runtime_api(); let at_hash = at.unwrap_or_else(|| self.client.info().best_hash); @@ -130,6 +166,8 @@ where encoded_xt: Bytes, at: Option, ) -> RpcResult> { + ensure_query_extrinsic_size(&encoded_xt, self.max_encoded_extrinsic_len)?; + let api = self.client.runtime_api(); let at_hash = at.unwrap_or_else(|| self.client.info().best_hash); @@ -174,3 +212,29 @@ where }) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn accepts_encoded_extrinsic_at_size_limit() { + let encoded = vec![0u8; 64]; + assert!(ensure_query_extrinsic_size(&encoded, 64).is_ok()); + assert!(ensure_query_extrinsic_size(&encoded, MAX_ENCODED_EXTRINSIC_LEN).is_ok()); + } + + #[test] + fn rejects_encoded_extrinsic_above_size_limit() { + let encoded = vec![0u8; 65]; + let err = ensure_query_extrinsic_size(&encoded, 64).unwrap_err(); + assert_eq!(err.code(), ErrorCode::InvalidParams.code()); + assert!(err.message().contains("exceeds maximum allowed for fee query")); + } + + #[test] + fn default_cap_matches_chain_max_block_length() { + // Must stay aligned with `RuntimeBlockLength` in runtime/src/configs/mod.rs. + assert_eq!(MAX_ENCODED_EXTRINSIC_LEN, 5 * 1024 * 1024); + } +} diff --git a/primitives/dilithium-crypto/Cargo.toml b/primitives/dilithium-crypto/Cargo.toml index 7578abbd4..2a9b8cff9 100644 --- a/primitives/dilithium-crypto/Cargo.toml +++ b/primitives/dilithium-crypto/Cargo.toml @@ -28,6 +28,7 @@ scale-info = { workspace = true, default-features = false } sp-core = { workspace = true, default-features = false } sp-runtime = { workspace = true, default-features = false } thiserror = { workspace = true, optional = true } +zeroize = { workspace = true, features = ["derive"] } [dev-dependencies] env_logger.workspace = true diff --git a/primitives/dilithium-crypto/src/pair.rs b/primitives/dilithium-crypto/src/pair.rs index 853326869..dfb56c3f6 100644 --- a/primitives/dilithium-crypto/src/pair.rs +++ b/primitives/dilithium-crypto/src/pair.rs @@ -116,16 +116,26 @@ impl Pair for DilithiumPair { phrase: &str, password: Option<&str>, ) -> Result<(Self, Self::Seed), SecretStringError> { - use qp_rusty_crystals_hdwallet::{derive_key_from_mnemonic, mnemonic_to_seed}; + use qp_rusty_crystals_hdwallet::{ + hderive::ExtendedPrivKey, mnemonic_to_seed, SensitiveBytes64, + }; // Default derivation path for Quantus: m/44'/189189'/0'/0'/0' const DEFAULT_PATH: &str = "m/44'/189189'/0'/0'/0'"; - let keypair = derive_key_from_mnemonic(phrase, password, DEFAULT_PATH) + let mut seed_bytes = mnemonic_to_seed(phrase.to_string(), password) .map_err(|_| SecretStringError::InvalidPhrase)?; - let pair = DilithiumPair::from_keypair(keypair); - let seed_bytes = mnemonic_to_seed(phrase.to_string(), password) - .map_err(|_| SecretStringError::InvalidPhrase)?; - let mut seed = [0u8; 32]; - seed.copy_from_slice(&seed_bytes[..32]); + // Wrap the BIP39 seed so that both the stack original (wiped by `from`) and the + // wrapped copy (`ZeroizeOnDrop`) are zeroized once derivation is done. + let seed_bytes = SensitiveBytes64::from(&mut seed_bytes); + // The returned seed must be the actual entropy of the returned pair, so that + // `from_seed(seed)` reconstructs the very same account (users back up the + // displayed seed as recovery material). That entropy is the 32-byte secret + // derived at the default HD path: `derive_key_from_mnemonic` internally runs + // `Keypair::generate` on exactly these bytes, which is also what `from_seed` + // does. + let xpriv = ExtendedPrivKey::derive(seed_bytes.as_bytes(), DEFAULT_PATH) + .map_err(|_| SecretStringError::InvalidPath)?; + let seed = xpriv.secret(); + let pair = DilithiumPair::from_seed(&seed).map_err(|_| SecretStringError::InvalidSeed)?; Ok((pair, seed)) } @@ -134,18 +144,8 @@ impl Pair for DilithiumPair { s: &str, password: Option<&str>, ) -> Result<(Self, Option), SecretStringError> { - use qp_rusty_crystals_hdwallet::derive_key_from_mnemonic; - // Default derivation path for Quantus: m/44'/189189'/0'/0'/0' - const DEFAULT_PATH: &str = "m/44'/189189'/0'/0'/0'"; - // For Dilithium, we use the string directly as entropy for key generation - // We combine the string with the password if provided - let keypair = derive_key_from_mnemonic(s, password, DEFAULT_PATH) - .map_err(|_| SecretStringError::InvalidPhrase)?; - let pair = DilithiumPair::from_keypair(keypair); - - // Return the pair with no seed since Dilithium doesn't use traditional seed-based - // generation - Ok((pair, None)) + let (pair, seed) = Self::from_phrase(s, password)?; + Ok((pair, Some(seed))) } } @@ -314,6 +314,83 @@ mod tests { ); } + const TEST_PHRASE: &str = + "sample split bamboo west visual approve brain fox arch impact relief smile"; + + /// The seed returned by `from_phrase` must control the very same account as the + /// returned pair: users back up the displayed "secret seed" and expect to recover + /// their account from it with `from_seed` if the mnemonic is lost. + #[test] + fn test_from_phrase_returned_seed_reconstructs_same_pair() { + let (pair, seed) = DilithiumPair::from_phrase(TEST_PHRASE, None).expect("valid phrase"); + let restored = DilithiumPair::from_seed(&seed).expect("valid seed"); + assert_eq!( + pair.public_bytes(), + restored.public_bytes(), + "seed returned by from_phrase must reconstruct the same account" + ); + assert_eq!(pair.secret_bytes(), restored.secret_bytes()); + } + + /// Same as above, with a password. + #[test] + fn test_from_phrase_returned_seed_reconstructs_same_pair_with_password() { + let password = Some("hunter2"); + let (pair, seed) = DilithiumPair::from_phrase(TEST_PHRASE, password).expect("valid phrase"); + let restored = DilithiumPair::from_seed(&seed).expect("valid seed"); + assert_eq!( + pair.public_bytes(), + restored.public_bytes(), + "seed returned by from_phrase must reconstruct the same account" + ); + } + + /// `from_string_with_seed` shall expose the same faithful seed for mnemonic inputs. + #[test] + fn test_from_string_with_seed_returns_matching_seed() { + let (pair, seed) = + DilithiumPair::from_string_with_seed(TEST_PHRASE, None).expect("valid phrase"); + let seed = seed.expect("a faithful seed is available for mnemonic inputs"); + let restored = DilithiumPair::from_seed(&seed).expect("valid seed"); + assert_eq!(pair.public_bytes(), restored.public_bytes()); + } + + /// Guard: `from_phrase` must keep deriving the keypair through the default HD path, + /// otherwise existing accounts created from mnemonics would change address. + #[test] + fn test_from_phrase_matches_hd_derivation() { + let keypair = qp_rusty_crystals_hdwallet::derive_key_from_mnemonic( + TEST_PHRASE, + None, + "m/44'/189189'/0'/0'/0'", + ) + .expect("valid phrase"); + let expected = DilithiumPair::from_keypair(keypair); + let (pair, _) = DilithiumPair::from_phrase(TEST_PHRASE, None).expect("valid phrase"); + assert_eq!( + pair.public_bytes(), + expected.public_bytes(), + "from_phrase must not change the account derived from a mnemonic" + ); + } + + /// `zeroize()` must scrub the secret key material. + #[test] + fn test_zeroize_clears_secret() { + use zeroize::Zeroize; + let mut pair = DilithiumPair::from_seed(&[7u8; 32]).expect("valid seed"); + assert!(pair.secret_bytes().iter().any(|b| *b != 0)); + pair.zeroize(); + assert!(pair.secret_bytes().iter().all(|b| *b == 0)); + } + + /// Compile-time guarantee that dropped pairs (including clones) wipe their secret. + #[test] + fn test_pair_zeroizes_on_drop() { + fn assert_zeroize_on_drop() {} + assert_zeroize_on_drop::(); + } + #[test] fn test_from_raw_matching_keys_succeeds() { let seed = [0u8; 32]; diff --git a/primitives/dilithium-crypto/src/types.rs b/primitives/dilithium-crypto/src/types.rs index 354232ea6..c08f9fe36 100644 --- a/primitives/dilithium-crypto/src/types.rs +++ b/primitives/dilithium-crypto/src/types.rs @@ -5,6 +5,7 @@ use sp_core::{ crypto::{PublicBytes, SignatureBytes}, ByteArray, RuntimeDebug, }; +use zeroize::{Zeroize, ZeroizeOnDrop}; /// Resonance Crypto Types /// @@ -31,8 +32,11 @@ pub struct DilithiumCryptoTag; /// Dilithium cryptographic key pair /// -/// Contains both secret and public key material for Dilithium ML-DSA-87 operations -#[derive(Clone, Eq, PartialEq)] +/// Contains both secret and public key material for Dilithium ML-DSA-87 operations. +/// +/// The secret key material is zeroized when an instance (including any clone) is +/// dropped, so released copies do not leave private-key bytes behind in memory. +#[derive(Clone, Eq, PartialEq, Zeroize, ZeroizeOnDrop)] pub struct DilithiumPair { pub(crate) secret: [u8; SECRETKEYBYTES], pub(crate) public: [u8; PUBLICKEYBYTES],