fatxpool: some more integration tests (#8152)

Some extra tests for `fatxpool`, including long term test:
- sending 5M transactions to relay/para,
- transaction gossiping tests (for network protocol evaluation),
- yet-another-parachain spemening test (2s / 7k),

The base directory can be specified by setting `TXPOOL_TEST_DIR` env
variable.

If set every individual test restults will be placed under this path in
a directory name formatted as `test_%Y%m%d_%H%M%S`. e.g.:
```
export TXPOOL_TEST_DIR=/home/miszka/test-results
cargo test  --release --test integration -- --ignored send_future_and_ready_from_many_accounts_to_parachain
...
2025-04-11T07:48:15.324966Z  INFO zombienet_orchestrator: 🧰 base_dir: "/home/miszka/test-results/test_20250411_094815"
...
```

---------

Co-authored-by: cmd[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Iulian Barbu <[email protected]>

Author: 3 people

Cargo.lock

@@ -1436,7 +1436,7 @@ trybuild = { version = "1.0.103" }
 tt-call = { version = "1.0.8" }
 tuplex = { version = "0.1", default-features = false }
 twox-hash = { version = "1.6.3", default-features = false }
-txtesttool = { version = "0.5.0", package = "substrate-txtesttool" }
+txtesttool = { version = "0.6.0", package = "substrate-txtesttool" }
 unsigned-varint = { version = "0.7.2" }
 url = { version = "2.5.4" }
 verifiable = { version = "0.1", default-features = false }
@@ -1465,6 +1465,7 @@ xcm-runtime-apis = { path = "polkadot/xcm/xcm-runtime-apis", default-features =
 xcm-simulator = { path = "polkadot/xcm/xcm-simulator", default-features = false }
 yet-another-parachain-runtime = { path = "cumulus/parachains/runtimes/testing/yet-another-parachain" }
 zeroize = { version = "1.7.0", default-features = false }
+zombienet-configuration = { version = "0.3.6" }
 zombienet-orchestrator = { version = "0.3.6" }
 zombienet-sdk = { version = "0.3.6" }
 zstd = { version = "0.12.4", default-features = false }

Cargo.toml

@@ -0,0 +1,7 @@
+title: '`fatxpool`: some more integration tests'
+doc:
+- audience: Node Dev
+  description: Some new test cases and improvements of zombienet integration tests for `fatxpool`. 
+crates:
+- name: sc-transaction-pool
+  bump: minor

prdoc/pr_8152.prdoc

@@ -48,9 +48,14 @@ tracing = { workspace = true, default-features = true }
 [dev-dependencies]
 anyhow = { workspace = true }
 assert_matches = { workspace = true }
+chrono = { workspace = true }
 criterion = { workspace = true, default-features = true }
+cumulus-zombienet-sdk-helpers = { workspace = true }
+env_logger = { workspace = true }
 rstest = { workspace = true }
 sc-block-builder = { workspace = true, default-features = true }
+serde = { workspace = true }
+serde_json = { workspace = true }
 sp-consensus = { workspace = true, default-features = true }
 substrate-test-runtime = { workspace = true }
 substrate-test-runtime-client = { workspace = true }
@@ -59,4 +64,5 @@ thiserror = { workspace = true }
 tokio = { workspace = true, features = ["rt-multi-thread"] }
 tracing-subscriber = { workspace = true }
 txtesttool = { workspace = true }
+zombienet-configuration = { workspace = true }
 zombienet-sdk = { workspace = true }

substrate/client/transaction-pool/Cargo.toml

@@ -16,26 +16,27 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-// Testsuite of fatp integration tests.
+//! Testsuite of transaction pool integration tests.
+
 pub mod zombienet;
 
+use std::time::Duration;
+
 use crate::zombienet::{
-	default_zn_scenario_builder,
-	relaychain_rococo_local_network_spec::{
-		parachain_asset_hub_network_spec::HIGH_POOL_LIMIT_FATP as PARACHAIN_HIGH_POOL_LIMIT_FATP,
-		HIGH_POOL_LIMIT_FATP as RELAYCHAIN_HIGH_POOL_LIMIT_FATP,
-	},
-	NetworkSpawner,
+	default_zn_scenario_builder, relaychain_rococo_local_network_spec as relay,
+	relaychain_rococo_local_network_spec::parachain_asset_hub_network_spec as para, NetworkSpawner,
 };
-use txtesttool::execution_log::ExecutionLog;
+use futures::future::join_all;
+use tracing::info;
+use txtesttool::{execution_log::ExecutionLog, scenario::ScenarioExecutor};
 use zombienet::DEFAULT_SEND_FUTURE_AND_READY_TXS_TESTS_TIMEOUT_IN_SECS;
 
 // Test which sends future and ready txs from many accounts
 // to an unlimited pool of a parachain collator based on the asset-hub-rococo runtime.
 #[tokio::test(flavor = "multi_thread")]
 #[ignore]
 async fn send_future_and_ready_from_many_accounts_to_parachain() {
-	let net = NetworkSpawner::from_toml_with_env_logger(PARACHAIN_HIGH_POOL_LIMIT_FATP)
+	let net = NetworkSpawner::from_toml_with_env_logger(para::HIGH_POOL_LIMIT_FATP)
 		.await
 		.unwrap();
 
@@ -86,7 +87,7 @@ async fn send_future_and_ready_from_many_accounts_to_parachain() {
 #[tokio::test(flavor = "multi_thread")]
 #[ignore]
 async fn send_future_and_ready_from_many_accounts_to_relaychain() {
-	let net = NetworkSpawner::from_toml_with_env_logger(RELAYCHAIN_HIGH_POOL_LIMIT_FATP)
+	let net = NetworkSpawner::from_toml_with_env_logger(relay::HIGH_POOL_LIMIT_FATP)
 		.await
 		.unwrap();
 
@@ -133,3 +134,241 @@ async fn send_future_and_ready_from_many_accounts_to_relaychain() {
 	assert_eq!(finalized_future, 10_000);
 	assert_eq!(finalized_ready, 10_000);
 }
+
+// Test which sends 5m transactions to parachain. Long execution time expected.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn send_5m_from_many_accounts_to_parachain() {
+	let net = NetworkSpawner::from_toml_with_env_logger(para::HIGH_POOL_LIMIT_FATP)
+		.await
+		.unwrap();
+
+	// Wait for the parachain collator to start block production.
+	net.wait_for_block_production("charlie").await.unwrap();
+
+	// Create txs executor.
+	let ws = net.node_rpc_uri("charlie").unwrap();
+	let executor = default_zn_scenario_builder(&net)
+		.with_rpc_uri(ws)
+		.with_start_id(0)
+		.with_last_id(999)
+		.with_txs_count(5_000)
+		.with_executor_id("txs-executor".to_string())
+		.with_send_threshold(7500)
+		.build()
+		.await;
+
+	// Execute transactions and fetch the execution logs.
+	let execution_logs = executor.execute().await;
+	let finalized_txs = execution_logs.values().filter_map(|tx_log| tx_log.finalized()).count();
+
+	assert_eq!(finalized_txs, 5_000_000);
+}
+
+// Test which sends 5m transactions to relaychain. Long execution time expected.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn send_5m_from_many_accounts_to_relaychain() {
+	let net = NetworkSpawner::from_toml_with_env_logger(relay::HIGH_POOL_LIMIT_FATP)
+		.await
+		.unwrap();
+
+	// Wait for the parachain collator to start block production.
+	net.wait_for_block_production("alice").await.unwrap();
+
+	// Create txs executor.
+	let ws = net.node_rpc_uri("alice").unwrap();
+	let executor = default_zn_scenario_builder(&net)
+		.with_rpc_uri(ws.clone())
+		.with_start_id(0)
+		.with_last_id(999)
+		.with_txs_count(5000)
+		.with_executor_id("txs-executor".to_string())
+		.with_send_threshold(7500)
+		.build()
+		.await;
+
+	// Execute transactions and fetch the execution logs.
+	let execution_logs = executor.execute().await;
+	let finalized_txs = execution_logs.values().filter_map(|tx_log| tx_log.finalized()).count();
+
+	assert_eq!(finalized_txs, 5_000_000);
+}
+
+/// Internal test that allows to observe how transcactions are gossiped in the network. Requires
+/// external tool to track transactions presence at nodes. Was used to evaluate some metrics of
+/// existing transaction protocol.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn gossiping() {
+	let net = NetworkSpawner::from_toml_with_env_logger(relay::HIGH_POOL_LIMIT_FATP_TRACE)
+		.await
+		.unwrap();
+
+	// Wait for the parachain collator to start block production.
+	net.wait_for_block_production("a00").await.unwrap();
+
+	// Create the txs executor.
+	let ws = net.node_rpc_uri("a00").unwrap();
+	let executor = default_zn_scenario_builder(&net)
+		.with_rpc_uri(ws)
+		.with_start_id(0)
+		.with_last_id(999)
+		.with_executor_id("txs-executor".to_string())
+		.build()
+		.await;
+
+	// Execute transactions and fetch the execution logs.
+	let execution_logs = executor.execute().await;
+	let finalized_txs = execution_logs.values().filter_map(|tx_log| tx_log.finalized()).count();
+
+	assert_eq!(finalized_txs, 1000);
+
+	tracing::info!("BASEDIR: {:?}", net.base_dir_path());
+}
+
+/// Creates new transaction scenario executor and sends given batch of ready transactions to the
+/// specified node. Single transaction is sent from single account.
+async fn send_batch(
+	net: &NetworkSpawner,
+	node_name: &str,
+	from: u32,
+	to: u32,
+	prio: u32,
+) -> ScenarioExecutor {
+	let ws = net.node_rpc_uri(node_name).unwrap();
+	info!(from, to, prio, "send_batch");
+	default_zn_scenario_builder(net)
+		.with_rpc_uri(ws)
+		.with_start_id(from)
+		.with_last_id(to)
+		.with_txs_count(1)
+		.with_tip(prio.into())
+		.with_executor_id(format!("txs-executor_{}_{}_{}", from, to, prio))
+		.with_send_threshold(usize::MAX)
+		.with_legacy_backend(true)
+		.build()
+		.await
+}
+
+/// Repeatedly sends batches of transactions to the specified node with priority provided by
+/// closure.
+///
+/// This function loops indefinitely, adjusting the priority of the transaction batch each time
+/// based on the provided function. Each batch is executed by an executor that times out after
+/// period duration if not completed.
+///
+/// The progress of transactions is intentionally not monitored; the utility is intended for
+/// transaction pool limits testing, where the accuracy of execution is challenging to monitor.
+async fn batch_loop<F>(
+	net: &NetworkSpawner,
+	node_name: &str,
+	from: u32,
+	to: u32,
+	priority: F,
+	period: std::time::Duration,
+) where
+	F: Fn(u32) -> u32,
+{
+	let mut prio = 0;
+	loop {
+		prio = priority(prio);
+		let executor = send_batch(&net, node_name, from, to, prio).await;
+		let start = std::time::Instant::now();
+		let _results = tokio::time::timeout(period, executor.execute()).await;
+		let elapsed = start.elapsed();
+		if elapsed < period {
+			tokio::time::sleep(period - elapsed).await;
+		}
+	}
+}
+
+/// Tests the transaction pool limits by continuously sending transaction batches to a parachain
+/// network node. This test checks the pool's behavior under high load by simulating multiple
+/// senders with increasing priorities.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn test_limits_increasing_prio_parachain() {
+	let net = NetworkSpawner::from_toml_with_env_logger(para::LOW_POOL_LIMIT_FATP)
+		.await
+		.unwrap();
+
+	net.wait_for_block_production("charlie").await.unwrap();
+
+	let mut executors = vec![];
+	let senders_count = 25;
+	let sender_batch = 2000;
+
+	for i in 0..senders_count {
+		let from = 0 + i * sender_batch;
+		let to = from + sender_batch - 1;
+		executors.push(batch_loop(
+			&net,
+			"charlie",
+			from,
+			to,
+			|prio| prio + 1,
+			Duration::from_secs(60),
+		));
+	}
+
+	let _results = join_all(executors).await;
+}
+
+/// Tests the transaction pool limits by continuously sending transaction batches to a relaychain
+/// network node. This test checks the pool's behavior under high load by simulating multiple
+/// senders with increasing priorities.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn test_limits_increasing_prio_relaychain() {
+	let net = NetworkSpawner::from_toml_with_env_logger(relay::LOW_POOL_LIMIT_FATP)
+		.await
+		.unwrap();
+
+	net.wait_for_block_production("alice").await.unwrap();
+
+	let mut executors = vec![];
+	//this looks like current limit of what we can handle. A bit choky but almost no empty blocks.
+	let senders_count = 50;
+	let sender_batch = 2000;
+
+	for i in 0..senders_count {
+		let from = 0 + i * sender_batch;
+		let to = from + sender_batch - 1;
+		executors.push(batch_loop(
+			&net,
+			"alice",
+			from,
+			to,
+			|prio| prio + 1,
+			Duration::from_secs(15),
+		));
+	}
+
+	let _results = join_all(executors).await;
+}
+
+/// Tests the transaction pool limits by continuously sending transaction batches to a relaychain
+/// network node. This test checks the pool's behavior under high load by simulating multiple
+/// senders with increasing priorities.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn test_limits_same_prio_relaychain() {
+	let net = NetworkSpawner::from_toml_with_env_logger(relay::LOW_POOL_LIMIT_FATP)
+		.await
+		.unwrap();
+
+	net.wait_for_block_production("alice").await.unwrap();
+
+	let mut executors = vec![];
+	let senders_count = 50;
+	let sender_batch = 2000;
+
+	for i in 0..senders_count {
+		let from = 0 + i * sender_batch;
+		let to = from + sender_batch - 1;
+		executors.push(batch_loop(&net, "alice", from, to, |prio| prio, Duration::from_secs(15)));
+	}
+
+	let _results = join_all(executors).await;
+}