Scaffold utility and library (#3)

Signed-off-by: Cyrill Leutwiler <bigcyrill@hotmail.com> Signed-off-by: xermicus <bigcyrill@hotmail.com>
2026-04-29 01:58:00 +00:00 · 2025-03-31 11:40:05 +02:00
parent 4b7af83be6
commit c590fa7bfd
36 changed files with 6720 additions and 0 deletions
@@ -0,0 +1,298 @@
+//! The go-ethereum node implementation.
+
+use std::{
+    fs::{File, create_dir_all, remove_dir_all},
+    io::{BufRead, BufReader, Read, Write},
+    path::PathBuf,
+    process::{Child, Command, Stdio},
+    sync::atomic::{AtomicU32, Ordering},
+    thread,
+    time::{Duration, Instant},
+};
+
+use alloy::{
+    network::EthereumWallet,
+    providers::{Provider, ProviderBuilder, ext::DebugApi},
+    rpc::types::{
+        TransactionReceipt, TransactionRequest,
+        trace::geth::{DiffMode, GethDebugTracingOptions, PreStateConfig, PreStateFrame},
+    },
+};
+use revive_dt_config::Arguments;
+use revive_dt_node_interaction::{
+    EthereumNode, trace::trace_transaction, transaction::execute_transaction,
+};
+
+use crate::Node;
+
+static NODE_COUNT: AtomicU32 = AtomicU32::new(0);
+
+/// The go-ethereum node instance implementation.
+///
+/// Implements helpers to initialize, spawn and wait the node.
+///
+/// Assumes dev mode and IPC only (`P2P`, `http`` etc. are kept disabled).
+///
+/// Prunes the child process and the base directory on drop.
+#[derive(Debug)]
+pub struct Instance {
+    connection_string: String,
+    base_directory: PathBuf,
+    data_directory: PathBuf,
+    geth: PathBuf,
+    id: u32,
+    handle: Option<Child>,
+    network_id: u64,
+    start_timeout: u64,
+    wallet: EthereumWallet,
+}
+
+impl Instance {
+    const BASE_DIRECTORY: &str = "geth";
+    const DATA_DIRECTORY: &str = "data";
+
+    const IPC_FILE: &str = "geth.ipc";
+    const GENESIS_JSON_FILE: &str = "genesis.json";
+
+    const READY_MARKER: &str = "IPC endpoint opened";
+    const ERROR_MARKER: &str = "Fatal:";
+
+    /// Create the node directory and call `geth init` to configure the genesis.
+    fn init(&mut self, genesis: String) -> anyhow::Result<&mut Self> {
+        create_dir_all(&self.base_directory)?;
+
+        let genesis_path = self.base_directory.join(Self::GENESIS_JSON_FILE);
+        File::create(&genesis_path)?.write_all(genesis.as_bytes())?;
+
+        let mut child = Command::new(&self.geth)
+            .arg("init")
+            .arg("--datadir")
+            .arg(&self.data_directory)
+            .arg(genesis_path)
+            .stderr(Stdio::piped())
+            .stdout(Stdio::null())
+            .spawn()?;
+
+        let mut stderr = String::new();
+        child
+            .stderr
+            .take()
+            .expect("should be piped")
+            .read_to_string(&mut stderr)?;
+
+        if !child.wait()?.success() {
+            anyhow::bail!("failed to initialize geth node #{:?}: {stderr}", &self.id);
+        }
+
+        Ok(self)
+    }
+
+    /// Spawn the go-ethereum node child process.
+    ///
+    /// [Instance::init] must be called priorly.
+    fn spawn_process(&mut self) -> anyhow::Result<&mut Self> {
+        self.handle = Command::new(&self.geth)
+            .arg("--dev")
+            .arg("--datadir")
+            .arg(&self.data_directory)
+            .arg("--ipcpath")
+            .arg(&self.connection_string)
+            .arg("--networkid")
+            .arg(self.network_id.to_string())
+            .arg("--nodiscover")
+            .arg("--maxpeers")
+            .arg("0")
+            .stderr(Stdio::piped())
+            .stdout(Stdio::null())
+            .spawn()?
+            .into();
+        Ok(self)
+    }
+
+    /// Wait for the g-ethereum node child process getting ready.
+    ///
+    /// [Instance::spawn_process] must be called priorly.
+    fn wait_ready(&mut self) -> anyhow::Result<&mut Self> {
+        // Thanks clippy but geth is a server; we don't `wait` but eventually kill it.
+        #[allow(clippy::zombie_processes)]
+        let mut child = self.handle.take().expect("should be spawned");
+        let start_time = Instant::now();
+        let maximum_wait_time = Duration::from_millis(self.start_timeout);
+        let mut stderr = BufReader::new(child.stderr.take().expect("should be piped")).lines();
+        let error = loop {
+            let Some(Ok(line)) = stderr.next() else {
+                break "child process stderr reading error".to_string();
+            };
+            if line.contains(Self::ERROR_MARKER) {
+                break line;
+            }
+            if line.contains(Self::READY_MARKER) {
+                // Keep stderr alive
+                // https://github.com/alloy-rs/alloy/issues/2091#issuecomment-2676134147
+                thread::spawn(move || for _ in stderr.by_ref() {});
+
+                self.handle = child.into();
+                return Ok(self);
+            }
+            if Instant::now().duration_since(start_time) > maximum_wait_time {
+                break "spawn timeout".to_string();
+            }
+        };
+
+        let _ = child.kill();
+        anyhow::bail!("geth node #{} spawn error: {error}", self.id)
+    }
+}
+
+impl EthereumNode for Instance {
+    fn execute_transaction(
+        &self,
+        transaction: TransactionRequest,
+    ) -> anyhow::Result<alloy::rpc::types::TransactionReceipt> {
+        let connection_string = self.connection_string();
+        let wallet = self.wallet.clone();
+
+        execute_transaction(Box::pin(async move {
+            Ok(ProviderBuilder::new()
+                .wallet(wallet)
+                .connect(&connection_string)
+                .await?
+                .send_transaction(transaction)
+                .await?
+                .get_receipt()
+                .await?)
+        }))
+    }
+
+    fn trace_transaction(
+        &self,
+        transaction: TransactionReceipt,
+    ) -> anyhow::Result<alloy::rpc::types::trace::geth::GethTrace> {
+        let connection_string = self.connection_string();
+        let trace_options = GethDebugTracingOptions::prestate_tracer(PreStateConfig {
+            diff_mode: Some(true),
+            disable_code: None,
+            disable_storage: None,
+        });
+        let wallet = self.wallet.clone();
+
+        trace_transaction(Box::pin(async move {
+            Ok(ProviderBuilder::new()
+                .wallet(wallet)
+                .connect(&connection_string)
+                .await?
+                .debug_trace_transaction(transaction.transaction_hash, trace_options)
+                .await?)
+        }))
+    }
+}
+
+impl Node for Instance {
+    fn new(config: &Arguments) -> Self {
+        let geth_directory = config.directory().join(Self::BASE_DIRECTORY);
+        let id = NODE_COUNT.fetch_add(1, Ordering::SeqCst);
+        let base_directory = geth_directory.join(id.to_string());
+
+        Self {
+            connection_string: base_directory.join(Self::IPC_FILE).display().to_string(),
+            data_directory: base_directory.join(Self::DATA_DIRECTORY),
+            base_directory,
+            geth: config.geth.clone(),
+            id,
+            handle: None,
+            network_id: config.network_id,
+            start_timeout: config.geth_start_timeout,
+            wallet: config.wallet(),
+        }
+    }
+
+    fn connection_string(&self) -> String {
+        self.connection_string.clone()
+    }
+
+    fn shutdown(self) -> anyhow::Result<()> {
+        Ok(())
+    }
+
+    fn spawn(&mut self, genesis: String) -> anyhow::Result<()> {
+        self.init(genesis)?.spawn_process()?.wait_ready()?;
+        Ok(())
+    }
+
+    fn state_diff(
+        &self,
+        transaction: alloy::rpc::types::TransactionReceipt,
+    ) -> anyhow::Result<DiffMode> {
+        match self
+            .trace_transaction(transaction)?
+            .try_into_pre_state_frame()?
+        {
+            PreStateFrame::Diff(diff) => Ok(diff),
+            _ => anyhow::bail!("expected a diff mode trace"),
+        }
+    }
+
+    fn version(&self) -> anyhow::Result<String> {
+        let output = Command::new(&self.geth)
+            .arg("--version")
+            .stdin(Stdio::null())
+            .stdout(Stdio::piped())
+            .stderr(Stdio::null())
+            .spawn()?
+            .wait_with_output()?
+            .stdout;
+        Ok(String::from_utf8_lossy(&output).into())
+    }
+}
+
+impl Drop for Instance {
+    fn drop(&mut self) {
+        if let Some(child) = self.handle.as_mut() {
+            let _ = child.kill();
+        }
+        if self.base_directory.exists() {
+            let _ = remove_dir_all(&self.base_directory);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use revive_dt_config::Arguments;
+    use temp_dir::TempDir;
+
+    use crate::{GENESIS_JSON, Node};
+
+    use super::Instance;
+
+    fn test_config() -> (Arguments, TempDir) {
+        let mut config = Arguments::default();
+        let temp_dir = TempDir::new().unwrap();
+        config.working_directory = temp_dir.path().to_path_buf().into();
+
+        (config, temp_dir)
+    }
+
+    #[test]
+    fn init_works() {
+        Instance::new(&test_config().0)
+            .init(GENESIS_JSON.to_string())
+            .unwrap();
+    }
+
+    #[test]
+    fn spawn_works() {
+        Instance::new(&test_config().0)
+            .spawn(GENESIS_JSON.to_string())
+            .unwrap();
+    }
+
+    #[test]
+    fn version_works() {
+        let version = Instance::new(&test_config().0).version().unwrap();
+        assert!(
+            version.starts_with("geth version"),
+            "expected version string, got: '{version}'"
+        );
+    }
+}
@@ -0,0 +1,36 @@
+//! This crate implements the testing nodes.
+
+use alloy::rpc::types::{TransactionReceipt, trace::geth::DiffMode};
+use revive_dt_config::Arguments;
+use revive_dt_node_interaction::EthereumNode;
+
+pub mod geth;
+pub mod pool;
+
+/// The default genesis configuration.
+pub const GENESIS_JSON: &str = include_str!("../../../genesis.json");
+
+/// An abstract interface for testing nodes.
+pub trait Node: EthereumNode {
+    /// Create a new uninitialized instance.
+    fn new(config: &Arguments) -> Self;
+
+    /// Spawns a node configured according to the genesis json.
+    ///
+    /// Blocking until it's ready to accept transactions.
+    fn spawn(&mut self, genesis: String) -> anyhow::Result<()>;
+
+    /// Prune the node instance and related data.
+    ///
+    /// Blocking until it's completely stopped.
+    fn shutdown(self) -> anyhow::Result<()>;
+
+    /// Returns the nodes connection string.
+    fn connection_string(&self) -> String;
+
+    /// Returns the state diff of the transaction hash in the [TransactionReceipt].
+    fn state_diff(&self, transaction: TransactionReceipt) -> anyhow::Result<DiffMode>;
+
+    /// Returns the node version.
+    fn version(&self) -> anyhow::Result<String>;
+}
@@ -0,0 +1,68 @@
+//! This crate implements concurrent handling of testing node.
+
+use std::{
+    fs::read_to_string,
+    sync::atomic::{AtomicUsize, Ordering},
+    thread,
+};
+
+use anyhow::Context;
+use revive_dt_config::Arguments;
+
+use crate::Node;
+
+/// The node pool starts one or more [Node] which then can be accessed
+/// in a round robbin fasion.
+pub struct NodePool<T> {
+    next: AtomicUsize,
+    nodes: Vec<T>,
+}
+
+impl<T> NodePool<T>
+where
+    T: Node + Send + 'static,
+{
+    /// Create a new Pool. This will start as many nodes as there are workers in `config`.
+    pub fn new(config: &Arguments) -> anyhow::Result<Self> {
+        let nodes = config.workers;
+        let genesis = read_to_string(&config.genesis_file).context(format!(
+            "can not read genesis file: {}",
+            config.genesis_file.display()
+        ))?;
+
+        let mut handles = Vec::with_capacity(nodes);
+        for _ in 0..nodes {
+            let config = config.clone();
+            let genesis = genesis.clone();
+            handles.push(thread::spawn(move || spawn_node::<T>(&config, genesis)));
+        }
+
+        let mut nodes = Vec::with_capacity(nodes);
+        for handle in handles {
+            nodes.push(
+                handle
+                    .join()
+                    .map_err(|error| anyhow::anyhow!("failed to spawn node: {:?}", error))?
+                    .map_err(|error| anyhow::anyhow!("node failed to spawn: {error}"))?,
+            );
+        }
+
+        Ok(Self {
+            nodes,
+            next: Default::default(),
+        })
+    }
+
+    /// Get a handle to the next node.
+    pub fn round_robbin(&self) -> &T {
+        let current = self.next.fetch_add(1, Ordering::SeqCst) % self.nodes.len();
+        self.nodes.get(current).unwrap()
+    }
+}
+
+fn spawn_node<T: Node + Send>(args: &Arguments, genesis: String) -> anyhow::Result<T> {
+    let mut node = T::new(args);
+    log::info!("starting node: {}", node.connection_string());
+    node.spawn(genesis)?;
+    Ok(node)
+}