Compare commits

..

5 Commits

Author SHA1 Message Date
Omar Abdulla 83c20b1be3 Fix tests 2025-07-14 21:30:35 +03:00
Omar Abdulla 075c8235a7 Merge remote-tracking branch 'origin/main' into bugfix/argument-encoding 2025-07-14 20:36:42 +03:00
Omar Abdulla 43e0d0e592 Remove reliance on the web3 crate 2025-07-14 18:27:38 +03:00
Omar Abdulla 2373872230 Avoid extra buffer allocation 2025-07-14 00:02:48 +03:00
Omar Abdulla e3723e780a Fix function selector and argument encoding 2025-07-13 19:52:06 +03:00
16 changed files with 283 additions and 444 deletions
Generated
-1
View File
@@ -4012,7 +4012,6 @@ version = "0.1.0"
dependencies = [
"alloy",
"anyhow",
"futures",
"once_cell",
"tokio",
"tracing",
-1
View File
@@ -25,7 +25,6 @@ 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"
+4 -8
View File
@@ -44,8 +44,6 @@ pub trait SolidityCompiler {
pub struct CompilerInput<T: PartialEq + Eq + Hash> {
pub extra_options: T,
pub input: SolcStandardJsonInput,
pub allow_paths: Vec<PathBuf>,
pub base_path: Option<PathBuf>,
}
/// The generic compilation output configuration.
@@ -85,8 +83,8 @@ where
pub struct Compiler<T: SolidityCompiler> {
input: SolcStandardJsonInput,
extra_options: T::Options,
allow_paths: Vec<PathBuf>,
base_path: Option<PathBuf>,
allow_paths: Vec<String>,
base_path: Option<String>,
}
impl Default for Compiler<solc::Solc> {
@@ -147,12 +145,12 @@ where
self
}
pub fn allow_path(mut self, path: PathBuf) -> Self {
pub fn allow_path(mut self, path: String) -> Self {
self.allow_paths.push(path);
self
}
pub fn base_path(mut self, base_path: PathBuf) -> Self {
pub fn base_path(mut self, base_path: String) -> Self {
self.base_path = Some(base_path);
self
}
@@ -161,8 +159,6 @@ where
T::new(solc_path).build(CompilerInput {
extra_options: self.extra_options,
input: self.input,
allow_paths: self.allow_paths,
base_path: self.base_path,
})
}
+5 -28
View File
@@ -23,27 +23,13 @@ impl SolidityCompiler for Resolc {
&self,
input: CompilerInput<Self::Options>,
) -> anyhow::Result<CompilerOutput<Self::Options>> {
let mut command = Command::new(&self.resolc_path);
command
let mut child = Command::new(&self.resolc_path)
.arg("--standard-json")
.args(&input.extra_options)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.arg("--standard-json");
if let Some(ref base_path) = input.base_path {
command.arg("--base-path").arg(base_path);
}
if !input.allow_paths.is_empty() {
command.arg("--allow-paths").arg(
input
.allow_paths
.iter()
.map(|path| path.display().to_string())
.collect::<Vec<_>>()
.join(","),
);
}
let mut child = command.spawn()?;
.spawn()?;
let stdin_pipe = child.stdin.as_mut().expect("stdin must be piped");
serde_json::to_writer(stdin_pipe, &input.input)?;
@@ -69,22 +55,13 @@ impl SolidityCompiler for Resolc {
});
}
let parsed = serde_json::from_slice::<SolcStandardJsonOutput>(&stdout).map_err(|e| {
let parsed: SolcStandardJsonOutput = serde_json::from_slice(&stdout).map_err(|e| {
anyhow::anyhow!(
"failed to parse resolc JSON output: {e}\nstderr: {}",
String::from_utf8_lossy(&stderr)
)
})?;
// Detecting if the compiler output contained errors and reporting them through logs and
// errors instead of returning the compiler output that might contain errors.
for error in parsed.errors.iter().flatten() {
if error.severity == "error" {
tracing::error!(?error, ?input, "Encountered an error in the compilation");
anyhow::bail!("Encountered an error in the compilation: {error}")
}
}
Ok(CompilerOutput {
input,
output: parsed,
+4 -37
View File
@@ -9,7 +9,6 @@ use std::{
use crate::{CompilerInput, CompilerOutput, SolidityCompiler};
use revive_dt_config::Arguments;
use revive_dt_solc_binaries::download_solc;
use revive_solc_json_interface::SolcStandardJsonOutput;
pub struct Solc {
solc_path: PathBuf,
@@ -22,27 +21,12 @@ impl SolidityCompiler for Solc {
&self,
input: CompilerInput<Self::Options>,
) -> anyhow::Result<CompilerOutput<Self::Options>> {
let mut command = Command::new(&self.solc_path);
command
let mut child = Command::new(&self.solc_path)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.arg("--standard-json");
if let Some(ref base_path) = input.base_path {
command.arg("--base-path").arg(base_path);
}
if !input.allow_paths.is_empty() {
command.arg("--allow-paths").arg(
input
.allow_paths
.iter()
.map(|path| path.display().to_string())
.collect::<Vec<_>>()
.join(","),
);
}
let mut child = command.spawn()?;
.arg("--standard-json")
.spawn()?;
let stdin = child.stdin.as_mut().expect("should be piped");
serde_json::to_writer(stdin, &input.input)?;
@@ -58,26 +42,9 @@ impl SolidityCompiler for Solc {
});
}
let parsed =
serde_json::from_slice::<SolcStandardJsonOutput>(&output.stdout).map_err(|e| {
anyhow::anyhow!(
"failed to parse resolc JSON output: {e}\nstderr: {}",
String::from_utf8_lossy(&output.stdout)
)
})?;
// Detecting if the compiler output contained errors and reporting them through logs and
// errors instead of returning the compiler output that might contain errors.
for error in parsed.errors.iter().flatten() {
if error.severity == "error" {
tracing::error!(?error, ?input, "Encountered an error in the compilation");
anyhow::bail!("Encountered an error in the compilation: {error}")
}
}
Ok(CompilerOutput {
input,
output: parsed,
output: serde_json::from_slice(&output.stdout)?,
error: None,
})
}
+13 -109
View File
@@ -69,13 +69,14 @@ where
anyhow::bail!("unsupported solc version: {:?}", &mode.solc_version);
};
let compiler = Compiler::<T::Compiler>::new()
.allow_path(metadata.directory()?)
let mut compiler = Compiler::<T::Compiler>::new()
.base_path(metadata.directory()?.display().to_string())
.solc_optimizer(mode.solc_optimize());
let compiler = FilesWithExtensionIterator::new(metadata.directory()?)
.with_allowed_extension("sol")
.try_fold(compiler, |compiler, path| compiler.with_source(&path))?;
for (file, _contract) in metadata.contract_sources()?.values() {
tracing::debug!("contract source {}", file.display());
compiler = compiler.with_source(file)?;
}
let mut task = CompilationTask {
json_input: compiler.input(),
@@ -179,15 +180,12 @@ where
}
pub fn deploy_contracts(&mut self, input: &Input, node: &T::Blockchain) -> anyhow::Result<()> {
let tracing_span = tracing::debug_span!(
"Deploying contracts",
?input,
node = std::any::type_name::<T>()
tracing::debug!(
"Deploying contracts {}, having address {} on node: {}",
&input.instance,
&input.caller,
std::any::type_name::<T>()
);
let _guard = tracing_span.enter();
tracing::debug!(number_of_contracts_to_deploy = self.contracts.len());
for output in self.contracts.values() {
let Some(contract_map) = &output.contracts else {
tracing::debug!(
@@ -219,17 +217,7 @@ where
continue;
};
let nonce = match node.fetch_add_nonce(input.caller) {
Ok(nonce) => nonce,
Err(error) => {
tracing::error!(
caller = ?input.caller,
?error,
"Failed to get the nonce for the caller"
);
return Err(error);
}
};
let nonce = node.fetch_add_nonce(input.caller)?;
tracing::debug!(
"Calculated nonce {}, for contract {}, having address {} on node: {}",
@@ -242,17 +230,7 @@ where
// 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 = match alloy::hex::decode(&code) {
Ok(code) => code,
Err(error) => {
tracing::error!(
code,
?error,
"Failed to hex-decode the code of the contract. (This could possibly mean that it contains '_' and therefore it requires linking to be performed)"
);
return Err(error.into());
}
};
let code = alloy::hex::decode(&code)?;
let tx = TransactionRequest::default()
.nonce(nonce)
.from(input.caller)
@@ -498,77 +476,3 @@ where
Ok(())
}
}
/// An iterator that finds files of a certain extension in the provided directory. You can think of
/// this a glob pattern similar to: `${path}/**/*.md`
struct FilesWithExtensionIterator {
/// The set of allowed extensions that that match the requirement and that should be returned
/// when found.
allowed_extensions: std::collections::HashSet<std::borrow::Cow<'static, str>>,
/// The set of directories to visit next. This iterator does BFS and so these directories will
/// only be visited if we can't find any files in our state.
directories_to_search: Vec<std::path::PathBuf>,
/// The set of files matching the allowed extensions that were found. If there are entries in
/// this vector then they will be returned when the [`Iterator::next`] method is called. If not
/// then we visit one of the next directories to visit.
///
/// [`Iterator`]: std::iter::Iterator
files_matching_allowed_extensions: Vec<std::path::PathBuf>,
}
impl FilesWithExtensionIterator {
fn new(root_directory: std::path::PathBuf) -> Self {
Self {
allowed_extensions: Default::default(),
directories_to_search: vec![root_directory],
files_matching_allowed_extensions: Default::default(),
}
}
fn with_allowed_extension(
mut self,
allowed_extension: impl Into<std::borrow::Cow<'static, str>>,
) -> Self {
self.allowed_extensions.insert(allowed_extension.into());
self
}
}
impl Iterator for FilesWithExtensionIterator {
type Item = std::path::PathBuf;
fn next(&mut self) -> Option<Self::Item> {
if let Some(file_path) = self.files_matching_allowed_extensions.pop() {
return Some(file_path);
};
let directory_to_search = self.directories_to_search.pop()?;
// Read all of the entries in the directory. If we failed to read this dir's entires then we
// elect to just ignore it and look in the next directory, we do that by calling the next
// method again on the iterator, which is an intentional decision that we made here instead
// of panicking.
let Ok(dir_entries) = std::fs::read_dir(directory_to_search) else {
return self.next();
};
for entry in dir_entries.flatten() {
let entry_path = entry.path();
if entry_path.is_dir() {
self.directories_to_search.push(entry_path)
} else if entry_path.is_file()
&& entry_path.extension().is_some_and(|ext| {
self.allowed_extensions
.iter()
.any(|allowed| ext.eq_ignore_ascii_case(allowed.as_ref()))
})
{
self.files_matching_allowed_extensions.push(entry_path)
}
}
self.next()
}
}
+3 -8
View File
@@ -89,14 +89,9 @@ impl Input {
return Ok(Bytes::default()); // fallback or deployer — no input
};
let Some(abi) = deployed_abis.get(&self.instance) else {
tracing::error!(
contract_name = self.instance,
available_abis = ?deployed_abis.keys().collect::<Vec<_>>(),
"Attempted to lookup ABI of contract but it wasn't found"
);
anyhow::bail!("ABI for instance '{}' not found", &self.instance);
};
let abi = deployed_abis
.get(&self.instance)
.ok_or_else(|| anyhow::anyhow!("ABI for instance '{}' not found", &self.instance))?;
tracing::trace!("ABI found for instance: {}", &self.instance);
-1
View File
@@ -11,7 +11,6 @@ rust-version.workspace = true
[dependencies]
alloy = { workspace = true }
anyhow = { workspace = true }
futures = { workspace = true }
tracing = { workspace = true }
once_cell = { workspace = true }
tokio = { workspace = true }
@@ -1,221 +0,0 @@
//! 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)
}
}
+5 -2
View File
@@ -3,9 +3,12 @@
use alloy::primitives::Address;
use alloy::rpc::types::trace::geth::{DiffMode, GethTrace};
use alloy::rpc::types::{TransactionReceipt, TransactionRequest};
use tokio_runtime::TO_TOKIO;
mod blocking_executor;
pub use blocking_executor::*;
pub mod nonce;
mod tokio_runtime;
pub mod trace;
pub mod transaction;
/// An interface for all interactions with Ethereum compatible nodes.
pub trait EthereumNode {
+55
View File
@@ -0,0 +1,55 @@
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}"))
}
@@ -0,0 +1,87 @@
//! 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(())
}
+43
View File
@@ -0,0 +1,43 @@
//! 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}"))
}
@@ -0,0 +1,46 @@
//! 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}"))
}
+9 -14
View File
@@ -23,7 +23,10 @@ use alloy::{
},
};
use revive_dt_config::Arguments;
use revive_dt_node_interaction::{BlockingExecutor, EthereumNode};
use revive_dt_node_interaction::{
EthereumNode, nonce::fetch_onchain_nonce, trace::trace_transaction,
transaction::execute_transaction,
};
use tracing::Level;
use crate::Node;
@@ -202,7 +205,7 @@ impl EthereumNode for Instance {
let connection_string = self.connection_string();
let wallet = self.wallet.clone();
BlockingExecutor::execute(async move {
execute_transaction(Box::pin(async move {
let outer_span = tracing::debug_span!("Submitting transaction", ?transaction,);
let _outer_guard = outer_span.enter();
@@ -281,7 +284,7 @@ impl EthereumNode for Instance {
}
}
}
})?
}))
}
#[tracing::instrument(skip_all, fields(geth_node_id = self.id))]
@@ -297,14 +300,14 @@ impl EthereumNode for Instance {
});
let wallet = self.wallet.clone();
BlockingExecutor::execute(async move {
trace_transaction(Box::pin(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))]
@@ -326,15 +329,7 @@ impl EthereumNode for Instance {
let connection_string = self.connection_string.clone();
let wallet = self.wallet.clone();
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 onchain_nonce = fetch_onchain_nonce(connection_string, wallet, address)?;
let mut nonces = self.nonces.lock().unwrap();
let current = nonces.entry(address).or_insert(onchain_nonce);
+9 -14
View File
@@ -27,7 +27,10 @@ use sp_runtime::AccountId32;
use tracing::Level;
use revive_dt_config::Arguments;
use revive_dt_node_interaction::{BlockingExecutor, EthereumNode};
use revive_dt_node_interaction::{
EthereumNode, nonce::fetch_onchain_nonce, trace::trace_transaction,
transaction::execute_transaction,
};
use crate::Node;
@@ -337,7 +340,7 @@ impl EthereumNode for KitchensinkNode {
tracing::debug!("Submitting transaction: {transaction:#?}");
tracing::info!("Submitting tx to kitchensink");
let receipt = BlockingExecutor::execute(async move {
let receipt = execute_transaction(Box::pin(async move {
Ok(ProviderBuilder::new()
.wallet(wallet)
.connect(&url)
@@ -346,7 +349,7 @@ impl EthereumNode for KitchensinkNode {
.await?
.get_receipt()
.await?)
})?;
}));
tracing::info!(?receipt, "Submitted tx to kitchensink");
receipt
}
@@ -365,14 +368,14 @@ impl EthereumNode for KitchensinkNode {
let wallet = self.wallet.clone();
BlockingExecutor::execute(async move {
trace_transaction(Box::pin(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))]
@@ -391,15 +394,7 @@ impl EthereumNode for KitchensinkNode {
let url = self.rpc_url.clone();
let wallet = self.wallet.clone();
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 onchain_nonce = fetch_onchain_nonce(url, wallet, address)?;
let mut nonces = self.nonces.lock().unwrap();
let current = nonces.entry(address).or_insert(onchain_nonce);