some improvements for testing pool limits

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "substrate-txtesttool"
-version = "0.5.0"
+version = "0.6.0"
 edition = "2021"
 description = "A library and CLI tool for sending transactions to substrate-based chains, enabling developers to test and monitor transaction scenarios."
 license = "Apache-2.0 OR GPL-3.0"

bin/main.rs

@@ -59,6 +59,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 			send_threshold,
 			remark,
 			tip,
+			use_legacy_backend,
 		} => match chain {
 			ChainType::Fake => {
 				unimplemented!()
@@ -71,6 +72,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 					.with_block_monitoring(*block_monitor)
 					.with_watched_txs(!unwatched)
 					.with_installed_ctrlc_stop_hook(true)
+					.with_legacy_backend(*use_legacy_backend)
 					.with_tip(*tip);
 
 				scenario_builder = populate_scenario_builder!(scenario_builder, scenario);
@@ -90,6 +92,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 					.with_block_monitoring(*block_monitor)
 					.with_watched_txs(!unwatched)
 					.with_installed_ctrlc_stop_hook(true)
+					.with_legacy_backend(*use_legacy_backend)
 					.with_tip(*tip);
 
 				scenario_builder = populate_scenario_builder!(scenario_builder, scenario);
@@ -118,6 +121,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 						desc,
 						generate_ecdsa_keypair,
 						None,
+						false,
 					)
 					.await;
 					let account =
@@ -136,6 +140,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 						desc,
 						generate_sr25519_keypair,
 						None,
+						false,
 					)
 					.await;
 					let account =
@@ -153,10 +158,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 				runtime_apis.call_raw("Metadata_metadata", None).await?;
 			println!("{meta:#?}");
 		},
-		CliCommand::BlockMonitor { chain, ws } => {
+		CliCommand::BlockMonitor { chain, ws, display } => {
 			match chain {
 				ChainType::Sub => {
-					let block_monitor = BlockMonitor::<PolkadotConfig>::new(ws).await;
+					let block_monitor =
+						BlockMonitor::<PolkadotConfig>::new_with_options(ws, *display).await;
 					async {
 						loop {
 							tokio::time::sleep(Duration::from_secs(10)).await
@@ -165,7 +171,8 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
 					.await;
 				},
 				ChainType::Eth => {
-					let block_monitor = BlockMonitor::<EthRuntimeConfig>::new(ws).await;
+					let block_monitor =
+						BlockMonitor::<EthRuntimeConfig>::new_with_options(ws, *display).await;
 					async {
 						loop {
 							tokio::time::sleep(Duration::from_secs(10)).await

src/block_monitor.rs

@@ -6,6 +6,7 @@ use std::{collections::HashMap, marker::PhantomData, pin::Pin};
 
 use crate::transaction::TransactionMonitor;
 use async_trait::async_trait;
+use clap::ValueEnum;
 use futures::Future;
 use subxt::{blocks::Block, OnlineClient};
 use subxt_core::config::Header;
@@ -23,6 +24,22 @@ type TxFoundListener<H> = oneshot::Receiver<H>;
 type TxFoundListenerTrigger<H> = oneshot::Sender<H>;
 type HashOf<C> = <C as subxt::Config>::Hash;
 
+#[derive(ValueEnum, Copy, Clone, Debug)]
+pub enum BlockMonitorDisplayOptions {
+	Best,
+	Finalized,
+	All,
+}
+
+impl BlockMonitorDisplayOptions {
+	fn display_best(&self) -> bool {
+		matches!(self, Self::Best | Self::All)
+	}
+	fn display_finalized(&self) -> bool {
+		matches!(self, Self::Finalized | Self::All)
+	}
+}
+
 #[derive(Clone)]
 pub struct BlockMonitor<C: subxt::Config> {
 	listener_request_tx: mpsc::Sender<(HashOf<C>, TxFoundListenerTrigger<HashOf<C>>)>,
@@ -45,7 +62,17 @@ impl<C: subxt::Config> BlockMonitor<C> {
 			.await
 			.expect("should connect to rpc client");
 		let (listener_request_tx, rx) = mpsc::channel(100);
-		tokio::spawn(async { Self::run(api, rx).await });
+		tokio::spawn(async { Self::run(api, rx, BlockMonitorDisplayOptions::All).await });
+		Self { listener_request_tx, _p: Default::default() }
+	}
+
+	pub async fn new_with_options(uri: &str, options: BlockMonitorDisplayOptions) -> Self {
+		trace!(uri, "BlockNumber::new");
+		let api = OnlineClient::<C>::from_insecure_url(uri)
+			.await
+			.expect("should connect to rpc client");
+		let (listener_request_tx, rx) = mpsc::channel(100);
+		tokio::spawn(async move { Self::run(api, rx, options).await });
 		Self { listener_request_tx, _p: Default::default() }
 	}
 
@@ -62,6 +89,7 @@ impl<C: subxt::Config> BlockMonitor<C> {
 	async fn handle_block(
 		callbacks: &mut HashMap<HashOf<C>, TxFoundListenerTrigger<HashOf<C>>>,
 		block: Block<C, OnlineClient<C>>,
+		options: BlockMonitorDisplayOptions,
 		finalized: bool,
 	) -> Result<(), Box<dyn std::error::Error>> {
 		let block_number: u64 = block.header().number().into();
@@ -77,8 +105,10 @@ impl<C: subxt::Config> BlockMonitor<C> {
 					trigger.send(block_hash).unwrap();
 				}
 			}
-			info!(block_number, extrinsics_count, "FINALIZED block");
-		} else {
+			if options.display_finalized() {
+				info!(block_number, extrinsics_count, "FINALIZED block");
+			}
+		} else if options.display_best() {
 			info!(block_number, extrinsics_count, "     BEST block");
 		}
 		Ok(())
@@ -87,6 +117,7 @@ impl<C: subxt::Config> BlockMonitor<C> {
 	async fn block_monitor_inner(
 		api: OnlineClient<C>,
 		mut listener_request_rx: mpsc::Receiver<(HashOf<C>, TxFoundListenerTrigger<HashOf<C>>)>,
+		options: BlockMonitorDisplayOptions,
 	) -> Result<(), Box<dyn std::error::Error>> {
 		let mut finalized_blocks_sub = api.blocks().subscribe_finalized().await?;
 		let mut best_blocks_sub = api.blocks().subscribe_best().await?;
@@ -95,11 +126,11 @@ impl<C: subxt::Config> BlockMonitor<C> {
 		loop {
 			select! {
 				Some(Ok(block)) = finalized_blocks_sub.next() => {
-					Self::handle_block(&mut callbacks, block, true).await?;
+					Self::handle_block(&mut callbacks, block, options, true).await?;
 				}
 
 				Some(Ok(block)) = best_blocks_sub.next() => {
-					Self::handle_block(&mut callbacks, block, false).await?;
+					Self::handle_block(&mut callbacks, block, options, false).await?;
 				}
 
 				Some((hash, tx)) = listener_request_rx.recv() => {
@@ -113,7 +144,8 @@ impl<C: subxt::Config> BlockMonitor<C> {
 	async fn run(
 		api: OnlineClient<C>,
 		listener_requrest_rx: mpsc::Receiver<(HashOf<C>, TxFoundListenerTrigger<HashOf<C>>)>,
+		options: BlockMonitorDisplayOptions,
 	) {
-		let _ = Self::block_monitor_inner(api, listener_requrest_rx).await;
+		let _ = Self::block_monitor_inner(api, listener_requrest_rx, options).await;
 	}
 }

src/cli.rs

@@ -2,10 +2,12 @@
 // This file is dual-licensed as Apache-2.0 or GPL-3.0.
 // see LICENSE for license details.
 
+use crate::{
+	block_monitor::BlockMonitorDisplayOptions,
+	scenario::{ChainType, ScenarioType},
+};
 use clap::{Parser, Subcommand};
 
-use crate::scenario::{ChainType, ScenarioType};
-
 #[derive(Parser)]
 #[clap(name = "txtt")]
 pub struct Cli {
@@ -48,6 +50,9 @@ pub enum CliCommand {
 		/// Accounts range used for building/seding transactions.
 		#[clap(subcommand)]
 		scenario: ScenarioType,
+		/// Use legacy backend
+		#[clap(long, default_value_t = false)]
+		use_legacy_backend: bool,
 	},
 	/// Check nonce for given account.
 	CheckNonce {
@@ -76,6 +81,8 @@ pub enum CliCommand {
 		/// The RPC endpoint of the node to be used.
 		#[clap(long, default_value = "ws://127.0.0.1:9933")]
 		ws: String,
+		#[clap(long, default_value = "all")]
+		display: BlockMonitorDisplayOptions,
 	},
 	/// Load and inspect existing log file.
 	LoadLog {

src/fake_transaction_sink.rs

@@ -59,7 +59,7 @@ impl TransactionsSink<FakeHash> for FakeTransactionsSink {
 	}
 
 	///Current count of transactions being processed by sink
-	async fn count(&self) -> usize {
+	async fn pending_extrinsics(&self) -> usize {
 		self.txs.read().len()
 	}
 
@@ -85,9 +85,9 @@ mod test {
 		let t: Box<dyn Transaction<HashType = FakeHash>> = Box::from(t);
 
 		let events = rpc.submit_and_watch(&*t).await.unwrap();
-		assert_eq!(rpc.count().await, 1);
+		assert_eq!(rpc.pending_extrinsics().await, 1);
 		let v = events.collect::<Vec<_>>().await;
-		assert_eq!(rpc.count().await, 0);
+		assert_eq!(rpc.pending_extrinsics().await, 0);
 		assert_eq!(
 			v,
 			vec![
@@ -149,14 +149,14 @@ mod test {
 
 		let f = || async {
 			tokio::time::sleep(Duration::from_millis(200)).await;
-			rpc.count().await
+			rpc.pending_extrinsics().await
 		};
 
 		let result = join3(rpc.submit(&*t1), rpc.submit(&*t2), f()).await;
 		let r1 = result.0;
 		let r2 = result.1;
 		assert_eq!(result.2, 2);
-		assert_eq!(rpc.count().await, 0);
+		assert_eq!(rpc.pending_extrinsics().await, 0);
 		assert_eq!(r1.unwrap(), 111u32.to_le_bytes().into());
 		assert_eq!(r2.unwrap(), 222u32.to_le_bytes().into());
 	}

src/helpers.rs

@@ -6,7 +6,7 @@ pub use jsonrpsee::{
 	client_transport::ws::{self, EitherStream, Url, WsTransportClientBuilder},
 	core::client::{Client, Error},
 };
-use std::pin::Pin;
+use std::{pin::Pin, time::Duration};
 use tokio_util::compat::Compat;
 
 /// Helper type for a futures stream.
@@ -22,7 +22,7 @@ pub(crate) async fn client(url: &str) -> Result<Client, Error> {
 	let (sender, receiver) = ws_transport(url).await?;
 	Ok(Client::builder()
 		.max_buffer_capacity_per_subscription(4096)
-		.max_concurrent_requests(128000)
+		.max_concurrent_requests(1280000)
 		.build_with_tokio(sender, receiver))
 }
 
@@ -31,6 +31,7 @@ async fn ws_transport(url: &str) -> Result<(Sender, Receiver), Error> {
 	WsTransportClientBuilder::default()
 		.max_request_size(400 * 1024 * 1024)
 		.max_response_size(400 * 1024 * 1024)
+		.connection_timeout(Duration::from_secs(600))
 		.build(url)
 		.await
 		.map_err(|e| Error::Transport(e.into()))

src/runner.rs

@@ -22,7 +22,7 @@ use tokio::{
 	select,
 	sync::mpsc::{channel, Receiver, Sender},
 };
-use tracing::{debug, info, instrument, trace, Span};
+use tracing::{debug, info, instrument, trace, warn, Span};
 
 const LOG_TARGET: &str = "runner";
 
@@ -122,9 +122,9 @@ impl<H: BlockHash, T: Transaction<HashType = H> + Send> TxTask for DefaultTxTask
 						continue;
 					}
 				}
-				//shall not happen to be here
-				panic!();
-				// ExecutionResult::Error(self.tx().hash())
+				//shall not happen to be here, return error.
+				warn!(target:LOG_TARGET,tx=?self,"stream error");
+				ExecutionResult::Error(self.tx().hash())
 			},
 			Err(e) => {
 				info!(nonce=?self.tx().nonce(),"submit_and_watch: error: {e:?}");
@@ -184,7 +184,7 @@ impl<T: Transaction> DefaultTxTask<T> {
 
 /// Holds the logic that handles multiple transactions execution on a specific chain.
 pub struct Runner<T: TxTask, Sink: TransactionsSink<TxTaskHash<T>>> {
-	initial_tasks: usize,
+	send_threshold: usize,
 	logs: Logs<T>,
 	transactions: Vec<T>,
 	done: Vec<TxTaskHash<T>>,
@@ -206,7 +206,7 @@ where
 {
 	/// Instantiates a new transactions [`Runner`].
 	pub fn new(
-		initial_tasks: usize,
+		send_threshold: usize,
 		rpc: Sink,
 		transactions: Vec<T>,
 		log_file_name: Option<String>,
@@ -229,7 +229,7 @@ where
 		(
 			tx,
 			Self {
-				initial_tasks,
+				send_threshold,
 				logs,
 				transactions,
 				rpc: rpc.into(),
@@ -257,11 +257,17 @@ where
 		&mut self,
 		workers: &mut FuturesUnordered<Pin<Box<dyn Future<Output = ExecutionResult<T>> + Send>>>,
 	) {
-		let current_count = self.rpc.count().await;
-		let to_consume = self
-			.initial_tasks
-			.saturating_sub(current_count)
-			.saturating_sub(self.event_counters.buffered());
+		let (to_consume, current_count) = if self.send_threshold == usize::MAX {
+			(usize::MAX, None)
+		} else {
+			let current_count = self.rpc.pending_extrinsics().await;
+			(
+				self.send_threshold
+					.saturating_sub(current_count)
+					.saturating_sub(self.event_counters.buffered()),
+				Some(current_count),
+			)
+		};
 		let mut pushed = 0;
 		let counters_displayed = format!("{}", self.event_counters);
 		let mut nonces = vec![];
@@ -286,9 +292,17 @@ where
 		} else {
 			true
 		};
+
 		if display {
 			self.last_displayed = Some(Instant::now());
-			info!(current_count, pushed, min_nonce_sent, "consume pending {}", counters_displayed,);
+			if let Some(current_count) = current_count {
+				info!(
+					current_count,
+					pushed, min_nonce_sent, "consume pending {}", counters_displayed,
+				);
+			} else {
+				info!(pushed, min_nonce_sent, "consume pending {}", counters_displayed,);
+			}
 		};
 	}
 
@@ -325,7 +339,7 @@ where
 		let original_transactions_count = self.transactions.len();
 		let mut workers = FuturesUnordered::new();
 
-		for _ in 0..self.initial_tasks {
+		for _ in 0..self.send_threshold {
 			if let Some(t) = self.transactions.pop() {
 				let t = Box::new(t);
 				let log = self.logs[&t.tx().hash()].clone();
@@ -482,7 +496,7 @@ mod tests {
 		// let api = OnlineClient::<EthRuntimeConfig>::from_insecure_url("ws://127.0.0.1:9933")
 		//     .await
 		//     .unwrap();
-		let api = subxt_api_connector::connect("ws://127.0.0.1:9933").await.unwrap();
+		let api = subxt_api_connector::connect("ws://127.0.0.1:9933", false).await.unwrap();
 
 		let rpc = EthTransactionsSink::new().await;