spawn geth node

Signed-off-by: Cyrill Leutwiler <bigcyrill@hotmail.com>

crates/node/src/geth.rs

@@ -1,17 +1,20 @@
 //! The go-ethereum node implementation.
 
 use std::{
-    fs::{File, create_dir, exists},
+    fs::{File, create_dir, exists, remove_dir_all},
+    io::{BufRead, BufReader, Read, Write},
     path::PathBuf,
-    process::{Command, Stdio},
+    process::{Child, Command, Stdio},
     sync::atomic::{AtomicU32, Ordering},
+    thread,
+    time::{Duration, Instant},
 };
 
-use alloy::{
-    genesis::Genesis,
-    rpc::types::trace::geth::{DiffMode, PreStateFrame},
+use alloy::rpc::types::{
+    TransactionReceipt, TransactionRequest,
+    trace::geth::{DiffMode, PreStateFrame},
 };
-use revive_dt_config::get_args;
+use revive_dt_config::Arguments;
 use revive_dt_node_interaction::{
     EthereumNode, trace::trace_transaction, transaction::execute_transaction,
 };
@@ -20,78 +23,157 @@ 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,
-    directory: PathBuf,
+    base_directory: PathBuf,
+    data_directory: PathBuf,
     geth: PathBuf,
     id: u32,
+    handle: Option<Child>,
+    network_id: u64,
+    start_timeout: u64,
 }
 
 impl Instance {
-    pub fn new() -> anyhow::Result<Self> {
-        let args = get_args();
+    const BASE_DIRECTORY: &str = "geth";
+    const DATA_DIRECTORY: &str = "data";
 
-        let geth_directory = args.working_directory.join("geth");
-        if !exists(&geth_directory)? {
-            create_dir(&geth_directory)?;
-        }
+    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 a new uninitialized instance.
+    pub fn new(config: &Arguments) -> anyhow::Result<Self> {
+        let geth_directory = PathBuf::from(&config.working_directory).join(Self::BASE_DIRECTORY);
         let id = NODE_COUNT.fetch_add(1, Ordering::SeqCst);
-        let directory = geth_directory.join(id.to_string());
-
-        let connection_string = directory.join("geth.ipc").display().to_string();
+        let base_directory = geth_directory.join(id.to_string());
 
         Ok(Self {
-            connection_string,
-            directory,
-            geth: args.geth.clone(),
+            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,
         })
     }
-}
 
-impl Instance {
-    /// Call `init` on the node to configure it's genesis.
-    fn init(&mut self, genesis: Genesis) -> anyhow::Result<()> {
-        let genesis_path = self.directory.join("genesis.json");
+    /// Create the node directory and call `geth init` to configure the genesis.
+    fn init(&mut self, genesis: String) -> anyhow::Result<&mut Self> {
+        let geth_directory = self.base_directory.parent().expect("the id should be set");
+        if !exists(geth_directory)? {
+            create_dir(geth_directory)?;
+        }
+        create_dir(&self.base_directory)?;
 
-        let mut file = File::create(&genesis_path)?;
-        serde_json::to_writer_pretty(&mut file, &genesis)?;
+        let genesis_path = self.base_directory.join(Self::GENESIS_JSON_FILE);
+        File::create(&genesis_path)?.write_all(genesis.as_bytes())?;
 
-        if !Command::new(&self.geth)
+        let mut child = Command::new(&self.geth)
+            .arg("init")
             .arg("--datadir")
-            .arg(self.directory.join("data"))
-            .stderr(Stdio::null())
+            .arg(&self.data_directory)
+            .arg(genesis_path)
+            .stderr(Stdio::piped())
             .stdout(Stdio::null())
-            .spawn()?
-            .wait()?
-            .success()
-        {
-            anyhow::bail!("failed to initialize geth node {:?}", &self);
+            .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(())
+        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_request: alloy::rpc::types::TransactionRequest,
+        transaction: TransactionRequest,
     ) -> anyhow::Result<alloy::rpc::types::TransactionReceipt> {
-        execute_transaction(transaction_request, self.connection_string())
+        execute_transaction(transaction, self.connection_string())
     }
 
     fn trace_transaction(
         &self,
-        transaction_receipt: alloy::rpc::types::TransactionReceipt,
+        transaction: TransactionReceipt,
     ) -> anyhow::Result<alloy::rpc::types::trace::geth::GethTrace> {
-        trace_transaction(
-            transaction_receipt,
-            Default::default(),
-            self.connection_string(),
-        )
+        trace_transaction(transaction, Default::default(), self.connection_string())
     }
 }
 
@@ -104,10 +186,9 @@ impl Node for Instance {
         Ok(())
     }
 
-    fn spawn(&mut self, genesis: Genesis) -> anyhow::Result<&mut Self> {
-        self.init(genesis)?;
-
-        Ok(self)
+    fn spawn(&mut self, genesis: String) -> anyhow::Result<()> {
+        self.init(genesis)?.spawn_process()?.wait_ready()?;
+        Ok(())
     }
 
     fn state_diff(
@@ -123,3 +204,40 @@ impl Node for Instance {
         }
     }
 }
+
+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 crate::{GENESIS_JSON, Node};
+
+    use super::Instance;
+
+    #[test]
+    fn init_works() {
+        Instance::new(&Arguments::default())
+            .unwrap()
+            .init(GENESIS_JSON.to_string())
+            .unwrap();
+    }
+
+    #[test]
+    fn spawn_works() {
+        Instance::new(&Arguments::default())
+            .unwrap()
+            .spawn(GENESIS_JSON.to_string())
+            .unwrap();
+    }
+}

crates/node/src/lib.rs

@@ -1,19 +1,19 @@
 //! This crate implements the testing nodes.
 
-use alloy::{
-    genesis::Genesis,
-    rpc::types::{TransactionReceipt, trace::geth::DiffMode},
-};
+use alloy::rpc::types::{TransactionReceipt, trace::geth::DiffMode};
 use revive_dt_node_interaction::EthereumNode;
 
 pub mod geth;
 
+/// The default genesis configuration.
+pub const GENESIS_JSON: &str = include_str!("../../../genesis.json");
+
 /// An abstract interface for testing nodes.
 pub trait Node: EthereumNode {
-    /// Spawns a node configured according to the [Genesis].
+    /// Spawns a node configured according to the genesis json.
     ///
     /// Blocking until it's ready to accept transactions.
-    fn spawn(&mut self, genesis: Genesis) -> anyhow::Result<&mut Self>;
+    fn spawn(&mut self, genesis: String) -> anyhow::Result<()>;
 
     /// Prune the node instance and related data.
     ///