Update the release profile

Co-authored-by: xermicus <cyrill@parity.io>

.github/workflows/test.yml

@@ -99,9 +99,28 @@ jobs:
       - name: Install Geth on Ubuntu
         if: matrix.os == 'ubuntu-24.04'
         run: |
-          sudo add-apt-repository -y ppa:ethereum/ethereum
           sudo apt-get update
-          sudo apt-get install -y ethereum protobuf-compiler
+          sudo apt-get install -y protobuf-compiler
+
+          # We were facing some issues in CI with the 1.16.* versions of geth, and specifically on
+          # Ubuntu. Eventually, we found out that the last version of geth that worked in our CI was
+          # version 1.15.11. Thus, this is the version that we want to use in CI. The PPA sadly does
+          # not have historic versions of Geth and therefore we need to resort to downloading pre
+          # built binaries for Geth and the surrounding tools which is what the following parts of 
+          # the script do.
+
+          sudo apt-get install -y wget ca-certificates tar
+          ARCH=$(uname -m)
+          if [ "$ARCH" = "x86_64" ]; then
+            URL="https://gethstore.blob.core.windows.net/builds/geth-alltools-linux-amd64-1.15.11-36b2371c.tar.gz"
+          elif [ "$ARCH" = "aarch64" ]; then
+            URL="https://gethstore.blob.core.windows.net/builds/geth-alltools-linux-arm64-1.15.11-36b2371c.tar.gz"
+          else
+            echo "Unsupported architecture: $ARCH"
+            exit 1
+          fi
+          wget -qO- "$URL" | sudo tar xz -C /usr/local/bin --strip-components=1
+          geth --version
 
       - name: Install Geth on macOS
         if: matrix.os == 'macos-14'

.gitignore

@@ -3,3 +3,7 @@
 .DS_Store
 node_modules
 /*.json
+
+# We do not want to commit any log files that we produce from running the code locally so this is 
+# added to the .gitignore file.
+*.log

Cargo.lock

@@ -67,9 +67,9 @@ checksum = "683d7910e743518b0e34f1186f92494becacb047c7b6bf616c96772180fef923"
 
 [[package]]
 name = "alloy"
-version = "1.0.9"
+version = "1.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0093d23bf026b580c1f66ed3a053d8209c104a446c5264d3ad99587f6edef24e"
+checksum = "ae58d888221eecf621595e2096836ce7cfc37be06bfa39d7f64aa6a3ea4c9e5b"
 dependencies = [
  "alloy-consensus",
  "alloy-contract",
@@ -162,9 +162,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-core"
-version = "1.1.2"
+version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a3c5a28f166629752f2e7246b813cdea3243cca59aab2d4264b1fd68392c10eb"
+checksum = "ad31216895d27d307369daa1393f5850b50bbbd372478a9fa951c095c210627e"
 dependencies = [
  "alloy-dyn-abi",
  "alloy-json-abi",
@@ -175,9 +175,9 @@ dependencies = [
 
 [[package]]
 name = "alloy-dyn-abi"
-version = "1.1.2"
+version = "1.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "18cc14d832bc3331ca22a1c7819de1ede99f58f61a7d123952af7dde8de124a6"
+checksum = "7b95b3deca680efc7e9cba781f1a1db352fa1ea50e6384a514944dcf4419e652"
 dependencies = [
  "alloy-json-abi",
  "alloy-primitives",
@@ -4002,6 +4002,7 @@ dependencies = [
  "sp-core",
  "sp-runtime",
  "temp-dir",
+ "tokio",
  "tracing",
 ]
 
@@ -4011,6 +4012,7 @@ version = "0.1.0"
 dependencies = [
  "alloy",
  "anyhow",
+ "futures",
  "once_cell",
  "tokio",
  "tracing",

Cargo.toml

@@ -25,6 +25,7 @@ alloy-primitives = "1.2.1"
 alloy-sol-types = "1.2.1"
 anyhow = "1.0"
 clap = { version = "4", features = ["derive"] }
+futures = { version = "0.3.31" }
 hex = "0.4.3"
 reqwest = { version = "0.12.15", features = ["blocking", "json"] }
 once_cell = "1.21"
@@ -68,6 +69,9 @@ features = [
     "rpc-types",
     "signer-local",
     "std",
+    "network",
+    "serde",
+    "rpc-types-eth",
 ]
 
 [profile.bench]

crates/config/src/lib.rs

@@ -1,4 +1,4 @@
-//! The global configuration used accross all revive differential testing crates.
+//! The global configuration used across all revive differential testing crates.
 
 use std::{
     fmt::Display,

crates/core/src/driver/mod.rs

@@ -1,11 +1,11 @@
 //! The test driver handles the compilation and execution of the test cases.
 
 use alloy::json_abi::JsonAbi;
-use alloy::primitives::Bytes;
+use alloy::network::TransactionBuilder;
+use alloy::rpc::types::TransactionReceipt;
 use alloy::rpc::types::trace::geth::GethTrace;
-use alloy::rpc::types::{TransactionInput, TransactionReceipt};
 use alloy::{
-    primitives::{Address, TxKind, map::HashMap},
+    primitives::{Address, map::HashMap},
     rpc::types::{
         TransactionRequest,
         trace::geth::{AccountState, DiffMode},
@@ -135,21 +135,17 @@ where
             std::any::type_name::<T>()
         );
 
-        let tx = match input.legacy_transaction(
-            self.config.network_id,
-            nonce,
-            &self.deployed_contracts,
-            &self.deployed_abis,
-        ) {
-            Ok(tx) => {
-                tracing::debug!("Legacy transaction data: {tx:#?}");
-                tx
-            }
-            Err(err) => {
-                tracing::error!("Failed to construct legacy transaction: {err:?}");
-                return Err(err);
-            }
-        };
+        let tx =
+            match input.legacy_transaction(nonce, &self.deployed_contracts, &self.deployed_abis) {
+                Ok(tx) => {
+                    tracing::debug!("Legacy transaction data: {tx:#?}");
+                    tx
+                }
+                Err(err) => {
+                    tracing::error!("Failed to construct legacy transaction: {err:?}");
+                    return Err(err);
+                }
+            };
 
         tracing::trace!("Executing transaction for input: {input:?}");
 
@@ -201,6 +197,9 @@ where
 
             for contracts in contract_map.values() {
                 for (contract_name, contract) in contracts {
+                    let tracing_span = tracing::info_span!("Deploying contract", contract_name);
+                    let _guard = tracing_span.enter();
+
                     tracing::debug!(
                         "Contract name is: {:?} and the input name is: {:?}",
                         &contract_name,
@@ -228,16 +227,14 @@ where
                         std::any::type_name::<T>()
                     );
 
-                    let tx = TransactionRequest {
-                        from: Some(input.caller),
-                        to: Some(TxKind::Create),
-                        gas_price: Some(5_000_000),
-                        gas: Some(5_000_000),
-                        chain_id: Some(self.config.network_id),
-                        nonce: Some(nonce),
-                        input: TransactionInput::new(Bytes::from(code.into_bytes())),
-                        ..Default::default()
-                    };
+                    // We are using alloy for building and submitting the transactions and it will
+                    // automatically fill in all of the missing fields from the provider that we
+                    // are using.
+                    let code = alloy::hex::decode(&code)?;
+                    let tx = TransactionRequest::default()
+                        .nonce(nonce)
+                        .from(input.caller)
+                        .with_deploy_code(code);
 
                     let receipt = match node.execute_transaction(tx) {
                         Ok(receipt) => receipt,

crates/format/src/input.rs

@@ -3,8 +3,9 @@ use std::collections::HashMap;
 use alloy::{
     hex,
     json_abi::{Function, JsonAbi},
-    primitives::{Address, Bytes, TxKind},
-    rpc::types::{TransactionInput, TransactionRequest},
+    network::TransactionBuilder,
+    primitives::{Address, Bytes},
+    rpc::types::TransactionRequest,
 };
 use alloy_primitives::U256;
 use alloy_sol_types::SolValue;
@@ -220,30 +221,18 @@ impl Input {
     /// Parse this input into a legacy transaction.
     pub fn legacy_transaction(
         &self,
-        chain_id: u64,
         nonce: u64,
         deployed_contracts: &HashMap<String, Address>,
         deployed_abis: &HashMap<String, JsonAbi>,
     ) -> anyhow::Result<TransactionRequest> {
-        let to = match self.method {
-            Method::Deployer => Some(TxKind::Create),
-            _ => Some(TxKind::Call(
-                self.instance_to_address(&self.instance, deployed_contracts)?,
-            )),
-        };
-
         let input_data = self.encoded_input(deployed_abis, deployed_contracts)?;
-
-        Ok(TransactionRequest {
-            from: Some(self.caller),
-            to,
-            nonce: Some(nonce),
-            chain_id: Some(chain_id),
-            gas_price: Some(5_000_000),
-            gas: Some(5_000_000),
-            input: TransactionInput::new(input_data),
-            ..Default::default()
-        })
+        let transaction_request = TransactionRequest::default().nonce(nonce);
+        match self.method {
+            Method::Deployer => Ok(transaction_request.with_deploy_code(input_data)),
+            _ => Ok(transaction_request
+                .to(self.instance_to_address(&self.instance, deployed_contracts)?)
+                .input(input_data.into())),
+        }
     }
 }
 

crates/node-interaction/Cargo.toml

@@ -11,6 +11,7 @@ rust-version.workspace = true
 [dependencies]
 alloy = { workspace = true }
 anyhow = { workspace = true }
+futures = { workspace = true }
 tracing = { workspace = true }
 once_cell = { workspace = true }
 tokio = { workspace = true }

crates/node-interaction/src/blocking_executor.rs

@@ -0,0 +1,221 @@
+//! The alloy crate __requires__ a tokio runtime.
+//! We contain any async rust right here.
+
+use std::{any::Any, panic::AssertUnwindSafe, pin::Pin, thread};
+
+use futures::FutureExt;
+use once_cell::sync::Lazy;
+use tokio::{
+    runtime::Builder,
+    sync::{mpsc::UnboundedSender, oneshot},
+};
+
+/// A blocking async executor.
+///
+/// This struct exposes the abstraction of a blocking async executor. It is a global and static
+/// executor which means that it doesn't require for new instances of it to be created, it's a
+/// singleton and can be accessed by any thread that wants to perform some async computation on the
+/// blocking executor thread.
+///
+/// The API of the blocking executor is created in a way so that it's very natural, simple to use,
+/// and unbounded to specific tasks or return types. The following is an example of using this
+/// executor to drive an async computation:
+///
+/// ```rust
+/// use revive_dt_node_interaction::*;
+///
+/// fn blocking_function() {
+///     let result = BlockingExecutor::execute(async move {
+///         tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+///         0xFFu8
+///     })
+///     .expect("Computation failed");
+///
+///     assert_eq!(result, 0xFF);
+/// }
+/// ```
+///
+/// Users get to pass in their async tasks without needing to worry about putting them in a [`Box`],
+/// [`Pin`], needing to perform down-casting, or the internal channel mechanism used by the runtime.
+/// To the user, it just looks like a function that converts some async code into sync code.
+///
+/// This struct also handled panics that occur in the passed futures and converts them into errors
+/// that can be handled by the user. This is done to allow the executor to be robust.
+///
+/// Internally, the executor communicates with the tokio runtime thread through channels which carry
+/// the [`TaskMessage`] and the results of the execution.
+pub struct BlockingExecutor;
+
+impl BlockingExecutor {
+    pub fn execute<R>(future: impl Future<Output = R> + Send + 'static) -> Result<R, anyhow::Error>
+    where
+        R: Send + 'static,
+    {
+        // Note: The blocking executor is a singleton and therefore we store its state in a static
+        // so that it's assigned only once. Additionally, when we set the state of the executor we
+        // spawn the thread where the async runtime runs.
+        static STATE: Lazy<ExecutorState> = Lazy::new(|| {
+            tracing::trace!("Initializing the BlockingExecutor state");
+
+            // All communication with the tokio runtime thread happens over mspc channels where the
+            // producers here are the threads that want to run async tasks and the consumer here is
+            // the tokio runtime thread.
+            let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel::<TaskMessage>();
+
+            thread::spawn(move || {
+                let runtime = Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                    .expect("Failed to create the async runtime");
+
+                runtime.block_on(async move {
+                    while let Some(TaskMessage {
+                        future: task,
+                        response_tx: response_channel,
+                    }) = rx.recv().await
+                    {
+                        tracing::trace!("Received a new future to execute");
+                        tokio::spawn(async move {
+                            // One of the things that the blocking executor does is that it allows
+                            // us to catch panics if they occur. By wrapping the given future in an
+                            // AssertUnwindSafe::catch_unwind we are able to catch all panic unwinds
+                            // in the given future and convert them into errors.
+                            let task = AssertUnwindSafe(task).catch_unwind();
+
+                            let result = task.await;
+                            let _ = response_channel.send(result);
+                        });
+                    }
+                })
+            });
+
+            ExecutorState { tx }
+        });
+
+        // We need to perform blocking synchronous communication between the current thread and the
+        // tokio runtime thread with the result of the async computation and the oneshot channels
+        // from tokio allows us to do that. The sender side of the channel will be given to the
+        // tokio runtime thread to send the result when the computation is completed and the receive
+        // side of the channel will be kept with this thread to await for the response of the async
+        // task to come back.
+        let (response_tx, response_rx) =
+            oneshot::channel::<Result<Box<dyn Any + Send>, Box<dyn Any + Send>>>();
+
+        // The tokio runtime thread expects a Future<Output = Box<dyn Any + Send>> + Send to be
+        // sent to it to execute. However, this function has a typed Future<Output = R> + Send and
+        // therefore we need to change the type of the future to fit what the runtime thread expects
+        // in the task message. In doing this conversion, we lose some of the type information since
+        // we're converting R => dyn Any. However, we will perform down-casting on the result to
+        // convert it back into R.
+        let future = Box::pin(async move { Box::new(future.await) as Box<dyn Any + Send> });
+
+        let task = TaskMessage::new(future, response_tx);
+        if let Err(error) = STATE.tx.send(task) {
+            tracing::error!(?error, "Failed to send the task to the blocking executor");
+            anyhow::bail!("Failed to send the task to the blocking executor: {error:?}")
+        }
+
+        let result = match response_rx.blocking_recv() {
+            Ok(result) => result,
+            Err(error) => {
+                tracing::error!(
+                    ?error,
+                    "Failed to get the response from the blocking executor"
+                );
+                anyhow::bail!("Failed to get the response from the blocking executor: {error:?}")
+            }
+        };
+
+        match result.map(|result| {
+            *result
+                .downcast::<R>()
+                .expect("Type mismatch in the downcast")
+        }) {
+            Ok(result) => Ok(result),
+            Err(error) => {
+                tracing::error!(
+                    ?error,
+                    "Failed to downcast the returned result into the expected type"
+                );
+                anyhow::bail!(
+                    "Failed to downcast the returned result into the expected type: {error:?}"
+                )
+            }
+        }
+    }
+}
+
+/// Represents the state of the async runtime. This runtime is designed to be a singleton runtime
+/// which means that in the current running program there's just a single thread that has an async
+/// runtime.
+struct ExecutorState {
+    /// The sending side of the task messages channel. This is used by all of the other threads to
+    /// communicate with the async runtime thread.
+    tx: UnboundedSender<TaskMessage>,
+}
+
+/// Represents a message that contains an asynchronous task that's to be executed by the runtime
+/// as well as a way for the runtime to report back on the result of the execution.
+struct TaskMessage {
+    /// The task that's being requested to run. This is a future that returns an object that does
+    /// implement [`Any`] and [`Send`] to allow it to be sent between the requesting thread and the
+    /// async thread.
+    future: Pin<Box<dyn Future<Output = Box<dyn Any + Send>> + Send>>,
+
+    /// A one shot sender channel where the sender of the task is expecting to hear back on the
+    /// result of the task.
+    response_tx: oneshot::Sender<Result<Box<dyn Any + Send>, Box<dyn Any + Send>>>,
+}
+
+impl TaskMessage {
+    pub fn new(
+        future: Pin<Box<dyn Future<Output = Box<dyn Any + Send>> + Send>>,
+        response_tx: oneshot::Sender<Result<Box<dyn Any + Send>, Box<dyn Any + Send>>>,
+    ) -> Self {
+        Self {
+            future,
+            response_tx,
+        }
+    }
+}
+
+#[cfg(test)]
+mod test {
+    use super::*;
+
+    #[test]
+    fn simple_future_works() {
+        // Act
+        let result = BlockingExecutor::execute(async move {
+            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+            0xFFu8
+        })
+        .unwrap();
+
+        // Assert
+        assert_eq!(result, 0xFFu8);
+    }
+
+    #[test]
+    #[allow(unreachable_code, clippy::unreachable)]
+    fn panics_in_futures_are_caught() {
+        // Act
+        let result = BlockingExecutor::execute(async move {
+            panic!("This is a panic!");
+            0xFFu8
+        });
+
+        // Assert
+        assert!(result.is_err());
+
+        // Act
+        let result = BlockingExecutor::execute(async move {
+            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+            0xFFu8
+        })
+        .unwrap();
+
+        // Assert
+        assert_eq!(result, 0xFFu8)
+    }
+}

crates/node-interaction/src/lib.rs

@@ -3,12 +3,9 @@
 use alloy::primitives::Address;
 use alloy::rpc::types::trace::geth::{DiffMode, GethTrace};
 use alloy::rpc::types::{TransactionReceipt, TransactionRequest};
-use tokio_runtime::TO_TOKIO;
 
-pub mod nonce;
-mod tokio_runtime;
-pub mod trace;
-pub mod transaction;
+mod blocking_executor;
+pub use blocking_executor::*;
 
 /// An interface for all interactions with Ethereum compatible nodes.
 pub trait EthereumNode {

crates/node-interaction/src/nonce.rs

@@ -1,55 +0,0 @@
-use std::pin::Pin;
-
-use alloy::{
-    primitives::Address,
-    providers::{Provider, ProviderBuilder},
-};
-use tokio::sync::oneshot;
-
-use crate::{TO_TOKIO, tokio_runtime::AsyncNodeInteraction};
-
-pub type Task = Pin<Box<dyn Future<Output = anyhow::Result<u64>> + Send>>;
-
-pub(crate) struct Nonce {
-    sender: oneshot::Sender<anyhow::Result<u64>>,
-    task: Task,
-}
-
-impl AsyncNodeInteraction for Nonce {
-    type Output = anyhow::Result<u64>;
-
-    fn split(
-        self,
-    ) -> (
-        std::pin::Pin<Box<dyn Future<Output = Self::Output> + Send>>,
-        oneshot::Sender<Self::Output>,
-    ) {
-        (self.task, self.sender)
-    }
-}
-
-/// This is like `trace_transaction`, just for nonces.
-pub fn fetch_onchain_nonce(
-    connection: String,
-    wallet: alloy::network::EthereumWallet,
-    address: Address,
-) -> anyhow::Result<u64> {
-    let sender = TO_TOKIO.lock().unwrap().nonce_sender.clone();
-
-    let (tx, rx) = oneshot::channel();
-    let task: Task = Box::pin(async move {
-        let provider = ProviderBuilder::new()
-            .wallet(wallet)
-            .connect(&connection)
-            .await?;
-        let onchain = provider.get_transaction_count(address).await?;
-        Ok(onchain)
-    });
-
-    sender
-        .blocking_send(Nonce { task, sender: tx })
-        .expect("not in async context");
-
-    rx.blocking_recv()
-        .unwrap_or_else(|err| anyhow::bail!("nonce fetch failed: {err}"))
-}

crates/node-interaction/src/tokio_runtime.rs

@@ -1,87 +0,0 @@
-//! The alloy crate __requires__ a tokio runtime.
-//! We contain any async rust right here.
-
-use once_cell::sync::Lazy;
-use std::pin::Pin;
-use std::sync::Mutex;
-use std::thread;
-use tokio::runtime::Runtime;
-use tokio::spawn;
-use tokio::sync::{mpsc, oneshot};
-use tokio::task::JoinError;
-
-use crate::nonce::Nonce;
-use crate::trace::Trace;
-use crate::transaction::Transaction;
-
-pub(crate) static TO_TOKIO: Lazy<Mutex<TokioRuntime>> =
-    Lazy::new(|| Mutex::new(TokioRuntime::spawn()));
-
-/// Common interface for executing async node interactions from a non-async context.
-#[allow(clippy::type_complexity)]
-pub(crate) trait AsyncNodeInteraction: Send + 'static {
-    type Output: Send;
-
-    //// Returns the task and the output sender.
-    fn split(
-        self,
-    ) -> (
-        Pin<Box<dyn Future<Output = Self::Output> + Send>>,
-        oneshot::Sender<Self::Output>,
-    );
-}
-
-pub(crate) struct TokioRuntime {
-    pub(crate) transaction_sender: mpsc::Sender<Transaction>,
-    pub(crate) trace_sender: mpsc::Sender<Trace>,
-    pub(crate) nonce_sender: mpsc::Sender<Nonce>,
-}
-
-impl TokioRuntime {
-    fn spawn() -> Self {
-        let rt = Runtime::new().expect("should be able to create the tokio runtime");
-        let (transaction_sender, transaction_receiver) = mpsc::channel::<Transaction>(1024);
-        let (trace_sender, trace_receiver) = mpsc::channel::<Trace>(1024);
-        let (nonce_sender, nonce_receiver) = mpsc::channel::<Nonce>(1024);
-
-        thread::spawn(move || {
-            rt.block_on(async move {
-                let transaction_task = spawn(interaction::<Transaction>(transaction_receiver));
-                let trace_task = spawn(interaction::<Trace>(trace_receiver));
-                let nonce_task = spawn(interaction::<Nonce>(nonce_receiver));
-
-                if let Err(error) = transaction_task.await {
-                    tracing::error!("tokio transaction task failed: {error}");
-                }
-                if let Err(error) = trace_task.await {
-                    tracing::error!("tokio trace transaction task failed: {error}");
-                }
-                if let Err(error) = nonce_task.await {
-                    tracing::error!("tokio nonce task failed: {error}");
-                }
-            });
-        });
-
-        Self {
-            transaction_sender,
-            trace_sender,
-            nonce_sender,
-        }
-    }
-}
-
-async fn interaction<T>(mut receiver: mpsc::Receiver<T>) -> Result<(), JoinError>
-where
-    T: AsyncNodeInteraction,
-{
-    while let Some(task) = receiver.recv().await {
-        spawn(async move {
-            let (task, sender) = task.split();
-            sender
-                .send(task.await)
-                .unwrap_or_else(|_| panic!("failed to send task output"));
-        });
-    }
-
-    Ok(())
-}

crates/node-interaction/src/trace.rs

@@ -1,43 +0,0 @@
-//! Trace transactions in a sync context.
-
-use std::pin::Pin;
-
-use alloy::rpc::types::trace::geth::GethTrace;
-use tokio::sync::oneshot;
-
-use crate::TO_TOKIO;
-use crate::tokio_runtime::AsyncNodeInteraction;
-
-pub type Task = Pin<Box<dyn Future<Output = anyhow::Result<GethTrace>> + Send>>;
-
-pub(crate) struct Trace {
-    sender: oneshot::Sender<anyhow::Result<GethTrace>>,
-    task: Task,
-}
-
-impl AsyncNodeInteraction for Trace {
-    type Output = anyhow::Result<GethTrace>;
-
-    fn split(
-        self,
-    ) -> (
-        std::pin::Pin<Box<dyn Future<Output = Self::Output> + Send>>,
-        oneshot::Sender<Self::Output>,
-    ) {
-        (self.task, self.sender)
-    }
-}
-
-/// Execute some [Task] that return a [GethTrace] result.
-pub fn trace_transaction(task: Task) -> anyhow::Result<GethTrace> {
-    let task_sender = TO_TOKIO.lock().unwrap().trace_sender.clone();
-    let (sender, receiver) = oneshot::channel();
-
-    task_sender
-        .blocking_send(Trace { task, sender })
-        .expect("we are not calling this from an async context");
-
-    receiver
-        .blocking_recv()
-        .unwrap_or_else(|error| anyhow::bail!("no trace received: {error}"))
-}

crates/node-interaction/src/transaction.rs

@@ -1,46 +0,0 @@
-//! Execute transactions in a sync context.
-
-use std::pin::Pin;
-
-use alloy::rpc::types::TransactionReceipt;
-use tokio::sync::oneshot;
-
-use crate::TO_TOKIO;
-use crate::tokio_runtime::AsyncNodeInteraction;
-
-pub type Task = Pin<Box<dyn Future<Output = anyhow::Result<TransactionReceipt>> + Send>>;
-
-pub(crate) struct Transaction {
-    receipt_sender: oneshot::Sender<anyhow::Result<TransactionReceipt>>,
-    task: Task,
-}
-
-impl AsyncNodeInteraction for Transaction {
-    type Output = anyhow::Result<TransactionReceipt>;
-
-    fn split(
-        self,
-    ) -> (
-        Pin<Box<dyn Future<Output = Self::Output> + Send>>,
-        oneshot::Sender<Self::Output>,
-    ) {
-        (self.task, self.receipt_sender)
-    }
-}
-
-/// Execute some [Task] that returns a [TransactionReceipt].
-pub fn execute_transaction(task: Task) -> anyhow::Result<TransactionReceipt> {
-    let request_sender = TO_TOKIO.lock().unwrap().transaction_sender.clone();
-    let (receipt_sender, receipt_receiver) = oneshot::channel();
-
-    request_sender
-        .blocking_send(Transaction {
-            receipt_sender,
-            task,
-        })
-        .expect("we are not calling this from an async context");
-
-    receipt_receiver
-        .blocking_recv()
-        .unwrap_or_else(|error| anyhow::bail!("no receipt received: {error}"))
-}

crates/node/Cargo.toml

@@ -12,6 +12,7 @@ rust-version.workspace = true
 anyhow = { workspace = true }
 alloy = { workspace = true }
 tracing = { workspace = true }
+tokio = { workspace = true }
 
 revive-dt-node-interaction = { workspace = true }
 revive-dt-config = { workspace = true }

crates/node/src/geth.rs

@@ -2,7 +2,7 @@
 
 use std::{
     collections::HashMap,
-    fs::{File, create_dir_all, remove_dir_all},
+    fs::{File, OpenOptions, create_dir_all, remove_dir_all},
     io::{BufRead, BufReader, Read, Write},
     path::PathBuf,
     process::{Child, Command, Stdio},
@@ -10,7 +10,6 @@ use std::{
         Mutex,
         atomic::{AtomicU32, Ordering},
     },
-    thread,
     time::{Duration, Instant},
 };
 
@@ -24,10 +23,8 @@ use alloy::{
     },
 };
 use revive_dt_config::Arguments;
-use revive_dt_node_interaction::{
-    EthereumNode, nonce::fetch_onchain_nonce, trace::trace_transaction,
-    transaction::execute_transaction,
-};
+use revive_dt_node_interaction::{BlockingExecutor, EthereumNode};
+use tracing::Level;
 
 use crate::Node;
 
@@ -45,6 +42,7 @@ pub struct Instance {
     connection_string: String,
     base_directory: PathBuf,
     data_directory: PathBuf,
+    logs_directory: PathBuf,
     geth: PathBuf,
     id: u32,
     handle: Option<Child>,
@@ -52,11 +50,17 @@ pub struct Instance {
     start_timeout: u64,
     wallet: EthereumWallet,
     nonces: Mutex<HashMap<Address, u64>>,
+    /// This vector stores [`File`] objects that we use for logging which we want to flush when the
+    /// node object is dropped. We do not store them in a structured fashion at the moment (in
+    /// separate fields) as the logic that we need to apply to them is all the same regardless of
+    /// what it belongs to, we just want to flush them on [`Drop`] of the node.
+    logs_file_to_flush: Vec<File>,
 }
 
 impl Instance {
     const BASE_DIRECTORY: &str = "geth";
     const DATA_DIRECTORY: &str = "data";
+    const LOGS_DIRECTORY: &str = "logs";
 
     const IPC_FILE: &str = "geth.ipc";
     const GENESIS_JSON_FILE: &str = "genesis.json";
@@ -64,9 +68,14 @@ impl Instance {
     const READY_MARKER: &str = "IPC endpoint opened";
     const ERROR_MARKER: &str = "Fatal:";
 
+    const GETH_STDOUT_LOG_FILE_NAME: &str = "node_stdout.log";
+    const GETH_STDERR_LOG_FILE_NAME: &str = "node_stderr.log";
+
     /// Create the node directory and call `geth init` to configure the genesis.
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     fn init(&mut self, genesis: String) -> anyhow::Result<&mut Self> {
         create_dir_all(&self.base_directory)?;
+        create_dir_all(&self.logs_directory)?;
 
         let genesis_path = self.base_directory.join(Self::GENESIS_JSON_FILE);
         File::create(&genesis_path)?.write_all(genesis.as_bytes())?;
@@ -96,8 +105,24 @@ impl Instance {
 
     /// Spawn the go-ethereum node child process.
     ///
-    /// [Instance::init] must be called priorly.
+    /// [Instance::init] must be called prior.
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     fn spawn_process(&mut self) -> anyhow::Result<&mut Self> {
+        // This is the `OpenOptions` that we wish to use for all of the log files that we will be
+        // opening in this method. We need to construct it in this way to:
+        // 1. Be consistent
+        // 2. Less verbose and more dry
+        // 3. Because the builder pattern uses mutable references so we need to get around that.
+        let open_options = {
+            let mut options = OpenOptions::new();
+            options.create(true).truncate(true).write(true);
+            options
+        };
+
+        let stdout_logs_file = open_options
+            .clone()
+            .open(self.geth_stdout_log_file_path())?;
+        let stderr_logs_file = open_options.open(self.geth_stderr_log_file_path())?;
         self.handle = Command::new(&self.geth)
             .arg("--dev")
             .arg("--datadir")
@@ -109,49 +134,67 @@ impl Instance {
             .arg("--nodiscover")
             .arg("--maxpeers")
             .arg("0")
-            .stderr(Stdio::piped())
-            .stdout(Stdio::null())
+            .stderr(stderr_logs_file.try_clone()?)
+            .stdout(stdout_logs_file.try_clone()?)
             .spawn()?
             .into();
+
+        if let Err(error) = self.wait_ready() {
+            tracing::error!(?error, "Failed to start geth, shutting down gracefully");
+            self.shutdown()?;
+            return Err(error);
+        }
+
+        self.logs_file_to_flush
+            .extend([stderr_logs_file, stdout_logs_file]);
+
         Ok(self)
     }
 
     /// Wait for the g-ethereum node child process getting ready.
     ///
     /// [Instance::spawn_process] must be called priorly.
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     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);
+        let logs_file = OpenOptions::new()
+            .read(true)
+            .write(false)
+            .append(false)
+            .truncate(false)
+            .open(self.geth_stderr_log_file_path())?;
+
+        let maximum_wait_time = Duration::from_millis(self.start_timeout);
+        let mut stderr = BufReader::new(logs_file).lines();
+        loop {
+            if let Some(Ok(line)) = stderr.next() {
+                if line.contains(Self::ERROR_MARKER) {
+                    anyhow::bail!("Failed to start geth {line}");
+                }
+                if line.contains(Self::READY_MARKER) {
+                    return Ok(self);
+                }
             }
             if Instant::now().duration_since(start_time) > maximum_wait_time {
-                break "spawn timeout".to_string();
+                anyhow::bail!("Timeout in starting geth");
             }
-        };
+        }
+    }
 
-        let _ = child.kill();
-        anyhow::bail!("geth node #{} spawn error: {error}", self.id)
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id), level = Level::TRACE)]
+    fn geth_stdout_log_file_path(&self) -> PathBuf {
+        self.logs_directory.join(Self::GETH_STDOUT_LOG_FILE_NAME)
+    }
+
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id), level = Level::TRACE)]
+    fn geth_stderr_log_file_path(&self) -> PathBuf {
+        self.logs_directory.join(Self::GETH_STDERR_LOG_FILE_NAME)
     }
 }
 
 impl EthereumNode for Instance {
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     fn execute_transaction(
         &self,
         transaction: TransactionRequest,
@@ -159,20 +202,89 @@ impl EthereumNode for Instance {
         let connection_string = self.connection_string();
         let wallet = self.wallet.clone();
 
-        tracing::debug!("Submitting transaction: {transaction:#?}");
+        BlockingExecutor::execute(async move {
+            let outer_span = tracing::debug_span!("Submitting transaction", ?transaction,);
+            let _outer_guard = outer_span.enter();
 
-        execute_transaction(Box::pin(async move {
-            Ok(ProviderBuilder::new()
+            let provider = ProviderBuilder::new()
                 .wallet(wallet)
                 .connect(&connection_string)
-                .await?
-                .send_transaction(transaction)
-                .await?
-                .get_receipt()
-                .await?)
-        }))
+                .await?;
+
+            let pending_transaction = provider.send_transaction(transaction).await?;
+            let transaction_hash = pending_transaction.tx_hash();
+
+            let span = tracing::info_span!("Awaiting transaction receipt", ?transaction_hash);
+            let _guard = span.enter();
+
+            // The following is a fix for the "transaction indexing is in progress" error that we
+            // used to get. You can find more information on this in the following GH issue in geth
+            // https://github.com/ethereum/go-ethereum/issues/28877. To summarize what's going on,
+            // before we can get the receipt of the transaction it needs to have been indexed by the
+            // node's indexer. Just because the transaction has been confirmed it doesn't mean that
+            // it has been indexed. When we call alloy's `get_receipt` it checks if the transaction
+            // was confirmed. If it has been, then it will call `eth_getTransactionReceipt` method
+            // which _might_ return the above error if the tx has not yet been indexed yet. So, we
+            // need to implement a retry mechanism for the receipt to keep retrying to get it until
+            // it eventually works, but we only do that if the error we get back is the "transaction
+            // indexing is in progress" error or if the receipt is None.
+            //
+            // At the moment we do not allow for the 60 seconds to be modified and we take it as
+            // being an implementation detail that's invisible to anything outside of this module.
+            //
+            // We allow a total of 60 retries for getting the receipt with one second between each
+            // retry and the next which means that we allow for a total of 60 seconds of waiting
+            // before we consider that we're unable to get the transaction receipt.
+            let mut retries = 0;
+            loop {
+                match provider.get_transaction_receipt(*transaction_hash).await {
+                    Ok(Some(receipt)) => {
+                        tracing::info!("Obtained the transaction receipt");
+                        break Ok(receipt);
+                    }
+                    Ok(None) => {
+                        if retries == 60 {
+                            tracing::error!(
+                                "Polled for transaction receipt for 60 seconds but failed to get it"
+                            );
+                            break Err(anyhow::anyhow!("Failed to get the transaction receipt"));
+                        } else {
+                            tracing::trace!(
+                                retries,
+                                "Sleeping for 1 second and trying to get the receipt again"
+                            );
+                            retries += 1;
+                            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                            continue;
+                        }
+                    }
+                    Err(error) => {
+                        let error_string = error.to_string();
+                        if error_string.contains("transaction indexing is in progress") {
+                            if retries == 60 {
+                                tracing::error!(
+                                    "Polled for transaction receipt for 60 seconds but failed to get it"
+                                );
+                                break Err(error.into());
+                            } else {
+                                tracing::trace!(
+                                    retries,
+                                    "Sleeping for 1 second and trying to get the receipt again"
+                                );
+                                retries += 1;
+                                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
+                                continue;
+                            }
+                        } else {
+                            break Err(error.into());
+                        }
+                    }
+                }
+            }
+        })?
     }
 
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     fn trace_transaction(
         &self,
         transaction: TransactionReceipt,
@@ -185,16 +297,17 @@ impl EthereumNode for Instance {
         });
         let wallet = self.wallet.clone();
 
-        trace_transaction(Box::pin(async move {
+        BlockingExecutor::execute(async move {
             Ok(ProviderBuilder::new()
                 .wallet(wallet)
                 .connect(&connection_string)
                 .await?
                 .debug_trace_transaction(transaction.transaction_hash, trace_options)
                 .await?)
-        }))
+        })?
     }
 
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     fn state_diff(
         &self,
         transaction: alloy::rpc::types::TransactionReceipt,
@@ -208,11 +321,20 @@ impl EthereumNode for Instance {
         }
     }
 
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     fn fetch_add_nonce(&self, address: Address) -> anyhow::Result<u64> {
         let connection_string = self.connection_string.clone();
         let wallet = self.wallet.clone();
 
-        let onchain_nonce = fetch_onchain_nonce(connection_string, wallet, address)?;
+        let onchain_nonce = BlockingExecutor::execute::<anyhow::Result<_>>(async move {
+            ProviderBuilder::new()
+                .wallet(wallet)
+                .connect(&connection_string)
+                .await?
+                .get_transaction_count(address)
+                .await
+                .map_err(Into::into)
+        })??;
 
         let mut nonces = self.nonces.lock().unwrap();
         let current = nonces.entry(address).or_insert(onchain_nonce);
@@ -231,6 +353,7 @@ impl Node for Instance {
         Self {
             connection_string: base_directory.join(Self::IPC_FILE).display().to_string(),
             data_directory: base_directory.join(Self::DATA_DIRECTORY),
+            logs_directory: base_directory.join(Self::LOGS_DIRECTORY),
             base_directory,
             geth: config.geth.clone(),
             id,
@@ -239,22 +362,46 @@ impl Node for Instance {
             start_timeout: config.geth_start_timeout,
             wallet: config.wallet(),
             nonces: Mutex::new(HashMap::new()),
+            // We know that we only need to be storing 2 files so we can specify that when creating
+            // the vector. It's the stdout and stderr of the geth node.
+            logs_file_to_flush: Vec::with_capacity(2),
         }
     }
 
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     fn connection_string(&self) -> String {
         self.connection_string.clone()
     }
 
-    fn shutdown(self) -> anyhow::Result<()> {
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
+    fn shutdown(&mut self) -> anyhow::Result<()> {
+        // Terminate the processes in a graceful manner to allow for the output to be flushed.
+        if let Some(mut child) = self.handle.take() {
+            child
+                .kill()
+                .map_err(|error| anyhow::anyhow!("Failed to kill the geth process: {error:?}"))?;
+        }
+
+        // Flushing the files that we're using for keeping the logs before shutdown.
+        for file in self.logs_file_to_flush.iter_mut() {
+            file.flush()?
+        }
+
+        // Remove the node's database so that subsequent runs do not run on the same database. We
+        // ignore the error just in case the directory didn't exist in the first place and therefore
+        // there's nothing to be deleted.
+        let _ = remove_dir_all(self.base_directory.join(Self::DATA_DIRECTORY));
+
         Ok(())
     }
 
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     fn spawn(&mut self, genesis: String) -> anyhow::Result<()> {
-        self.init(genesis)?.spawn_process()?.wait_ready()?;
+        self.init(genesis)?.spawn_process()?;
         Ok(())
     }
 
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     fn version(&self) -> anyhow::Result<String> {
         let output = Command::new(&self.geth)
             .arg("--version")
@@ -269,13 +416,9 @@ impl Node for Instance {
 }
 
 impl Drop for Instance {
+    #[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
     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);
-        }
+        self.shutdown().expect("Failed to shutdown")
     }
 }
 

crates/node/src/kitchensink.rs

@@ -1,8 +1,8 @@
 use std::{
     collections::HashMap,
-    fs::create_dir_all,
-    io::BufRead,
-    path::PathBuf,
+    fs::{File, OpenOptions, create_dir_all, remove_dir_all},
+    io::{BufRead, Write},
+    path::{Path, PathBuf},
     process::{Child, Command, Stdio},
     sync::{
         Mutex,
@@ -24,12 +24,10 @@ use alloy::{
 use serde_json::{Value as JsonValue, json};
 use sp_core::crypto::Ss58Codec;
 use sp_runtime::AccountId32;
+use tracing::Level;
 
 use revive_dt_config::Arguments;
-use revive_dt_node_interaction::{
-    EthereumNode, nonce::fetch_onchain_nonce, trace::trace_transaction,
-    transaction::execute_transaction,
-};
+use revive_dt_node_interaction::{BlockingExecutor, EthereumNode};
 
 use crate::Node;
 
@@ -43,13 +41,22 @@ pub struct KitchensinkNode {
     rpc_url: String,
     wallet: EthereumWallet,
     base_directory: PathBuf,
+    logs_directory: PathBuf,
     process_substrate: Option<Child>,
     process_proxy: Option<Child>,
     nonces: Mutex<HashMap<Address, u64>>,
+    /// This vector stores [`File`] objects that we use for logging which we want to flush when the
+    /// node object is dropped. We do not store them in a structured fashion at the moment (in
+    /// separate fields) as the logic that we need to apply to them is all the same regardless of
+    /// what it belongs to, we just want to flush them on [`Drop`] of the node.
+    logs_file_to_flush: Vec<File>,
 }
 
 impl KitchensinkNode {
     const BASE_DIRECTORY: &str = "kitchensink";
+    const LOGS_DIRECTORY: &str = "logs";
+    const DATA_DIRECTORY: &str = "chains";
+
     const SUBSTRATE_READY_MARKER: &str = "Running JSON-RPC server";
     const ETH_PROXY_READY_MARKER: &str = "Running JSON-RPC server";
     const CHAIN_SPEC_JSON_FILE: &str = "template_chainspec.json";
@@ -59,11 +66,21 @@ impl KitchensinkNode {
     const SUBSTRATE_LOG_ENV: &str = "error,evm=debug,sc_rpc_server=info,runtime::revive=debug";
     const PROXY_LOG_ENV: &str = "info,eth-rpc=debug";
 
+    const KITCHENSINK_STDOUT_LOG_FILE_NAME: &str = "node_stdout.log";
+    const KITCHENSINK_STDERR_LOG_FILE_NAME: &str = "node_stderr.log";
+
+    const PROXY_STDOUT_LOG_FILE_NAME: &str = "proxy_stdout.log";
+    const PROXY_STDERR_LOG_FILE_NAME: &str = "proxy_stderr.log";
+
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn init(&mut self, genesis: &str) -> anyhow::Result<&mut Self> {
         create_dir_all(&self.base_directory)?;
+        create_dir_all(&self.logs_directory)?;
 
         let template_chainspec_path = self.base_directory.join(Self::CHAIN_SPEC_JSON_FILE);
 
+        // Note: we do not pipe the logs of this process to a separate file since this is just a
+        // once-off export of the default chain spec and not part of the long-running node process.
         let output = Command::new(&self.substrate_binary)
             .arg("export-chain-spec")
             .arg("--chain")
@@ -112,6 +129,7 @@ impl KitchensinkNode {
         Ok(self)
     }
 
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn spawn_process(&mut self) -> anyhow::Result<()> {
         let substrate_rpc_port = Self::BASE_SUBSTRATE_RPC_PORT + self.id as u16;
         let proxy_rpc_port = Self::BASE_PROXY_RPC_PORT + self.id as u16;
@@ -120,8 +138,25 @@ impl KitchensinkNode {
 
         let chainspec_path = self.base_directory.join(Self::CHAIN_SPEC_JSON_FILE);
 
+        // This is the `OpenOptions` that we wish to use for all of the log files that we will be
+        // opening in this method. We need to construct it in this way to:
+        // 1. Be consistent
+        // 2. Less verbose and more dry
+        // 3. Because the builder pattern uses mutable references so we need to get around that.
+        let open_options = {
+            let mut options = OpenOptions::new();
+            options.create(true).truncate(true).write(true);
+            options
+        };
+
         // Start Substrate node
-        let mut substrate_process = Command::new(&self.substrate_binary)
+        let kitchensink_stdout_logs_file = open_options
+            .clone()
+            .open(self.kitchensink_stdout_log_file_path())?;
+        let kitchensink_stderr_logs_file = open_options
+            .clone()
+            .open(self.kitchensink_stderr_log_file_path())?;
+        self.process_substrate = Command::new(&self.substrate_binary)
             .arg("--chain")
             .arg(chainspec_path)
             .arg("--base-path")
@@ -136,40 +171,61 @@ impl KitchensinkNode {
             .arg("--rpc-cors")
             .arg("all")
             .env("RUST_LOG", Self::SUBSTRATE_LOG_ENV)
-            .stdout(Stdio::null())
-            .stderr(Stdio::piped())
-            .spawn()?;
+            .stdout(kitchensink_stdout_logs_file.try_clone()?)
+            .stderr(kitchensink_stderr_logs_file.try_clone()?)
+            .spawn()?
+            .into();
 
         // Give the node a moment to boot
-        Self::wait_ready(
-            &mut substrate_process,
+        if let Err(error) = Self::wait_ready(
+            self.kitchensink_stderr_log_file_path().as_path(),
             Self::SUBSTRATE_READY_MARKER,
             Duration::from_secs(30),
-        )?;
+        ) {
+            tracing::error!(
+                ?error,
+                "Failed to start substrate, shutting down gracefully"
+            );
+            self.shutdown()?;
+            return Err(error);
+        };
 
-        let mut proxy_process = Command::new(&self.eth_proxy_binary)
+        let eth_proxy_stdout_logs_file = open_options
+            .clone()
+            .open(self.proxy_stdout_log_file_path())?;
+        let eth_proxy_stderr_logs_file = open_options.open(self.proxy_stderr_log_file_path())?;
+        self.process_proxy = Command::new(&self.eth_proxy_binary)
             .arg("--dev")
             .arg("--rpc-port")
             .arg(proxy_rpc_port.to_string())
             .arg("--node-rpc-url")
             .arg(format!("ws://127.0.0.1:{substrate_rpc_port}"))
             .env("RUST_LOG", Self::PROXY_LOG_ENV)
-            .stdout(Stdio::null())
-            .stderr(Stdio::piped())
-            .spawn()?;
+            .stdout(eth_proxy_stdout_logs_file.try_clone()?)
+            .stderr(eth_proxy_stderr_logs_file.try_clone()?)
+            .spawn()?
+            .into();
 
-        Self::wait_ready(
-            &mut proxy_process,
+        if let Err(error) = Self::wait_ready(
+            self.proxy_stderr_log_file_path().as_path(),
             Self::ETH_PROXY_READY_MARKER,
             Duration::from_secs(30),
-        )?;
+        ) {
+            tracing::error!(?error, "Failed to start proxy, shutting down gracefully");
+            self.shutdown()?;
+            return Err(error);
+        };
 
-        self.process_substrate = Some(substrate_process);
-        self.process_proxy = Some(proxy_process);
+        self.logs_file_to_flush.extend([
+            kitchensink_stdout_logs_file,
+            kitchensink_stderr_logs_file,
+            eth_proxy_stdout_logs_file,
+            eth_proxy_stderr_logs_file,
+        ]);
 
         Ok(())
     }
-
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn extract_balance_from_genesis_file(
         &self,
         genesis_str: &str,
@@ -210,27 +266,30 @@ impl KitchensinkNode {
         Ok(account_id.to_ss58check())
     }
 
-    fn wait_ready(child: &mut Child, marker: &str, timeout: Duration) -> anyhow::Result<()> {
+    fn wait_ready(logs_file_path: &Path, marker: &str, timeout: Duration) -> anyhow::Result<()> {
         let start_time = std::time::Instant::now();
-        let stderr = child.stderr.take().expect("stderr must be piped");
+        let logs_file = OpenOptions::new()
+            .read(true)
+            .write(false)
+            .append(false)
+            .truncate(false)
+            .open(logs_file_path)?;
 
-        let mut lines = std::io::BufReader::new(stderr).lines();
+        let mut lines = std::io::BufReader::new(logs_file).lines();
         loop {
             if let Some(Ok(line)) = lines.next() {
-                println!("Kitchensink log: {line:?}");
                 if line.contains(marker) {
-                    std::thread::spawn(move || for _ in lines.by_ref() {});
                     return Ok(());
                 }
             }
 
             if start_time.elapsed() > timeout {
-                let _ = child.kill();
                 anyhow::bail!("Timeout waiting for process readiness: {marker}");
             }
         }
     }
 
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     pub fn eth_rpc_version(&self) -> anyhow::Result<String> {
         let output = Command::new(&self.eth_proxy_binary)
             .arg("--version")
@@ -242,9 +301,32 @@ impl KitchensinkNode {
             .stdout;
         Ok(String::from_utf8_lossy(&output).trim().to_string())
     }
+
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id), level = Level::TRACE)]
+    fn kitchensink_stdout_log_file_path(&self) -> PathBuf {
+        self.logs_directory
+            .join(Self::KITCHENSINK_STDOUT_LOG_FILE_NAME)
+    }
+
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id), level = Level::TRACE)]
+    fn kitchensink_stderr_log_file_path(&self) -> PathBuf {
+        self.logs_directory
+            .join(Self::KITCHENSINK_STDERR_LOG_FILE_NAME)
+    }
+
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id), level = Level::TRACE)]
+    fn proxy_stdout_log_file_path(&self) -> PathBuf {
+        self.logs_directory.join(Self::PROXY_STDOUT_LOG_FILE_NAME)
+    }
+
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id), level = Level::TRACE)]
+    fn proxy_stderr_log_file_path(&self) -> PathBuf {
+        self.logs_directory.join(Self::PROXY_STDERR_LOG_FILE_NAME)
+    }
 }
 
 impl EthereumNode for KitchensinkNode {
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn execute_transaction(
         &self,
         transaction: alloy::rpc::types::TransactionRequest,
@@ -254,7 +336,8 @@ impl EthereumNode for KitchensinkNode {
 
         tracing::debug!("Submitting transaction: {transaction:#?}");
 
-        execute_transaction(Box::pin(async move {
+        tracing::info!("Submitting tx to kitchensink");
+        let receipt = BlockingExecutor::execute(async move {
             Ok(ProviderBuilder::new()
                 .wallet(wallet)
                 .connect(&url)
@@ -263,9 +346,12 @@ impl EthereumNode for KitchensinkNode {
                 .await?
                 .get_receipt()
                 .await?)
-        }))
+        })?;
+        tracing::info!(?receipt, "Submitted tx to kitchensink");
+        receipt
     }
 
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn trace_transaction(
         &self,
         transaction: TransactionReceipt,
@@ -279,16 +365,17 @@ impl EthereumNode for KitchensinkNode {
 
         let wallet = self.wallet.clone();
 
-        trace_transaction(Box::pin(async move {
+        BlockingExecutor::execute(async move {
             Ok(ProviderBuilder::new()
                 .wallet(wallet)
                 .connect(&url)
                 .await?
                 .debug_trace_transaction(transaction.transaction_hash, trace_options)
                 .await?)
-        }))
+        })?
     }
 
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn state_diff(&self, transaction: TransactionReceipt) -> anyhow::Result<DiffMode> {
         match self
             .trace_transaction(transaction)?
@@ -299,11 +386,20 @@ impl EthereumNode for KitchensinkNode {
         }
     }
 
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn fetch_add_nonce(&self, address: Address) -> anyhow::Result<u64> {
         let url = self.rpc_url.clone();
         let wallet = self.wallet.clone();
 
-        let onchain_nonce = fetch_onchain_nonce(url, wallet, address)?;
+        let onchain_nonce = BlockingExecutor::execute::<anyhow::Result<_>>(async move {
+            ProviderBuilder::new()
+                .wallet(wallet)
+                .connect(&url)
+                .await?
+                .get_transaction_count(address)
+                .await
+                .map_err(Into::into)
+        })??;
 
         let mut nonces = self.nonces.lock().unwrap();
         let current = nonces.entry(address).or_insert(onchain_nonce);
@@ -318,6 +414,7 @@ impl Node for KitchensinkNode {
         let kitchensink_directory = config.directory().join(Self::BASE_DIRECTORY);
         let id = NODE_COUNT.fetch_add(1, Ordering::SeqCst);
         let base_directory = kitchensink_directory.join(id.to_string());
+        let logs_directory = base_directory.join(Self::LOGS_DIRECTORY);
 
         Self {
             id,
@@ -326,30 +423,54 @@ impl Node for KitchensinkNode {
             rpc_url: String::new(),
             wallet: config.wallet(),
             base_directory,
+            logs_directory,
             process_substrate: None,
             process_proxy: None,
             nonces: Mutex::new(HashMap::new()),
+            // We know that we only need to be storing 4 files so we can specify that when creating
+            // the vector. It's the stdout and stderr of the substrate-node and the eth-rpc.
+            logs_file_to_flush: Vec::with_capacity(4),
         }
     }
 
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn connection_string(&self) -> String {
         self.rpc_url.clone()
     }
 
-    fn shutdown(mut self) -> anyhow::Result<()> {
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
+    fn shutdown(&mut self) -> anyhow::Result<()> {
+        // Terminate the processes in a graceful manner to allow for the output to be flushed.
         if let Some(mut child) = self.process_proxy.take() {
-            let _ = child.kill();
+            child
+                .kill()
+                .map_err(|error| anyhow::anyhow!("Failed to kill the proxy process: {error:?}"))?;
         }
         if let Some(mut child) = self.process_substrate.take() {
-            let _ = child.kill();
+            child.kill().map_err(|error| {
+                anyhow::anyhow!("Failed to kill the substrate process: {error:?}")
+            })?;
         }
+
+        // Flushing the files that we're using for keeping the logs before shutdown.
+        for file in self.logs_file_to_flush.iter_mut() {
+            file.flush()?
+        }
+
+        // Remove the node's database so that subsequent runs do not run on the same database. We
+        // ignore the error just in case the directory didn't exist in the first place and therefore
+        // there's nothing to be deleted.
+        let _ = remove_dir_all(self.base_directory.join(Self::DATA_DIRECTORY));
+
         Ok(())
     }
 
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn spawn(&mut self, genesis: String) -> anyhow::Result<()> {
         self.init(&genesis)?.spawn_process()
     }
 
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn version(&self) -> anyhow::Result<String> {
         let output = Command::new(&self.substrate_binary)
             .arg("--version")
@@ -364,13 +485,9 @@ impl Node for KitchensinkNode {
 }
 
 impl Drop for KitchensinkNode {
+    #[tracing::instrument(skip_all, fields(kitchensink_node_id = self.id))]
     fn drop(&mut self) {
-        if let Some(mut child) = self.process_proxy.take() {
-            let _ = child.kill();
-        }
-        if let Some(mut child) = self.process_substrate.take() {
-            let _ = child.kill();
-        }
+        self.shutdown().expect("Failed to shutdown")
     }
 }
 

crates/node/src/lib.rs

@@ -23,7 +23,7 @@ pub trait Node: EthereumNode {
     /// Prune the node instance and related data.
     ///
     /// Blocking until it's completely stopped.
-    fn shutdown(self) -> anyhow::Result<()>;
+    fn shutdown(&mut self) -> anyhow::Result<()>;
 
     /// Returns the nodes connection string.
     fn connection_string(&self) -> String;