feat(infra): concurrent materializer tests

fendermint/testing/materializer/src/concurrency.rs

@@ -0,0 +1,13 @@
+pub struct Config {
+    pub parallelism_level: usize
+}
+
+impl Config {
+    pub fn with_parallelism_level(v: usize) -> Self {
+        Self { parallelism_level: v }
+    }
+}
+
+
+
+

fendermint/testing/materializer/src/lib.rs

@@ -20,6 +20,7 @@ pub mod validation;
 
 #[cfg(feature = "arb")]
 mod arb;
+pub mod concurrency;
 
 /// An ID identifying a resource within its parent.
 #[derive(Clone, Serialize, PartialEq, Eq, PartialOrd, Ord)]

fendermint/testing/materializer/src/testnet.rs

@@ -140,6 +140,15 @@ where
             .ok_or_else(|| anyhow!("account {id} does not exist"))
     }
 
+    pub fn account_mod_nth(&self, v: usize) -> &M::Account {
+        let nth = v % self.accounts.len();
+        self.accounts
+            .iter()
+            .nth(nth)
+            .map(|(_, account)| account)
+            .unwrap()
+    }
+
     /// Get a node by name.
     pub fn node(&self, name: &NodeName) -> anyhow::Result<&M::Node> {
         self.nodes

fendermint/testing/materializer/tests/docker.rs

@@ -16,13 +16,8 @@ use std::{
 
 use anyhow::{anyhow, Context};
 use ethers::providers::Middleware;
-use fendermint_materializer::{
-    docker::{DockerMaterializer, DockerMaterials},
-    manifest::Manifest,
-    testnet::Testnet,
-    HasCometBftApi, HasEthApi, TestnetName,
-};
-use futures::Future;
+use fendermint_materializer::{concurrency, docker::{DockerMaterializer, DockerMaterials}, manifest::Manifest, testnet::Testnet, HasCometBftApi, HasEthApi, TestnetName};
+use futures::{future, Future};
 use lazy_static::lazy_static;
 use tendermint_rpc::Client;
 
@@ -59,13 +54,14 @@ fn read_manifest(file_name: &str) -> anyhow::Result<Manifest> {
 
 /// Parse a manifest file in the `manifests` directory, clean up any corresponding
 /// testnet resources, then materialize a testnet and run some tests.
-pub async fn with_testnet<F, G>(manifest_file_name: &str, alter: G, test: F) -> anyhow::Result<()>
+pub async fn with_testnet<F, G>(manifest_file_name: &str, concurrency: Option<concurrency::Config>, alter: G, test: F) -> anyhow::Result<()>
 where
-    // https://users.rust-lang.org/t/function-that-takes-a-closure-with-mutable-reference-that-returns-a-future/54324
-    F: for<'a> FnOnce(
+// https://users.rust-lang.org/t/function-that-takes-a-closure-with-mutable-reference-that-returns-a-future/54324
+    F: for<'a> Fn(
         &Manifest,
-        &mut DockerMaterializer,
-        &'a mut DockerTestnet,
+        &DockerMaterializer,
+        &'a DockerTestnet,
+        usize
     ) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + 'a>>,
     G: FnOnce(&mut Manifest),
 {
@@ -98,14 +94,40 @@ where
         .await
         .context("failed to remove testnet")?;
 
-    let mut testnet = Testnet::setup(&mut materializer, &testnet_name, &manifest)
+    let testnet = Testnet::setup(&mut materializer, &testnet_name, &manifest)
         .await
         .context("failed to set up testnet")?;
 
     let started = wait_for_startup(&testnet).await?;
 
     let res = if started {
-        test(&manifest, &mut materializer, &mut testnet).await
+        match concurrency {
+            None => test(&manifest, &materializer, &testnet, 0).await,
+            Some(cfg) => {
+                let mut futures = Vec::new();
+                let mut test_ids = Vec::new();
+                for i in 0..cfg.parallelism_level {
+                    let test_id = i;
+                    let task = test(&manifest, &materializer, &testnet, test_id);
+                    futures.push(task);
+                    test_ids.push(test_id);
+                }
+
+                let results: Vec<Result<(), anyhow::Error>> = future::join_all(futures).await;
+                let mut err = None;
+                for (i, result) in results.into_iter().enumerate() {
+                    let test_id = test_ids[i];
+                    match result {
+                        Ok(_) => println!("test completed successfully (test_id={})", test_id),
+                        Err(e) => {
+                            println!("test failed: {} (test_id={})", e, test_id);
+                            err = Some(e);
+                        },
+                    }
+                }
+                err.map_or(Ok(()), Err)
+            }
+        }
     } else {
         Err(anyhow!("the startup sequence timed out"))
     };

fendermint/testing/materializer/tests/docker_tests/layer2.rs

@@ -27,6 +27,7 @@ const MAX_RETRIES: u32 = 5;
 async fn test_topdown_and_bottomup() {
     with_testnet(
         MANIFEST,
+        None,
         |manifest| {
             // Try to make sure the bottom-up checkpoint period is quick enough for reasonable test runtime.
             let subnet = manifest
@@ -36,7 +37,7 @@ async fn test_topdown_and_bottomup() {
 
             subnet.bottom_up_checkpoint.period = CHECKPOINT_PERIOD;
         },
-        |_, _, testnet| {
+        |_, _, testnet, _| {
             let test = async {
                 let brussels = testnet.node(&testnet.root().node("brussels"))?;
                 let london = testnet.node(&testnet.root().subnet("england").node("london"))?;

fendermint/testing/materializer/tests/docker_tests/root_only.rs

@@ -16,8 +16,9 @@ const MANIFEST: &str = "root-only.yaml";
 async fn test_full_node_sync() {
     with_testnet(
         MANIFEST,
+        None,
         |_| {},
-        |_, _, testnet| {
+        |_, _, testnet, _| {
             let test = async {
                 // Allow a little bit of time for node-2 to catch up with node-1.
                 tokio::time::sleep(Duration::from_secs(5)).await;

fendermint/testing/materializer/tests/docker_tests/standalone.rs

@@ -11,10 +11,9 @@ use ethers::{
     signers::{Signer, Wallet},
     types::{transaction::eip2718::TypedTransaction, Eip1559TransactionRequest, H160},
 };
-use fendermint_materializer::{manifest::Rootnet, materials::DefaultAccount, HasEthApi};
+use fendermint_materializer::{concurrency, manifest::Rootnet, materials::DefaultAccount, HasEthApi};
 use futures::FutureExt;
-
-use crate::with_testnet;
+use crate::{with_testnet};
 
 const MANIFEST: &str = "standalone.yaml";
 
@@ -46,30 +45,31 @@ where
 async fn test_sent_tx_found_in_mempool() {
     with_testnet(
         MANIFEST,
+        Some(concurrency::Config::with_parallelism_level(3)),
         |manifest| {
             // Slow down consensus to where we can see the effect of the transaction not being found by Ethereum hash.
             if let Rootnet::New { ref mut env, .. } = manifest.rootnet {
                 env.insert("CMT_CONSENSUS_TIMEOUT_COMMIT".into(), "10s".into());
             };
         },
-        |_, _, testnet| {
-            let test = async {
-                let bob = testnet.account("bob")?;
-                let charlie = testnet.account("charlie")?;
-
+        |_, _, testnet, test_id| {
+            let sender = testnet.account_mod_nth(test_id);
+            let recipient = testnet.account_mod_nth(test_id + 1);
+            let test = async move {
+                println!("running (test_id={})", test_id);
                 let pangea = testnet.node(&testnet.root().node("pangea"))?;
                 let provider = pangea
                     .ethapi_http_provider()?
                     .expect("ethapi should be enabled");
 
-                let middleware = make_middleware(provider, bob)
+                let middleware = make_middleware(provider, sender)
                     .await
                     .context("failed to set up middleware")?;
 
-                eprintln!("middleware ready, pending tests");
+                println!("middleware ready, pending tests (test_id={})", test_id);
 
                 // Create the simplest transaction possible: send tokens between accounts.
-                let to: H160 = charlie.eth_addr().into();
+                let to: H160 = recipient.eth_addr().into();
                 let transfer = Eip1559TransactionRequest::new().to(to).value(1);
 
                 let pending: PendingTransaction<_> = middleware
@@ -79,7 +79,7 @@ async fn test_sent_tx_found_in_mempool() {
 
                 let tx_hash = pending.tx_hash();
 
-                eprintln!("sent pending txn {:?}", tx_hash);
+                println!("sent pending txn {:?} (test_id={})", tx_hash, test_id);
 
                 // We expect that the transaction is pending, however it should not return an error.
                 match middleware.get_transaction(tx_hash).await {
@@ -111,8 +111,9 @@ async fn test_out_of_order_mempool() {
 
     with_testnet(
         MANIFEST,
+        None,
         |_| {},
-        |_, _, testnet| {
+        |_, _, testnet, _| {
             let test = async {
                 let bob = testnet.account("bob")?;
                 let charlie = testnet.account("charlie")?;