fatxpool: added mortal transactions integration test (#8887)

# Description

Based on michalkucharczyk/tx-test-tool#43.

## Integration

N/A

## Review Notes

- added tests with future mortal txs that are dropped due to not being
included in blocks within their lifetime, but also future mortal txs
that have a sufficient lifetime to be included in blocks
- added test which uses priorities to delay mortal txs inclusion and
make them invalid

---------

Signed-off-by: Iulian Barbu <iulian.barbu@parity.io>
Co-authored-by: cmd[bot] <41898282+github-actions[bot]@users.noreply.github.com>
Co-authored-by: Sebastian Kunert <mail@skunert.dev>

Author: 3 people

Cargo.lock

@@ -1463,7 +1463,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.6.0", package = "substrate-txtesttool" }
+txtesttool = { version = "0.7.0", package = "substrate-txtesttool" }
 unsigned-varint = { version = "0.7.2" }
 url = { version = "2.5.4" }
 verifiable = { version = "0.1", default-features = false }

Cargo.toml

@@ -24,7 +24,8 @@ use std::time::Duration;
 
 use crate::zombienet::{
 	default_zn_scenario_builder, relaychain_rococo_local_network_spec as relay,
-	relaychain_rococo_local_network_spec::parachain_asset_hub_network_spec as para, NetworkSpawner,
+	relaychain_rococo_local_network_spec::parachain_asset_hub_network_spec as para,
+	BlockSubscriptionType, NetworkSpawner,
 };
 use futures::future::join_all;
 use tracing::info;
@@ -41,7 +42,7 @@ async fn send_future_and_ready_from_many_accounts_to_parachain() {
 		.unwrap();
 
 	// Wait for the parachain collator to start block production.
-	net.wait_for_block_production("charlie").await.unwrap();
+	net.wait_for_block("charlie", BlockSubscriptionType::Best).await.unwrap();
 
 	// Create future & ready txs executors.
 	let ws = net.node_rpc_uri("charlie").unwrap();
@@ -93,7 +94,7 @@ async fn send_future_and_ready_from_many_accounts_to_relaychain() {
 
 	// Wait for the paracha validator to start block production & have its genesis block
 	// finalized.
-	net.wait_for_block_production("alice").await.unwrap();
+	net.wait_for_block("alice", BlockSubscriptionType::Best).await.unwrap();
 
 	// Create future & ready txs executors.
 	let ws = net.node_rpc_uri("alice").unwrap();
@@ -135,6 +136,171 @@ async fn send_future_and_ready_from_many_accounts_to_relaychain() {
 	assert_eq!(finalized_ready, 10_000);
 }
 
+// Send immortal and mortal txs. Some of the mortal txs are configured to get dropped
+// while others to succeed. Mortal txs are future so not being able to become ready in time and
+// included in blocks result in their dropping.
+//
+// Block length for rococo for user txs is 75% of maximum 5MB (per frame-system setup),
+// so we get 3750KB. In the test scenario we aim for 5 txs per block roughly (not precesily)
+// so to fill a block each user tx must have around 750kb.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn send_future_mortal_txs() {
+	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("alice", BlockSubscriptionType::Finalized).await.unwrap();
+
+	// Create txs executors.
+	let ws = net.node_rpc_uri("alice").unwrap();
+	let ready_scenario_executor = default_zn_scenario_builder(&net)
+		.with_rpc_uri(ws.clone())
+		.with_start_id(0)
+		.with_nonce_from(Some(0))
+		.with_txs_count(50)
+		// Block length for rococo for user txs is 75% of maximum 5MB (per frame-system setup),
+		// so we get 3750KB. In the test scenario we aim for 5 txs per block roughly (not precesily)
+		// so to fill a block each user tx must have around 750kb. We aim for 5 txs per block
+		// because we send 50 ready txs which we want to distribute over 10 blocks, so mortal txs
+		// with lifetime lower than 10 should be declared invalid after the ready txs finalize,
+		// while mortal txs with bigger lifetime should be finalized.
+		.with_remark_recipe(750)
+		.with_executor_id("ready-txs-executor".to_string())
+		.build()
+		.await;
+
+	let mortal_scenario_invalid = default_zn_scenario_builder(&net)
+		.with_rpc_uri(ws.clone())
+		.with_start_id(0)
+		.with_nonce_from(Some(60))
+		.with_txs_count(10)
+		.with_executor_id("mortal-tx-executor-invalid".to_string())
+		.with_mortality(5)
+		.build()
+		.await;
+
+	let mortal_scenario_success = default_zn_scenario_builder(&net)
+		.with_rpc_uri(ws)
+		.with_start_id(0)
+		.with_nonce_from(Some(50))
+		.with_txs_count(10)
+		.with_executor_id("mortal-tx-executor-success".to_string())
+		.with_mortality(25)
+		.build()
+		.await;
+
+	// Execute transactions and fetch the execution logs.
+	let (mortal_invalid_logs, ready_logs, mortal_succes_logs) = tokio::join!(
+		mortal_scenario_invalid.execute(),
+		ready_scenario_executor.execute(),
+		mortal_scenario_success.execute(),
+	);
+
+	let mortal_invalid = mortal_invalid_logs
+		.values()
+		.filter(|default_log| default_log.get_invalid_reason().len() > 0)
+		.count();
+	let mortal_succesfull = mortal_succes_logs
+		.values()
+		.filter_map(|default_log| default_log.finalized())
+		.count();
+	let finalized_ready =
+		ready_logs.values().filter_map(|default_log| default_log.finalized()).count();
+
+	assert_eq!(mortal_invalid, 10);
+	assert_eq!(mortal_succesfull, 10);
+	assert_eq!(finalized_ready, 50);
+}
+
+// Send immortal and mortal txs. Some mortal txs have lower priority so they shouldn't get into
+// blocks during their lifetime, and will be considered invalid, while other mortal txs have
+// sufficient lifetime to be included in blocks, and are finalized successfully.
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+async fn send_lower_priority_mortal_txs() {
+	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("alice", BlockSubscriptionType::Finalized).await.unwrap();
+
+	// Create txs executors.
+	let ws = net.node_rpc_uri("alice").unwrap();
+	let ready_scenario_executor = default_zn_scenario_builder(&net)
+		.with_rpc_uri(ws.clone())
+		.with_start_id(0)
+		.with_nonce_from(Some(0))
+		.with_txs_count(50)
+		.with_executor_id("ready-txs-executor".to_string())
+		// Block length for rococo for user txs is 75% of maximum 5MB (per frame-system setup),
+		// so we get 3750KB. In the test scenario we aim for 5 txs per block roughly (not precesily)
+		// so to fill a block each user tx must have around 750kb. We aim for 5 txs per block
+		// because we send 50 ready txs which we want to distribute over 10 blocks, so mortal txs
+		// with lifetime lower than 10 should be declared invalid after the ready txs finalize,
+		// while mortal txs with bigger lifetime should be finalized.
+		.with_remark_recipe(750)
+		.with_tip(150)
+		.build()
+		.await;
+
+	let mortal_scenario_invalid = default_zn_scenario_builder(&net)
+		.with_rpc_uri(ws.clone())
+		.with_start_id(1)
+		.with_nonce_from(Some(0))
+		.with_txs_count(10)
+		.with_executor_id("mortal-tx-executor-invalid".to_string())
+		.with_mortality(5)
+		// Make it very hard for these mortal txs to be included in blocks, by making them big
+		// enough to not let other txs be part of the same block as them, but also make sure they
+		// have the lowest priority so that they are not included in a single tx block over other
+		// txs. At some point they'll be starved and their lifetime will pass.
+		.with_remark_recipe(3500)
+		.with_tip(50)
+		.build()
+		.await;
+
+	let mortal_scenario_success = default_zn_scenario_builder(&net)
+		.with_rpc_uri(ws)
+		.with_start_id(2)
+		.with_nonce_from(Some(0))
+		.with_txs_count(10)
+		.with_executor_id("mortal-tx-executor-success".to_string())
+		.with_mortality(20)
+		// Same reasoning as for the ready immortal txs, we want these txs to be similarly big as
+		// the immortal txs, so at all times if it comes to pick a ready txs to include it in a
+		// block, an immortal tx should be picked instead (leaving these mortal txs to be picked
+		// only after, which is fine for these mortal txs).
+		.with_remark_recipe(750)
+		.with_tip(100)
+		.build()
+		.await;
+
+	// Execute transactions and fetch the execution logs.
+	let (mortal_invalid_logs, ready_logs, mortal_success_logs) = tokio::join!(
+		mortal_scenario_invalid.execute(),
+		ready_scenario_executor.execute(),
+		mortal_scenario_success.execute(),
+	);
+
+	let mortal_invalid = mortal_invalid_logs
+		.values()
+		.filter(|default_log| default_log.get_invalid_reason().len() > 0)
+		.count();
+	let mortal_succesfull = mortal_success_logs
+		.values()
+		.filter_map(|default_log| default_log.finalized())
+		.count();
+	let finalized_ready =
+		ready_logs.values().filter_map(|default_log| default_log.finalized()).count();
+
+	assert_eq!(mortal_invalid, 10);
+	assert_eq!(mortal_succesfull, 10);
+	assert_eq!(finalized_ready, 50);
+}
+
 // Test which sends 5m transactions to parachain. Long execution time expected.
 #[tokio::test(flavor = "multi_thread")]
 #[ignore]
@@ -144,7 +310,7 @@ async fn send_5m_from_many_accounts_to_parachain() {
 		.unwrap();
 
 	// Wait for the parachain collator to start block production.
-	net.wait_for_block_production("charlie").await.unwrap();
+	net.wait_for_block("charlie", BlockSubscriptionType::Best).await.unwrap();
 
 	// Create txs executor.
 	let ws = net.node_rpc_uri("charlie").unwrap();
@@ -174,7 +340,7 @@ async fn send_5m_from_many_accounts_to_relaychain() {
 		.unwrap();
 
 	// Wait for the parachain collator to start block production.
-	net.wait_for_block_production("alice").await.unwrap();
+	net.wait_for_block("alice", BlockSubscriptionType::Best).await.unwrap();
 
 	// Create txs executor.
 	let ws = net.node_rpc_uri("alice").unwrap();
@@ -206,7 +372,7 @@ async fn gossiping() {
 		.unwrap();
 
 	// Wait for the parachain collator to start block production.
-	net.wait_for_block_production("a00").await.unwrap();
+	net.wait_for_block("a00", BlockSubscriptionType::Best).await.unwrap();
 
 	// Create the txs executor.
 	let ws = net.node_rpc_uri("a00").unwrap();
@@ -293,7 +459,7 @@ async fn test_limits_increasing_prio_parachain() {
 		.await
 		.unwrap();
 
-	net.wait_for_block_production("charlie").await.unwrap();
+	net.wait_for_block("charlie", BlockSubscriptionType::Best).await.unwrap();
 
 	let mut executors = vec![];
 	let senders_count = 25;
@@ -325,7 +491,7 @@ async fn test_limits_increasing_prio_relaychain() {
 		.await
 		.unwrap();
 
-	net.wait_for_block_production("alice").await.unwrap();
+	net.wait_for_block("alice", BlockSubscriptionType::Best).await.unwrap();
 
 	let mut executors = vec![];
 	//this looks like current limit of what we can handle. A bit choky but almost no empty blocks.
@@ -358,7 +524,7 @@ async fn test_limits_same_prio_relaychain() {
 		.await
 		.unwrap();
 
-	net.wait_for_block_production("alice").await.unwrap();
+	net.wait_for_block("alice", BlockSubscriptionType::Best).await.unwrap();
 
 	let mut executors = vec![];
 	let senders_count = 50;

substrate/client/transaction-pool/tests/integration.rs

@@ -71,6 +71,12 @@ pub type Result<T> = std::result::Result<T, Error>;
 /// Environment variable defining the location of zombienet network base dir.
 const TXPOOL_TEST_DIR_ENV: &str = "TXPOOL_TEST_DIR";
 
+/// Type for block subscription modes.
+pub enum BlockSubscriptionType {
+	Finalized,
+	Best,
+}
+
 /// Provides logic to spawn a network based on a Zombienet toml file.
 pub struct NetworkSpawner {
 	network: Network<LocalFileSystem>,
@@ -152,7 +158,14 @@ impl NetworkSpawner {
 	}
 
 	/// Waits for blocks production/import to kick-off on given node.
-	pub async fn wait_for_block_production(&self, node_name: &str) -> Result<()> {
+	///
+	/// It subscribes to best/finalized blocks on the given node to determine whether
+	/// the blocks were considered as best/finalized.
+	pub async fn wait_for_block(
+		&self,
+		node_name: &str,
+		subscription_type: BlockSubscriptionType,
+	) -> Result<()> {
 		let node = self
 			.network
 			.get_node(node_name)
@@ -161,23 +174,31 @@ impl NetworkSpawner {
 			.wait_client::<SubstrateConfig>()
 			.await
 			.map_err(|_| Error::FailedToGetOnlineClinet)?;
-		let mut stream = client
-			.blocks()
-			.subscribe_best()
-			.await
-			.map_err(|_| Error::FailedToGetBlocksStream)?;
+		let mut stream = match subscription_type {
+			BlockSubscriptionType::Best => client
+				.blocks()
+				.subscribe_finalized()
+				.await
+				.map_err(|_| Error::FailedToGetBlocksStream)?,
+			BlockSubscriptionType::Finalized => client
+				.blocks()
+				.subscribe_best()
+				.await
+				.map_err(|_| Error::FailedToGetBlocksStream)?,
+		};
+
 		// It should take at most two iterations to return with the best block, if any.
 		for _ in 0..=1 {
 			let Some(block) = stream.next().await else {
 				continue;
 			};
 
 			if let Some(block) = block.ok().filter(|block| block.number() == 1) {
-				tracing::info!("[{node_name}] found first best block: {:#?}", block.hash());
+				tracing::info!("[{node_name}] found first block: {:#?}", block.hash());
 				break;
 			}
 
-			tracing::info!("[{node_name}] waiting for first best block");
+			tracing::info!("[{node_name}] waiting for first block");
 		}
 		Ok(())
 	}

substrate/client/transaction-pool/tests/zombienet/mod.rs

@@ -10,7 +10,7 @@ default_args = [
 	"--pool-limit 500000",
 	"--pool-type=fork-aware",
 	"--rpc-max-connections 15000",
-	"--rpc-max-response-size 150",
+	"--rpc-max-response-size 1500",
 	"--rpc-max-subscriptions-per-connection=128000",
 	"--state-pruning=1024",
 	"-lsync=info",

substrate/client/transaction-pool/tests/zombienet/network-specs/rococo-local-high-pool-limit-fatp.toml

@@ -21,7 +21,7 @@
 // and patched as in:
 // https://github.com/paritytech/polkadot-sdk/pull/7220#issuecomment-2808830472
 
-use crate::zombienet::{NetworkSpawner, ScenarioBuilderSharedParams};
+use crate::zombienet::{BlockSubscriptionType, NetworkSpawner, ScenarioBuilderSharedParams};
 use cumulus_zombienet_sdk_helpers::create_assign_core_call;
 use serde_json::json;
 use txtesttool::{execution_log::ExecutionLog, scenario::ScenarioBuilder};
@@ -111,7 +111,7 @@ async fn slot_based_3cores_test() -> Result<(), anyhow::Error> {
 	tracing::info!("2 more cores assigned to the parachain");
 
 	// Wait for the parachain collator to start block production.
-	spawner.wait_for_block_production("dave").await.unwrap();
+	spawner.wait_for_block("dave", BlockSubscriptionType::Best).await.unwrap();
 
 	// Create txs executor.
 	let ws = spawner.node_rpc_uri("dave").unwrap();