mirror of
https://github.com/pezkuwichain/revive-differential-tests.git
synced 2026-04-27 20:47:58 +00:00
use polkadot-sdk rustfmt
This commit is contained in:
pgherveou
+132
-151
@@ -1,9 +1,9 @@
|
||||
use std::{
|
||||
fs::{File, OpenOptions},
|
||||
io::{BufRead, BufReader, Write},
|
||||
path::Path,
|
||||
process::{Child, Command},
|
||||
time::{Duration, Instant},
|
||||
fs::{File, OpenOptions},
|
||||
io::{BufRead, BufReader, Write},
|
||||
path::Path,
|
||||
process::{Child, Command},
|
||||
time::{Duration, Instant},
|
||||
};
|
||||
|
||||
use anyhow::{Context, Result, bail};
|
||||
@@ -12,180 +12,161 @@ use anyhow::{Context, Result, bail};
|
||||
/// when the process is dropped.
|
||||
#[derive(Debug)]
|
||||
pub struct Process {
|
||||
/// The handle of the child process.
|
||||
child: Child,
|
||||
/// The handle of the child process.
|
||||
child: Child,
|
||||
|
||||
/// The file that stdout is being logged to.
|
||||
stdout_logs_file: File,
|
||||
/// The file that stdout is being logged to.
|
||||
stdout_logs_file: File,
|
||||
|
||||
/// The file that stderr is being logged to.
|
||||
stderr_logs_file: File,
|
||||
/// The file that stderr is being logged to.
|
||||
stderr_logs_file: File,
|
||||
}
|
||||
|
||||
impl Process {
|
||||
pub fn new(
|
||||
log_file_prefix: impl Into<Option<&'static str>>,
|
||||
logs_directory: impl AsRef<Path>,
|
||||
binary_path: impl AsRef<Path>,
|
||||
command_building_callback: impl FnOnce(&mut Command, File, File),
|
||||
process_readiness_wait_behavior: ProcessReadinessWaitBehavior,
|
||||
) -> Result<Self> {
|
||||
let log_file_prefix = log_file_prefix.into();
|
||||
pub fn new(
|
||||
log_file_prefix: impl Into<Option<&'static str>>,
|
||||
logs_directory: impl AsRef<Path>,
|
||||
binary_path: impl AsRef<Path>,
|
||||
command_building_callback: impl FnOnce(&mut Command, File, File),
|
||||
process_readiness_wait_behavior: ProcessReadinessWaitBehavior,
|
||||
) -> Result<Self> {
|
||||
let log_file_prefix = log_file_prefix.into();
|
||||
|
||||
let (stdout_file_name, stderr_file_name) = match log_file_prefix {
|
||||
Some(prefix) => (
|
||||
format!("{prefix}_stdout.log"),
|
||||
format!("{prefix}_stderr.log"),
|
||||
),
|
||||
None => ("stdout.log".to_string(), "stderr.log".to_string()),
|
||||
};
|
||||
let (stdout_file_name, stderr_file_name) = match log_file_prefix {
|
||||
Some(prefix) => (format!("{prefix}_stdout.log"), format!("{prefix}_stderr.log")),
|
||||
None => ("stdout.log".to_string(), "stderr.log".to_string()),
|
||||
};
|
||||
|
||||
let stdout_logs_file_path = logs_directory.as_ref().join(stdout_file_name);
|
||||
let stderr_logs_file_path = logs_directory.as_ref().join(stderr_file_name);
|
||||
let stdout_logs_file_path = logs_directory.as_ref().join(stdout_file_name);
|
||||
let stderr_logs_file_path = logs_directory.as_ref().join(stderr_file_name);
|
||||
|
||||
let stdout_logs_file = OpenOptions::new()
|
||||
.write(true)
|
||||
.truncate(true)
|
||||
.create(true)
|
||||
.open(stdout_logs_file_path.as_path())
|
||||
.context("Failed to open the stdout logs file")?;
|
||||
let stderr_logs_file = OpenOptions::new()
|
||||
.write(true)
|
||||
.truncate(true)
|
||||
.create(true)
|
||||
.open(stderr_logs_file_path.as_path())
|
||||
.context("Failed to open the stderr logs file")?;
|
||||
let stdout_logs_file = OpenOptions::new()
|
||||
.write(true)
|
||||
.truncate(true)
|
||||
.create(true)
|
||||
.open(stdout_logs_file_path.as_path())
|
||||
.context("Failed to open the stdout logs file")?;
|
||||
let stderr_logs_file = OpenOptions::new()
|
||||
.write(true)
|
||||
.truncate(true)
|
||||
.create(true)
|
||||
.open(stderr_logs_file_path.as_path())
|
||||
.context("Failed to open the stderr logs file")?;
|
||||
|
||||
let mut command = {
|
||||
let stdout_logs_file = stdout_logs_file
|
||||
.try_clone()
|
||||
.context("Failed to clone the stdout logs file")?;
|
||||
let stderr_logs_file = stderr_logs_file
|
||||
.try_clone()
|
||||
.context("Failed to clone the stderr logs file")?;
|
||||
let mut command = {
|
||||
let stdout_logs_file =
|
||||
stdout_logs_file.try_clone().context("Failed to clone the stdout logs file")?;
|
||||
let stderr_logs_file =
|
||||
stderr_logs_file.try_clone().context("Failed to clone the stderr logs file")?;
|
||||
|
||||
let mut command = Command::new(binary_path.as_ref());
|
||||
command_building_callback(&mut command, stdout_logs_file, stderr_logs_file);
|
||||
command
|
||||
};
|
||||
let mut child = command
|
||||
.spawn()
|
||||
.context("Failed to spawn the built command")?;
|
||||
let mut command = Command::new(binary_path.as_ref());
|
||||
command_building_callback(&mut command, stdout_logs_file, stderr_logs_file);
|
||||
command
|
||||
};
|
||||
let mut child = command.spawn().context("Failed to spawn the built command")?;
|
||||
|
||||
match process_readiness_wait_behavior {
|
||||
ProcessReadinessWaitBehavior::NoStartupWait => {}
|
||||
ProcessReadinessWaitBehavior::WaitDuration(duration) => std::thread::sleep(duration),
|
||||
ProcessReadinessWaitBehavior::TimeBoundedWaitFunction {
|
||||
max_wait_duration,
|
||||
mut check_function,
|
||||
} => {
|
||||
let spawn_time = Instant::now();
|
||||
match process_readiness_wait_behavior {
|
||||
ProcessReadinessWaitBehavior::NoStartupWait => {},
|
||||
ProcessReadinessWaitBehavior::WaitDuration(duration) => std::thread::sleep(duration),
|
||||
ProcessReadinessWaitBehavior::TimeBoundedWaitFunction {
|
||||
max_wait_duration,
|
||||
mut check_function,
|
||||
} => {
|
||||
let spawn_time = Instant::now();
|
||||
|
||||
let stdout_logs_file = OpenOptions::new()
|
||||
.read(true)
|
||||
.open(stdout_logs_file_path)
|
||||
.context("Failed to open the stdout logs file")?;
|
||||
let stderr_logs_file = OpenOptions::new()
|
||||
.read(true)
|
||||
.open(stderr_logs_file_path)
|
||||
.context("Failed to open the stderr logs file")?;
|
||||
let stdout_logs_file = OpenOptions::new()
|
||||
.read(true)
|
||||
.open(stdout_logs_file_path)
|
||||
.context("Failed to open the stdout logs file")?;
|
||||
let stderr_logs_file = OpenOptions::new()
|
||||
.read(true)
|
||||
.open(stderr_logs_file_path)
|
||||
.context("Failed to open the stderr logs file")?;
|
||||
|
||||
let mut stdout_lines = BufReader::new(stdout_logs_file).lines();
|
||||
let mut stderr_lines = BufReader::new(stderr_logs_file).lines();
|
||||
let mut stdout_lines = BufReader::new(stdout_logs_file).lines();
|
||||
let mut stderr_lines = BufReader::new(stderr_logs_file).lines();
|
||||
|
||||
let mut stdout = String::new();
|
||||
let mut stderr = String::new();
|
||||
let mut stdout = String::new();
|
||||
let mut stderr = String::new();
|
||||
|
||||
loop {
|
||||
let stdout_line = stdout_lines.next().and_then(Result::ok);
|
||||
let stderr_line = stderr_lines.next().and_then(Result::ok);
|
||||
loop {
|
||||
let stdout_line = stdout_lines.next().and_then(Result::ok);
|
||||
let stderr_line = stderr_lines.next().and_then(Result::ok);
|
||||
|
||||
if let Some(stdout_line) = stdout_line.as_ref() {
|
||||
stdout.push_str(stdout_line);
|
||||
stdout.push('\n');
|
||||
}
|
||||
if let Some(stderr_line) = stderr_line.as_ref() {
|
||||
stderr.push_str(stderr_line);
|
||||
stderr.push('\n');
|
||||
}
|
||||
if let Some(stdout_line) = stdout_line.as_ref() {
|
||||
stdout.push_str(stdout_line);
|
||||
stdout.push('\n');
|
||||
}
|
||||
if let Some(stderr_line) = stderr_line.as_ref() {
|
||||
stderr.push_str(stderr_line);
|
||||
stderr.push('\n');
|
||||
}
|
||||
|
||||
let check_result =
|
||||
check_function(stdout_line.as_deref(), stderr_line.as_deref()).context(
|
||||
format!(
|
||||
"Failed to wait for the process to be ready - {stdout} - {stderr}"
|
||||
),
|
||||
)?;
|
||||
let check_result =
|
||||
check_function(stdout_line.as_deref(), stderr_line.as_deref()).context(
|
||||
format!(
|
||||
"Failed to wait for the process to be ready - {stdout} - {stderr}"
|
||||
),
|
||||
)?;
|
||||
|
||||
if check_result {
|
||||
break;
|
||||
}
|
||||
if check_result {
|
||||
break;
|
||||
}
|
||||
|
||||
if Instant::now().duration_since(spawn_time) > max_wait_duration {
|
||||
bail!(
|
||||
"Waited for the process to start but it failed to start in time. stderr {stderr} - stdout {stdout}"
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
ProcessReadinessWaitBehavior::WaitForCommandToExit => {
|
||||
if !child
|
||||
.wait()
|
||||
.context("Failed waiting for process to finish")?
|
||||
.success()
|
||||
{
|
||||
anyhow::bail!("Failed to spawn command");
|
||||
}
|
||||
}
|
||||
}
|
||||
if Instant::now().duration_since(spawn_time) > max_wait_duration {
|
||||
bail!(
|
||||
"Waited for the process to start but it failed to start in time. stderr {stderr} - stdout {stdout}"
|
||||
)
|
||||
}
|
||||
}
|
||||
},
|
||||
ProcessReadinessWaitBehavior::WaitForCommandToExit => {
|
||||
if !child.wait().context("Failed waiting for process to finish")?.success() {
|
||||
anyhow::bail!("Failed to spawn command");
|
||||
}
|
||||
},
|
||||
}
|
||||
|
||||
Ok(Self {
|
||||
child,
|
||||
stdout_logs_file,
|
||||
stderr_logs_file,
|
||||
})
|
||||
}
|
||||
Ok(Self { child, stdout_logs_file, stderr_logs_file })
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for Process {
|
||||
fn drop(&mut self) {
|
||||
self.child.kill().expect("Failed to kill the process");
|
||||
self.stdout_logs_file
|
||||
.flush()
|
||||
.expect("Failed to flush the stdout logs file");
|
||||
self.stderr_logs_file
|
||||
.flush()
|
||||
.expect("Failed to flush the stderr logs file");
|
||||
}
|
||||
fn drop(&mut self) {
|
||||
self.child.kill().expect("Failed to kill the process");
|
||||
self.stdout_logs_file.flush().expect("Failed to flush the stdout logs file");
|
||||
self.stderr_logs_file.flush().expect("Failed to flush the stderr logs file");
|
||||
}
|
||||
}
|
||||
|
||||
pub enum ProcessReadinessWaitBehavior {
|
||||
/// The process does not require any kind of wait after it's been spawned and can be used
|
||||
/// straight away.
|
||||
NoStartupWait,
|
||||
/// The process does not require any kind of wait after it's been spawned and can be used
|
||||
/// straight away.
|
||||
NoStartupWait,
|
||||
|
||||
/// Waits for the command to exit.
|
||||
WaitForCommandToExit,
|
||||
/// Waits for the command to exit.
|
||||
WaitForCommandToExit,
|
||||
|
||||
/// The process does require some amount of wait duration after it's been started.
|
||||
WaitDuration(Duration),
|
||||
/// The process does require some amount of wait duration after it's been started.
|
||||
WaitDuration(Duration),
|
||||
|
||||
/// The process requires a time bounded wait function which is a function of the lines that
|
||||
/// appear in the log files.
|
||||
TimeBoundedWaitFunction {
|
||||
/// The maximum amount of time to wait for the check function to return true.
|
||||
max_wait_duration: Duration,
|
||||
/// The process requires a time bounded wait function which is a function of the lines that
|
||||
/// appear in the log files.
|
||||
TimeBoundedWaitFunction {
|
||||
/// The maximum amount of time to wait for the check function to return true.
|
||||
max_wait_duration: Duration,
|
||||
|
||||
/// The function to use to check if the process spawned is ready to use or not. This
|
||||
/// function should return the following in the following cases:
|
||||
///
|
||||
/// - `Ok(true)`: Returned when the condition the process is waiting for has been fulfilled
|
||||
/// and the wait is completed.
|
||||
/// - `Ok(false)`: The process is not ready yet but it might be ready in the future.
|
||||
/// - `Err`: The process is not ready yet and will not be ready in the future as it appears
|
||||
/// that it has encountered an error when it was being spawned.
|
||||
///
|
||||
/// The first argument is a line from stdout and the second argument is a line from stderr.
|
||||
#[allow(clippy::type_complexity)]
|
||||
check_function: Box<dyn FnMut(Option<&str>, Option<&str>) -> anyhow::Result<bool>>,
|
||||
},
|
||||
/// The function to use to check if the process spawned is ready to use or not. This
|
||||
/// function should return the following in the following cases:
|
||||
///
|
||||
/// - `Ok(true)`: Returned when the condition the process is waiting for has been fulfilled
|
||||
/// and the wait is completed.
|
||||
/// - `Ok(false)`: The process is not ready yet but it might be ready in the future.
|
||||
/// - `Err`: The process is not ready yet and will not be ready in the future as it appears
|
||||
/// that it has encountered an error when it was being spawned.
|
||||
///
|
||||
/// The first argument is a line from stdout and the second argument is a line from stderr.
|
||||
#[allow(clippy::type_complexity)]
|
||||
check_function: Box<dyn FnMut(Option<&str>, Option<&str>) -> anyhow::Result<bool>>,
|
||||
},
|
||||
}
|
||||
|
||||
+10
-10
@@ -10,16 +10,16 @@ pub mod provider_utils;
|
||||
|
||||
/// An abstract interface for testing nodes.
|
||||
pub trait Node: EthereumNode {
|
||||
/// 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<()>;
|
||||
/// 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<()>;
|
||||
|
||||
/// Prune the node instance and related data.
|
||||
///
|
||||
/// Blocking until it's completely stopped.
|
||||
fn shutdown(&mut self) -> anyhow::Result<()>;
|
||||
/// Prune the node instance and related data.
|
||||
///
|
||||
/// Blocking until it's completely stopped.
|
||||
fn shutdown(&mut self) -> anyhow::Result<()>;
|
||||
|
||||
/// Returns the node version.
|
||||
fn version(&self) -> anyhow::Result<String>;
|
||||
/// Returns the node version.
|
||||
fn version(&self) -> anyhow::Result<String>;
|
||||
}
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
@@ -6,64 +6,56 @@ use tower::{Layer, Service};
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct ConcurrencyLimiterLayer {
|
||||
semaphore: Arc<Semaphore>,
|
||||
semaphore: Arc<Semaphore>,
|
||||
}
|
||||
|
||||
impl ConcurrencyLimiterLayer {
|
||||
pub fn new(permit_count: usize) -> Self {
|
||||
Self {
|
||||
semaphore: Arc::new(Semaphore::new(permit_count)),
|
||||
}
|
||||
}
|
||||
pub fn new(permit_count: usize) -> Self {
|
||||
Self { semaphore: Arc::new(Semaphore::new(permit_count)) }
|
||||
}
|
||||
}
|
||||
|
||||
impl<S> Layer<S> for ConcurrencyLimiterLayer {
|
||||
type Service = ConcurrencyLimiterService<S>;
|
||||
type Service = ConcurrencyLimiterService<S>;
|
||||
|
||||
fn layer(&self, inner: S) -> Self::Service {
|
||||
ConcurrencyLimiterService {
|
||||
service: inner,
|
||||
semaphore: self.semaphore.clone(),
|
||||
}
|
||||
}
|
||||
fn layer(&self, inner: S) -> Self::Service {
|
||||
ConcurrencyLimiterService { service: inner, semaphore: self.semaphore.clone() }
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
pub struct ConcurrencyLimiterService<S> {
|
||||
service: S,
|
||||
semaphore: Arc<Semaphore>,
|
||||
service: S,
|
||||
semaphore: Arc<Semaphore>,
|
||||
}
|
||||
|
||||
impl<S, Request> Service<Request> for ConcurrencyLimiterService<S>
|
||||
where
|
||||
S: Service<Request> + Send,
|
||||
S::Future: Send + 'static,
|
||||
S: Service<Request> + Send,
|
||||
S::Future: Send + 'static,
|
||||
{
|
||||
type Response = S::Response;
|
||||
type Error = S::Error;
|
||||
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
|
||||
type Response = S::Response;
|
||||
type Error = S::Error;
|
||||
type Future = BoxFuture<'static, Result<Self::Response, Self::Error>>;
|
||||
|
||||
fn poll_ready(
|
||||
&mut self,
|
||||
cx: &mut std::task::Context<'_>,
|
||||
) -> std::task::Poll<Result<(), Self::Error>> {
|
||||
self.service.poll_ready(cx)
|
||||
}
|
||||
fn poll_ready(
|
||||
&mut self,
|
||||
cx: &mut std::task::Context<'_>,
|
||||
) -> std::task::Poll<Result<(), Self::Error>> {
|
||||
self.service.poll_ready(cx)
|
||||
}
|
||||
|
||||
fn call(&mut self, req: Request) -> Self::Future {
|
||||
let semaphore = self.semaphore.clone();
|
||||
let future = self.service.call(req);
|
||||
fn call(&mut self, req: Request) -> Self::Future {
|
||||
let semaphore = self.semaphore.clone();
|
||||
let future = self.service.call(req);
|
||||
|
||||
Box::pin(async move {
|
||||
let _permit = semaphore
|
||||
.acquire()
|
||||
.await
|
||||
.expect("Semaphore has been closed");
|
||||
tracing::debug!(
|
||||
available_permits = semaphore.available_permits(),
|
||||
"Acquired Semaphore Permit"
|
||||
);
|
||||
future.await
|
||||
})
|
||||
}
|
||||
Box::pin(async move {
|
||||
let _permit = semaphore.acquire().await.expect("Semaphore has been closed");
|
||||
tracing::debug!(
|
||||
available_permits = semaphore.available_permits(),
|
||||
"Acquired Semaphore Permit"
|
||||
);
|
||||
future.await
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,84 +1,79 @@
|
||||
use alloy::{
|
||||
network::{Network, TransactionBuilder},
|
||||
providers::{
|
||||
Provider, SendableTx,
|
||||
fillers::{GasFiller, TxFiller},
|
||||
},
|
||||
transports::TransportResult,
|
||||
network::{Network, TransactionBuilder},
|
||||
providers::{
|
||||
Provider, SendableTx,
|
||||
fillers::{GasFiller, TxFiller},
|
||||
},
|
||||
transports::TransportResult,
|
||||
};
|
||||
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct FallbackGasFiller {
|
||||
inner: GasFiller,
|
||||
default_gas_limit: u64,
|
||||
default_max_fee_per_gas: u128,
|
||||
default_priority_fee: u128,
|
||||
inner: GasFiller,
|
||||
default_gas_limit: u64,
|
||||
default_max_fee_per_gas: u128,
|
||||
default_priority_fee: u128,
|
||||
}
|
||||
|
||||
impl FallbackGasFiller {
|
||||
pub fn new(
|
||||
default_gas_limit: u64,
|
||||
default_max_fee_per_gas: u128,
|
||||
default_priority_fee: u128,
|
||||
) -> Self {
|
||||
Self {
|
||||
inner: GasFiller,
|
||||
default_gas_limit,
|
||||
default_max_fee_per_gas,
|
||||
default_priority_fee,
|
||||
}
|
||||
}
|
||||
pub fn new(
|
||||
default_gas_limit: u64,
|
||||
default_max_fee_per_gas: u128,
|
||||
default_priority_fee: u128,
|
||||
) -> Self {
|
||||
Self { inner: GasFiller, default_gas_limit, default_max_fee_per_gas, default_priority_fee }
|
||||
}
|
||||
}
|
||||
|
||||
impl Default for FallbackGasFiller {
|
||||
fn default() -> Self {
|
||||
FallbackGasFiller::new(25_000_000, 1_000_000_000, 1_000_000_000)
|
||||
}
|
||||
fn default() -> Self {
|
||||
FallbackGasFiller::new(25_000_000, 1_000_000_000, 1_000_000_000)
|
||||
}
|
||||
}
|
||||
|
||||
impl<N> TxFiller<N> for FallbackGasFiller
|
||||
where
|
||||
N: Network,
|
||||
N: Network,
|
||||
{
|
||||
type Fillable = Option<<GasFiller as TxFiller<N>>::Fillable>;
|
||||
type Fillable = Option<<GasFiller as TxFiller<N>>::Fillable>;
|
||||
|
||||
fn status(
|
||||
&self,
|
||||
tx: &<N as Network>::TransactionRequest,
|
||||
) -> alloy::providers::fillers::FillerControlFlow {
|
||||
<GasFiller as TxFiller<N>>::status(&self.inner, tx)
|
||||
}
|
||||
fn status(
|
||||
&self,
|
||||
tx: &<N as Network>::TransactionRequest,
|
||||
) -> alloy::providers::fillers::FillerControlFlow {
|
||||
<GasFiller as TxFiller<N>>::status(&self.inner, tx)
|
||||
}
|
||||
|
||||
fn fill_sync(&self, _: &mut alloy::providers::SendableTx<N>) {}
|
||||
fn fill_sync(&self, _: &mut alloy::providers::SendableTx<N>) {}
|
||||
|
||||
async fn prepare<P: Provider<N>>(
|
||||
&self,
|
||||
provider: &P,
|
||||
tx: &<N as Network>::TransactionRequest,
|
||||
) -> TransportResult<Self::Fillable> {
|
||||
// Try to fetch GasFiller’s “fillable” (gas_price, base_fee, estimate_gas, …)
|
||||
// If it errors (i.e. tx would revert under eth_estimateGas), swallow it.
|
||||
match self.inner.prepare(provider, tx).await {
|
||||
Ok(fill) => Ok(Some(fill)),
|
||||
Err(_) => Ok(None),
|
||||
}
|
||||
}
|
||||
async fn prepare<P: Provider<N>>(
|
||||
&self,
|
||||
provider: &P,
|
||||
tx: &<N as Network>::TransactionRequest,
|
||||
) -> TransportResult<Self::Fillable> {
|
||||
// Try to fetch GasFiller’s “fillable” (gas_price, base_fee, estimate_gas, …)
|
||||
// If it errors (i.e. tx would revert under eth_estimateGas), swallow it.
|
||||
match self.inner.prepare(provider, tx).await {
|
||||
Ok(fill) => Ok(Some(fill)),
|
||||
Err(_) => Ok(None),
|
||||
}
|
||||
}
|
||||
|
||||
async fn fill(
|
||||
&self,
|
||||
fillable: Self::Fillable,
|
||||
mut tx: alloy::providers::SendableTx<N>,
|
||||
) -> TransportResult<SendableTx<N>> {
|
||||
if let Some(fill) = fillable {
|
||||
// our inner GasFiller succeeded — use it
|
||||
self.inner.fill(fill, tx).await
|
||||
} else {
|
||||
if let Some(builder) = tx.as_mut_builder() {
|
||||
builder.set_gas_limit(self.default_gas_limit);
|
||||
builder.set_max_fee_per_gas(self.default_max_fee_per_gas);
|
||||
builder.set_max_priority_fee_per_gas(self.default_priority_fee);
|
||||
}
|
||||
Ok(tx)
|
||||
}
|
||||
}
|
||||
async fn fill(
|
||||
&self,
|
||||
fillable: Self::Fillable,
|
||||
mut tx: alloy::providers::SendableTx<N>,
|
||||
) -> TransportResult<SendableTx<N>> {
|
||||
if let Some(fill) = fillable {
|
||||
// our inner GasFiller succeeded — use it
|
||||
self.inner.fill(fill, tx).await
|
||||
} else {
|
||||
if let Some(builder) = tx.as_mut_builder() {
|
||||
builder.set_gas_limit(self.default_gas_limit);
|
||||
builder.set_max_fee_per_gas(self.default_max_fee_per_gas);
|
||||
builder.set_max_priority_fee_per_gas(self.default_priority_fee);
|
||||
}
|
||||
Ok(tx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,12 +1,12 @@
|
||||
use std::{ops::ControlFlow, sync::LazyLock, time::Duration};
|
||||
|
||||
use alloy::{
|
||||
network::{Ethereum, Network, NetworkWallet, TransactionBuilder4844},
|
||||
providers::{
|
||||
Identity, PendingTransactionBuilder, Provider, ProviderBuilder, RootProvider,
|
||||
fillers::{ChainIdFiller, FillProvider, JoinFill, NonceFiller, TxFiller, WalletFiller},
|
||||
},
|
||||
rpc::client::ClientBuilder,
|
||||
network::{Ethereum, Network, NetworkWallet, TransactionBuilder4844},
|
||||
providers::{
|
||||
Identity, PendingTransactionBuilder, Provider, ProviderBuilder, RootProvider,
|
||||
fillers::{ChainIdFiller, FillProvider, JoinFill, NonceFiller, TxFiller, WalletFiller},
|
||||
},
|
||||
rpc::client::ClientBuilder,
|
||||
};
|
||||
use anyhow::{Context, Result};
|
||||
use revive_dt_common::futures::{PollingWaitBehavior, poll};
|
||||
@@ -15,114 +15,109 @@ use tracing::debug;
|
||||
use crate::provider_utils::{ConcurrencyLimiterLayer, FallbackGasFiller};
|
||||
|
||||
pub type ConcreteProvider<N, W> = FillProvider<
|
||||
JoinFill<
|
||||
JoinFill<JoinFill<JoinFill<Identity, FallbackGasFiller>, ChainIdFiller>, NonceFiller>,
|
||||
WalletFiller<W>,
|
||||
>,
|
||||
RootProvider<N>,
|
||||
N,
|
||||
JoinFill<
|
||||
JoinFill<JoinFill<JoinFill<Identity, FallbackGasFiller>, ChainIdFiller>, NonceFiller>,
|
||||
WalletFiller<W>,
|
||||
>,
|
||||
RootProvider<N>,
|
||||
N,
|
||||
>;
|
||||
|
||||
pub async fn construct_concurrency_limited_provider<N, W>(
|
||||
rpc_url: &str,
|
||||
fallback_gas_filler: FallbackGasFiller,
|
||||
chain_id_filler: ChainIdFiller,
|
||||
nonce_filler: NonceFiller,
|
||||
wallet: W,
|
||||
rpc_url: &str,
|
||||
fallback_gas_filler: FallbackGasFiller,
|
||||
chain_id_filler: ChainIdFiller,
|
||||
nonce_filler: NonceFiller,
|
||||
wallet: W,
|
||||
) -> Result<ConcreteProvider<N, W>>
|
||||
where
|
||||
N: Network<TransactionRequest: TransactionBuilder4844>,
|
||||
W: NetworkWallet<N>,
|
||||
Identity: TxFiller<N>,
|
||||
FallbackGasFiller: TxFiller<N>,
|
||||
ChainIdFiller: TxFiller<N>,
|
||||
NonceFiller: TxFiller<N>,
|
||||
WalletFiller<W>: TxFiller<N>,
|
||||
N: Network<TransactionRequest: TransactionBuilder4844>,
|
||||
W: NetworkWallet<N>,
|
||||
Identity: TxFiller<N>,
|
||||
FallbackGasFiller: TxFiller<N>,
|
||||
ChainIdFiller: TxFiller<N>,
|
||||
NonceFiller: TxFiller<N>,
|
||||
WalletFiller<W>: TxFiller<N>,
|
||||
{
|
||||
// This is a global limit on the RPC concurrency that applies to all of the providers created
|
||||
// by the framework. With this limit, it means that we can have a maximum of N concurrent
|
||||
// requests at any point of time and no more than that. This is done in an effort to stabilize
|
||||
// the framework from some of the interment issues that we've been seeing related to RPC calls.
|
||||
static GLOBAL_CONCURRENCY_LIMITER_LAYER: LazyLock<ConcurrencyLimiterLayer> =
|
||||
LazyLock::new(|| ConcurrencyLimiterLayer::new(10));
|
||||
// This is a global limit on the RPC concurrency that applies to all of the providers created
|
||||
// by the framework. With this limit, it means that we can have a maximum of N concurrent
|
||||
// requests at any point of time and no more than that. This is done in an effort to stabilize
|
||||
// the framework from some of the interment issues that we've been seeing related to RPC calls.
|
||||
static GLOBAL_CONCURRENCY_LIMITER_LAYER: LazyLock<ConcurrencyLimiterLayer> =
|
||||
LazyLock::new(|| ConcurrencyLimiterLayer::new(10));
|
||||
|
||||
let client = ClientBuilder::default()
|
||||
.layer(GLOBAL_CONCURRENCY_LIMITER_LAYER.clone())
|
||||
.connect(rpc_url)
|
||||
.await
|
||||
.context("Failed to construct the RPC client")?;
|
||||
let client = ClientBuilder::default()
|
||||
.layer(GLOBAL_CONCURRENCY_LIMITER_LAYER.clone())
|
||||
.connect(rpc_url)
|
||||
.await
|
||||
.context("Failed to construct the RPC client")?;
|
||||
|
||||
let provider = ProviderBuilder::new()
|
||||
.disable_recommended_fillers()
|
||||
.network::<N>()
|
||||
.filler(fallback_gas_filler)
|
||||
.filler(chain_id_filler)
|
||||
.filler(nonce_filler)
|
||||
.wallet(wallet)
|
||||
.connect_client(client);
|
||||
let provider = ProviderBuilder::new()
|
||||
.disable_recommended_fillers()
|
||||
.network::<N>()
|
||||
.filler(fallback_gas_filler)
|
||||
.filler(chain_id_filler)
|
||||
.filler(nonce_filler)
|
||||
.wallet(wallet)
|
||||
.connect_client(client);
|
||||
|
||||
Ok(provider)
|
||||
Ok(provider)
|
||||
}
|
||||
|
||||
pub async fn execute_transaction<N, W>(
|
||||
provider: ConcreteProvider<N, W>,
|
||||
transaction: N::TransactionRequest,
|
||||
provider: ConcreteProvider<N, W>,
|
||||
transaction: N::TransactionRequest,
|
||||
) -> Result<N::ReceiptResponse>
|
||||
where
|
||||
N: Network<
|
||||
TransactionRequest: TransactionBuilder4844,
|
||||
TxEnvelope = <Ethereum as Network>::TxEnvelope,
|
||||
>,
|
||||
W: NetworkWallet<N>,
|
||||
Identity: TxFiller<N>,
|
||||
FallbackGasFiller: TxFiller<N>,
|
||||
ChainIdFiller: TxFiller<N>,
|
||||
NonceFiller: TxFiller<N>,
|
||||
WalletFiller<W>: TxFiller<N>,
|
||||
N: Network<
|
||||
TransactionRequest: TransactionBuilder4844,
|
||||
TxEnvelope = <Ethereum as Network>::TxEnvelope,
|
||||
>,
|
||||
W: NetworkWallet<N>,
|
||||
Identity: TxFiller<N>,
|
||||
FallbackGasFiller: TxFiller<N>,
|
||||
ChainIdFiller: TxFiller<N>,
|
||||
NonceFiller: TxFiller<N>,
|
||||
WalletFiller<W>: TxFiller<N>,
|
||||
{
|
||||
let sendable_transaction = provider
|
||||
.fill(transaction)
|
||||
.await
|
||||
.context("Failed to fill transaction")?;
|
||||
let sendable_transaction =
|
||||
provider.fill(transaction).await.context("Failed to fill transaction")?;
|
||||
|
||||
let transaction_envelope = sendable_transaction
|
||||
.try_into_envelope()
|
||||
.context("Failed to convert transaction into an envelope")?;
|
||||
let tx_hash = *transaction_envelope.tx_hash();
|
||||
let transaction_envelope = sendable_transaction
|
||||
.try_into_envelope()
|
||||
.context("Failed to convert transaction into an envelope")?;
|
||||
let tx_hash = *transaction_envelope.tx_hash();
|
||||
|
||||
let mut pending_transaction = match provider.send_tx_envelope(transaction_envelope).await {
|
||||
Ok(pending_transaction) => pending_transaction,
|
||||
Err(error) => {
|
||||
let error_string = error.to_string();
|
||||
let mut pending_transaction = match provider.send_tx_envelope(transaction_envelope).await {
|
||||
Ok(pending_transaction) => pending_transaction,
|
||||
Err(error) => {
|
||||
let error_string = error.to_string();
|
||||
|
||||
if error_string.contains("Transaction Already Imported") {
|
||||
PendingTransactionBuilder::<N>::new(provider.root().clone(), tx_hash)
|
||||
} else {
|
||||
return Err(error).context(format!("Failed to submit transaction {tx_hash}"));
|
||||
}
|
||||
}
|
||||
};
|
||||
debug!(%tx_hash, "Submitted Transaction");
|
||||
if error_string.contains("Transaction Already Imported") {
|
||||
PendingTransactionBuilder::<N>::new(provider.root().clone(), tx_hash)
|
||||
} else {
|
||||
return Err(error).context(format!("Failed to submit transaction {tx_hash}"));
|
||||
}
|
||||
},
|
||||
};
|
||||
debug!(%tx_hash, "Submitted Transaction");
|
||||
|
||||
pending_transaction.set_timeout(Some(Duration::from_secs(120)));
|
||||
let tx_hash = pending_transaction.watch().await.context(format!(
|
||||
"Transaction inclusion watching timeout for {tx_hash}"
|
||||
))?;
|
||||
pending_transaction.set_timeout(Some(Duration::from_secs(120)));
|
||||
let tx_hash = pending_transaction
|
||||
.watch()
|
||||
.await
|
||||
.context(format!("Transaction inclusion watching timeout for {tx_hash}"))?;
|
||||
|
||||
poll(
|
||||
Duration::from_secs(60),
|
||||
PollingWaitBehavior::Constant(Duration::from_secs(3)),
|
||||
|| {
|
||||
let provider = provider.clone();
|
||||
poll(Duration::from_secs(60), PollingWaitBehavior::Constant(Duration::from_secs(3)), || {
|
||||
let provider = provider.clone();
|
||||
|
||||
async move {
|
||||
match provider.get_transaction_receipt(tx_hash).await {
|
||||
Ok(Some(receipt)) => Ok(ControlFlow::Break(receipt)),
|
||||
_ => Ok(ControlFlow::Continue(())),
|
||||
}
|
||||
}
|
||||
},
|
||||
)
|
||||
.await
|
||||
.context(format!("Polling for receipt failed for {tx_hash}"))
|
||||
async move {
|
||||
match provider.get_transaction_receipt(tx_hash).await {
|
||||
Ok(Some(receipt)) => Ok(ControlFlow::Break(receipt)),
|
||||
_ => Ok(ControlFlow::Continue(())),
|
||||
}
|
||||
}
|
||||
})
|
||||
.await
|
||||
.context(format!("Polling for receipt failed for {tx_hash}"))
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user