Zombienet & Benchmarks Cleanups (#184)

* Minor zombienet cleanups * Remove un-necessary trace call from the benchmark driver * Improve the benchmarks driver * Ignore the lighthouse tests * Allow for the consensus to be specified for the revive dev node * Ignore the zombienet tests for the time being
2026-04-22 17:17:58 +00:00 · 2025-10-09 14:41:01 +03:00
parent 8c412dc924
commit 21e25f09e6
14 changed files with 292 additions and 227 deletions
@@ -1131,6 +1131,7 @@ mod tests {
    }

    #[tokio::test]
+    #[ignore = "Ignored since they take a long time to run"]
    async fn node_mines_simple_transfer_transaction_and_returns_receipt() {
        // Arrange
        let (context, node) = new_node();
@@ -36,7 +36,8 @@ use alloy::{
    },
 };
 use anyhow::Context as _;
-use futures::{Stream, StreamExt};
+use async_stream::stream;
+use futures::Stream;
 use revive_common::EVMVersion;
 use revive_dt_common::fs::clear_directory;
 use revive_dt_format::traits::ResolverApi;
@@ -80,6 +81,7 @@ pub struct SubstrateNode {
    wallet: Arc<EthereumWallet>,
    nonce_manager: CachedNonceManager,
    provider: OnceCell<ConcreteProvider<ReviveNetwork, Arc<EthereumWallet>>>,
+    consensus: Option<String>,
 }

 impl SubstrateNode {
@@ -102,6 +104,7 @@ impl SubstrateNode {
    pub fn new(
        node_path: PathBuf,
        export_chainspec_command: &str,
+        consensus: Option<String>,
        context: impl AsRef<WorkingDirectoryConfiguration>
        + AsRef<EthRpcConfiguration>
        + AsRef<WalletConfiguration>,
@@ -131,6 +134,7 @@ impl SubstrateNode {
            wallet: wallet.clone(),
            nonce_manager: Default::default(),
            provider: Default::default(),
+            consensus,
        }
    }

@@ -228,7 +232,7 @@ impl SubstrateNode {
            self.logs_directory.as_path(),
            self.node_binary.as_path(),
            |command, stdout_file, stderr_file| {
-                command
+                let cmd = command
                    .arg("--dev")
                    .arg("--chain")
                    .arg(chainspec_path)
@@ -245,9 +249,16 @@ impl SubstrateNode {
                    .arg("all")
                    .arg("--rpc-max-connections")
                    .arg(u32::MAX.to_string())
+                    .arg("--pool-limit")
+                    .arg(u32::MAX.to_string())
+                    .arg("--pool-kbytes")
+                    .arg(u32::MAX.to_string())
                    .env("RUST_LOG", Self::SUBSTRATE_LOG_ENV)
                    .stdout(stdout_file)
                    .stderr(stderr_file);
+                if let Some(consensus) = self.consensus.as_ref() {
+                    cmd.arg("--consensus").arg(consensus.clone());
+                }
            },
            ProcessReadinessWaitBehavior::TimeBoundedWaitFunction {
                max_wait_duration: Duration::from_secs(30),
@@ -508,37 +519,46 @@ impl EthereumNode for SubstrateNode {
                + '_,
        >,
    > {
+        fn create_stream(
+            provider: ConcreteProvider<ReviveNetwork, Arc<EthereumWallet>>,
+        ) -> impl Stream<Item = MinedBlockInformation> {
+            stream! {
+                let mut block_number = provider.get_block_number().await.expect("Failed to get the block number");
+                loop {
+                    tokio::time::sleep(Duration::from_secs(1)).await;
+
+                    let Ok(Some(block)) = provider.get_block_by_number(BlockNumberOrTag::Number(block_number)).await
+                    else {
+                        continue;
+                    };
+
+                    block_number += 1;
+                    yield MinedBlockInformation {
+                        block_number: block.number(),
+                        block_timestamp: block.header.timestamp,
+                        mined_gas: block.header.gas_used as _,
+                        block_gas_limit: block.header.gas_limit,
+                        transaction_hashes: block
+                            .transactions
+                            .into_hashes()
+                            .as_hashes()
+                            .expect("Must be hashes")
+                            .to_vec(),
+                    };
+                };
+            }
+        }
+
        Box::pin(async move {
            let provider = self
                .provider()
                .await
-                .context("Failed to create the provider for block subscription")?;
-            let mut block_subscription = provider
-                .watch_full_blocks()
-                .await
-                .context("Failed to create the blocks stream")?;
-            block_subscription.set_channel_size(0xFFFF);
-            block_subscription.set_poll_interval(Duration::from_secs(1));
-            let block_stream = block_subscription.into_stream();
+                .context("Failed to create the provider for a block subscription")?;

-            let mined_block_information_stream = block_stream.filter_map(|block| async {
-                let block = block.ok()?;
-                Some(MinedBlockInformation {
-                    block_number: block.number(),
-                    block_timestamp: block.header.timestamp,
-                    mined_gas: block.header.gas_used as _,
-                    block_gas_limit: block.header.gas_limit,
-                    transaction_hashes: block
-                        .transactions
-                        .into_hashes()
-                        .as_hashes()
-                        .expect("Must be hashes")
-                        .to_vec(),
-                })
-            });
+            let stream = Box::pin(create_stream(provider))
+                as Pin<Box<dyn Stream<Item = MinedBlockInformation>>>;

-            Ok(Box::pin(mined_block_information_stream)
-                as Pin<Box<dyn Stream<Item = MinedBlockInformation>>>)
+            Ok(stream)
        })
    }
 }
@@ -1170,6 +1190,7 @@ mod tests {
        let mut node = SubstrateNode::new(
            context.kitchensink_configuration.path.clone(),
            SubstrateNode::KITCHENSINK_EXPORT_CHAINSPEC_COMMAND,
+            None,
            &context,
        );
        node.init(context.genesis_configuration.genesis().unwrap().clone())
@@ -1235,6 +1256,7 @@ mod tests {
        let mut dummy_node = SubstrateNode::new(
            context.kitchensink_configuration.path.clone(),
            SubstrateNode::KITCHENSINK_EXPORT_CHAINSPEC_COMMAND,
+            None,
            &context,
        );

@@ -1287,6 +1309,7 @@ mod tests {
        let node = SubstrateNode::new(
            context.kitchensink_configuration.path.clone(),
            SubstrateNode::KITCHENSINK_EXPORT_CHAINSPEC_COMMAND,
+            None,
            &context,
        );

@@ -55,7 +55,8 @@ use alloy::{
 };

 use anyhow::Context as _;
-use futures::{Stream, StreamExt};
+use async_stream::stream;
+use futures::Stream;
 use revive_common::EVMVersion;
 use revive_dt_common::fs::clear_directory;
 use revive_dt_config::*;
@@ -73,16 +74,19 @@ use crate::{
    constants::INITIAL_BALANCE,
    helpers::{Process, ProcessReadinessWaitBehavior},
    node_implementations::substrate::ReviveNetwork,
-    provider_utils::{ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider},
+    provider_utils::{
+        ConcreteProvider, FallbackGasFiller, construct_concurrency_limited_provider,
+        execute_transaction,
+    },
 };

 static NODE_COUNT: AtomicU32 = AtomicU32::new(0);

-/// A Zombienet network where collator is `polkadot-parachain` node with `eth-rpc`
-/// [`ZombieNode`] abstracts away the details of managing the zombienet network and provides
-/// an interface to interact with the parachain's Ethereum RPC.
+/// A Zombienet network where collator is `polkadot-parachain` node with `eth-rpc` [`ZombieNode`]
+/// abstracts away the details of managing the zombienet network and provides an interface to
+/// interact with the parachain's Ethereum RPC.
 #[derive(Debug, Default)]
-pub struct ZombieNode {
+pub struct ZombienetNode {
    /* Node Identifier */
    id: u32,
    connection_string: String,
@@ -110,7 +114,7 @@ pub struct ZombieNode {
    provider: OnceCell<ConcreteProvider<ReviveNetwork, Arc<EthereumWallet>>>,
 }

-impl ZombieNode {
+impl ZombienetNode {
    const BASE_DIRECTORY: &str = "zombienet";
    const DATA_DIRECTORY: &str = "data";
    const LOGS_DIRECTORY: &str = "logs";
@@ -119,6 +123,8 @@ impl ZombieNode {
    const PARACHAIN_ID: u32 = 100;
    const ETH_RPC_BASE_PORT: u16 = 8545;

+    const PROXY_LOG_ENV: &str = "info,eth-rpc=debug";
+
    const ETH_RPC_READY_MARKER: &str = "Running JSON-RPC server";

    const EXPORT_CHAINSPEC_COMMAND: &str = "build-spec";
@@ -178,25 +184,35 @@ impl ZombieNode {
        let node_rpc_port = Self::NODE_BASE_RPC_PORT + self.id as u16;

        let network_config = NetworkConfigBuilder::new()
-            .with_relaychain(|r| {
-                r.with_chain("westend-local")
+            .with_relaychain(|relay_chain| {
+                relay_chain
+                    .with_chain("westend-local")
                    .with_default_command("polkadot")
                    .with_node(|node| node.with_name("alice"))
                    .with_node(|node| node.with_name("bob"))
            })
-            .with_global_settings(|g| g.with_base_dir(&self.base_directory))
-            .with_parachain(|p| {
-                p.with_id(Self::PARACHAIN_ID)
-                    .with_chain_spec_path(template_chainspec_path.to_str().unwrap())
+            .with_global_settings(|global_settings| {
+                // global_settings.with_base_dir(&self.base_directory)
+                global_settings
+            })
+            .with_parachain(|parachain| {
+                parachain
+                    .with_id(Self::PARACHAIN_ID)
+                    .with_chain_spec_path(template_chainspec_path.to_path_buf())
                    .with_chain("asset-hub-westend-local")
-                    .with_collator(|n| {
-                        n.with_name("Collator")
+                    .with_collator(|node_config| {
+                        node_config
+                            .with_name("Collator")
                            .with_command(polkadot_parachain_path)
                            .with_rpc_port(node_rpc_port)
+                            .with_args(vec![
+                                ("--pool-limit", u32::MAX.to_string().as_str()).into(),
+                                ("--pool-kbytes", u32::MAX.to_string().as_str()).into(),
+                            ])
                    })
            })
            .build()
-            .map_err(|e| anyhow::anyhow!("Failed to build zombienet network config: {e:?}"))?;
+            .map_err(|err| anyhow::anyhow!("Failed to build zombienet network config: {err:?}"))?;

        self.node_rpc_port = Some(node_rpc_port);
        self.network_config = Some(network_config);
@@ -210,6 +226,9 @@ impl ZombieNode {
            .clone()
            .context("Node not initialized, call init() first")?;

+        // TODO: Look into the possibility of removing this in the future, perhaps by reintroducing
+        // the blocking runtime abstraction and making it available to the entire program so that we
+        // don't need to be spawning multiple different runtimes.
        let rt = tokio::runtime::Runtime::new().unwrap();
        let network = rt.block_on(async {
            network_config
@@ -237,6 +256,7 @@ impl ZombieNode {
                    .arg(u32::MAX.to_string())
                    .arg("--rpc-port")
                    .arg(eth_rpc_port.to_string())
+                    .env("RUST_LOG", Self::PROXY_LOG_ENV)
                    .stdout(stdout_file)
                    .stderr(stderr_file);
            },
@@ -272,12 +292,13 @@ impl ZombieNode {
        template_chainspec_path: PathBuf,
        mut genesis: Genesis,
    ) -> anyhow::Result<()> {
-        let mut cmd: Command = std::process::Command::new(&self.polkadot_parachain_path);
-        cmd.arg(Self::EXPORT_CHAINSPEC_COMMAND)
+        let output = Command::new(self.polkadot_parachain_path.as_path())
+            .arg(Self::EXPORT_CHAINSPEC_COMMAND)
            .arg("--chain")
-            .arg("asset-hub-westend-local");
-
-        let output = cmd.output().context("Failed to export the chain-spec")?;
+            .arg("asset-hub-westend-local")
+            .env_remove("RUST_LOG")
+            .output()
+            .context("Failed to export the chainspec of the chain")?;

        if !output.status.success() {
            anyhow::bail!(
@@ -385,7 +406,7 @@ impl ZombieNode {
            .get_or_try_init(|| async move {
                construct_concurrency_limited_provider::<ReviveNetwork, _>(
                    self.connection_string.as_str(),
-                    FallbackGasFiller::new(250_000_000, 5_000_000_000, 1_000_000_000),
+                    FallbackGasFiller::new(u64::MAX, 5_000_000_000, 1_000_000_000),
                    ChainIdFiller::default(), // TODO: use CHAIN_ID constant
                    NonceFiller::new(self.nonce_manager.clone()),
                    self.wallet.clone(),
@@ -398,7 +419,7 @@ impl ZombieNode {
    }
 }

-impl EthereumNode for ZombieNode {
+impl EthereumNode for ZombienetNode {
    fn pre_transactions(&mut self) -> Pin<Box<dyn Future<Output = anyhow::Result<()>> + '_>> {
        Box::pin(async move { Ok(()) })
    }
@@ -448,17 +469,11 @@ impl EthereumNode for ZombieNode {
        transaction: alloy::rpc::types::TransactionRequest,
    ) -> Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + '_>> {
        Box::pin(async move {
-            let receipt = self
+            let provider = self
                .provider()
                .await
-                .context("Failed to create provider for transaction submission")?
-                .send_transaction(transaction)
-                .await
-                .context("Failed to submit transaction to proxy")?
-                .get_receipt()
-                .await
-                .context("Failed to fetch transaction receipt from proxy")?;
-            Ok(receipt)
+                .context("Failed to create the provider")?;
+            execute_transaction(provider, transaction).await
        })
    }

@@ -552,37 +567,46 @@ impl EthereumNode for ZombieNode {
                + '_,
        >,
    > {
+        fn create_stream(
+            provider: ConcreteProvider<ReviveNetwork, Arc<EthereumWallet>>,
+        ) -> impl Stream<Item = MinedBlockInformation> {
+            stream! {
+                let mut block_number = provider.get_block_number().await.expect("Failed to get the block number");
+                loop {
+                    tokio::time::sleep(Duration::from_secs(1)).await;
+
+                    let Ok(Some(block)) = provider.get_block_by_number(BlockNumberOrTag::Number(block_number)).await
+                    else {
+                        continue;
+                    };
+
+                    block_number += 1;
+                    yield MinedBlockInformation {
+                        block_number: block.number(),
+                        block_timestamp: block.header.timestamp,
+                        mined_gas: block.header.gas_used as _,
+                        block_gas_limit: block.header.gas_limit,
+                        transaction_hashes: block
+                            .transactions
+                            .into_hashes()
+                            .as_hashes()
+                            .expect("Must be hashes")
+                            .to_vec(),
+                    };
+                };
+            }
+        }
+
        Box::pin(async move {
            let provider = self
                .provider()
                .await
-                .context("Failed to create the provider for block subscription")?;
-            let mut block_subscription = provider
-                .watch_full_blocks()
-                .await
-                .context("Failed to create the blocks stream")?;
-            block_subscription.set_channel_size(0xFFFF);
-            block_subscription.set_poll_interval(Duration::from_secs(1));
-            let block_stream = block_subscription.into_stream();
+                .context("Failed to create the provider for a block subscription")?;

-            let mined_block_information_stream = block_stream.filter_map(|block| async {
-                let block = block.ok()?;
-                Some(MinedBlockInformation {
-                    block_number: block.number(),
-                    block_timestamp: block.header.timestamp,
-                    mined_gas: block.header.gas_used as _,
-                    block_gas_limit: block.header.gas_limit,
-                    transaction_hashes: block
-                        .transactions
-                        .into_hashes()
-                        .as_hashes()
-                        .expect("Must be hashes")
-                        .to_vec(),
-                })
-            });
+            let stream = Box::pin(create_stream(provider))
+                as Pin<Box<dyn Stream<Item = MinedBlockInformation>>>;

-            Ok(Box::pin(mined_block_information_stream)
-                as Pin<Box<dyn Stream<Item = MinedBlockInformation>>>)
+            Ok(stream)
        })
    }
 }
@@ -717,7 +741,7 @@ impl<F: TxFiller<ReviveNetwork>, P: Provider<ReviveNetwork>> ResolverApi
    }
 }

-impl Node for ZombieNode {
+impl Node for ZombienetNode {
    fn shutdown(&mut self) -> anyhow::Result<()> {
        // Kill the eth_rpc process
        drop(self.eth_rpc_process.take());
@@ -762,7 +786,7 @@ impl Node for ZombieNode {
    }
 }

-impl Drop for ZombieNode {
+impl Drop for ZombienetNode {
    fn drop(&mut self) {
        let _ = self.shutdown();
    }
@@ -786,9 +810,9 @@ mod tests {
            TestExecutionContext::default()
        }

-        pub async fn new_node() -> (TestExecutionContext, ZombieNode) {
+        pub async fn new_node() -> (TestExecutionContext, ZombienetNode) {
            let context = test_config();
-            let mut node = ZombieNode::new(
+            let mut node = ZombienetNode::new(
                context.polkadot_parachain_configuration.path.clone(),
                &context,
            );
@@ -806,8 +830,9 @@ mod tests {
            (context, node)
        }

-        pub async fn shared_state() -> &'static (TestExecutionContext, Arc<ZombieNode>) {
-            static NODE: OnceCell<(TestExecutionContext, Arc<ZombieNode>)> = OnceCell::const_new();
+        pub async fn shared_state() -> &'static (TestExecutionContext, Arc<ZombienetNode>) {
+            static NODE: OnceCell<(TestExecutionContext, Arc<ZombienetNode>)> =
+                OnceCell::const_new();

            NODE.get_or_init(|| async {
                let (context, node) = new_node().await;
@@ -816,13 +841,14 @@ mod tests {
            .await
        }

-        pub async fn shared_node() -> &'static Arc<ZombieNode> {
+        pub async fn shared_node() -> &'static Arc<ZombienetNode> {
            &shared_state().await.1
        }
    }
    use utils::{new_node, test_config};

    #[tokio::test]
+    #[ignore = "Ignored for the time being"]
    async fn test_transfer_transaction_should_return_receipt() {
        let (ctx, node) = new_node().await;

@@ -856,7 +882,7 @@ mod tests {
        "#;

        let context = test_config();
-        let mut node = ZombieNode::new(
+        let mut node = ZombienetNode::new(
            context.polkadot_parachain_configuration.path.clone(),
            &context,
        );
@@ -866,17 +892,19 @@ mod tests {
            .expect("init failed");

        // Check that the patched chainspec file was generated
-        let final_chainspec_path = node.base_directory.join(ZombieNode::CHAIN_SPEC_JSON_FILE);
+        let final_chainspec_path = node
+            .base_directory
+            .join(ZombienetNode::CHAIN_SPEC_JSON_FILE);
        assert!(final_chainspec_path.exists(), "Chainspec file should exist");

        let contents =
            std::fs::read_to_string(&final_chainspec_path).expect("Failed to read chainspec");

        // Validate that the Polkadot addresses derived from the Ethereum addresses are in the file
-        let first_eth_addr = ZombieNode::eth_to_polkadot_address(
+        let first_eth_addr = ZombienetNode::eth_to_polkadot_address(
            &"90F8bf6A479f320ead074411a4B0e7944Ea8c9C1".parse().unwrap(),
        );
-        let second_eth_addr = ZombieNode::eth_to_polkadot_address(
+        let second_eth_addr = ZombienetNode::eth_to_polkadot_address(
            &"Ab8483F64d9C6d1EcF9b849Ae677dD3315835cb2".parse().unwrap(),
        );

@@ -904,7 +932,7 @@ mod tests {
        "#;

        let context = test_config();
-        let node = ZombieNode::new(
+        let node = ZombienetNode::new(
            context.polkadot_parachain_configuration.path.clone(),
            &context,
        );
@@ -940,7 +968,7 @@ mod tests {
        ];

        for eth_addr in eth_addresses {
-            let ss58 = ZombieNode::eth_to_polkadot_address(&eth_addr.parse().unwrap());
+            let ss58 = ZombienetNode::eth_to_polkadot_address(&eth_addr.parse().unwrap());

            println!("Ethereum: {eth_addr} -> Polkadot SS58: {ss58}");
        }
@@ -968,7 +996,7 @@ mod tests {
        ];

        for (eth_addr, expected_ss58) in cases {
-            let result = ZombieNode::eth_to_polkadot_address(&eth_addr.parse().unwrap());
+            let result = ZombienetNode::eth_to_polkadot_address(&eth_addr.parse().unwrap());
            assert_eq!(
                result, expected_ss58,
                "Mismatch for Ethereum address {eth_addr}"
@@ -980,7 +1008,7 @@ mod tests {
    fn eth_rpc_version_works() {
        // Arrange
        let context = test_config();
-        let node = ZombieNode::new(
+        let node = ZombienetNode::new(
            context.polkadot_parachain_configuration.path.clone(),
            &context,
        );
@@ -999,7 +1027,7 @@ mod tests {
    fn version_works() {
        // Arrange
        let context = test_config();
-        let node = ZombieNode::new(
+        let node = ZombienetNode::new(
            context.polkadot_parachain_configuration.path.clone(),
            &context,
        );