V12 dilithium

client/cli/src/params/transaction_pool_params.rs

@@ -43,11 +43,15 @@ impl Into<sc_transaction_pool::TransactionPoolType> 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<u64>,
 
 	/// 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,
 }
 

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

@@ -63,11 +63,34 @@ pub struct FullChainApi<Client, Block> {
 	validate_transaction_maintained_stats: DurationSlidingStats,
 }
 
+/// Boxed validation job scheduled on a worker lane.
+type ValidationTask = Pin<Box<dyn Future<Output = ()> + 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<Mutex<mpsc::Receiver<ValidationTask>>>,
+	receiver_maintained: Arc<Mutex<mpsc::Receiver<ValidationTask>>>,
+) -> Option<ValidationTask> {
+	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<Mutex<mpsc::Receiver<Pin<Box<dyn Future<Output = ()> + Send>>>>>,
-	receiver_maintained: Arc<Mutex<mpsc::Receiver<Pin<Box<dyn Future<Output = ()> + Send>>>>>,
+	receiver_normal: Arc<Mutex<mpsc::Receiver<ValidationTask>>>,
+	receiver_maintained: Arc<Mutex<mpsc::Receiver<ValidationTask>>>,
 	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"
+			);
+		}
+	}
+}

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 = <Block as sp_runtime::traits::Block>::Hash;
+
+	/// Common ancestor of the two forks, at height 1.
+	fn ancestor() -> HashAndNumber<Block> {
+		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<Block> {
+		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<HashAndNumber<Block>> {
+		(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<Block> {
+		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<Option<NumberFor<Block>>, 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<Block>,
+		expected_best_block: HashAndNumber<Block>,
+		expected_finalized_block: HashAndNumber<Block>,
+	) {
+		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<TreeRoute<Block>, 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<TreeRoute<Block>, 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<TreeRoute<Block>, 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());
+	}
+}